Repository: skiptests/astraea
Pull request #368 (2022-05-26 17:17:16)

Add database client which is implemented by JDBC
新增簡單的database client，包含：
1. 建立表格
2. 查詢表格
3. 刪除表格


Patch:
diff --git a/app/build.gradle b/app/build.gradle
index 8f3b28bc..55cca03a 100644
--- a/app/build.gradle
+++ b/app/build.gradle
@@ -17,7 +17,8 @@ def versions = [
     slf4j: project.properties['slf4j.version'] ?: "1.7.32",
     zookeeper: project.properties['zookeeper.version'] ?: "3.6.3",
     gson: project.properties['gson.version'] ?: "2.9.0",
-    mysql: project.properties['mysql.version'] ?: "4.6.2",
+    wix: project.properties['wix.version'] ?: "4.6.2",
+    mysql: project.properties['mysql.version'] ?: "8.0.29",
 ]
 
 dependencies {
@@ -26,7 +27,8 @@ dependencies {
     testImplementation "org.mockito:mockito-inline:${versions["mockito"]}"
     testImplementation "org.apache.zookeeper:zookeeper:${versions["zookeeper"]}"
     testImplementation "org.apache.kafka:kafka_2.12:${versions["kafka"]}"
-    testImplementation "com.wix:wix-embedded-mysql:${versions["mysql"]}"
+    testImplementation "com.wix:wix-embedded-mysql:${versions["wix"]}"
+    testImplementation "mysql:mysql-connector-java:${versions["mysql"]}"
 
 
     implementation "org.apache.kafka:kafka-clients:${versions["kafka"]}"
diff --git a/app/src/main/java/org/astraea/database/Builder.java b/app/src/main/java/org/astraea/database/Builder.java
new file mode 100644
index 00000000..24c8f18b
--- /dev/null
+++ b/app/src/main/java/org/astraea/database/Builder.java
@@ -0,0 +1,231 @@
+package org.astraea.database;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.astraea.Utils;
+
+public class Builder {
+  private String url;
+  private String user;
+  private String password;
+
+  Builder() {}
+
+  public Builder url(String url) {
+    this.url = url;
+    return this;
+  }
+
+  public Builder user(String user) {
+    this.user = user;
+    return this;
+  }
+
+  public Builder password(String password) {
+    this.password = password;
+    return this;
+  }
+
+  public DatabaseClient build() {
+    return new DatabaseClient() {
+      private final Connection connection =
+          Utils.handleException(
+              () ->
+                  DriverManager.getConnection(
+                      Objects.requireNonNull(url, "url must be defined"),
+                      Objects.requireNonNull(user, "user must be defined"),
+                      Objects.requireNonNull(password, "password must be defined")));
+
+      @Override
+      public TableQuery query() {
+        return new TableQuery() {
+          private String catalog;
+          private String schema;
+          private String tableName;
+
+          @Override
+          public TableQuery catalog(String catalog) {
+            this.catalog = catalog;
+            return this;
+          }
+
+          @Override
+          public TableQuery schema(String schema) {
+            this.schema = schema;
+            return this;
+          }
+
+          @Override
+          public TableQuery tableName(String tableName) {
+            this.tableName = tableName;
+            return this;
+          }
+
+          @Override
+          public Collection<TableInfo> run() {
+            return Utils.handleException(
+                () -> {
+                  var md = connection.getMetaData();
+                  return catalogAndNames(md.getTables(catalog, schema, tableName, null)).stream()
+                      .map(
+                          e -> {
+                            var catalog = e.getKey();
+                            var tableName = e.getValue();
+                            var pks =
+                                primaryKeys(
+                                    Utils.handleException(
+                                        () -> md.getPrimaryKeys(catalog, null, tableName)));
+                            var columnNameAndTypes =
+                                columnNameAndTypes(
+                                    Utils.handleException(
+                                        () -> md.getColumns(catalog, null, tableName, null)));
+                            return new TableInfo(
+                                tableName,
+                                columnNameAndTypes.entrySet().stream()
+                                    .map(
+                                        c ->
+                                            new ColumnInfo(
+                                                c.getKey(), c.getValue(), pks.contains(c.getKey())))
+                                    .collect(Collectors.toUnmodifiableList()));
+                          })
+                      .collect(Collectors.toUnmodifiableList());
+                });
+          }
+        };
+      }
+
+      @Override
+      public TableCreator tableCreator() {
+        return new TableCreator() {
+          private String tableName;
+          private final Map<String, String> nameAndType = new LinkedHashMap<>();
+          private final Set<String> pks = new LinkedHashSet<>();
+
+          @Override
+          public TableCreator name(String name) {
+            this.tableName = name;
+            return this;
+          }
+
+          @Override
+          public TableCreator column(String name, String type) {
+            nameAndType.put(name, type);
+            return this;
+          }
+
+          @Override
+          public TableCreator primaryKey(String name, String type) {
+            nameAndType.put(name, type);
+            pks.add(name);
+            return this;
+          }
+
+          @Override
+          public void run() {
+            if (pks.isEmpty()) throw new IllegalArgumentException("primary keys must be defined");
+            var sql =
+                "CREATE TABLE \""
+                    + Objects.requireNonNull(tableName, "table name must be defined")
+                    + "\" ( "
+                    + nameAndType.entrySet().stream()
+                        .map(e -> "\"" + e.getKey() + "\" " + e.getValue())
+                        .collect(Collectors.joining(","))
+                    + ", PRIMARY KEY ( "
+                    + pks.stream().map(c -> "\"" + c + "\"").collect(Collectors.joining(","))
+                    + " ))";
+            execute(sql);
+          }
+        };
+      }
+
+      @Override
+      public void deleteTable(String name) {
+        execute("DROP TABLE \"" + name + "\"");
+      }
+
+      @Override
+      public void close() {
+        Utils.close(connection);
+      }
+
+      private void execute(String sql) {
+        try (var statement = connection.createStatement()) {
+          statement.execute(sql);
+        } catch (SQLException e) {
+          throw new RuntimeException(e);
+        }
+      }
+    };
+  }
+
+  private static Map<String, String> columnNameAndTypes(ResultSet rs) {
+    try (rs) {
+      var result = new HashMap<String, String>();
+      while (rs.next()) result.put(columnName(rs), columnType(rs));
+      return Collections.unmodifiableMap(result);
+    } catch (SQLException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private static Set<String> primaryKeys(ResultSet rs) {
+    try (rs) {
+      var result = new HashSet<String>();
+      while (rs.next()) result.add(columnName(rs));
+      return Collections.unmodifiableSet(result);
+    } catch (SQLException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private static Collection<Map.Entry<String, String>> catalogAndNames(ResultSet rs) {
+    try (rs) {
+      var result = new ArrayList<Map.Entry<String, String>>();
+      while (rs.next()) {
+        if (isNormalTable(rs)) result.add(Map.entry(catalog(rs), tableName(rs)));
+      }
+      return Collections.unmodifiableList(result);
+    } catch (SQLException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private static String catalog(ResultSet rs) {
+    return Utils.handleException(() -> rs.getString("TABLE_CAT"));
+  }
+
+  private static String tableName(ResultSet rs) {
+    return Utils.handleException(() -> rs.getString("TABLE_NAME"));
+  }
+
+  private static String columnName(ResultSet rs) {
+    return Utils.handleException(() -> rs.getString("COLUMN_NAME"));
+  }
+
+  private static String columnType(ResultSet rs) {
+    return Utils.handleException(() -> rs.getString("TYPE_NAME"));
+  }
+
+  private static boolean isNormalTable(ResultSet rs) {
+    return Utils.handleException(
+        () -> {
+          var r = rs.getString("TABLE_TYPE");
+          if (r == null) return true;
+          return !Arrays.asList(r.split(" ")).contains("SYSTEM");
+        });
+  }
+}
diff --git a/app/src/main/java/org/astraea/database/ColumnInfo.java b/app/src/main/java/org/astraea/database/ColumnInfo.java
new file mode 100644
index 00000000..efd5ad9c
--- /dev/null
+++ b/app/src/main/java/org/astraea/database/ColumnInfo.java
@@ -0,0 +1,46 @@
+package org.astraea.database;
+
+import java.util.Objects;
+
+public class ColumnInfo {
+
+  private final String name;
+  private final String type;
+  private final boolean pk;
+
+  public ColumnInfo(String name, String type, boolean pk) {
+    this.name = name;
+    this.type = type;
+    this.pk = pk;
+  }
+
+  public String name() {
+    return name;
+  }
+
+  public String type() {
+    return type;
+  }
+
+  public boolean pk() {
+    return pk;
+  }
+
+  @Override
+  public String toString() {
+    return "ColumnInfo{" + "name='" + name + '\'' + ", type='" + type + '\'' + ", pk=" + pk + '}';
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+    ColumnInfo that = (ColumnInfo) o;
+    return pk == that.pk && Objects.equals(name, that.name) && Objects.equals(type, that.type);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(name, type, pk);
+  }
+}
diff --git a/app/src/main/java/org/astraea/database/DatabaseClient.java b/app/src/main/java/org/astraea/database/DatabaseClient.java
new file mode 100644
index 00000000..2db85383
--- /dev/null
+++ b/app/src/main/java/org/astraea/database/DatabaseClient.java
@@ -0,0 +1,28 @@
+package org.astraea.database;
+
+public interface DatabaseClient extends AutoCloseable {
+
+  static Builder builder() {
+    return new Builder();
+  }
+
+  /**
+   * start to query tables
+   *
+   * @return Table Query
+   */
+  TableQuery query();
+
+  /**
+   * start to create new table
+   *
+   * @return Table Creator
+   */
+  TableCreator tableCreator();
+
+  /** @param name to delete */
+  void deleteTable(String name);
+
+  @Override
+  void close();
+}
diff --git a/app/src/main/java/org/astraea/database/TableCreator.java b/app/src/main/java/org/astraea/database/TableCreator.java
new file mode 100644
index 00000000..7758ef47
--- /dev/null
+++ b/app/src/main/java/org/astraea/database/TableCreator.java
@@ -0,0 +1,30 @@
+package org.astraea.database;
+
+public interface TableCreator {
+
+  /**
+   * @param name table name
+   * @return this creator
+   */
+  TableCreator name(String name);
+
+  /**
+   * create a normal column
+   *
+   * @param name column name
+   * @param type column type
+   * @return this creator
+   */
+  TableCreator column(String name, String type);
+
+  /**
+   * create a primary key
+   *
+   * @param name primary key name
+   * @param type primary key type
+   * @return this creator
+   */
+  TableCreator primaryKey(String name, String type);
+
+  void run();
+}
diff --git a/app/src/main/java/org/astraea/database/TableInfo.java b/app/src/main/java/org/astraea/database/TableInfo.java
new file mode 100644
index 00000000..7c382c48
--- /dev/null
+++ b/app/src/main/java/org/astraea/database/TableInfo.java
@@ -0,0 +1,40 @@
+package org.astraea.database;
+
+import java.util.Collection;
+import java.util.Objects;
+
+public class TableInfo {
+  private final String name;
+  private final Collection<ColumnInfo> columns;
+
+  public TableInfo(String name, Collection<ColumnInfo> columns) {
+    this.name = name;
+    this.columns = columns;
+  }
+
+  public String name() {
+    return name;
+  }
+
+  public Collection<ColumnInfo> columns() {
+    return columns;
+  }
+
+  @Override
+  public String toString() {
+    return "TableInfo{" + "name='" + name + '\'' + ", columns=" + columns + '}';
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+    TableInfo tableInfo = (TableInfo) o;
+    return Objects.equals(name, tableInfo.name) && Objects.equals(columns, tableInfo.columns);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(name, columns);
+  }
+}
diff --git a/app/src/main/java/org/astraea/database/TableQuery.java b/app/src/main/java/org/astraea/database/TableQuery.java
new file mode 100644
index 00000000..9d88cb73
--- /dev/null
+++ b/app/src/main/java/org/astraea/database/TableQuery.java
@@ -0,0 +1,31 @@
+package org.astraea.database;
+
+import java.util.Collection;
+
+public interface TableQuery {
+
+  /**
+   * Normally, catalog in database is server instance or database instance.
+   *
+   * @param catalog to search
+   * @return this query
+   */
+  TableQuery catalog(String catalog);
+
+  /**
+   * Normally, schema in database is namespace.
+   *
+   * @param schema to search
+   * @return this query
+   */
+  TableQuery schema(String schema);
+
+  /**
+   * @param tableName to search
+   * @return this query
+   */
+  TableQuery tableName(String tableName);
+
+  /** @return the tables matched to this query. */
+  Collection<TableInfo> run();
+}


Test patch:
diff --git a/app/src/test/java/org/astraea/database/DatabaseClientTest.java b/app/src/test/java/org/astraea/database/DatabaseClientTest.java
new file mode 100644
index 00000000..a952aa02
--- /dev/null
+++ b/app/src/test/java/org/astraea/database/DatabaseClientTest.java
@@ -0,0 +1,75 @@
+package org.astraea.database;
+
+import org.astraea.Utils;
+import org.astraea.service.Database;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class DatabaseClientTest {
+  @Test
+  void testCreate() {
+    try (var database = Database.builder().build();
+        var client =
+            DatabaseClient.builder()
+                .user(database.user())
+                .password(database.password())
+                .url(database.url())
+                .build()) {
+      var tableCount = client.query().run().size();
+      var tableName = Utils.randomString(10);
+      var c0 = new ColumnInfo(Utils.randomString(5), "INT", true);
+      var c1 = new ColumnInfo(Utils.randomString(5), "INT", false);
+      client
+          .tableCreator()
+          .name(tableName)
+          .primaryKey(c0.name(), c0.type())
+          .column(c1.name(), c1.type())
+          .run();
+
+      Assertions.assertEquals(tableCount + 1, client.query().run().size());
+      Assertions.assertEquals(1, client.query().tableName(tableName).run().size());
+      var table = client.query().tableName(tableName).run().iterator().next();
+      Assertions.assertEquals(2, table.columns().size());
+      Assertions.assertEquals(
+          c0, table.columns().stream().filter(c -> c.name().equals(c0.name())).findFirst().get());
+      Assertions.assertEquals(
+          c1, table.columns().stream().filter(c -> c.name().equals(c1.name())).findFirst().get());
+    }
+  }
+
+  @Test
+  void testDelete() {
+    try (var database = Database.builder().build();
+        var client =
+            DatabaseClient.builder()
+                .user(database.user())
+                .password(database.password())
+                .url(database.url())
+                .build()) {
+      var tableName = Utils.randomString(10);
+      var tableCount = client.query().run().size();
+      client.tableCreator().name(tableName).primaryKey(Utils.randomString(5), "INT").run();
+      Assertions.assertEquals(tableCount + 1, client.query().run().size());
+      client.deleteTable(tableName);
+      Assertions.assertEquals(tableCount, client.query().run().size());
+    }
+  }
+
+  @Test
+  void testNoPrimaryKeyOrTableName() {
+    try (var database = Database.builder().build();
+        var client =
+            DatabaseClient.builder()
+                .user(database.user())
+                .password(database.password())
+                .url(database.url())
+                .build()) {
+      Assertions.assertThrows(
+          NullPointerException.class,
+          () -> client.tableCreator().primaryKey(Utils.randomString(5), "INT").run());
+      Assertions.assertThrows(
+          IllegalArgumentException.class,
+          () -> client.tableCreator().name("a").column(Utils.randomString(5), "INT").run());
+    }
+  }
+}