[#12070] fix(jdbc-clickhouse): complete V3 type compatibility

catalogs-contrib/catalog-jdbc-clickhouse/src/main/java/org/apache/gravitino/catalog/clickhouse/converter/ClickHouseTypeConverter.java

@@ -18,6 +18,7 @@
  */
 package org.apache.gravitino.catalog.clickhouse.converter;
 
+import java.util.Optional;
 import org.apache.gravitino.catalog.jdbc.converter.JdbcTypeConverter;
 import org.apache.gravitino.rel.types.Type;
 import org.apache.gravitino.rel.types.Types;
@@ -73,6 +74,8 @@ public class ClickHouseTypeConverter extends JdbcTypeConverter {
   static final String JSON = "JSON";
 
   private static final int MAX_PRECISION = 76;
+  private static final int MAX_DATETIME_PRECISION = 9;
+  private static final String UTC = "UTC";
 
   @Override
   public Type toGravitino(JdbcTypeBean typeBean) {
@@ -83,6 +86,13 @@ public Type toGravitino(JdbcTypeBean typeBean) {
 
     Integer dateTimePrecision = TypeUtils.extractDateTimePrecision(typeName);
     if (dateTimePrecision != null) {
+      Optional<String> timeZone = TypeUtils.extractDateTimeTimeZone(typeName);
+      if (timeZone.isPresent()) {
+        if (UTC.equalsIgnoreCase(timeZone.get())) {
+          return Types.TimestampType.withTimeZone(dateTimePrecision);
+        }
+        return Types.ExternalType.of(typeBean.getTypeName());
+      }
       return Types.TimestampType.withoutTimeZone(dateTimePrecision);
     }
 
@@ -177,10 +187,17 @@ public String fromGravitino(Type type) {
     } else if (type instanceof Types.DateType) {
       return DATE;
     } else if (type instanceof Types.TimestampType timestampType) {
-      // Gravitino timestamp type maps to ClickHouse DateTime with precision 0.
-      // For precision > 0, use DateTime64(N).
-      if (timestampType.precision() > 0) {
-        return DATETIME64 + "(" + timestampType.precision() + ")";
+      int precision = timestampType.hasPrecisionSet() ? timestampType.precision() : 0;
+      if (precision > MAX_DATETIME_PRECISION) {
+        throw unsupportedType(type, "DateTime64 precision must be between 0 and 9");
+      }
+      if (timestampType.hasTimeZone()) {
+        return precision > 0
+            ? String.format("%s(%d, '%s')", DATETIME64, precision, UTC)
+            : String.format("%s('%s')", DATETIME, UTC);
+      }
+      if (precision > 0) {
+        return DATETIME64 + "(" + precision + ")";
       }
       return DATETIME;
     } else if (type instanceof Types.TimeType) {
@@ -195,11 +212,32 @@ public String fromGravitino(Type type) {
       return BOOL;
     } else if (type instanceof Types.UUIDType) {
       return UUID;
+    } else if (type instanceof Types.VariantType) {
+      throw unsupportedType(
+          type,
+          "ClickHouse Variant requires a closed list of alternative types and cannot preserve "
+              + "the open-ended Gravitino Variant contract");
+    } else if (type instanceof Types.NullType) {
+      throw unsupportedType(type, "the null-only placeholder has no ClickHouse column type");
+    } else if (type instanceof Types.GeometryType) {
+      throw unsupportedType(
+          type, "ClickHouse Geo types do not preserve Gravitino Geometry CRS metadata");
+    } else if (type instanceof Types.GeographyType) {
+      throw unsupportedType(
+          type,
+          "ClickHouse Geo types do not preserve Gravitino Geography CRS and edge-algorithm "
+              + "metadata");
     } else if (type instanceof Types.ExternalType) {
       return ((Types.ExternalType) type).catalogString();
     }
     throw new IllegalArgumentException(
         String.format(
             "Couldn't convert Gravitino type %s to ClickHouse type", type.simpleString()));
   }
+
+  private static IllegalArgumentException unsupportedType(Type type, String reason) {
+    return new IllegalArgumentException(
+        String.format(
+            "ClickHouse does not support Gravitino type %s: %s", type.simpleString(), reason));
+  }
 }

catalogs-contrib/catalog-jdbc-clickhouse/src/main/java/org/apache/gravitino/catalog/clickhouse/converter/TypeUtils.java

@@ -18,11 +18,17 @@
  */
 package org.apache.gravitino.catalog.clickhouse.converter;
 
+import java.util.Optional;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
 public class TypeUtils {
 
+  private static final Pattern DATETIME_PATTERN =
+      Pattern.compile("^DateTime(?:\\(\\s*'([^']+)'\\s*\\))?$");
+  private static final Pattern DATETIME64_PATTERN =
+      Pattern.compile("^DateTime64\\(\\s*(\\d+)\\s*(?:,\\s*'([^']+)'\\s*)?\\)$");
+
   private TypeUtils() {}
 
   public static String stripNullable(String typeName) {
@@ -34,10 +40,25 @@ public static String stripLowCardinality(String typeName) {
   }
 
   public static Integer extractDateTimePrecision(String typeName) {
-    Matcher matcher = Pattern.compile("^DateTime(?:64)?\\((\\d+)\\)$").matcher(typeName);
-    if (matcher.matches()) {
-      return Integer.parseInt(matcher.group(1));
+    Matcher dateTime64Matcher = DATETIME64_PATTERN.matcher(typeName);
+    if (dateTime64Matcher.matches()) {
+      return Integer.parseInt(dateTime64Matcher.group(1));
+    }
+    if (DATETIME_PATTERN.matcher(typeName).matches()) {
+      return 0;
     }
     return null;
   }
+
+  static Optional<String> extractDateTimeTimeZone(String typeName) {
+    Matcher dateTime64Matcher = DATETIME64_PATTERN.matcher(typeName);
+    if (dateTime64Matcher.matches()) {
+      return Optional.ofNullable(dateTime64Matcher.group(2));
+    }
+    Matcher dateTimeMatcher = DATETIME_PATTERN.matcher(typeName);
+    if (dateTimeMatcher.matches()) {
+      return Optional.ofNullable(dateTimeMatcher.group(1));
+    }
+    return Optional.empty();
+  }
 }

catalogs-contrib/catalog-jdbc-clickhouse/src/test/java/org/apache/gravitino/catalog/clickhouse/converter/TestClickHouseTypeConverter.java

@@ -107,6 +107,17 @@ public void testToGravitinoType() {
     Assertions.assertEquals(
         Types.TimestampType.withoutTimeZone(3),
         CLICKHOUSE_TYPE_CONVERTER.toGravitino(dateTime64WithPrecision));
+    checkJdbcTypeToGravitinoType(
+        Types.TimestampType.withoutTimeZone(9), "DateTime64(9)", null, null);
+    checkJdbcTypeToGravitinoType(
+        Types.TimestampType.withTimeZone(9), "DateTime64(9, 'UTC')", null, null);
+    checkJdbcTypeToGravitinoType(
+        Types.TimestampType.withTimeZone(0), "DateTime('UTC')", null, null);
+    checkJdbcTypeToGravitinoType(
+        Types.ExternalType.of("DateTime64(9, 'America/Los_Angeles')"),
+        "DateTime64(9, 'America/Los_Angeles')",
+        null,
+        null);
 
     // LowCardinality(Nullable(String)) should map to StringType
     JdbcTypeConverter.JdbcTypeBean lowCardNullable =
@@ -154,13 +165,74 @@ public void testFromGravitinoType() {
     checkGravitinoTypeToJdbcType("DateTime", Types.TimestampType.withoutTimeZone(0));
     // DateTime64(3) round-trip
     checkGravitinoTypeToJdbcType(DATETIME64 + "(3)", Types.TimestampType.withoutTimeZone(3));
+    checkGravitinoTypeToJdbcType(DATETIME64 + "(9)", Types.TimestampType.withoutTimeZone(9));
+    checkGravitinoTypeToJdbcType(DATETIME64 + "(9, 'UTC')", Types.TimestampType.withTimeZone(9));
+    checkGravitinoTypeToJdbcType(DATETIME + "('UTC')", Types.TimestampType.withTimeZone(0));
     // IPv4/IPv6 round-trip
     checkGravitinoTypeToJdbcType(IPV4, Types.ExternalType.of(IPV4));
     checkGravitinoTypeToJdbcType(IPV6, Types.ExternalType.of(IPV6));
     checkGravitinoTypeToJdbcType(TIME, Types.TimeType.get());
     Assertions.assertThrows(
         IllegalArgumentException.class,
         () -> CLICKHOUSE_TYPE_CONVERTER.fromGravitino(Types.UnparsedType.of(USER_DEFINED_TYPE)));
+    IllegalArgumentException precisionException =
+        Assertions.assertThrows(
+            IllegalArgumentException.class,
+            () -> CLICKHOUSE_TYPE_CONVERTER.fromGravitino(Types.TimestampType.withoutTimeZone(10)));
+    Assertions.assertTrue(
+        precisionException.getMessage().contains("DateTime64 precision must be between 0 and 9"));
+  }
+
+  @Test
+  public void testRejectVariantType() {
+    IllegalArgumentException exception =
+        Assertions.assertThrows(
+            IllegalArgumentException.class,
+            () -> CLICKHOUSE_TYPE_CONVERTER.fromGravitino(Types.VariantType.get()));
+    Assertions.assertTrue(
+        exception
+            .getMessage()
+            .contains(
+                "ClickHouse Variant requires a closed list of alternative types and cannot "
+                    + "preserve the open-ended Gravitino Variant contract"));
+  }
+
+  @Test
+  public void testRejectNullType() {
+    IllegalArgumentException exception =
+        Assertions.assertThrows(
+            IllegalArgumentException.class,
+            () -> CLICKHOUSE_TYPE_CONVERTER.fromGravitino(Types.NullType.get()));
+    Assertions.assertTrue(
+        exception.getMessage().contains("the null-only placeholder has no ClickHouse column type"));
+  }
+
+  @Test
+  public void testRejectGeometryType() {
+    IllegalArgumentException exception =
+        Assertions.assertThrows(
+            IllegalArgumentException.class,
+            () -> CLICKHOUSE_TYPE_CONVERTER.fromGravitino(Types.GeometryType.of("SRID:3857")));
+    Assertions.assertTrue(
+        exception
+            .getMessage()
+            .contains("ClickHouse Geo types do not preserve Gravitino Geometry CRS metadata"));
+  }
+
+  @Test
+  public void testRejectGeographyType() {
+    IllegalArgumentException exception =
+        Assertions.assertThrows(
+            IllegalArgumentException.class,
+            () ->
+                CLICKHOUSE_TYPE_CONVERTER.fromGravitino(
+                    Types.GeographyType.of("EPSG:4326", "karney")));
+    Assertions.assertTrue(
+        exception
+            .getMessage()
+            .contains(
+                "ClickHouse Geo types do not preserve Gravitino Geography CRS and "
+                    + "edge-algorithm metadata"));
   }
 
   protected void checkGravitinoTypeToJdbcType(String jdbcTypeName, Type gravitinoType) {

catalogs-contrib/catalog-jdbc-clickhouse/src/test/java/org/apache/gravitino/catalog/clickhouse/integration/test/CatalogClickHouseIT.java

@@ -76,6 +76,7 @@
 import org.apache.gravitino.rel.indexes.Index;
 import org.apache.gravitino.rel.indexes.Indexes;
 import org.apache.gravitino.rel.types.Decimal;
+import org.apache.gravitino.rel.types.Type;
 import org.apache.gravitino.rel.types.Types;
 import org.apache.gravitino.utils.RandomNameUtils;
 import org.junit.jupiter.api.AfterAll;
@@ -2482,4 +2483,106 @@ void testLoadTableWithoutSettings() {
     // SHOW CREATE TABLE output, so one settings key is expected even without explicit SETTINGS.
     Assertions.assertEquals(1, settingsCount);
   }
+
+  @Test
+  void testNanosecondTimestampRoundTrip() {
+    NameIdentifier tableIdentifier =
+        NameIdentifier.of(schemaName, GravitinoITUtils.genRandomName("timestamp_ns_round_trip"));
+    Column[] columns = {
+      Column.of("id", Types.IntegerType.get(), "sort column", false, false, DEFAULT_VALUE_NOT_SET),
+      Column.of(
+          "timestamp_ns",
+          Types.TimestampType.withoutTimeZone(9),
+          "nanosecond timestamp",
+          false,
+          false,
+          DEFAULT_VALUE_NOT_SET),
+      Column.of(
+          "timestamp_tz_ns",
+          Types.TimestampType.withTimeZone(9),
+          "nanosecond timestamp with time zone",
+          false,
+          false,
+          DEFAULT_VALUE_NOT_SET)
+    };
+
+    catalog
+        .asTableCatalog()
+        .createTable(
+            tableIdentifier,
+            columns,
+            "nanosecond timestamp round-trip",
+            createProperties(),
+            Distributions.NONE,
+            getSortOrders("id"));
+
+    Table loaded = catalog.asTableCatalog().loadTable(tableIdentifier);
+    Assertions.assertEquals(Types.TimestampType.withoutTimeZone(9), loaded.columns()[1].dataType());
+    Assertions.assertEquals(Types.TimestampType.withTimeZone(9), loaded.columns()[2].dataType());
+
+    String createSql =
+        clickhouseService.executeQueryForResult(
+            String.format("SHOW CREATE TABLE `%s`.`%s`", schemaName, tableIdentifier.name()));
+    Assertions.assertTrue(createSql.contains("DateTime64(9)"), createSql);
+    Assertions.assertTrue(createSql.contains("DateTime64(9, 'UTC')"), createSql);
+  }
+
+  @Test
+  void testRejectVariantWithoutSideEffects() {
+    assertUnsupportedV3TypeDoesNotCreateTable(
+        "test_variant",
+        Types.VariantType.get(),
+        "ClickHouse Variant requires a closed list of alternative types");
+  }
+
+  @Test
+  void testRejectNullTypeWithoutSideEffects() {
+    assertUnsupportedV3TypeDoesNotCreateTable(
+        "test_null",
+        Types.NullType.get(),
+        "the null-only placeholder has no ClickHouse column type");
+  }
+
+  @Test
+  void testRejectGeometryWithoutSideEffects() {
+    assertUnsupportedV3TypeDoesNotCreateTable(
+        "test_geometry",
+        Types.GeometryType.of("SRID:3857"),
+        "ClickHouse Geo types do not preserve Gravitino Geometry CRS metadata");
+  }
+
+  @Test
+  void testRejectGeographyWithoutSideEffects() {
+    assertUnsupportedV3TypeDoesNotCreateTable(
+        "test_geography",
+        Types.GeographyType.of("EPSG:4326", "karney"),
+        "ClickHouse Geo types do not preserve Gravitino Geography CRS and edge-algorithm metadata");
+  }
+
+  private void assertUnsupportedV3TypeDoesNotCreateTable(
+      String tablePrefix, Type type, String expectedMessage) {
+    NameIdentifier tableIdentifier =
+        NameIdentifier.of(schemaName, GravitinoITUtils.genRandomName(tablePrefix));
+    Column[] columns = {
+      Column.of("id", Types.IntegerType.get(), "sort column", false, false, DEFAULT_VALUE_NOT_SET),
+      Column.of("v3_column", type, "V3 column", false, false, DEFAULT_VALUE_NOT_SET)
+    };
+
+    IllegalArgumentException exception =
+        assertThrows(
+            IllegalArgumentException.class,
+            () ->
+                catalog
+                    .asTableCatalog()
+                    .createTable(
+                        tableIdentifier,
+                        columns,
+                        null,
+                        createProperties(),
+                        Distributions.NONE,
+                        getSortOrders("id")));
+
+    Assertions.assertTrue(exception.getMessage().contains(expectedMessage), exception::getMessage);
+    Assertions.assertFalse(catalog.asTableCatalog().tableExists(tableIdentifier));
+  }
 }

docs/jdbc-clickhouse-catalog.md

@@ -195,12 +195,31 @@ See [Manage Relational Metadata Using Gravitino](./manage-relational-metadata-us
 | `String`/`VarChar`  | `String`                               |
 | `FixedChar(n)`      | `FixedString(n)`                       |
 | `Date`              | `Date`                                 |
-| `Timestamp[(p)]`    | `DateTime` (precision defaults to `0`) |
+| `Timestamp[(p)]`    | `DateTime` for p=0; `DateTime64(p)` for p in [1, 9] |
+| `Timestamp_tz[(p)]` | `DateTime('UTC')` for p=0; `DateTime64(p, 'UTC')` for p in [1, 9] |
 | `BOOLEAN`           | `Bool`                                 |
 | `UUID`              | `UUID`                                 |
 
 Other ClickHouse types are exposed as [External Type](./manage-relational-metadata-using-gravitino.md#external-type).
 
+ClickHouse stores explicit time zones in column metadata. The catalog uses `UTC` for Gravitino
+`Timestamp_tz` so the type can be recovered exactly on load. Native ClickHouse timestamp columns
+with a different explicit time zone remain `ExternalType`, preserving the detailed zone metadata.
+Timestamp precision from 10 through 12 is rejected before executing DDL because ClickHouse
+`DateTime64` supports at most nanoseconds.
+
+Gravitino `Variant` is rejected before executing DDL. ClickHouse `Variant(T1, T2, ...)` requires a
+closed list of alternative types and cannot preserve Gravitino's open-ended Variant contract.
+
+Gravitino `Null` is rejected before executing DDL because ClickHouse has no column type for a
+null-only placeholder whose concrete type is not yet known.
+
+Gravitino `Geometry` is rejected before executing DDL. The Geo types in the supported ClickHouse
+24.8 line are shape-specific and do not preserve Gravitino's CRS metadata.
+
+Gravitino `Geography` is rejected before executing DDL because ClickHouse Geo types do not preserve
+its CRS and edge-algorithm metadata.
+
 ### Table Properties
 
 :::note