[#6656 ] fix(trino-connector): support read MySQL time/datetime/timestamp columns with precision

api/src/main/java/org/apache/gravitino/rel/types/Types.java

@@ -330,14 +330,31 @@ public String simpleString() {
 
   /** The time type in Gravitino. */
   public static class TimeType extends Type.DateTimeType {
-    private static final TimeType INSTANCE = new TimeType();
+    private static final TimeType INSTANCE = new TimeType(0);
 
     /** @return The singleton instance of {@link TimeType}. */
     public static TimeType get() {
       return INSTANCE;
     }
 
-    private TimeType() {}
+    /**
+     * @param precision The precision of the time type.
+     * @return A {@link TimeType} with the given precision.
+     */
+    public static TimeType of(int precision) {
+      return new TimeType(precision);
+    }
+
+    private final int precision;
+
+    private TimeType(int precision) {
+      this.precision = precision;
+    }
+
+    /** @return The precision of the time type. */
+    public int precision() {
+      return precision;
+    }
 
     @Override
     public Name name() {
@@ -346,14 +363,27 @@ public Name name() {
 
     @Override
     public String simpleString() {
-      return "time";
+      return precision == 0 ? "time" : String.format("time(%d)", precision);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) return true;
+      if (!(o instanceof TimeType)) return false;
+      TimeType that = (TimeType) o;
+      return precision == that.precision;
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(precision);
     }
   }
 
   /** The timestamp type in Gravitino. */
   public static class TimestampType extends Type.DateTimeType {
-    private static final TimestampType INSTANCE_WITHOUT_TIME_ZONE = new TimestampType(false);
-    private static final TimestampType INSTANCE_WITH_TIME_ZONE = new TimestampType(true);
+    private static final TimestampType INSTANCE_WITHOUT_TIME_ZONE = new TimestampType(false, 0);
+    private static final TimestampType INSTANCE_WITH_TIME_ZONE = new TimestampType(true, 0);
 
     /** @return A {@link TimestampType} with time zone. */
     public static TimestampType withTimeZone() {
@@ -365,17 +395,40 @@ public static TimestampType withoutTimeZone() {
       return INSTANCE_WITHOUT_TIME_ZONE;
     }
 
+    /**
+     * @param precision The precision of the timestamp type.
+     * @return A {@link TimestampType} with the given precision and time zone.
+     */
+    public static TimestampType withTimeZone(int precision) {
+      return new TimestampType(true, precision);
+    }
+
+    /**
+     * @param precision The precision of the timestamp type.
+     * @return A {@link TimestampType} with the given precision and without time zone.
+     */
+    public static TimestampType withoutTimeZone(int precision) {
+      return new TimestampType(false, precision);
+    }
+
     private final boolean withTimeZone;
+    private final int precision;
 
-    private TimestampType(boolean withTimeZone) {
+    private TimestampType(boolean withTimeZone, int precision) {
       this.withTimeZone = withTimeZone;
+      this.precision = precision;
     }
 
     /** @return True if the timestamp type has time zone, false otherwise. */
     public boolean hasTimeZone() {
       return withTimeZone;
     }
 
+    /** @return The precision of the timestamp type. */
+    public int precision() {
+      return precision;
+    }
+
     @Override
     public Name name() {
       return Name.TIMESTAMP;
@@ -384,7 +437,25 @@ public Name name() {
     /** @return The simple string representation of the timestamp type. */
     @Override
     public String simpleString() {
-      return withTimeZone ? "timestamp_tz" : "timestamp";
+      if (precision == 0) {
+        return withTimeZone ? "timestamp_tz" : "timestamp";
+      }
+      return withTimeZone
+          ? String.format("timestamp_tz(%d)", precision)
+          : String.format("timestamp(%d)", precision);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) return true;
+      if (!(o instanceof TimestampType)) return false;
+      TimestampType that = (TimestampType) o;
+      return withTimeZone == that.withTimeZone && precision == that.precision;
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(withTimeZone, precision);
     }
   }
 

catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/converter/JdbcTypeConverter.java

@@ -40,6 +40,8 @@ public static class JdbcTypeBean {
     /** Scale. For example: 2 in decimal (10,2). */
     private Integer scale;
 
+    private Integer datetimePrecision;
+
     public JdbcTypeBean(String typeName) {
       this.typeName = typeName;
     }
@@ -68,19 +70,28 @@ public void setScale(Integer scale) {
       this.scale = scale;
     }
 
+    public Integer getDatetimePrecision() {
+      return datetimePrecision;
+    }
+
+    public void setDatetimePrecision(Integer datetimePrecision) {
+      this.datetimePrecision = datetimePrecision;
+    }
+
     @Override
     public boolean equals(Object o) {
       if (this == o) return true;
       if (!(o instanceof JdbcTypeBean)) return false;
       JdbcTypeBean typeBean = (JdbcTypeBean) o;
       return Objects.equals(typeName, typeBean.typeName)
           && Objects.equals(columnSize, typeBean.columnSize)
-          && Objects.equals(scale, typeBean.scale);
+          && Objects.equals(scale, typeBean.scale)
+          && Objects.equals(datetimePrecision, typeBean.datetimePrecision);
     }
 
     @Override
     public int hashCode() {
-      return Objects.hash(typeName, columnSize, scale);
+      return Objects.hash(typeName, columnSize, scale, datetimePrecision);
     }
 
     @Override
@@ -95,6 +106,9 @@ public String toString() {
           + ", scale='"
           + scale
           + '\''
+          + ", datetimePrecision='"
+          + datetimePrecision
+          + '\''
           + '}';
     }
   }

catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/operation/JdbcTableOperations.java

@@ -585,10 +585,14 @@ protected JdbcTable.Builder getBasicJdbcTableInfo(ResultSet table) throws SQLExc
   }
 
   protected JdbcColumn.Builder getBasicJdbcColumnInfo(ResultSet column) throws SQLException {
-    JdbcTypeConverter.JdbcTypeBean typeBean =
-        new JdbcTypeConverter.JdbcTypeBean(column.getString("TYPE_NAME"));
-    typeBean.setColumnSize(column.getInt("COLUMN_SIZE"));
+    String typeName = column.getString("TYPE_NAME");
+    int columnSize = column.getInt("COLUMN_SIZE");
+    JdbcTypeConverter.JdbcTypeBean typeBean = new JdbcTypeConverter.JdbcTypeBean(typeName);
+    typeBean.setColumnSize(columnSize);
     typeBean.setScale(column.getInt("DECIMAL_DIGITS"));
+    int datetimePrecision = calculateDatetimePrecision(typeName, columnSize);
+    typeBean.setDatetimePrecision(datetimePrecision);
+
     String comment = column.getString("REMARKS");
     boolean nullable = column.getBoolean("NULLABLE");
 
@@ -604,4 +608,15 @@ protected JdbcColumn.Builder getBasicJdbcColumnInfo(ResultSet column) throws SQL
         .withNullable(nullable)
         .withDefaultValue(defaultValue);
   }
+
+  /**
+   * Calculate the precision for time/datetime/timestamp types.
+   *
+   * @param typeName the type name from database
+   * @param columnSize the column size from database
+   * @return the precision of the time/datetime/timestamp type
+   */
+  public int calculateDatetimePrecision(String typeName, int columnSize) {
+    return 0;
+  }
 }

catalogs/catalog-jdbc-mysql/src/main/java/org/apache/gravitino/catalog/mysql/converter/MysqlColumnDefaultValueConverter.java

@@ -49,7 +49,8 @@ public Expression toGravitino(
     }
 
     if (isExpression) {
-      if (columnDefaultValue.equals(CURRENT_TIMESTAMP)) {
+      if (columnDefaultValue.equals(CURRENT_TIMESTAMP)
+          || columnDefaultValue.startsWith(CURRENT_TIMESTAMP + "(")) {
         return DEFAULT_VALUE_OF_CURRENT_TIMESTAMP;
       }
       // The parsing of MySQL expressions is complex, so we are not currently undertaking the
@@ -88,6 +89,7 @@ public Expression toGravitino(
       case JdbcTypeConverter.TIMESTAMP:
       case MysqlTypeConverter.DATETIME:
         return CURRENT_TIMESTAMP.equals(columnDefaultValue)
+                || columnDefaultValue.startsWith(CURRENT_TIMESTAMP + "(")
             ? DEFAULT_VALUE_OF_CURRENT_TIMESTAMP
             : Literals.timestampLiteral(
                 LocalDateTime.parse(columnDefaultValue, DATE_TIME_FORMATTER));

catalogs/catalog-jdbc-mysql/src/main/java/org/apache/gravitino/catalog/mysql/converter/MysqlTypeConverter.java

@@ -72,15 +72,18 @@ public Type toGravitino(JdbcTypeBean typeBean) {
       case DATE:
         return Types.DateType.get();
       case TIME:
-        return Types.TimeType.get();
+        return Types.TimeType.of(
+            typeBean.getDatetimePrecision() != null ? typeBean.getDatetimePrecision() : 0);
         // MySQL converts TIMESTAMP values from the current time zone to UTC for storage, and back
         // from UTC to the current time zone for retrieval. (This does not occur for other types
         // such as DATETIME.) see more details:
         // https://dev.mysql.com/doc/refman/8.0/en/datetime.html
       case TIMESTAMP:
-        return Types.TimestampType.withTimeZone();
+        return Types.TimestampType.withTimeZone(
+            typeBean.getDatetimePrecision() != null ? typeBean.getDatetimePrecision() : 0);
       case DATETIME:
-        return Types.TimestampType.withoutTimeZone();
+        return Types.TimestampType.withoutTimeZone(
+            typeBean.getDatetimePrecision() != null ? typeBean.getDatetimePrecision() : 0);
       case DECIMAL:
         return Types.DecimalType.of(typeBean.getColumnSize(), typeBean.getScale());
       case VARCHAR:
@@ -136,7 +139,11 @@ public String fromGravitino(Type type) {
       // MySQL converts TIMESTAMP values from the current time zone to UTC for storage, and back
       // from UTC to the current time zone for retrieval. (This does not occur for other types
       // such as DATETIME.) see more details: https://dev.mysql.com/doc/refman/8.0/en/datetime.html
-      return ((Types.TimestampType) type).hasTimeZone() ? TIMESTAMP : DATETIME;
+      Types.TimestampType timestampType = (Types.TimestampType) type;
+      String baseType = timestampType.hasTimeZone() ? TIMESTAMP : DATETIME;
+      return timestampType.precision() > 0
+          ? String.format("%s(%d)", baseType, timestampType.precision())
+          : baseType;
     } else if (type instanceof Types.DecimalType) {
       return type.simpleString();
     } else if (type instanceof Types.VarCharType) {

catalogs/catalog-jdbc-mysql/src/main/java/org/apache/gravitino/catalog/mysql/operation/MysqlTableOperations.java

@@ -672,4 +672,17 @@ private static void validateIndexes(Index[] indexes, JdbcColumn[] columns) {
       }
     }
   }
+
+  @Override
+  public int calculateDatetimePrecision(String typeName, int columnSize) {
+    String upperTypeName = typeName.toUpperCase();
+    switch (upperTypeName) {
+      case "TIME":
+        return columnSize > 8 ? columnSize - 9 : 0;
+      case "TIMESTAMP":
+      case "DATETIME":
+        return columnSize > 19 ? columnSize - 20 : 0;
+    }
+    return 0;
+  }
 }

catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/converter/TestMysqlTypeConverter.java

@@ -47,22 +47,34 @@ public class TestMysqlTypeConverter {
 
   @Test
   public void testToGravitinoType() {
-    checkJdbcTypeToGravitinoType(Types.ByteType.get(), TINYINT, null, null);
-    checkJdbcTypeToGravitinoType(Types.IntegerType.get(), INT, null, null);
-    checkJdbcTypeToGravitinoType(Types.LongType.get(), BIGINT, null, null);
-    checkJdbcTypeToGravitinoType(Types.FloatType.get(), FLOAT, null, null);
-    checkJdbcTypeToGravitinoType(Types.DoubleType.get(), DOUBLE, null, null);
-    checkJdbcTypeToGravitinoType(Types.DateType.get(), DATE, null, null);
-    checkJdbcTypeToGravitinoType(Types.TimeType.get(), TIME, null, null);
-    checkJdbcTypeToGravitinoType(Types.TimestampType.withoutTimeZone(), DATETIME, null, null);
-    checkJdbcTypeToGravitinoType(Types.TimestampType.withTimeZone(), TIMESTAMP, null, null);
-    checkJdbcTypeToGravitinoType(Types.DecimalType.of(10, 2), DECIMAL, 10, 2);
-    checkJdbcTypeToGravitinoType(Types.VarCharType.of(20), VARCHAR, 20, null);
-    checkJdbcTypeToGravitinoType(Types.FixedCharType.of(20), CHAR, 20, null);
-    checkJdbcTypeToGravitinoType(Types.StringType.get(), TEXT, null, null);
-    checkJdbcTypeToGravitinoType(Types.BinaryType.get(), BINARY, null, null);
+    checkJdbcTypeToGravitinoType(Types.ByteType.get(), TINYINT, null, null, 0);
+    checkJdbcTypeToGravitinoType(Types.IntegerType.get(), INT, null, null, 0);
+    checkJdbcTypeToGravitinoType(Types.LongType.get(), BIGINT, null, null, 0);
+    checkJdbcTypeToGravitinoType(Types.FloatType.get(), FLOAT, null, null, 0);
+    checkJdbcTypeToGravitinoType(Types.DoubleType.get(), DOUBLE, null, null, 0);
+    checkJdbcTypeToGravitinoType(Types.DateType.get(), DATE, null, null, 0);
+    checkJdbcTypeToGravitinoType(Types.TimeType.get(), TIME, null, null, 0);
+    checkJdbcTypeToGravitinoType(Types.TimeType.of(0), TIME, 8, null, 0);
+    checkJdbcTypeToGravitinoType(Types.TimeType.of(3), TIME, 12, null, 3);
+    checkJdbcTypeToGravitinoType(Types.TimeType.of(6), TIME, 15, null, 6);
+    checkJdbcTypeToGravitinoType(Types.TimeType.of(9), TIME, 18, null, 9);
+    checkJdbcTypeToGravitinoType(Types.TimestampType.withoutTimeZone(), DATETIME, null, null, 0);
+    checkJdbcTypeToGravitinoType(Types.TimestampType.withoutTimeZone(0), DATETIME, 19, null, 0);
+    checkJdbcTypeToGravitinoType(Types.TimestampType.withoutTimeZone(3), DATETIME, 23, null, 3);
+    checkJdbcTypeToGravitinoType(Types.TimestampType.withoutTimeZone(6), DATETIME, 26, null, 6);
+    checkJdbcTypeToGravitinoType(Types.TimestampType.withoutTimeZone(9), DATETIME, 29, null, 9);
+    checkJdbcTypeToGravitinoType(Types.TimestampType.withTimeZone(), TIMESTAMP, null, null, 0);
+    checkJdbcTypeToGravitinoType(Types.TimestampType.withTimeZone(0), TIMESTAMP, null, null, 0);
+    checkJdbcTypeToGravitinoType(Types.TimestampType.withTimeZone(3), TIMESTAMP, 23, null, 3);
+    checkJdbcTypeToGravitinoType(Types.TimestampType.withTimeZone(6), TIMESTAMP, 26, null, 6);
+    checkJdbcTypeToGravitinoType(Types.TimestampType.withTimeZone(9), TIMESTAMP, 29, null, 9);
+    checkJdbcTypeToGravitinoType(Types.DecimalType.of(10, 2), DECIMAL, 10, 2, 0);
+    checkJdbcTypeToGravitinoType(Types.VarCharType.of(20), VARCHAR, 20, null, 0);
+    checkJdbcTypeToGravitinoType(Types.FixedCharType.of(20), CHAR, 20, null, 0);
+    checkJdbcTypeToGravitinoType(Types.StringType.get(), TEXT, null, null, 0);
+    checkJdbcTypeToGravitinoType(Types.BinaryType.get(), BINARY, null, null, 0);
     checkJdbcTypeToGravitinoType(
-        Types.ExternalType.of(USER_DEFINED_TYPE), USER_DEFINED_TYPE, null, null);
+        Types.ExternalType.of(USER_DEFINED_TYPE), USER_DEFINED_TYPE, null, null, 0);
   }
 
   @Test
@@ -92,17 +104,23 @@ protected void checkGravitinoTypeToJdbcType(String jdbcTypeName, Type gravitinoT
   }
 
   protected void checkJdbcTypeToGravitinoType(
-      Type gravitinoType, String jdbcTypeName, Integer columnSize, Integer scale) {
-    JdbcTypeConverter.JdbcTypeBean typeBean = createTypeBean(jdbcTypeName, columnSize, scale);
+      Type gravitinoType,
+      String jdbcTypeName,
+      Integer columnSize,
+      Integer scale,
+      Integer datetimePrecision) {
+    JdbcTypeConverter.JdbcTypeBean typeBean =
+        createTypeBean(jdbcTypeName, columnSize, scale, datetimePrecision);
     Assertions.assertEquals(gravitinoType, MYSQL_TYPE_CONVERTER.toGravitino(typeBean));
   }
 
   protected static JdbcTypeConverter.JdbcTypeBean createTypeBean(
-      String typeName, Integer columnSize, Integer scale) {
+      String typeName, Integer columnSize, Integer scale, Integer datetimePrecision) {
     return new JdbcTypeConverter.JdbcTypeBean(typeName) {
       {
         setColumnSize(columnSize);
         setScale(scale);
+        setDatetimePrecision(datetimePrecision);
       }
     };
   }

catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/integration/test/CatalogMysqlIT.java

@@ -620,8 +620,7 @@ void testColumnDefaultValueConverter() {
               Literals.timestampLiteral("1983-09-05T00:00:00"), column.defaultValue());
           break;
         case "timestamp_col_3":
-          Assertions.assertEquals(
-              UnparsedExpression.of("CURRENT_TIMESTAMP(6)"), column.defaultValue());
+          Assertions.assertEquals(DEFAULT_VALUE_OF_CURRENT_TIMESTAMP, column.defaultValue());
           break;
         case "decimal_6_2_col_1":
           Assertions.assertEquals(
@@ -863,7 +862,7 @@ void testUpdateColumnDefaultValue() {
             TableChange.updateColumnDefaultValue(
                 new String[] {columns[0].name()}, Literals.of("1.2345", Types.FloatType.get())),
             TableChange.updateColumnDefaultValue(
-                new String[] {columns[1].name()}, FunctionExpression.of("current_timestamp")),
+                new String[] {columns[1].name()}, DEFAULT_VALUE_OF_CURRENT_TIMESTAMP),
             TableChange.updateColumnDefaultValue(
                 new String[] {columns[2].name()}, Literals.of("hello", Types.VarCharType.of(255))),
             TableChange.updateColumnDefaultValue(