This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new c27520bd81 [#11803] fix(clickhouse): add ON CLUSTER support for ALTER
TABLE (#11807)
c27520bd81 is described below
commit c27520bd817e9090cb1e9865444bd656988f89fd
Author: StormSpirit <[email protected]>
AuthorDate: Tue Jun 30 19:16:35 2026 +0800
[#11803] fix(clickhouse): add ON CLUSTER support for ALTER TABLE (#11807)
### What changes were proposed in this pull request?
- `generateAlterTableSql()` now checks `ON_CLUSTER` and `CLUSTER_NAME`
table properties and injects `ON CLUSTER <clusterName>` into the ALTER
TABLE SQL when both are set.
- Adds `getOrCreateTable()` call for change types that do not otherwise
require loading the table, so cluster metadata is available.
- Adds null-safety for `properties()` access, consistent with
`generateDropTableSql()`.
- Omits `SYNC` (unlike `generateDropTableSql()`) because ClickHouse does
not support synchronous ALTER operations.
### Why are the changes needed?
Fix: #11803
Previously, `generateAlterTableSql()` always generated `ALTER TABLE ...`
without `ON CLUSTER`. For tables created with `ON CLUSTER`, subsequent
ALTER operations only affected the local node, causing schema divergence
across cluster nodes.
### Does this PR introduce _any_ user-facing change?
Yes. ALTER TABLE operations on clustered tables now propagate to all
nodes in the cluster, preventing schema divergence.
### How was this patch tested?
- `./gradlew :catalogs-contrib:catalog-jdbc-clickhouse:test --tests
"org.apache.gravitino.catalog.clickhouse.operations.TestClickHouseTableOperationsCluster"
-PskipITs`
- `./gradlew :catalogs-contrib:catalog-jdbc-clickhouse:test --tests
"org.apache.gravitino.catalog.clickhouse.integration.test.CatalogClickHouseClusterIT"
-PskipDockerTests=false`
---------
Signed-off-by: jiangxt2 <[email protected]>
Co-authored-by: Chang-Tong <[email protected]>
Co-authored-by: ArtificialIdoit <[email protected]>
Co-authored-by: cwq222 <[email protected]>
---
.../operations/ClickHouseTableOperations.java | 24 +++-
.../test/CatalogClickHouseClusterIT.java | 88 +++++++++++++++
.../TestClickHouseTableOperationsCluster.java | 123 +++++++++++++++++++++
3 files changed, 232 insertions(+), 3 deletions(-)
diff --git
a/catalogs-contrib/catalog-jdbc-clickhouse/src/main/java/org/apache/gravitino/catalog/clickhouse/operations/ClickHouseTableOperations.java
b/catalogs-contrib/catalog-jdbc-clickhouse/src/main/java/org/apache/gravitino/catalog/clickhouse/operations/ClickHouseTableOperations.java
index fc823ef78c..e3cd9378e1 100644
---
a/catalogs-contrib/catalog-jdbc-clickhouse/src/main/java/org/apache/gravitino/catalog/clickhouse/operations/ClickHouseTableOperations.java
+++
b/catalogs-contrib/catalog-jdbc-clickhouse/src/main/java/org/apache/gravitino/catalog/clickhouse/operations/ClickHouseTableOperations.java
@@ -817,10 +817,28 @@ public class ClickHouseTableOperations extends
JdbcTableOperations {
return "";
}
+ // Check if the table is on a cluster, so that ALTER TABLE includes ON
CLUSTER
+ lazyLoadTable = getOrCreateTable(databaseName, tableName, lazyLoadTable);
+ Map<String, String> props = lazyLoadTable.properties();
+ String clusterName = props == null ? null :
props.get(ClusterConstants.CLUSTER_NAME);
+ boolean onCluster =
+ props != null
+ &&
Boolean.parseBoolean(props.getOrDefault(ClusterConstants.ON_CLUSTER, "false"));
+
// Return the generated SQL statement
- String result =
- "ALTER TABLE %s \n%s;"
- .formatted(quoteIdentifier(tableName), String.join(",\n",
nonEmptySQLs));
+ String result;
+ if (onCluster && StringUtils.isNotBlank(clusterName)) {
+ result =
+ "ALTER TABLE %s ON CLUSTER %s \n%s;"
+ .formatted(
+ quoteIdentifier(tableName),
+ quoteIdentifier(clusterName),
+ String.join(",\n", nonEmptySQLs));
+ } else {
+ result =
+ "ALTER TABLE %s \n%s;"
+ .formatted(quoteIdentifier(tableName), String.join(",\n",
nonEmptySQLs));
+ }
LOG.info("Generated alter table:{} sql: {}", databaseName + "." +
tableName, result);
return result;
}
diff --git
a/catalogs-contrib/catalog-jdbc-clickhouse/src/test/java/org/apache/gravitino/catalog/clickhouse/integration/test/CatalogClickHouseClusterIT.java
b/catalogs-contrib/catalog-jdbc-clickhouse/src/test/java/org/apache/gravitino/catalog/clickhouse/integration/test/CatalogClickHouseClusterIT.java
index 7e7187277a..48d6ee8f89 100644
---
a/catalogs-contrib/catalog-jdbc-clickhouse/src/test/java/org/apache/gravitino/catalog/clickhouse/integration/test/CatalogClickHouseClusterIT.java
+++
b/catalogs-contrib/catalog-jdbc-clickhouse/src/test/java/org/apache/gravitino/catalog/clickhouse/integration/test/CatalogClickHouseClusterIT.java
@@ -745,4 +745,92 @@ public class CatalogClickHouseClusterIT extends BaseIT {
schemaName, name, ClickHouseContainer.DEFAULT_CLUSTER_NAME));
}
}
+
+ /**
+ * Verifies that ALTER TABLE on a cluster table includes ON CLUSTER in the
executed SQL, while
+ * ALTER TABLE on a non-cluster table does not. This validates the fix for
generating correct
+ * ALTER TABLE statements with ON CLUSTER support.
+ */
+ @Test
+ public void testAlterTableOnClusterSqlGeneration() throws Exception {
+ String clusterTbl = GravitinoITUtils.genRandomName("ck_alter_on_cluster");
+ String nonClusterTbl =
GravitinoITUtils.genRandomName("ck_alter_no_cluster");
+ NameIdentifier clusterIdent = NameIdentifier.of(schemaName, clusterTbl);
+ NameIdentifier nonClusterIdent = NameIdentifier.of(schemaName,
nonClusterTbl);
+ TableCatalog tableCatalog = catalog.asTableCatalog();
+
+ // Create a cluster table and a non-cluster table
+ tableCatalog.createTable(
+ clusterIdent,
+ createColumns(),
+ tableComment,
+ clusterMergeTreeProperties(),
+ Transforms.EMPTY_TRANSFORM,
+ Distributions.NONE,
+ getSortOrders("col_3"),
+ Indexes.EMPTY_INDEXES);
+
+ tableCatalog.createTable(
+ nonClusterIdent,
+ createColumns(),
+ tableComment,
+ Collections.singletonMap(GRAVITINO_ENGINE_KEY,
ENGINE.MERGETREE.getValue()),
+ Transforms.EMPTY_TRANSFORM,
+ Distributions.NONE,
+ getSortOrders("col_3"),
+ Indexes.EMPTY_INDEXES);
+
+ // ALTER both tables — add a column
+ tableCatalog.alterTable(
+ clusterIdent, TableChange.addColumn(new String[] {"new_col"},
Types.StringType.get()));
+ tableCatalog.alterTable(
+ nonClusterIdent, TableChange.addColumn(new String[] {"new_col"},
Types.StringType.get()));
+
+ // Flush query_log and verify the executed SQL
+ try (Connection connection =
+ DriverManager.getConnection(
+ clickHouseClusterContainer.getJdbcUrl(TEST_DB_NAME),
+ clickHouseClusterContainer.getUsername(),
+ clickHouseClusterContainer.getPassword());
+ Statement statement = connection.createStatement()) {
+
+ statement.execute("SYSTEM FLUSH LOGS");
+
+ // Verify cluster table ALTER includes ON CLUSTER
+ try (ResultSet rs =
+ statement.executeQuery(
+ String.format(
+ "SELECT query FROM system.query_log "
+ + "WHERE type = 'QueryFinish' "
+ + "AND query_kind = 'Alter' "
+ + "AND query LIKE '%%`%s`%%' "
+ + "AND query LIKE '%%ADD COLUMN%%' "
+ + "ORDER BY event_time DESC LIMIT 1",
+ clusterTbl))) {
+ Assertions.assertTrue(rs.next(), "Should find ALTER query for cluster
table");
+ String sql = rs.getString("query");
+ Assertions.assertTrue(
+ sql.contains("ON CLUSTER"),
+ "ALTER TABLE on cluster table must include ON CLUSTER, actual: " +
sql);
+ }
+
+ // Verify non-cluster table ALTER does NOT include ON CLUSTER
+ try (ResultSet rs =
+ statement.executeQuery(
+ String.format(
+ "SELECT query FROM system.query_log "
+ + "WHERE type = 'QueryFinish' "
+ + "AND query_kind = 'Alter' "
+ + "AND query LIKE '%%`%s`%%' "
+ + "AND query LIKE '%%ADD COLUMN%%' "
+ + "ORDER BY event_time DESC LIMIT 1",
+ nonClusterTbl))) {
+ Assertions.assertTrue(rs.next(), "Should find ALTER query for
non-cluster table");
+ String sql = rs.getString("query");
+ Assertions.assertFalse(
+ sql.contains("ON CLUSTER"),
+ "ALTER TABLE on non-cluster table must NOT include ON CLUSTER,
actual: " + sql);
+ }
+ }
+ }
}
diff --git
a/catalogs-contrib/catalog-jdbc-clickhouse/src/test/java/org/apache/gravitino/catalog/clickhouse/operations/TestClickHouseTableOperationsCluster.java
b/catalogs-contrib/catalog-jdbc-clickhouse/src/test/java/org/apache/gravitino/catalog/clickhouse/operations/TestClickHouseTableOperationsCluster.java
index f231b9d4d3..e20e476a8f 100644
---
a/catalogs-contrib/catalog-jdbc-clickhouse/src/test/java/org/apache/gravitino/catalog/clickhouse/operations/TestClickHouseTableOperationsCluster.java
+++
b/catalogs-contrib/catalog-jdbc-clickhouse/src/test/java/org/apache/gravitino/catalog/clickhouse/operations/TestClickHouseTableOperationsCluster.java
@@ -31,6 +31,8 @@ import
org.apache.gravitino.catalog.clickhouse.converter.ClickHouseColumnDefault
import
org.apache.gravitino.catalog.clickhouse.converter.ClickHouseExceptionConverter;
import
org.apache.gravitino.catalog.clickhouse.converter.ClickHouseTypeConverter;
import org.apache.gravitino.catalog.jdbc.JdbcColumn;
+import org.apache.gravitino.catalog.jdbc.JdbcTable;
+import org.apache.gravitino.rel.TableChange;
import org.apache.gravitino.rel.expressions.NamedReference;
import org.apache.gravitino.rel.expressions.distributions.Distribution;
import org.apache.gravitino.rel.expressions.distributions.Distributions;
@@ -360,6 +362,77 @@ class TestClickHouseTableOperationsCluster {
Assertions.assertEquals("", stripClusterMetadata(stored));
}
+ /** ALTER TABLE with ON CLUSTER=true should include ON CLUSTER in SQL. */
+ @Test
+ void testAlterTableWithOnCluster() {
+ StubClickHouseTableOperations ops = new StubClickHouseTableOperations();
+ ops.initialize(
+ null,
+ new ClickHouseExceptionConverter(),
+ new ClickHouseTypeConverter(),
+ new ClickHouseColumnDefaultValueConverter(),
+ new HashMap<>());
+ ops.setTable(buildStubTableWithCluster("ck_cluster", true));
+
+ String sql =
+ ops.buildAlterSql(
+ "default",
+ "orders",
+ new TableChange[] {
+ TableChange.addColumn(new String[] {"new_col"},
Types.IntegerType.get())
+ });
+
+ Assertions.assertTrue(sql.contains("ON CLUSTER"), "ALTER TABLE should
include ON CLUSTER");
+ Assertions.assertTrue(sql.contains("`ck_cluster`"), "ALTER TABLE should
include cluster name");
+ }
+
+ /** ALTER TABLE with ON CLUSTER=false should NOT include ON CLUSTER in SQL.
*/
+ @Test
+ void testAlterTableWithoutOnCluster() {
+ StubClickHouseTableOperations ops = new StubClickHouseTableOperations();
+ ops.initialize(
+ null,
+ new ClickHouseExceptionConverter(),
+ new ClickHouseTypeConverter(),
+ new ClickHouseColumnDefaultValueConverter(),
+ new HashMap<>());
+ ops.setTable(buildStubTableWithCluster("ck_cluster", false));
+
+ String sql =
+ ops.buildAlterSql(
+ "default",
+ "orders",
+ new TableChange[] {
+ TableChange.addColumn(new String[] {"new_col"},
Types.IntegerType.get())
+ });
+
+ Assertions.assertFalse(sql.contains("ON CLUSTER"), "ALTER TABLE should NOT
include ON CLUSTER");
+ }
+
+ /** ALTER TABLE with null properties should NOT throw NPE. */
+ @Test
+ void testAlterTableWithNullProperties() {
+ StubClickHouseTableOperations ops = new StubClickHouseTableOperations();
+ ops.initialize(
+ null,
+ new ClickHouseExceptionConverter(),
+ new ClickHouseTypeConverter(),
+ new ClickHouseColumnDefaultValueConverter(),
+ new HashMap<>());
+ ops.setTable(buildStubTableWithNullProperties());
+
+ String sql =
+ ops.buildAlterSql(
+ "default",
+ "orders",
+ new TableChange[] {
+ TableChange.addColumn(new String[] {"new_col"},
Types.IntegerType.get())
+ });
+
+ Assertions.assertFalse(sql.contains("ON CLUSTER"), "ALTER TABLE should NOT
include ON CLUSTER");
+ Assertions.assertTrue(sql.contains("ADD COLUMN"), "ALTER TABLE should
contain ADD COLUMN");
+ }
+
private static class TestableClickHouseTableOperations extends
ClickHouseTableOperations {
String buildCreateSql(
String tableName,
@@ -378,4 +451,54 @@ class TestClickHouseTableOperationsCluster {
return generateDropTableSql(tableName, properties);
}
}
+
+ private static final class StubClickHouseTableOperations extends
ClickHouseTableOperations {
+ private JdbcTable table;
+
+ void setTable(JdbcTable table) {
+ this.table = table;
+ }
+
+ @Override
+ protected JdbcTable getOrCreateTable(
+ String databaseName, String tableName, JdbcTable lazyLoadCreateTable) {
+ return table;
+ }
+
+ String buildAlterSql(String db, String tableName, TableChange[] changes) {
+ return generateAlterTableSql(db, tableName, changes);
+ }
+ }
+
+ private static JdbcTable buildStubTableWithCluster(String clusterName,
boolean onCluster) {
+ JdbcColumn c1 =
+ JdbcColumn.builder()
+ .withName("id")
+ .withType(Types.IntegerType.get())
+ .withNullable(false)
+ .build();
+ Map<String, String> props = new HashMap<>();
+ props.put(ClusterConstants.CLUSTER_NAME, clusterName);
+ props.put(ClusterConstants.ON_CLUSTER, String.valueOf(onCluster));
+ return JdbcTable.builder()
+ .withName("orders")
+ .withColumns(new JdbcColumn[] {c1})
+ .withProperties(props)
+ .withTableOperation(null)
+ .build();
+ }
+
+ private static JdbcTable buildStubTableWithNullProperties() {
+ JdbcColumn c1 =
+ JdbcColumn.builder()
+ .withName("id")
+ .withType(Types.IntegerType.get())
+ .withNullable(false)
+ .build();
+ return JdbcTable.builder()
+ .withName("orders")
+ .withColumns(new JdbcColumn[] {c1})
+ .withTableOperation(null)
+ .build();
+ }
}
