[#1339] feat(jdbc-doris): Support Apache Doris Catalog

build.gradle.kts

@@ -673,6 +673,7 @@ tasks {
       ":catalogs:catalog-lakehouse-iceberg:copyLibAndConfig",
       ":catalogs:catalog-jdbc-mysql:copyLibAndConfig",
       ":catalogs:catalog-jdbc-postgresql:copyLibAndConfig",
+      ":catalogs:catalog-jdbc-doris:copyLibAndConfig",
       ":catalogs:catalog-hadoop:copyLibAndConfig",
       "catalogs:catalog-messaging-kafka:copyLibAndConfig"
     )

catalogs/catalog-jdbc-common/src/main/java/com/datastrato/gravitino/catalog/jdbc/JdbcCatalogOperations.java

@@ -35,7 +35,6 @@
 import com.datastrato.gravitino.rel.TableCatalog;
 import com.datastrato.gravitino.rel.TableChange;
 import com.datastrato.gravitino.rel.expressions.distributions.Distribution;
-import com.datastrato.gravitino.rel.expressions.distributions.Distributions;
 import com.datastrato.gravitino.rel.expressions.sorts.SortOrder;
 import com.datastrato.gravitino.rel.expressions.transforms.Transform;
 import com.datastrato.gravitino.rel.indexes.Index;
@@ -370,9 +369,6 @@ public Table createTable(
       SortOrder[] sortOrders,
       Index[] indexes)
       throws NoSuchSchemaException, TableAlreadyExistsException {
-    Preconditions.checkArgument(
-        null == distribution || distribution == Distributions.NONE,
-        "jdbc-catalog does not support distribution");
     Preconditions.checkArgument(
         null == sortOrders || sortOrders.length == 0, "jdbc-catalog does not support sort orders");
 
@@ -406,6 +402,7 @@ public Table createTable(
         StringIdentifier.addToComment(identifier, comment),
         resultProperties,
         partitioning,
+        distribution,
         indexes);
 
     return new JdbcTable.Builder()

catalogs/catalog-jdbc-common/src/main/java/com/datastrato/gravitino/catalog/jdbc/operation/JdbcTableOperations.java

@@ -18,6 +18,7 @@
 import com.datastrato.gravitino.meta.AuditInfo;
 import com.datastrato.gravitino.rel.TableChange;
 import com.datastrato.gravitino.rel.expressions.Expression;
+import com.datastrato.gravitino.rel.expressions.distributions.Distribution;
 import com.datastrato.gravitino.rel.expressions.literals.Literals;
 import com.datastrato.gravitino.rel.expressions.transforms.Transform;
 import com.datastrato.gravitino.rel.indexes.Index;
@@ -79,13 +80,15 @@ public void create(
       String comment,
       Map<String, String> properties,
       Transform[] partitioning,
+      Distribution distribution,
       Index[] indexes)
       throws TableAlreadyExistsException {
     LOG.info("Attempting to create table {} in database {}", tableName, databaseName);
     try (Connection connection = getConnection(databaseName)) {
       JdbcConnectorUtils.executeUpdate(
           connection,
-          generateCreateTableSql(tableName, columns, comment, properties, partitioning, indexes));
+          generateCreateTableSql(
+              tableName, columns, comment, properties, partitioning, distribution, indexes));
       LOG.info("Created table {} in database {}", tableName, databaseName);
     } catch (final SQLException se) {
       throw this.exceptionMapper.toGravitinoException(se);
@@ -172,7 +175,7 @@ public JdbcTable load(String databaseName, String tableName) throws NoSuchTableE
       jdbcTableBuilder.withProperties(tableProperties);
 
       // 5.Leave the information to the bottom layer to append the table
-      correctJdbcTableFields(connection, tableName, jdbcTableBuilder);
+      correctJdbcTableFields(connection, databaseName, tableName, jdbcTableBuilder);
       return jdbcTableBuilder.build();
     } catch (SQLException e) {
       throw exceptionMapper.toGravitinoException(e);
@@ -272,7 +275,10 @@ protected ResultSet getColumns(Connection connection, String databaseName, Strin
    * @throws SQLException
    */
   protected void correctJdbcTableFields(
-      Connection connection, String tableName, JdbcTable.Builder jdbcTableBuilder)
+      Connection connection,
+      String databaseName,
+      String tableName,
+      JdbcTable.Builder jdbcTableBuilder)
       throws SQLException {
     // nothing to do
   }
@@ -357,6 +363,7 @@ protected abstract String generateCreateTableSql(
       String comment,
       Map<String, String> properties,
       Transform[] partitioning,
+      Distribution distribution,
       Index[] indexes);
 
   protected abstract String generateRenameTableSql(String oldTableName, String newTableName);

catalogs/catalog-jdbc-common/src/main/java/com/datastrato/gravitino/catalog/jdbc/operation/TableOperation.java

@@ -14,6 +14,7 @@
 import com.datastrato.gravitino.exceptions.NoSuchTableException;
 import com.datastrato.gravitino.exceptions.TableAlreadyExistsException;
 import com.datastrato.gravitino.rel.TableChange;
+import com.datastrato.gravitino.rel.expressions.distributions.Distribution;
 import com.datastrato.gravitino.rel.expressions.transforms.Transform;
 import com.datastrato.gravitino.rel.indexes.Index;
 import java.util.List;
@@ -53,6 +54,7 @@ void create(
       String comment,
       Map<String, String> properties,
       Transform[] partitioning,
+      Distribution distribution,
       Index[] indexes)
       throws TableAlreadyExistsException;
 

catalogs/catalog-jdbc-common/src/test/java/com/datastrato/gravitino/catalog/jdbc/operation/SqliteTableOperations.java

@@ -9,6 +9,7 @@
 import com.datastrato.gravitino.exceptions.NoSuchSchemaException;
 import com.datastrato.gravitino.rel.Column;
 import com.datastrato.gravitino.rel.TableChange;
+import com.datastrato.gravitino.rel.expressions.distributions.Distribution;
 import com.datastrato.gravitino.rel.expressions.transforms.Transform;
 import com.datastrato.gravitino.rel.indexes.Index;
 import com.google.common.collect.Lists;
@@ -27,6 +28,7 @@ protected String generateCreateTableSql(
       String comment,
       Map<String, String> properties,
       Transform[] partitioning,
+      Distribution distribution,
       Index[] indexes) {
     StringBuilder sqlBuilder = new StringBuilder();
     sqlBuilder.append("CREATE TABLE ").append(tableName).append(" (");

catalogs/catalog-jdbc-common/src/test/java/com/datastrato/gravitino/catalog/jdbc/operation/TestJdbcTableOperations.java

@@ -134,7 +134,14 @@ public void testOperationTable() {
     Assertions.assertDoesNotThrow(
         () ->
             JDBC_TABLE_OPERATIONS.create(
-                DATABASE_NAME, table1, jdbcColumns, null, properties, null, Indexes.EMPTY_INDEXES));
+                DATABASE_NAME,
+                table1,
+                jdbcColumns,
+                null,
+                properties,
+                null,
+                null,
+                Indexes.EMPTY_INDEXES));
 
     // list table.
     List<String> allTables = JDBC_TABLE_OPERATIONS.listTables(DATABASE_NAME);

catalogs/catalog-jdbc-doris/build.gradle.kts

@@ -0,0 +1,91 @@
+/*
+ * Copyright 2024 Datastrato Pvt Ltd.
+ * This software is licensed under the Apache License version 2.
+ */
+description = "catalog-jdbc-doris"
+
+plugins {
+  `maven-publish`
+  id("java")
+  id("idea")
+}
+
+dependencies {
+  implementation(project(":api"))
+  implementation(project(":catalogs:catalog-jdbc-common"))
+  implementation(project(":common"))
+  implementation(project(":core"))
+
+  implementation(libs.bundles.log4j)
+  implementation(libs.commons.collections4)
+  implementation(libs.commons.lang3)
+  implementation(libs.guava)
+  implementation(libs.jsqlparser)
+  implementation(libs.slf4j.api)
+
+  testImplementation(project(":catalogs:catalog-jdbc-common", "testArtifacts"))
+  testImplementation(project(":clients:client-java"))
+  testImplementation(project(":integration-test-common", "testArtifacts"))
+  testImplementation(project(":server"))
+  testImplementation(project(":server-common"))
+
+  testImplementation(libs.commons.lang3)
+  testImplementation(libs.guava)
+  testImplementation(libs.junit.jupiter.api)
+  testImplementation(libs.junit.jupiter.params)
+  testImplementation(libs.mysql.driver)
+  testImplementation(libs.testcontainers)
+  testImplementation(libs.testcontainers.mysql)
+
+  testRuntimeOnly(libs.junit.jupiter.engine)
+}
+
+tasks {
+  val runtimeJars by registering(Copy::class) {
+    from(configurations.runtimeClasspath)
+    into("build/libs")
+  }
+  val copyCatalogLibs by registering(Copy::class) {
+    dependsOn("jar", "runtimeJars")
+    from("build/libs")
+    into("$rootDir/distribution/package/catalogs/jdbc-doris/libs")
+  }
+
+  val copyCatalogConfig by registering(Copy::class) {
+    from("src/main/resources")
+    into("$rootDir/distribution/package/catalogs/jdbc-doris/conf")
+
+    include("jdbc-doris.conf")
+
+    exclude { details ->
+      details.file.isDirectory()
+    }
+  }
+
+  register("copyLibAndConfig", Copy::class) {
+    dependsOn(copyCatalogLibs, copyCatalogConfig)
+  }
+}
+
+tasks.test {
+  val skipUTs = project.hasProperty("skipTests")
+  if (skipUTs) {
+    // Only run integration tests
+    include("**/integration/**")
+  }
+
+  val skipITs = project.hasProperty("skipITs")
+  if (skipITs) {
+    // Exclude integration tests
+    exclude("**/integration/**")
+  } else {
+    dependsOn(tasks.jar)
+
+    doFirst {
+      environment("GRAVITINO_CI_DORIS_DOCKER_IMAGE", "datastrato/gravitino-ci-doris:0.1.0")
+    }
+
+    val init = project.extra.get("initIntegrationTest") as (Test) -> Unit
+    init(this)
+  }
+}

catalogs/catalog-jdbc-doris/src/main/java/com/datastrato/gravitino/catalog/doris/DorisCatalog.java

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2024 Datastrato Pvt Ltd.
+ * This software is licensed under the Apache License version 2.
+ */
+package com.datastrato.gravitino.catalog.doris;
+
+import com.datastrato.gravitino.catalog.doris.converter.DorisColumnDefaultValueConverter;
+import com.datastrato.gravitino.catalog.doris.converter.DorisExceptionConverter;
+import com.datastrato.gravitino.catalog.doris.converter.DorisTypeConverter;
+import com.datastrato.gravitino.catalog.doris.operation.DorisDatabaseOperations;
+import com.datastrato.gravitino.catalog.doris.operation.DorisTableOperations;
+import com.datastrato.gravitino.catalog.jdbc.JdbcCatalog;
+import com.datastrato.gravitino.catalog.jdbc.converter.JdbcColumnDefaultValueConverter;
+import com.datastrato.gravitino.catalog.jdbc.converter.JdbcExceptionConverter;
+import com.datastrato.gravitino.catalog.jdbc.converter.JdbcTypeConverter;
+import com.datastrato.gravitino.catalog.jdbc.operation.JdbcDatabaseOperations;
+import com.datastrato.gravitino.catalog.jdbc.operation.JdbcTableOperations;
+
+/** Implementation of a Doris catalog in Gravitino. */
+public class DorisCatalog extends JdbcCatalog {
+
+  @Override
+  public String shortName() {
+    return "jdbc-doris";
+  }
+
+  @Override
+  protected JdbcExceptionConverter createExceptionConverter() {
+    return new DorisExceptionConverter();
+  }
+
+  @Override
+  protected JdbcTypeConverter createJdbcTypeConverter() {
+    return new DorisTypeConverter();
+  }
+
+  @Override
+  protected JdbcDatabaseOperations createJdbcDatabaseOperations() {
+    return new DorisDatabaseOperations();
+  }
+
+  @Override
+  protected JdbcTableOperations createJdbcTableOperations() {
+    return new DorisTableOperations();
+  }
+
+  @Override
+  protected JdbcColumnDefaultValueConverter createJdbcColumnDefaultValueConverter() {
+    return new DorisColumnDefaultValueConverter();
+  }
+}

catalogs/catalog-jdbc-doris/src/main/java/com/datastrato/gravitino/catalog/doris/converter/DorisColumnDefaultValueConverter.java

@@ -0,0 +1,96 @@
+/*
+ * Copyright 2024 Datastrato Pvt Ltd.
+ * This software is licensed under the Apache License version 2.
+ */
+package com.datastrato.gravitino.catalog.doris.converter;
+
+import static com.datastrato.gravitino.catalog.doris.converter.DorisTypeConverter.BIGINT;
+import static com.datastrato.gravitino.catalog.doris.converter.DorisTypeConverter.CHAR;
+import static com.datastrato.gravitino.catalog.doris.converter.DorisTypeConverter.DATETIME;
+import static com.datastrato.gravitino.catalog.doris.converter.DorisTypeConverter.DECIMAL;
+import static com.datastrato.gravitino.catalog.doris.converter.DorisTypeConverter.DOUBLE;
+import static com.datastrato.gravitino.catalog.doris.converter.DorisTypeConverter.FLOAT;
+import static com.datastrato.gravitino.catalog.doris.converter.DorisTypeConverter.INT;
+import static com.datastrato.gravitino.catalog.doris.converter.DorisTypeConverter.SMALLINT;
+import static com.datastrato.gravitino.catalog.doris.converter.DorisTypeConverter.TINYINT;
+import static com.datastrato.gravitino.rel.Column.DEFAULT_VALUE_NOT_SET;
+import static com.datastrato.gravitino.rel.Column.DEFAULT_VALUE_OF_CURRENT_TIMESTAMP;
+
+import com.datastrato.gravitino.catalog.jdbc.converter.JdbcColumnDefaultValueConverter;
+import com.datastrato.gravitino.catalog.jdbc.converter.JdbcTypeConverter;
+import com.datastrato.gravitino.rel.expressions.Expression;
+import com.datastrato.gravitino.rel.expressions.UnparsedExpression;
+import com.datastrato.gravitino.rel.expressions.literals.Literals;
+import com.datastrato.gravitino.rel.types.Decimal;
+import com.datastrato.gravitino.rel.types.Types;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+
+public class DorisColumnDefaultValueConverter extends JdbcColumnDefaultValueConverter {
+
+  @Override
+  public Expression toGravitino(
+      JdbcTypeConverter.JdbcTypeBean type,
+      String columnDefaultValue,
+      boolean isExpression,
+      boolean nullable) {
+    if (columnDefaultValue == null) {
+      return nullable ? Literals.NULL : DEFAULT_VALUE_NOT_SET;
+    }
+
+    if (columnDefaultValue.equalsIgnoreCase(NULL)) {
+      return Literals.NULL;
+    }
+
+    if (isExpression) {
+      if (columnDefaultValue.equals(CURRENT_TIMESTAMP)) {
+        return DEFAULT_VALUE_OF_CURRENT_TIMESTAMP;
+      }
+      // The parsing of Doris expressions is complex, so we are not currently undertaking the
+      // parsing.
+      return UnparsedExpression.of(columnDefaultValue);
+    }
+
+    switch (type.getTypeName().toLowerCase()) {
+      case TINYINT:
+        return Literals.byteLiteral(Byte.valueOf(columnDefaultValue));
+      case SMALLINT:
+        return Literals.shortLiteral(Short.valueOf(columnDefaultValue));
+      case INT:
+        return Literals.integerLiteral(Integer.valueOf(columnDefaultValue));
+      case BIGINT:
+        return Literals.longLiteral(Long.valueOf(columnDefaultValue));
+      case FLOAT:
+        return Literals.floatLiteral(Float.valueOf(columnDefaultValue));
+      case DOUBLE:
+        return Literals.doubleLiteral(Double.valueOf(columnDefaultValue));
+      case DECIMAL:
+        return Literals.decimalLiteral(
+            Decimal.of(
+                columnDefaultValue,
+                Integer.parseInt(type.getColumnSize()),
+                Integer.parseInt(type.getScale())));
+      case JdbcTypeConverter.DATE:
+        return Literals.dateLiteral(LocalDate.parse(columnDefaultValue, DATE_TIME_FORMATTER));
+      case JdbcTypeConverter.TIME:
+        return Literals.timeLiteral(LocalTime.parse(columnDefaultValue, DATE_TIME_FORMATTER));
+      case JdbcTypeConverter.TIMESTAMP:
+      case DATETIME:
+        return CURRENT_TIMESTAMP.equals(columnDefaultValue)
+            ? DEFAULT_VALUE_OF_CURRENT_TIMESTAMP
+            : Literals.timestampLiteral(
+                LocalDateTime.parse(columnDefaultValue, DATE_TIME_FORMATTER));
+      case JdbcTypeConverter.VARCHAR:
+        return Literals.of(
+            columnDefaultValue, Types.VarCharType.of(Integer.parseInt(type.getColumnSize())));
+      case CHAR:
+        return Literals.of(
+            columnDefaultValue, Types.FixedCharType.of(Integer.parseInt(type.getColumnSize())));
+      case JdbcTypeConverter.TEXT:
+        return Literals.stringLiteral(columnDefaultValue);
+      default:
+        throw new IllegalArgumentException("Unknown data type for literal: " + type);
+    }
+  }
+}