[#12069] fix(jdbc-starrocks): complete V3 type compatibility

catalogs/catalog-jdbc-starrocks/src/main/java/org/apache/gravitino/catalog/starrocks/converter/StarRocksTypeConverter.java

@@ -146,7 +146,36 @@ public String fromGravitino(Type type) {
     } else if (type instanceof Types.DateType) {
       return DATE;
     } else if (type instanceof Types.TimestampType) {
+      Types.TimestampType timestampType = (Types.TimestampType) type;
+      if (timestampType.hasTimeZone()) {
+        throw new IllegalArgumentException(
+            String.format(
+                "StarRocks DATETIME does not preserve time-zone semantics; cannot convert Gravitino type %s",
+                type.simpleString()));
+      }
+      if (timestampType.hasPrecisionSet()) {
+        throw new IllegalArgumentException(
+            String.format(
+                "StarRocks DATETIME columns do not preserve declared fractional precision; cannot convert Gravitino type %s",
+                type.simpleString()));
+      }
       return DATETIME;
+    } else if (type instanceof Types.VariantType) {
+      throw new IllegalArgumentException(
+          "StarRocks JSON is not an exact representation of Gravitino Variant; cannot convert Gravitino type variant");
+    } else if (type instanceof Types.NullType) {
+      throw new IllegalArgumentException(
+          "StarRocks table columns cannot represent Gravitino Unknown (NullType); cannot convert Gravitino type null");
+    } else if (type instanceof Types.GeometryType) {
+      throw new IllegalArgumentException(
+          String.format(
+              "StarRocks has no storable GEOMETRY column type with CRS metadata; cannot convert Gravitino type %s",
+              type.simpleString()));
+    } else if (type instanceof Types.GeographyType) {
+      throw new IllegalArgumentException(
+          String.format(
+              "StarRocks has no storable GEOGRAPHY column type with CRS and edge-algorithm metadata; cannot convert Gravitino type %s",
+              type.simpleString()));
     }
     throw new IllegalArgumentException(
         String.format("Couldn't convert Gravitino type %s to StarRocks type", type.simpleString()));

catalogs/catalog-jdbc-starrocks/src/test/java/org/apache/gravitino/catalog/starrocks/converter/TestStarRocksTypeConverter.java

@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gravitino.catalog.starrocks.converter;
+
+import org.apache.gravitino.catalog.jdbc.converter.JdbcTypeConverter;
+import org.apache.gravitino.rel.types.Type;
+import org.apache.gravitino.rel.types.Types;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class TestStarRocksTypeConverter {
+
+  private static final StarRocksTypeConverter TYPE_CONVERTER = new StarRocksTypeConverter();
+
+  @Test
+  void testTimestampMapping() {
+    Assertions.assertEquals(Types.TimestampType.withoutTimeZone(), toGravitino("datetime", null));
+    Assertions.assertEquals(Types.TimestampType.withoutTimeZone(), toGravitino("datetime", 0));
+    Assertions.assertEquals(Types.TimestampType.withoutTimeZone(), toGravitino("datetime", 6));
+
+    Assertions.assertEquals(
+        "datetime", TYPE_CONVERTER.fromGravitino(Types.TimestampType.withoutTimeZone()));
+  }
+
+  @Test
+  void testRejectPrecisionQualifiedTimestampTypes() {
+    Type[] timestampTypes = {
+      Types.TimestampType.withoutTimeZone(3), Types.TimestampType.withoutTimeZone(9)
+    };
+    for (Type timestampType : timestampTypes) {
+      IllegalArgumentException timestampException =
+          Assertions.assertThrows(
+              IllegalArgumentException.class, () -> TYPE_CONVERTER.fromGravitino(timestampType));
+      Assertions.assertTrue(
+          timestampException
+              .getMessage()
+              .contains(
+                  "StarRocks DATETIME columns do not preserve declared fractional precision"));
+    }
+
+    IllegalArgumentException timestampTzException =
+        Assertions.assertThrows(
+            IllegalArgumentException.class,
+            () -> TYPE_CONVERTER.fromGravitino(Types.TimestampType.withTimeZone(9)));
+    Assertions.assertTrue(
+        timestampTzException
+            .getMessage()
+            .contains("StarRocks DATETIME does not preserve time-zone semantics"));
+  }
+
+  @Test
+  void testRejectVariantType() {
+    Assertions.assertEquals(Types.ExternalType.of("json"), toGravitino("json", null));
+
+    IllegalArgumentException exception =
+        Assertions.assertThrows(
+            IllegalArgumentException.class,
+            () -> TYPE_CONVERTER.fromGravitino(Types.VariantType.get()));
+    Assertions.assertTrue(
+        exception
+            .getMessage()
+            .contains("StarRocks JSON is not an exact representation of Gravitino Variant"));
+  }
+
+  @Test
+  void testRejectUnknownType() {
+    Assertions.assertEquals(Types.ExternalType.of("unknown"), toGravitino("unknown", null));
+
+    IllegalArgumentException exception =
+        Assertions.assertThrows(
+            IllegalArgumentException.class,
+            () -> TYPE_CONVERTER.fromGravitino(Types.NullType.get()));
+    Assertions.assertTrue(
+        exception
+            .getMessage()
+            .contains("StarRocks table columns cannot represent Gravitino Unknown (NullType)"));
+  }
+
+  @Test
+  void testRejectGeometryType() {
+    Assertions.assertEquals(Types.ExternalType.of("geometry"), toGravitino("geometry", null));
+
+    IllegalArgumentException exception =
+        Assertions.assertThrows(
+            IllegalArgumentException.class,
+            () -> TYPE_CONVERTER.fromGravitino(Types.GeometryType.of("EPSG:3857")));
+    Assertions.assertTrue(
+        exception
+            .getMessage()
+            .contains("StarRocks has no storable GEOMETRY column type with CRS metadata"));
+  }
+
+  @Test
+  void testRejectGeographyType() {
+    Assertions.assertEquals(Types.ExternalType.of("geography"), toGravitino("geography", null));
+
+    IllegalArgumentException exception =
+        Assertions.assertThrows(
+            IllegalArgumentException.class,
+            () -> TYPE_CONVERTER.fromGravitino(Types.GeographyType.of("EPSG:4326", "karney")));
+    Assertions.assertTrue(
+        exception
+            .getMessage()
+            .contains(
+                "StarRocks has no storable GEOGRAPHY column type with CRS and edge-algorithm metadata"));
+  }
+
+  private static Type toGravitino(String typeName, Integer datetimePrecision) {
+    JdbcTypeConverter.JdbcTypeBean typeBean = new JdbcTypeConverter.JdbcTypeBean(typeName);
+    typeBean.setDatetimePrecision(datetimePrecision);
+    return TYPE_CONVERTER.toGravitino(typeBean);
+  }
+}

catalogs/catalog-jdbc-starrocks/src/test/java/org/apache/gravitino/catalog/starrocks/integration/test/CatalogStarRocksIT.java

@@ -74,6 +74,7 @@
 import org.apache.gravitino.rel.partitions.Partition;
 import org.apache.gravitino.rel.partitions.Partitions;
 import org.apache.gravitino.rel.partitions.RangePartition;
+import org.apache.gravitino.rel.types.Type;
 import org.apache.gravitino.rel.types.Types;
 import org.apache.gravitino.utils.RandomNameUtils;
 import org.awaitility.Awaitility;
@@ -808,6 +809,54 @@ void testTableWithTimeStampColumn() {
     Assertions.assertEquals(Types.TimestampType.withoutTimeZone(), timestampColumn.dataType());
   }
 
+  @Test
+  void testRejectPrecisionQualifiedTimestampTypesWithoutSideEffects() {
+    assertCreateRejectedWithoutSideEffects(
+        "timestamp_3",
+        Types.TimestampType.withoutTimeZone(3),
+        "StarRocks DATETIME columns do not preserve declared fractional precision");
+    assertCreateRejectedWithoutSideEffects(
+        "timestamp_ns",
+        Types.TimestampType.withoutTimeZone(9),
+        "StarRocks DATETIME columns do not preserve declared fractional precision");
+    assertCreateRejectedWithoutSideEffects(
+        "timestamptz_ns",
+        Types.TimestampType.withTimeZone(9),
+        "StarRocks DATETIME does not preserve time-zone semantics");
+  }
+
+  @Test
+  void testRejectVariantWithoutSideEffects() {
+    assertCreateRejectedWithoutSideEffects(
+        "variant",
+        Types.VariantType.get(),
+        "StarRocks JSON is not an exact representation of Gravitino Variant");
+  }
+
+  @Test
+  void testRejectUnknownWithoutSideEffects() {
+    assertCreateRejectedWithoutSideEffects(
+        "unknown",
+        Types.NullType.get(),
+        "StarRocks table columns cannot represent Gravitino Unknown (NullType)");
+  }
+
+  @Test
+  void testRejectGeometryWithoutSideEffects() {
+    assertCreateRejectedWithoutSideEffects(
+        "geometry",
+        Types.GeometryType.of("EPSG:3857"),
+        "StarRocks has no storable GEOMETRY column type with CRS metadata");
+  }
+
+  @Test
+  void testRejectGeographyWithoutSideEffects() {
+    assertCreateRejectedWithoutSideEffects(
+        "geography",
+        Types.GeographyType.of("EPSG:4326", "karney"),
+        "StarRocks has no storable GEOGRAPHY column type with CRS and edge-algorithm metadata");
+  }
+
   @Test
   void testNonPartitionedTable() {
     // create a non-partitioned table
@@ -1000,4 +1049,29 @@ void testColumnDefaultValue() {
         colDefaultValues[1].defaultValue());
     Assertions.assertEquals(DEFAULT_VALUE_OF_CURRENT_TIMESTAMP, colDefaultValues[2].defaultValue());
   }
+
+  private void assertCreateRejectedWithoutSideEffects(
+      String typeName, Type type, String expectedMessage) {
+    String rejectedTableName = GravitinoITUtils.genRandomName("rejected_" + typeName);
+    NameIdentifier tableIdentifier = NameIdentifier.of(schemaName, rejectedTableName);
+    TableCatalog tableCatalog = catalog.asTableCatalog();
+    Distribution distribution = Distributions.hash(2, NamedReference.field("unsupported_col"));
+
+    IllegalArgumentException exception =
+        assertThrows(
+            IllegalArgumentException.class,
+            () ->
+                tableCatalog.createTable(
+                    tableIdentifier,
+                    new Column[] {Column.of("unsupported_col", type)},
+                    null,
+                    ImmutableMap.of(),
+                    Transforms.EMPTY_TRANSFORM,
+                    distribution,
+                    null,
+                    null));
+
+    Assertions.assertTrue(exception.getMessage().contains(expectedMessage));
+    Assertions.assertFalse(tableCatalog.tableExists(tableIdentifier));
+  }
 }

catalogs/catalog-jdbc-starrocks/src/test/java/org/apache/gravitino/catalog/starrocks/operation/TestStarRocksTableOperationsSqlGeneration.java

@@ -120,4 +120,112 @@ public void testCreateTableWithWhitespaceDefaultValue() {
         sql.contains("DEFAULT " + converter.fromGravitino(col1.defaultValue())),
         "Should contain DEFAULT '   ' but was: " + sql);
   }
+
+  @Test
+  public void testCreateTableUsesUnparameterizedDatetime() {
+    TestableStarRocksTableOperations ops = new TestableStarRocksTableOperations();
+    JdbcColumn column =
+        JdbcColumn.builder()
+            .withName("timestamp_col")
+            .withType(Types.TimestampType.withoutTimeZone())
+            .build();
+    Distribution distribution = Distributions.hash(1, NamedReference.field("timestamp_col"));
+
+    String sql = ops.createTableSql("test_table", new JdbcColumn[] {column}, distribution);
+
+    Assertions.assertTrue(sql.contains("`timestamp_col` datetime"));
+  }
+
+  @Test
+  public void testCreateTableRejectsPrecisionQualifiedTimestampTypesBeforeSqlGeneration() {
+    TestableStarRocksTableOperations ops = new TestableStarRocksTableOperations();
+    Types.TimestampType[] unsupportedTypes = {
+      Types.TimestampType.withoutTimeZone(3),
+      Types.TimestampType.withoutTimeZone(9),
+      Types.TimestampType.withTimeZone(9)
+    };
+    Distribution distribution = Distributions.hash(1, NamedReference.field("unsupported_col"));
+
+    for (Types.TimestampType type : unsupportedTypes) {
+      JdbcColumn column = JdbcColumn.builder().withName("unsupported_col").withType(type).build();
+      Assertions.assertThrows(
+          IllegalArgumentException.class,
+          () -> ops.createTableSql("test_table", new JdbcColumn[] {column}, distribution));
+    }
+  }
+
+  @Test
+  public void testCreateTableRejectsVariantBeforeSqlGeneration() {
+    TestableStarRocksTableOperations ops = new TestableStarRocksTableOperations();
+    JdbcColumn column =
+        JdbcColumn.builder().withName("unsupported_col").withType(Types.VariantType.get()).build();
+    Distribution distribution = Distributions.hash(1, NamedReference.field("unsupported_col"));
+
+    IllegalArgumentException exception =
+        Assertions.assertThrows(
+            IllegalArgumentException.class,
+            () -> ops.createTableSql("test_table", new JdbcColumn[] {column}, distribution));
+    Assertions.assertTrue(
+        exception
+            .getMessage()
+            .contains("StarRocks JSON is not an exact representation of Gravitino Variant"));
+  }
+
+  @Test
+  public void testCreateTableRejectsUnknownBeforeSqlGeneration() {
+    TestableStarRocksTableOperations ops = new TestableStarRocksTableOperations();
+    JdbcColumn column =
+        JdbcColumn.builder().withName("unsupported_col").withType(Types.NullType.get()).build();
+    Distribution distribution = Distributions.hash(1, NamedReference.field("unsupported_col"));
+
+    IllegalArgumentException exception =
+        Assertions.assertThrows(
+            IllegalArgumentException.class,
+            () -> ops.createTableSql("test_table", new JdbcColumn[] {column}, distribution));
+    Assertions.assertTrue(
+        exception
+            .getMessage()
+            .contains("StarRocks table columns cannot represent Gravitino Unknown (NullType)"));
+  }
+
+  @Test
+  public void testCreateTableRejectsGeometryBeforeSqlGeneration() {
+    TestableStarRocksTableOperations ops = new TestableStarRocksTableOperations();
+    JdbcColumn column =
+        JdbcColumn.builder()
+            .withName("unsupported_col")
+            .withType(Types.GeometryType.of("EPSG:3857"))
+            .build();
+    Distribution distribution = Distributions.hash(1, NamedReference.field("unsupported_col"));
+
+    IllegalArgumentException exception =
+        Assertions.assertThrows(
+            IllegalArgumentException.class,
+            () -> ops.createTableSql("test_table", new JdbcColumn[] {column}, distribution));
+    Assertions.assertTrue(
+        exception
+            .getMessage()
+            .contains("StarRocks has no storable GEOMETRY column type with CRS metadata"));
+  }
+
+  @Test
+  public void testCreateTableRejectsGeographyBeforeSqlGeneration() {
+    TestableStarRocksTableOperations ops = new TestableStarRocksTableOperations();
+    JdbcColumn column =
+        JdbcColumn.builder()
+            .withName("unsupported_col")
+            .withType(Types.GeographyType.of("EPSG:4326", "karney"))
+            .build();
+    Distribution distribution = Distributions.hash(1, NamedReference.field("unsupported_col"));
+
+    IllegalArgumentException exception =
+        Assertions.assertThrows(
+            IllegalArgumentException.class,
+            () -> ops.createTableSql("test_table", new JdbcColumn[] {column}, distribution));
+    Assertions.assertTrue(
+        exception
+            .getMessage()
+            .contains(
+                "StarRocks has no storable GEOGRAPHY column type with CRS and edge-algorithm metadata"));
+  }
 }