[#11910] fix(clickhouse): fix GraphiteMergeTree classification and add missing type mappings

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

@@ -47,6 +47,7 @@ private TableConstants() {}
     public static final String ENGINE = "engine";
     public static final String ENGINE_UPPER = "ENGINE";
     public static final String SETTINGS_PREFIX = "settings.";
+    public static final String GRAPHITE_CONFIG = "graphite.config";
   }
 
   public static final class IndexConstants {

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

@@ -67,6 +67,15 @@ public class ClickHouseTablePropertiesMetadata extends JdbcTablePropertiesMetada
           "",
           false);
 
+  // The following is for ClickHouse GraphiteMergeTree engine
+  public static final PropertyEntry<String> GRAPHITE_CONFIG_PROPERTY_ENTRY =
+      stringOptionalPropertyEntry(
+          TableConstants.GRAPHITE_CONFIG,
+          "The graphite_rollup config element name for GraphiteMergeTree engine",
+          false,
+          "",
+          false);
+
   // The following three are for ClickHouse Distributed engine
   public static final PropertyEntry<String> CLUSTER_REMOTE_DATABASE_PROPERTY_ENTRY =
       stringOptionalPropertyEntry(
@@ -107,6 +116,8 @@ private static Map<String, PropertyEntry<?>> createPropertiesMetadata() {
     // For all tables with different engines
     map.put(COMMENT_PROPERTY_ENTRY.getName(), COMMENT_PROPERTY_ENTRY);
     map.put(ENGINE_PROPERTY_ENTRY.getName(), ENGINE_PROPERTY_ENTRY);
+    // For ClickHouse GraphiteMergeTree engine
+    map.put(GRAPHITE_CONFIG_PROPERTY_ENTRY.getName(), GRAPHITE_CONFIG_PROPERTY_ENTRY);
     // For ClickHouse Distributed engine
     map.put(ON_CLUSTER_PROPERTY_ENTRY.getName(), ON_CLUSTER_PROPERTY_ENTRY);
     map.put(CLUSTER_NAME_PROPERTY_ENTRY.getName(), CLUSTER_NAME_PROPERTY_ENTRY);
@@ -127,7 +138,7 @@ public enum ENGINE {
     AGGREGATINGMERGETREE("AggregatingMergeTree", true, true),
     COLLAPSINGMERGETREE("CollapsingMergeTree", true, true),
     VERSIONEDCOLLAPSINGMERGETREE("VersionedCollapsingMergeTree", true, true),
-    GRAPHITEMERGETREE("GraphiteMergeTree"),
+    GRAPHITEMERGETREE("GraphiteMergeTree", true, true),
 
     // Log
     TINYLOG("TinyLog"),

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

@@ -94,6 +94,16 @@ public Type toGravitino(JdbcTypeBean typeBean) {
       typeName = "FixedString";
     }
 
+    // ClickHouse normalizes Enum to Enum8/Enum16, so "Enum8(...)" never matches "Enum".
+    // Return directly as ExternalType to preserve the full parameterized form.
+    // Use stripped typeName (not typeBean.getTypeName()) to avoid preserving
+    // Nullable/LowCardinality
+    // wrappers that were already stripped above. Consistent with wide integer ExternalType
+    // handling.
+    if (typeName.startsWith(ENUM)) {
+      return Types.ExternalType.of(typeName);
+    }
+
     switch (typeName) {
       case INT8:
         return Types.ByteType.get();
@@ -111,10 +121,20 @@ public Type toGravitino(JdbcTypeBean typeBean) {
         return Types.IntegerType.unsigned();
       case UINT64:
         return Types.LongType.unsigned();
+      case INT128:
+      case INT256:
+      case UINT128:
+      case UINT256:
+        // ClickHouse native wide integer types with no corresponding Gravitino built-in type.
+        return Types.ExternalType.of(typeName);
       case FLOAT32:
         return Types.FloatType.get();
       case FLOAT64:
         return Types.DoubleType.get();
+      case BFLOAT16:
+        // Lossy: Gravitino has no half-precision float type. Use ExternalType to preserve
+        // round-trip (BFloat16 → ExternalType("BFloat16") → BFloat16).
+        return Types.ExternalType.of(BFLOAT16);
       case DECIMAL:
         int precision = typeBean.getColumnSize();
         int scale = typeBean.getScale();
@@ -141,7 +161,9 @@ public Type toGravitino(JdbcTypeBean typeBean) {
         return Types.FixedCharType.of(typeBean.getColumnSize());
       case DATE:
         return Types.DateType.get();
-        // No type mapping for date32, we will use external type to handle it.
+      case DATE32:
+        // Date32 supports 1900-2299 vs Date's 1970-2149. Use ExternalType to preserve round-trip.
+        return Types.ExternalType.of(DATE32);
       case DATETIME:
         // Default is 0 precision
         return Types.TimestampType.withoutTimeZone(0);

catalogs-contrib/catalog-jdbc-clickhouse/src/main/java/org/apache/gravitino/catalog/clickhouse/operations/ClickHouseTableOperations.java

@@ -325,6 +325,18 @@ private ClickHouseTablePropertiesMetadata.ENGINE appendTableEngine(
       return engine;
     }
 
+    if (engine == ENGINE.GRAPHITEMERGETREE) {
+      String config = properties.get(TableConstants.GRAPHITE_CONFIG);
+      Preconditions.checkArgument(
+          StringUtils.isNotBlank(config),
+          "GraphiteMergeTree requires '%s' property referencing a <graphite_rollup> config element",
+          TableConstants.GRAPHITE_CONFIG);
+      // Escape single quotes to prevent SQL injection
+      String escapedConfig = config.replace("'", "''");
+      sqlBuilder.append("\n ENGINE = GraphiteMergeTree('%s')".formatted(escapedConfig));
+      return engine;
+    }
+
     sqlBuilder.append("\n ENGINE = %s".formatted(engine.getValue()));
     return engine;
   }

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

@@ -18,6 +18,7 @@
  */
 package org.apache.gravitino.catalog.clickhouse.converter;
 
+import static org.apache.gravitino.catalog.clickhouse.converter.ClickHouseTypeConverter.BFLOAT16;
 import static org.apache.gravitino.catalog.clickhouse.converter.ClickHouseTypeConverter.BOOL;
 import static org.apache.gravitino.catalog.clickhouse.converter.ClickHouseTypeConverter.DATE;
 import static org.apache.gravitino.catalog.clickhouse.converter.ClickHouseTypeConverter.DATE32;
@@ -27,14 +28,18 @@
 import static org.apache.gravitino.catalog.clickhouse.converter.ClickHouseTypeConverter.FIXEDSTRING;
 import static org.apache.gravitino.catalog.clickhouse.converter.ClickHouseTypeConverter.FLOAT32;
 import static org.apache.gravitino.catalog.clickhouse.converter.ClickHouseTypeConverter.FLOAT64;
+import static org.apache.gravitino.catalog.clickhouse.converter.ClickHouseTypeConverter.INT128;
 import static org.apache.gravitino.catalog.clickhouse.converter.ClickHouseTypeConverter.INT16;
+import static org.apache.gravitino.catalog.clickhouse.converter.ClickHouseTypeConverter.INT256;
 import static org.apache.gravitino.catalog.clickhouse.converter.ClickHouseTypeConverter.INT32;
 import static org.apache.gravitino.catalog.clickhouse.converter.ClickHouseTypeConverter.INT64;
 import static org.apache.gravitino.catalog.clickhouse.converter.ClickHouseTypeConverter.INT8;
 import static org.apache.gravitino.catalog.clickhouse.converter.ClickHouseTypeConverter.IPV4;
 import static org.apache.gravitino.catalog.clickhouse.converter.ClickHouseTypeConverter.IPV6;
 import static org.apache.gravitino.catalog.clickhouse.converter.ClickHouseTypeConverter.STRING;
+import static org.apache.gravitino.catalog.clickhouse.converter.ClickHouseTypeConverter.UINT128;
 import static org.apache.gravitino.catalog.clickhouse.converter.ClickHouseTypeConverter.UINT16;
+import static org.apache.gravitino.catalog.clickhouse.converter.ClickHouseTypeConverter.UINT256;
 import static org.apache.gravitino.catalog.clickhouse.converter.ClickHouseTypeConverter.UINT32;
 import static org.apache.gravitino.catalog.clickhouse.converter.ClickHouseTypeConverter.UINT64;
 import static org.apache.gravitino.catalog.clickhouse.converter.ClickHouseTypeConverter.UINT8;
@@ -67,13 +72,18 @@ public void testToGravitinoType() {
     checkJdbcTypeToGravitinoType(Types.FloatType.get(), FLOAT32, null, null);
     checkJdbcTypeToGravitinoType(Types.DoubleType.get(), FLOAT64, null, null);
     checkJdbcTypeToGravitinoType(Types.DateType.get(), DATE, null, null);
-    checkJdbcTypeToGravitinoType(Types.ExternalType.of("Date32"), DATE32, null, null);
+    checkJdbcTypeToGravitinoType(Types.ExternalType.of(DATE32), DATE32, null, null);
     checkJdbcTypeToGravitinoType(Types.TimestampType.withoutTimeZone(0), DATETIME, null, null);
     checkJdbcTypeToGravitinoType(Types.DecimalType.of(10, 2), DECIMAL, 10, 2);
     checkJdbcTypeToGravitinoType(Types.StringType.get(), STRING, 20, null);
     checkJdbcTypeToGravitinoType(Types.FixedCharType.of(20), FIXEDSTRING, 20, null);
     checkJdbcTypeToGravitinoType(Types.BooleanType.get(), BOOL, 20, null);
     checkJdbcTypeToGravitinoType(Types.UUIDType.get(), UUID, 20, null);
+    checkJdbcTypeToGravitinoType(Types.ExternalType.of(BFLOAT16), BFLOAT16, null, null);
+    checkJdbcTypeToGravitinoType(Types.ExternalType.of(INT128), INT128, null, null);
+    checkJdbcTypeToGravitinoType(Types.ExternalType.of(INT256), INT256, null, null);
+    checkJdbcTypeToGravitinoType(Types.ExternalType.of(UINT128), UINT128, null, null);
+    checkJdbcTypeToGravitinoType(Types.ExternalType.of(UINT256), UINT256, null, null);
     checkJdbcTypeToGravitinoType(
         Types.ExternalType.of(USER_DEFINED_TYPE), USER_DEFINED_TYPE, null, null);
 
@@ -93,6 +103,18 @@ public void testToGravitinoType() {
     Assertions.assertEquals(
         Types.ExternalType.of(DATE32), CLICKHOUSE_TYPE_CONVERTER.toGravitino(date32));
 
+    // Enum8/Enum16: ClickHouse normalizes Enum to Enum8, so "Enum8('active'=1)" must match
+    JdbcTypeConverter.JdbcTypeBean enum8 =
+        createTypeBean("Enum8('active'=1,'inactive'=2)", null, null);
+    Assertions.assertEquals(
+        Types.ExternalType.of("Enum8('active'=1,'inactive'=2)"),
+        CLICKHOUSE_TYPE_CONVERTER.toGravitino(enum8));
+
+    JdbcTypeConverter.JdbcTypeBean enum16 = createTypeBean("Enum16('x'=1,'y'=2)", null, null);
+    Assertions.assertEquals(
+        Types.ExternalType.of("Enum16('x'=1,'y'=2)"),
+        CLICKHOUSE_TYPE_CONVERTER.toGravitino(enum16));
+
     JdbcTypeConverter.JdbcTypeBean ipv4 = createTypeBean("IPv4", null, null);
     Assertions.assertEquals(
         Types.ExternalType.of("IPv4"), CLICKHOUSE_TYPE_CONVERTER.toGravitino(ipv4));
@@ -157,6 +179,18 @@ public void testFromGravitinoType() {
     // IPv4/IPv6 round-trip
     checkGravitinoTypeToJdbcType(IPV4, Types.ExternalType.of(IPV4));
     checkGravitinoTypeToJdbcType(IPV6, Types.ExternalType.of(IPV6));
+    // Wide integer round-trip (ExternalType passthrough)
+    checkGravitinoTypeToJdbcType(INT128, Types.ExternalType.of(INT128));
+    checkGravitinoTypeToJdbcType(INT256, Types.ExternalType.of(INT256));
+    checkGravitinoTypeToJdbcType(UINT128, Types.ExternalType.of(UINT128));
+    checkGravitinoTypeToJdbcType(UINT256, Types.ExternalType.of(UINT256));
+    // BFloat16 round-trip (ExternalType passthrough)
+    checkGravitinoTypeToJdbcType(BFLOAT16, Types.ExternalType.of(BFLOAT16));
+    // Enum8 round-trip (ExternalType passthrough)
+    String enum8Type = "Enum8('active'=1,'inactive'=2)";
+    checkGravitinoTypeToJdbcType(enum8Type, Types.ExternalType.of(enum8Type));
+    // DATE32 round-trip (ExternalType passthrough)
+    checkGravitinoTypeToJdbcType(DATE32, Types.ExternalType.of(DATE32));
     checkGravitinoTypeToJdbcType(TIME, Types.TimeType.get());
     Assertions.assertThrows(
         IllegalArgumentException.class,

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

@@ -18,6 +18,7 @@
  */
 package org.apache.gravitino.catalog.clickhouse.integration.test;
 
+import static org.apache.gravitino.catalog.clickhouse.ClickHouseTablePropertiesMetadata.ENGINE;
 import static org.apache.gravitino.catalog.clickhouse.ClickHouseTablePropertiesMetadata.ENGINE.MERGETREE;
 import static org.apache.gravitino.catalog.clickhouse.ClickHouseTablePropertiesMetadata.GRAVITINO_ENGINE_KEY;
 import static org.apache.gravitino.catalog.clickhouse.ClickHouseUtils.getSortOrders;
@@ -2261,4 +2262,69 @@ void testAlterCatalogProperties() throws SQLException {
     loadCatalog.asSchemas().dropSchema("test", true);
     metalake.dropCatalog(testCatalogName, true);
   }
+
+  /**
+   * Verifies round-trip for new ClickHouse types (Int128/256, UInt128/256, Enum8/16, Date32) via
+   * SQL CREATE TABLE + Gravitino loadTable. BFloat16 is excluded because it requires ClickHouse
+   * 24.12+ (the Gravitino CI image is 24.8.14). The BFloat16 converter mapping is verified in
+   * {@link org.apache.gravitino.catalog.clickhouse.converter.TestClickHouseTypeConverter}.
+   */
+  @Test
+  void testCreateTableWithNewTypeMappings() {
+    String tableName = GravitinoITUtils.genRandomName("test_new_types");
+    clickhouseService.executeQuery(
+        String.format(
+            "CREATE TABLE %s.%s ("
+                + "c_int128 Int128, c_int256 Int256, "
+                + "c_uint128 UInt128, c_uint256 UInt256, "
+                + "c_enum8 Enum8('a'=1,'b'=2), "
+                + "c_enum16 Enum16('x'=1,'y'=2), "
+                + "c_date32 Date32"
+                + ") ORDER BY c_int128",
+            schemaName, tableName));
+    Table loadedTable =
+        catalog.asTableCatalog().loadTable(NameIdentifier.of(schemaName, tableName));
+    Assertions.assertEquals(Types.ExternalType.of("Int128"), loadedTable.columns()[0].dataType());
+    Assertions.assertEquals(Types.ExternalType.of("Int256"), loadedTable.columns()[1].dataType());
+    Assertions.assertEquals(Types.ExternalType.of("UInt128"), loadedTable.columns()[2].dataType());
+    Assertions.assertEquals(Types.ExternalType.of("UInt256"), loadedTable.columns()[3].dataType());
+    // Enum types use instanceof check (not exact value match) because the ClickHouse JDBC driver
+    // may normalize the enum definition format (e.g., spacing around '=' and ',').
+    Assertions.assertTrue(loadedTable.columns()[4].dataType() instanceof Types.ExternalType);
+    Assertions.assertTrue(loadedTable.columns()[5].dataType() instanceof Types.ExternalType);
+    Assertions.assertEquals(Types.ExternalType.of("Date32"), loadedTable.columns()[6].dataType());
+  }
+
+  /**
+   * Tests that GraphiteMergeTree engine creation fails when graphite.config property is missing.
+   * Note: Positive path test (successful creation with valid config) is not included because it
+   * requires a pre-configured graphite_rollup element on the ClickHouse server side, which is not
+   * available in the standard test container.
+   */
+  @Test
+  void testGraphiteMergeTreeEngineCreation() {
+    String tableName = GravitinoITUtils.genRandomName("test_graphite");
+    Column[] columns =
+        new Column[] {
+          Column.of("id", Types.IntegerType.get(), "pk", false, false, DEFAULT_VALUE_NOT_SET),
+          Column.of("val", Types.StringType.get(), "data", true, false, DEFAULT_VALUE_NOT_SET)
+        };
+    Map<String, String> properties = new HashMap<>();
+    properties.put(GRAVITINO_ENGINE_KEY, ENGINE.GRAPHITEMERGETREE.getValue());
+    // GraphiteMergeTree requires graphite.config property
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () ->
+            catalog
+                .asTableCatalog()
+                .createTable(
+                    NameIdentifier.of(schemaName, tableName),
+                    columns,
+                    table_comment,
+                    properties,
+                    Transforms.EMPTY_TRANSFORM,
+                    Distributions.NONE,
+                    getSortOrders("id"),
+                    Indexes.EMPTY_INDEXES));
+  }
 }

integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/ClickHouseContainer.java

@@ -134,13 +134,19 @@ public String getPassword() {
     return PASSWORD;
   }
 
+  /**
+   * Returns the JDBC URL for connecting to ClickHouse from the host machine. Uses localhost +
+   * mapped port for macOS Docker Desktop compatibility (container internal IP is not reachable from
+   * host). Note: This is not suitable for container-to-container communication scenarios.
+   */
   public String getJdbcUrl() {
-    return format("jdbc:clickhouse://%s:%d", getContainerIpAddress(), CLICKHOUSE_PORT);
+    return format("jdbc:clickhouse://localhost:%d", container.getMappedPort(CLICKHOUSE_PORT));
   }
 
   public String getJdbcUrl(TestDatabaseName testDatabaseName) {
     return format(
-        "jdbc:clickhouse://%s:%d/%s", getContainerIpAddress(), CLICKHOUSE_PORT, testDatabaseName);
+        "jdbc:clickhouse://localhost:%d/%s",
+        container.getMappedPort(CLICKHOUSE_PORT), testDatabaseName);
   }
 
   public String getDriverClassName(TestDatabaseName testDatabaseName) throws SQLException {