[#2572] feat(catalog-jdbc-doris): support table operation for Doris catalog

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

@@ -167,15 +167,15 @@ public JdbcTable load(String databaseName, String tableName) throws NoSuchTableE
       jdbcTableBuilder.withColumns(jdbcColumns.toArray(new JdbcColumn[0]));
 
       // 3.Get index information
-      List<Index> indexes = getIndexes(databaseName, tableName, connection.getMetaData());
+      List<Index> indexes = getIndexes(connection, databaseName, tableName);
       jdbcTableBuilder.withIndexes(indexes.toArray(new Index[0]));
 
       // 4.Get table properties
       Map<String, String> tableProperties = getTableProperties(connection, tableName);
       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);
@@ -275,13 +275,17 @@ 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
   }
 
-  protected List<Index> getIndexes(String databaseName, String tableName, DatabaseMetaData metaData)
+  protected List<Index> getIndexes(Connection connection, String databaseName, String tableName)
       throws SQLException {
+    DatabaseMetaData metaData = connection.getMetaData();
     List<Index> indexes = new ArrayList<>();
 
     // Get primary key information

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

@@ -4,20 +4,94 @@
  */
 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 columnType,
       String columnDefaultValue,
       boolean isExpression,
       boolean nullable) {
-    // TODO: add implementation for doris catalog
-    return DEFAULT_VALUE_NOT_SET;
+    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 (columnType.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(columnType.getColumnSize()),
+                Integer.parseInt(columnType.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(columnType.getColumnSize())));
+      case CHAR:
+        return Literals.of(
+            columnDefaultValue,
+            Types.FixedCharType.of(Integer.parseInt(columnType.getColumnSize())));
+      case JdbcTypeConverter.TEXT:
+        return Literals.stringLiteral(columnDefaultValue);
+      default:
+        throw new IllegalArgumentException("Unknown data columnType for literal: " + columnType);
+    }
   }
 }

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

@@ -35,6 +35,12 @@ public class DorisExceptionConverter extends JdbcExceptionConverter {
   private static final Pattern DATABASE_ALREADY_EXISTS_PATTERN =
       Pattern.compile(DATABASE_ALREADY_EXISTS_PATTERN_STRING);
 
+  private static final String TABLE_NOT_EXIST_PATTERN_STRING =
+      ".*detailMessage = Unknown table '.*' in .*:.*";
+
+  private static final Pattern TABLE_NOT_EXIST_PATTERN =
+      Pattern.compile(TABLE_NOT_EXIST_PATTERN_STRING);
+
   @SuppressWarnings("FormatStringAnnotation")
   @Override
   public GravitinoRuntimeException toGravitinoException(SQLException se) {
@@ -70,6 +76,10 @@ static int getErrorCodeFromMessage(String message) {
       return CODE_DATABASE_EXISTS;
     }
 
+    if (TABLE_NOT_EXIST_PATTERN.matcher(message).matches()) {
+      return CODE_NO_SUCH_TABLE;
+    }
+
     return CODE_OTHER;
   }
 }

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

@@ -10,15 +10,102 @@
 
 /** Type converter for Doris. */
 public class DorisTypeConverter extends JdbcTypeConverter<String> {
-  // TODO: add implementation for doris catalog
+  static final String BOOLEAN = "boolean";
+  static final String TINYINT = "tinyint";
+  static final String SMALLINT = "smallint";
+  static final String INT = "int";
+  static final String BIGINT = "bigint";
+  static final String FLOAT = "float";
+  static final String DOUBLE = "double";
+  static final String DECIMAL = "decimal";
+  static final String DATETIME = "datetime";
+  static final String CHAR = "char";
+  static final String STRING = "string";
 
   @Override
   public Type toGravitinoType(JdbcTypeBean typeBean) {
-    return Types.UnparsedType.of(typeBean.getTypeName());
+    switch (typeBean.getTypeName().toLowerCase()) {
+      case BOOLEAN:
+        return Types.BooleanType.get();
+      case TINYINT:
+        return Types.ByteType.get();
+      case SMALLINT:
+        return Types.ShortType.get();
+      case INT:
+        return Types.IntegerType.get();
+      case BIGINT:
+        return Types.LongType.get();
+      case FLOAT:
+        return Types.FloatType.get();
+      case DOUBLE:
+        return Types.DoubleType.get();
+      case DECIMAL:
+        return Types.DecimalType.of(
+            Integer.parseInt(typeBean.getColumnSize()), Integer.parseInt(typeBean.getScale()));
+      case DATE:
+        return Types.DateType.get();
+      case DATETIME:
+        return Types.TimestampType.withTimeZone();
+      case CHAR:
+        return Types.FixedCharType.of(Integer.parseInt(typeBean.getColumnSize()));
+      case VARCHAR:
+        return Types.VarCharType.of(Integer.parseInt(typeBean.getColumnSize()));
+      case STRING:
+      case TEXT:
+        return Types.StringType.get();
+      default:
+        throw new IllegalArgumentException("Not a supported type: " + typeBean);
+    }
   }
 
   @Override
   public String fromGravitinoType(Type type) {
-    throw new IllegalArgumentException("unsupported type: " + type.simpleString());
+    if (type instanceof Types.BooleanType) {
+      return BOOLEAN;
+    } else if (type instanceof Types.ByteType) {
+      return TINYINT;
+    } else if (type instanceof Types.ShortType) {
+      return SMALLINT;
+    } else if (type instanceof Types.IntegerType) {
+      return INT;
+    } else if (type instanceof Types.LongType) {
+      return BIGINT;
+    } else if (type instanceof Types.FloatType) {
+      return FLOAT;
+    } else if (type instanceof Types.DoubleType) {
+      return DOUBLE;
+    } else if (type instanceof Types.DecimalType) {
+      return DECIMAL
+          + "("
+          + ((Types.DecimalType) type).precision()
+          + ","
+          + ((Types.DecimalType) type).scale()
+          + ")";
+    } else if (type instanceof Types.DateType) {
+      return DATE;
+    } else if (type instanceof Types.TimestampType) {
+      return DATETIME;
+    } else if (type instanceof Types.VarCharType) {
+      int length = ((Types.VarCharType) type).length();
+      if (length < 1 || length > 65533) {
+        throw new IllegalArgumentException(
+            String.format(
+                "Type %s is invalid, length should be between 1 and 65533", type.simpleString()));
+      }
+      return VARCHAR + "(" + ((Types.VarCharType) type).length() + ")";
+    } else if (type instanceof Types.FixedCharType) {
+      int length = ((Types.FixedCharType) type).length();
+      if (length < 1 || length > 255) {
+        throw new IllegalArgumentException(
+            String.format(
+                "Type %s is invalid, length should be between 1 and 255", type.simpleString()));
+      }
+
+      return CHAR + "(" + ((Types.FixedCharType) type).length() + ")";
+    } else if (type instanceof Types.StringType) {
+      return STRING;
+    }
+    throw new IllegalArgumentException(
+        String.format("Couldn't convert Gravitino type %s to Doris type", type.simpleString()));
   }
 }