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 0080137efe [#12761] fix(clickhouse): Propagate clustered table rename
(#12763)
0080137efe is described below
commit 0080137efe82a539d1d12db43496c3c42453d5c1
Author: StormSpirit <[email protected]>
AuthorDate: Tue Sep 1 19:08:18 2026 +0800
[#12761] fix(clickhouse): Propagate clustered table rename (#12763)
### What changes were proposed in this pull request?
- Overrides `ClickHouseTableOperations.rename()` so tables with trusted
Gravitino cluster metadata use `RENAME TABLE ... ON CLUSTER ...`, while
local and unmarked external tables keep the existing local rename
behavior.
- Reads the old table metadata from `system.tables` with an exact
current-database and table-name predicate, and rejects a present but
blank cluster marker before executing DDL.
- Preserves the existing ClickHouse exception mapping and identifier
quoting contract without changing the common JDBC rename path.
- Expands the repository ClickHouse cluster fixture from one
self-referencing node to three independently addressable nodes.
- Adds focused unit coverage, local SQL regression coverage, and a
three-node lifecycle test that verifies the initiating query, all-node
old/new state, comment metadata preservation, and cleanup after drop.
### Why are the changes needed?
Fix: #12761
The common JDBC rename path generates a local `RENAME TABLE old_name TO
new_name` statement. For ClickHouse tables created through Gravitino
with `ON CLUSTER`, that statement renames only the JDBC connection node
and leaves the old name on the other nodes, silently splitting cluster
metadata and making later DDL inconsistent.
### Does this PR introduce _any_ user-facing change?
Yes. Renaming a Gravitino-created ClickHouse table with trusted cluster
metadata now propagates to every configured cluster node. Local tables
and unmarked external tables retain local rename behavior. This PR does
not add or change catalog-facing APIs or property keys.
### How was this patch tested?
- `./gradlew :catalogs-contrib:catalog-jdbc-clickhouse:spotlessCheck` —
passed.
- `./gradlew rat` — passed.
- `./gradlew :catalogs-contrib:catalog-jdbc-clickhouse:test -PskipITs` —
passed with no skipped, failed, or errored tests.
- `./gradlew :catalogs-contrib:catalog-jdbc-clickhouse:test --tests
"org.apache.gravitino.catalog.clickhouse.integration.test.CatalogClickHouseIT.testAlterAndDropClickhouseTable"
-PskipDockerTests=false` — passed with `tests=1 skipped=0 failures=0
errors=0`.
- `./gradlew :catalogs-contrib:catalog-jdbc-clickhouse:test --tests
"org.apache.gravitino.catalog.clickhouse.integration.test.CatalogClickHouseClusterIT"
-PskipDockerTests=false` — passed on three ClickHouse 24.8.14 nodes with
`tests=19 skipped=0 failures=0 errors=0`.
- `./gradlew :catalogs-contrib:catalog-jdbc-clickhouse:build -x test` —
passed.
- `python3 ~/GitHub/bin/gravitino-pr-precheck.py --worktree
~/GitHub/workspace/gravitino-fix-clickhouse-clustered-table-rename` —
all checks passed.
Signed-off-by: jiangxt2 <[email protected]>
---
.../operations/ClickHouseClusterUtils.java | 14 ++
.../operations/ClickHouseTableOperations.java | 176 +++++++++++++++------
.../test/CatalogClickHouseClusterIT.java | 144 ++++++++++++++++-
.../integration/test/CatalogClickHouseIT.java | 17 ++
.../TestClickHouseTableOperationsCluster.java | 37 +++++
.../TestClickHouseTableOperationsUnit.java | 119 +++++++++++++-
.../src/test/resources/remote_servers.xml | 10 +-
.../integration/test/container/ContainerSuite.java | 57 ++++---
8 files changed, 499 insertions(+), 75 deletions(-)
diff --git
a/catalogs-contrib/catalog-jdbc-clickhouse/src/main/java/org/apache/gravitino/catalog/clickhouse/operations/ClickHouseClusterUtils.java
b/catalogs-contrib/catalog-jdbc-clickhouse/src/main/java/org/apache/gravitino/catalog/clickhouse/operations/ClickHouseClusterUtils.java
index b225102650..010c6dc4a5 100644
---
a/catalogs-contrib/catalog-jdbc-clickhouse/src/main/java/org/apache/gravitino/catalog/clickhouse/operations/ClickHouseClusterUtils.java
+++
b/catalogs-contrib/catalog-jdbc-clickhouse/src/main/java/org/apache/gravitino/catalog/clickhouse/operations/ClickHouseClusterUtils.java
@@ -141,6 +141,20 @@ public final class ClickHouseClusterUtils {
return storedComment.substring(idx + CLUSTER_META_PREFIX.length());
}
+ /**
+ * Returns whether {@code storedComment} contains the Gravitino cluster
metadata marker.
+ *
+ * <p>This deliberately distinguishes an absent marker from a present marker
with a blank value.
+ * Callers performing cluster-wide DDL can therefore keep unmarked external
objects local while
+ * rejecting corrupted Gravitino metadata before mutation.
+ *
+ * @param storedComment The raw comment as stored in ClickHouse.
+ * @return {@code true} if the Gravitino cluster metadata marker is present.
+ */
+ public static boolean hasClusterMetadata(String storedComment) {
+ return storedComment != null &&
storedComment.contains(CLUSTER_META_PREFIX);
+ }
+
/**
* Returns the user-visible portion of the stored comment, stripping any
embedded cluster metadata
* suffix. Returns {@code null} if {@code storedComment} is {@code null}.
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 77115b9113..3bb6a6d561 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
@@ -681,68 +681,116 @@ public class ClickHouseTableOperations extends
JdbcTableOperations {
}
}
+ @Override
+ public void rename(String databaseName, String oldTableName, String
newTableName)
+ throws NoSuchTableException {
+ LOG.info(
+ "Attempting to rename table {}/{} to {}/{}",
+ databaseName,
+ oldTableName,
+ databaseName,
+ newTableName);
+ try (Connection connection = getConnection(databaseName)) {
+ TablePropertiesWithClusterMetadata metadata =
+ loadTablePropertiesWithClusterMetadata(connection, oldTableName);
+ JdbcConnectorUtils.executeUpdate(
+ connection,
+ generateRenameTableSql(
+ oldTableName, newTableName, metadata.hasClusterMetadata(),
metadata.clusterName()));
+ LOG.info(
+ "Renamed table {}/{} to {}/{}", databaseName, oldTableName,
databaseName, newTableName);
+ } catch (final SQLException se) {
+ throw exceptionMapper.toGravitinoException(se);
+ }
+ }
+
+ @VisibleForTesting
+ String generateRenameTableSql(
+ String oldTableName,
+ String newTableName,
+ boolean hasClusterMetadata,
+ @Nullable String clusterName) {
+ if (hasClusterMetadata) {
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(clusterName),
+ "ClickHouse cluster metadata for table %s is missing a cluster name",
+ oldTableName);
+ return "RENAME TABLE %s TO %s ON CLUSTER %s"
+ .formatted(
+ quoteIdentifier(oldTableName),
+ quoteIdentifier(newTableName),
+ quoteIdentifier(clusterName));
+ }
+ return "RENAME TABLE %s TO %s"
+ .formatted(quoteIdentifier(oldTableName),
quoteIdentifier(newTableName));
+ }
+
@Override
protected Map<String, String> getTableProperties(Connection connection,
String tableName)
throws SQLException {
+ return loadTablePropertiesWithClusterMetadata(connection,
tableName).properties();
+ }
+
+ private TablePropertiesWithClusterMetadata
loadTablePropertiesWithClusterMetadata(
+ Connection connection, String tableName) throws SQLException {
try (PreparedStatement statement =
- connection.prepareStatement("select * from system.tables where name =
? ")) {
+ connection.prepareStatement(
+ "SELECT comment, engine, engine_full FROM system.tables "
+ + "WHERE database = currentDatabase() AND name = ?")) {
statement.setString(1, tableName);
try (ResultSet resultSet = statement.executeQuery()) {
- while (resultSet.next()) {
- String name = resultSet.getString("name");
- if (Objects.equals(name, tableName)) {
- Map<String, String> tableProperties = new HashMap<>();
+ if (!resultSet.next()) {
+ throw new NoSuchTableException(
+ "Table %s does not exist in %s.", tableName,
connection.getCatalog());
+ }
- // Extract cluster name embedded in the COMMENT at create time.
- // SHOW CREATE TABLE does not include ON CLUSTER (see
ClickHouseClusterUtils).
- String storedComment = resultSet.getString(COMMENT);
- String clusterName =
ClickHouseClusterUtils.extractClusterFromComment(storedComment);
+ Map<String, String> tableProperties = new HashMap<>();
+
+ // Extract cluster name embedded in the COMMENT at create time.
+ // SHOW CREATE TABLE does not include ON CLUSTER (see
ClickHouseClusterUtils).
+ String storedComment = resultSet.getString(COMMENT);
+ boolean hasClusterMetadata =
ClickHouseClusterUtils.hasClusterMetadata(storedComment);
+ String clusterName =
ClickHouseClusterUtils.extractClusterFromComment(storedComment);
+ tableProperties.put(COMMENT,
ClickHouseClusterUtils.stripClusterMetadata(storedComment));
+ String engine = resultSet.getString(CLICKHOUSE_ENGINE_KEY);
+ String engineFull = resultSet.getString("engine_full");
+ tableProperties.put(GRAVITINO_ENGINE_KEY, engine);
+ if (StringUtils.isNotBlank(clusterName)) {
+ tableProperties.put(ClusterConstants.ON_CLUSTER,
String.valueOf(true));
+ tableProperties.put(ClusterConstants.CLUSTER_NAME, clusterName);
+ } else {
+ tableProperties.put(ClusterConstants.ON_CLUSTER,
String.valueOf(false));
+ }
+
+ if (StringUtils.equalsIgnoreCase(engine,
ENGINE.DISTRIBUTED.getValue())) {
+ Matcher distributedEngineMatcher =
+
DISTRIBUTED_ENGINE_PATTERN.matcher(StringUtils.trimToEmpty(engineFull));
+ if (distributedEngineMatcher.matches()) {
+ String distributedClusterName =
unquote(distributedEngineMatcher.group(1));
+ tableProperties.put(ClusterConstants.CLUSTER_NAME,
distributedClusterName);
+ tableProperties.put(
+ DistributedTableConstants.REMOTE_DATABASE,
+ unquote(distributedEngineMatcher.group(2)));
tableProperties.put(
- COMMENT,
ClickHouseClusterUtils.stripClusterMetadata(storedComment));
- String engine = resultSet.getString(CLICKHOUSE_ENGINE_KEY);
- String engineFull = resultSet.getString("engine_full");
- tableProperties.put(GRAVITINO_ENGINE_KEY, engine);
- if (StringUtils.isNotBlank(clusterName)) {
- tableProperties.put(ClusterConstants.ON_CLUSTER,
String.valueOf(true));
- tableProperties.put(ClusterConstants.CLUSTER_NAME, clusterName);
- } else {
- tableProperties.put(ClusterConstants.ON_CLUSTER,
String.valueOf(false));
- }
-
- if (StringUtils.equalsIgnoreCase(engine,
ENGINE.DISTRIBUTED.getValue())) {
- Matcher distributedEngineMatcher =
-
DISTRIBUTED_ENGINE_PATTERN.matcher(StringUtils.trimToEmpty(engineFull));
- if (distributedEngineMatcher.matches()) {
- String distributedClusterName =
unquote(distributedEngineMatcher.group(1));
- tableProperties.put(ClusterConstants.CLUSTER_NAME,
distributedClusterName);
- tableProperties.put(
- DistributedTableConstants.REMOTE_DATABASE,
- unquote(distributedEngineMatcher.group(2)));
- tableProperties.put(
- DistributedTableConstants.REMOTE_TABLE,
- unquote(distributedEngineMatcher.group(3)));
- tableProperties.put(
- DistributedTableConstants.SHARDING_KEY,
- StringUtils.trim(distributedEngineMatcher.group(4)));
- }
- } else if (StringUtils.equalsIgnoreCase(engine,
ENGINE.GRAPHITEMERGETREE.getValue())) {
- String graphiteConfig = extractGraphiteConfig(engineFull);
- if (StringUtils.isNotBlank(graphiteConfig)) {
- tableProperties.put(TableConstants.GRAPHITE_CONFIG,
graphiteConfig);
- }
- } else if (isGenericEngineParameterEngine(engine)) {
- String engineParams = extractEngineParams(engine, engineFull);
- if (StringUtils.isNotBlank(engineParams)) {
- tableProperties.put(TableConstants.ENGINE_PARAMETERS,
engineParams);
- }
- }
-
- return Collections.unmodifiableMap(tableProperties);
+ DistributedTableConstants.REMOTE_TABLE,
unquote(distributedEngineMatcher.group(3)));
+ tableProperties.put(
+ DistributedTableConstants.SHARDING_KEY,
+ StringUtils.trim(distributedEngineMatcher.group(4)));
+ }
+ } else if (StringUtils.equalsIgnoreCase(engine,
ENGINE.GRAPHITEMERGETREE.getValue())) {
+ String graphiteConfig = extractGraphiteConfig(engineFull);
+ if (StringUtils.isNotBlank(graphiteConfig)) {
+ tableProperties.put(TableConstants.GRAPHITE_CONFIG,
graphiteConfig);
+ }
+ } else if (isGenericEngineParameterEngine(engine)) {
+ String engineParams = extractEngineParams(engine, engineFull);
+ if (StringUtils.isNotBlank(engineParams)) {
+ tableProperties.put(TableConstants.ENGINE_PARAMETERS,
engineParams);
}
}
- throw new NoSuchTableException(
- "Table %s does not exist in %s.", tableName,
connection.getCatalog());
+ return new TablePropertiesWithClusterMetadata(
+ Collections.unmodifiableMap(tableProperties), hasClusterMetadata,
clusterName);
}
}
}
@@ -1582,6 +1630,32 @@ public class ClickHouseTableOperations extends
JdbcTableOperations {
private Transform[] partitioning = Transforms.EMPTY_TRANSFORM;
}
+ private static final class TablePropertiesWithClusterMetadata {
+ private final Map<String, String> properties;
+ private final boolean hasClusterMetadata;
+ @Nullable private final String clusterName;
+
+ private TablePropertiesWithClusterMetadata(
+ Map<String, String> properties, boolean hasClusterMetadata, @Nullable
String clusterName) {
+ this.properties = properties;
+ this.hasClusterMetadata = hasClusterMetadata;
+ this.clusterName = clusterName;
+ }
+
+ private Map<String, String> properties() {
+ return properties;
+ }
+
+ private boolean hasClusterMetadata() {
+ return hasClusterMetadata;
+ }
+
+ @Nullable
+ private String clusterName() {
+ return clusterName;
+ }
+ }
+
@VisibleForTesting
String[][] parseIndexFields(String expression) {
return ClickHouseTableSqlUtils.parseIndexFields(expression);
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 0d2ba7bb4c..910268e87b 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
@@ -31,11 +31,15 @@ import com.google.common.collect.Maps;
import java.nio.file.Paths;
import java.sql.Connection;
import java.sql.DriverManager;
+import java.sql.PreparedStatement;
import java.sql.ResultSet;
+import java.sql.SQLException;
import java.sql.Statement;
+import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.Objects;
import org.apache.commons.lang3.StringUtils;
@@ -43,7 +47,9 @@ import org.apache.gravitino.Catalog;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.Schema;
+import org.apache.gravitino.StringIdentifier;
import
org.apache.gravitino.catalog.clickhouse.integration.test.service.ClickHouseService;
+import
org.apache.gravitino.catalog.clickhouse.operations.ClickHouseClusterUtils;
import org.apache.gravitino.catalog.jdbc.config.JdbcConfig;
import org.apache.gravitino.client.GravitinoMetalake;
import org.apache.gravitino.integration.test.container.ClickHouseContainer;
@@ -64,6 +70,7 @@ import
org.apache.gravitino.rel.expressions.transforms.Transforms;
import org.apache.gravitino.rel.indexes.Index;
import org.apache.gravitino.rel.indexes.Indexes;
import org.apache.gravitino.rel.types.Types;
+import org.awaitility.Awaitility;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
@@ -98,6 +105,7 @@ public class CatalogClickHouseClusterIT extends BaseIT {
private Catalog catalog;
private ClickHouseService clickHouseService;
private ClickHouseContainer clickHouseClusterContainer;
+ private List<ClickHouseContainer> clickHouseClusterContainers;
private final TestDatabaseName TEST_DB_NAME =
TestDatabaseName.CLICKHOUSE_CLUSTER_CLICKHOUSE_IT;
@BeforeAll
@@ -106,6 +114,8 @@ public class CatalogClickHouseClusterIT extends BaseIT {
Paths.get("src", "test", "resources",
"remote_servers.xml").toAbsolutePath().toString();
containerSuite.startClickHouseClusterContainer(TEST_DB_NAME,
remoteServersConfig);
clickHouseClusterContainer =
containerSuite.getClickHouseClusterContainer();
+ clickHouseClusterContainers =
containerSuite.getClickHouseClusterContainers();
+ Assertions.assertEquals(3, clickHouseClusterContainers.size());
clickHouseService = new ClickHouseService(clickHouseClusterContainer,
TEST_DB_NAME);
createMetalake();
@@ -162,8 +172,10 @@ public class CatalogClickHouseClusterIT extends BaseIT {
}
private void createSchema() {
- Schema createdSchema =
- catalog.asSchemas().createSchema(schemaName, null,
Collections.emptyMap());
+ Map<String, String> properties = new HashMap<>();
+ properties.put(CLUSTER_NAME, ClickHouseContainer.DEFAULT_CLUSTER_NAME);
+ properties.put(ON_CLUSTER, String.valueOf(true));
+ Schema createdSchema = catalog.asSchemas().createSchema(schemaName, null,
properties);
Schema loadSchema = catalog.asSchemas().loadSchema(schemaName);
Assertions.assertEquals(createdSchema.name(), loadSchema.name());
}
@@ -778,6 +790,57 @@ public class CatalogClickHouseClusterIT extends BaseIT {
}
}
+ @Test
+ public void testRenameTableOnClusterPropagatesToEveryNode() throws
SQLException {
+ String oldName = GravitinoITUtils.genRandomName("ck_cluster_rename_old");
+ String newName = GravitinoITUtils.genRandomName("ck_cluster_rename_new");
+ NameIdentifier oldIdent = NameIdentifier.of(schemaName, oldName);
+ NameIdentifier newIdent = NameIdentifier.of(schemaName, newName);
+ TableCatalog tableCatalog = catalog.asTableCatalog();
+
+ try {
+ tableCatalog.createTable(
+ oldIdent,
+ createColumns(),
+ "cluster rename comment",
+ clusterMergeTreeProperties(),
+ Transforms.EMPTY_TRANSFORM,
+ Distributions.NONE,
+ getSortOrders("col_3"),
+ Indexes.EMPTY_INDEXES);
+
+ Table renamed = tableCatalog.alterTable(oldIdent,
TableChange.rename(newName));
+ Assertions.assertEquals(newName, renamed.name());
+ Assertions.assertEquals(String.valueOf(true),
renamed.properties().get(ON_CLUSTER));
+ Assertions.assertEquals(
+ ClickHouseContainer.DEFAULT_CLUSTER_NAME,
renamed.properties().get(CLUSTER_NAME));
+
Assertions.assertTrue(renamed.properties().containsKey(StringIdentifier.ID_KEY));
+
+ awaitTableStateOnEveryNode(oldName, false, newName, true);
+ assertRenameQueryUsesOnCluster(oldName, newName);
+ for (ClickHouseContainer container : clickHouseClusterContainers) {
+ String storedComment = loadStoredComment(container, newName);
+ Assertions.assertNotNull(StringIdentifier.fromComment(storedComment));
+ Assertions.assertEquals(
+ ClickHouseContainer.DEFAULT_CLUSTER_NAME,
+ ClickHouseClusterUtils.extractClusterFromComment(storedComment));
+ }
+
+ Assertions.assertTrue(tableCatalog.dropTable(newIdent));
+ awaitTableStateOnEveryNode(oldName, false, newName, false);
+ } finally {
+ try {
+ clickHouseService.executeQuery(
+ "DROP TABLE IF EXISTS `%s`.`%s` ON CLUSTER `%s` SYNC"
+ .formatted(schemaName, oldName,
ClickHouseContainer.DEFAULT_CLUSTER_NAME));
+ } finally {
+ clickHouseService.executeQuery(
+ "DROP TABLE IF EXISTS `%s`.`%s` ON CLUSTER `%s` SYNC"
+ .formatted(schemaName, newName,
ClickHouseContainer.DEFAULT_CLUSTER_NAME));
+ }
+ }
+ }
+
/**
* When a user updates the table comment via Gravitino, the cluster metadata
embedded in the
* ClickHouse COMMENT field must be preserved. Without re-embedding, the
next loadTable call would
@@ -1193,4 +1256,81 @@ public class CatalogClickHouseClusterIT extends BaseIT {
tableCatalog.dropTable(localIdent);
}
}
+
+ private void awaitTableStateOnEveryNode(
+ String oldTableName, boolean oldTableExists, String newTableName,
boolean newTableExists) {
+ Awaitility.await()
+ .atMost(Duration.ofSeconds(30))
+ .pollInterval(Duration.ofMillis(250))
+ .untilAsserted(
+ () -> {
+ for (ClickHouseContainer container :
clickHouseClusterContainers) {
+ Assertions.assertEquals(
+ oldTableExists,
+ tableExists(container, oldTableName),
+ "Unexpected old-table state on " +
container.getContainerIpAddress());
+ Assertions.assertEquals(
+ newTableExists,
+ tableExists(container, newTableName),
+ "Unexpected new-table state on " +
container.getContainerIpAddress());
+ }
+ });
+ }
+
+ private boolean tableExists(ClickHouseContainer container, String tableName)
throws SQLException {
+ try (Connection connection =
+ DriverManager.getConnection(
+ container.getJdbcUrl(TEST_DB_NAME),
+ container.getUsername(),
+ container.getPassword());
+ PreparedStatement statement =
+ connection.prepareStatement(
+ "SELECT count() FROM system.tables WHERE database = ? AND name
= ?")) {
+ statement.setString(1, schemaName);
+ statement.setString(2, tableName);
+ try (ResultSet resultSet = statement.executeQuery()) {
+ Assertions.assertTrue(resultSet.next());
+ return resultSet.getLong(1) == 1;
+ }
+ }
+ }
+
+ private String loadStoredComment(ClickHouseContainer container, String
tableName)
+ throws SQLException {
+ try (Connection connection =
+ DriverManager.getConnection(
+ container.getJdbcUrl(TEST_DB_NAME),
+ container.getUsername(),
+ container.getPassword());
+ PreparedStatement statement =
+ connection.prepareStatement(
+ "SELECT comment FROM system.tables WHERE database = ? AND name
= ?")) {
+ statement.setString(1, schemaName);
+ statement.setString(2, tableName);
+ try (ResultSet resultSet = statement.executeQuery()) {
+ Assertions.assertTrue(resultSet.next());
+ return resultSet.getString(1);
+ }
+ }
+ }
+
+ private void assertRenameQueryUsesOnCluster(String oldTableName, String
newTableName) {
+ clickHouseService.executeQuery("SYSTEM FLUSH LOGS");
+ String query =
+ clickHouseService.executeQueryForResult(
+ String.format(
+ "SELECT query FROM system.query_log "
+ + "WHERE type = 'QueryFinish' "
+ + "AND startsWith(query, 'RENAME TABLE') "
+ + "AND query LIKE '%%`%s`%%' "
+ + "ORDER BY event_time DESC LIMIT 1",
+ oldTableName));
+
+ Assertions.assertNotNull(query, "The initiating RENAME query must be
present in query_log");
+ Assertions.assertTrue(
+ query.contains("RENAME TABLE `%s` TO `%s`".formatted(oldTableName,
newTableName)));
+ Assertions.assertTrue(
+ query.contains("ON CLUSTER
`%s`".formatted(ClickHouseContainer.DEFAULT_CLUSTER_NAME)));
+ Assertions.assertEquals(1, StringUtils.countMatches(query, "ON CLUSTER"));
+ }
}
diff --git
a/catalogs-contrib/catalog-jdbc-clickhouse/src/test/java/org/apache/gravitino/catalog/clickhouse/integration/test/CatalogClickHouseIT.java
b/catalogs-contrib/catalog-jdbc-clickhouse/src/test/java/org/apache/gravitino/catalog/clickhouse/integration/test/CatalogClickHouseIT.java
index 72731536f4..7c5abf9463 100644
---
a/catalogs-contrib/catalog-jdbc-clickhouse/src/test/java/org/apache/gravitino/catalog/clickhouse/integration/test/CatalogClickHouseIT.java
+++
b/catalogs-contrib/catalog-jdbc-clickhouse/src/test/java/org/apache/gravitino/catalog/clickhouse/integration/test/CatalogClickHouseIT.java
@@ -48,6 +48,7 @@ import org.apache.gravitino.CatalogChange;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.Schema;
+import org.apache.gravitino.StringIdentifier;
import org.apache.gravitino.SupportsSchemas;
import org.apache.gravitino.auth.AuthConstants;
import
org.apache.gravitino.catalog.clickhouse.ClickHouseConstants.TableConstants;
@@ -1417,6 +1418,21 @@ public class CatalogClickHouseIT extends BaseIT {
.asTableCatalog()
.alterTable(NameIdentifier.of(schemaName, tableName),
TableChange.rename(alertTableName));
+ clickhouseService.executeQuery("SYSTEM FLUSH LOGS");
+ String renameQuery =
+ clickhouseService.executeQueryForResult(
+ String.format(
+ "SELECT query FROM system.query_log "
+ + "WHERE type = 'QueryFinish' "
+ + "AND startsWith(query, 'RENAME TABLE') "
+ + "AND query LIKE '%%`%s`%%' "
+ + "ORDER BY event_time DESC LIMIT 1",
+ tableName));
+ Assertions.assertNotNull(renameQuery);
+ Assertions.assertTrue(
+ renameQuery.contains("RENAME TABLE `%s` TO `%s`".formatted(tableName,
alertTableName)));
+ Assertions.assertFalse(renameQuery.contains("ON CLUSTER"));
+
catalog
.asTableCatalog()
.alterTable(
@@ -1444,6 +1460,7 @@ public class CatalogClickHouseIT extends BaseIT {
Table table =
catalog.asTableCatalog().loadTable(NameIdentifier.of(schemaName,
alertTableName));
Assertions.assertEquals(alertTableName, table.name());
+
Assertions.assertTrue(table.properties().containsKey(StringIdentifier.ID_KEY));
Assertions.assertEquals(CLICKHOUSE_COL_NAME1, table.columns()[0].name());
Assertions.assertEquals(Types.IntegerType.get(),
table.columns()[0].dataType());
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 5978fe212c..aa1120cebf 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
@@ -21,6 +21,7 @@ package org.apache.gravitino.catalog.clickhouse.operations;
import static
org.apache.gravitino.catalog.clickhouse.ClickHouseTablePropertiesMetadata.GRAVITINO_ENGINE_KEY;
import static
org.apache.gravitino.catalog.clickhouse.operations.ClickHouseClusterUtils.CLUSTER_META_PREFIX;
import static
org.apache.gravitino.catalog.clickhouse.operations.ClickHouseClusterUtils.extractClusterFromComment;
+import static
org.apache.gravitino.catalog.clickhouse.operations.ClickHouseClusterUtils.hasClusterMetadata;
import static
org.apache.gravitino.catalog.clickhouse.operations.ClickHouseClusterUtils.stripClusterMetadata;
import java.util.HashMap;
@@ -253,6 +254,30 @@ class TestClickHouseTableOperationsCluster {
Assertions.assertEquals("DROP TABLE `orders` ON CLUSTER `ck_cluster`
SYNC", sql);
}
+ @Test
+ void testGenerateRenameTableSqlWithoutClusterMetadata() {
+ String sql = ops.buildRenameSql("old-table", "new table", false, null);
+
+ Assertions.assertEquals("RENAME TABLE `old-table` TO `new table`", sql);
+ }
+
+ @Test
+ void testGenerateRenameTableSqlWithClusterMetadata() {
+ String sql = ops.buildRenameSql("old-table", "new table", true,
"ck-cluster");
+
+ Assertions.assertEquals("RENAME TABLE `old-table` TO `new table` ON
CLUSTER `ck-cluster`", sql);
+ }
+
+ @Test
+ void testGenerateRenameTableSqlRejectsBlankClusterMetadata() {
+ IllegalArgumentException exception =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> ops.buildRenameSql("orders", "renamed_orders", true, " "));
+
+ Assertions.assertTrue(exception.getMessage().contains("missing a cluster
name"));
+ }
+
@Test
void testGenerateDropTableSqlOnClusterWithoutClusterName() {
// on-cluster=true but no cluster-name → fall back to plain DROP TABLE
@@ -364,6 +389,13 @@ class TestClickHouseTableOperationsCluster {
Assertions.assertEquals("", stripClusterMetadata(stored));
}
+ @Test
+ void testClusterMetadataPresenceDistinguishesAbsentAndBlankMarker() {
+ Assertions.assertFalse(hasClusterMetadata("plain comment"));
+ Assertions.assertTrue(hasClusterMetadata("comment" + CLUSTER_META_PREFIX));
+ Assertions.assertEquals("", extractClusterFromComment("comment" +
CLUSTER_META_PREFIX));
+ }
+
/** ALTER TABLE with ON CLUSTER=true should include ON CLUSTER in SQL. */
@Test
void testAlterTableWithOnCluster() {
@@ -827,6 +859,11 @@ class TestClickHouseTableOperationsCluster {
String buildDropSql(String tableName, Map<String, String> properties) {
return generateDropTableSql(tableName, properties);
}
+
+ String buildRenameSql(
+ String oldTableName, String newTableName, boolean hasClusterMetadata,
String clusterName) {
+ return generateRenameTableSql(oldTableName, newTableName,
hasClusterMetadata, clusterName);
+ }
}
private static final class StubClickHouseTableOperations extends
ClickHouseTableOperations {
diff --git
a/catalogs-contrib/catalog-jdbc-clickhouse/src/test/java/org/apache/gravitino/catalog/clickhouse/operations/TestClickHouseTableOperationsUnit.java
b/catalogs-contrib/catalog-jdbc-clickhouse/src/test/java/org/apache/gravitino/catalog/clickhouse/operations/TestClickHouseTableOperationsUnit.java
index 8dfe14b204..3b80110d23 100644
---
a/catalogs-contrib/catalog-jdbc-clickhouse/src/test/java/org/apache/gravitino/catalog/clickhouse/operations/TestClickHouseTableOperationsUnit.java
+++
b/catalogs-contrib/catalog-jdbc-clickhouse/src/test/java/org/apache/gravitino/catalog/clickhouse/operations/TestClickHouseTableOperationsUnit.java
@@ -24,9 +24,11 @@ import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
+import java.sql.Statement;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import javax.sql.DataSource;
import
org.apache.gravitino.catalog.clickhouse.ClickHouseConstants.TableConstants;
import
org.apache.gravitino.catalog.clickhouse.ClickHouseTablePropertiesMetadata.ENGINE;
import
org.apache.gravitino.catalog.clickhouse.converter.ClickHouseColumnDefaultValueConverter;
@@ -88,9 +90,13 @@ public class TestClickHouseTableOperationsUnit {
}
private ExposedClickHouseTableOperations newOps() {
+ return newOps(null);
+ }
+
+ private ExposedClickHouseTableOperations newOps(DataSource dataSource) {
ExposedClickHouseTableOperations ops = new
ExposedClickHouseTableOperations();
ops.initialize(
- null,
+ dataSource,
new ClickHouseExceptionConverter(),
new ClickHouseTypeConverter(),
new ClickHouseColumnDefaultValueConverter(),
@@ -103,7 +109,6 @@ public class TestClickHouseTableOperationsUnit {
PreparedStatement statement = Mockito.mock(PreparedStatement.class);
ResultSet resultSet = Mockito.mock(ResultSet.class);
Mockito.when(resultSet.next()).thenReturn(true);
- Mockito.when(resultSet.getString("name")).thenReturn("test_table");
Mockito.when(resultSet.getString("COMMENT")).thenReturn("");
Mockito.when(resultSet.getString("ENGINE")).thenReturn(engine);
Mockito.when(resultSet.getString("engine_full")).thenReturn(engineFull);
@@ -264,6 +269,82 @@ public class TestClickHouseTableOperationsUnit {
Assertions.assertTrue(exception.getMessage().contains("db_name"));
}
+ @Test
+ void testGetTablePropertiesScopesMetadataToCurrentDatabase() throws
Exception {
+ ExposedClickHouseTableOperations ops = newOps();
+ Connection connection = Mockito.mock(Connection.class);
+ PreparedStatement statement = Mockito.mock(PreparedStatement.class);
+ ResultSet resultSet = Mockito.mock(ResultSet.class);
+ ArgumentCaptor<String> sqlCaptor = ArgumentCaptor.forClass(String.class);
+
+
Mockito.when(connection.prepareStatement(sqlCaptor.capture())).thenReturn(statement);
+ Mockito.when(statement.executeQuery()).thenReturn(resultSet);
+ Mockito.when(resultSet.next()).thenReturn(true);
+ Mockito.when(resultSet.getString("COMMENT")).thenReturn("table comment");
+
Mockito.when(resultSet.getString("ENGINE")).thenReturn(ENGINE.MERGETREE.getValue());
+ Mockito.when(resultSet.getString("engine_full")).thenReturn("MergeTree
ORDER BY id");
+
+ ops.callGetTableProperties(connection, "same_name");
+
+ Assertions.assertEquals(
+ "SELECT comment, engine, engine_full FROM system.tables "
+ + "WHERE database = currentDatabase() AND name = ?",
+ sqlCaptor.getValue());
+ Mockito.verify(statement).setString(1, "same_name");
+ }
+
+ @Test
+ void testRenameUsesTrustedClusterMetadata() throws Exception {
+ RenameMocks mocks = renameMocks("comment\n[Gravitino]
ch.cluster=ck_cluster", "MergeTree");
+ ExposedClickHouseTableOperations ops = newOps(mocks.dataSource);
+
+ ops.rename("db_name", "old-table", "new table");
+
+ Mockito.verify(mocks.connection).setCatalog("db_name");
+ Mockito.verify(mocks.updateStatement)
+ .executeUpdate("RENAME TABLE `old-table` TO `new table` ON CLUSTER
`ck_cluster`");
+ }
+
+ @Test
+ void testRenameDoesNotPromoteUnmarkedDistributedTableToClusterScope() throws
Exception {
+ // The Distributed engine contains a cluster name, but only the Gravitino
comment marker may
+ // authorize cluster-wide DDL.
+ RenameMocks mocks =
+ renameMocks("external table", "Distributed('ck_cluster', 'db',
'remote', id)");
+ ExposedClickHouseTableOperations ops = newOps(mocks.dataSource);
+
+ ops.rename("db_name", "old_table", "new_table");
+
+ Mockito.verify(mocks.updateStatement).executeUpdate("RENAME TABLE
`old_table` TO `new_table`");
+ }
+
+ @Test
+ void testRenameRejectsCorruptedClusterMetadataBeforeMutation() throws
Exception {
+ RenameMocks mocks = renameMocks("comment\n[Gravitino] ch.cluster= ",
"MergeTree");
+ ExposedClickHouseTableOperations ops = newOps(mocks.dataSource);
+
+ IllegalArgumentException exception =
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () -> ops.rename("db_name",
"old_table", "new_table"));
+
+ Assertions.assertTrue(exception.getMessage().contains("missing a cluster
name"));
+ Mockito.verify(mocks.connection, Mockito.never()).createStatement();
+ }
+
+ @Test
+ void testRenameMapsSqlException() throws Exception {
+ RenameMocks mocks = renameMocks("comment\n[Gravitino]
ch.cluster=ck_cluster", "MergeTree");
+ SQLException sqlException = new SQLException("rename failed");
+
Mockito.when(mocks.updateStatement.executeUpdate(Mockito.anyString())).thenThrow(sqlException);
+ ExposedClickHouseTableOperations ops = newOps(mocks.dataSource);
+
+ GravitinoRuntimeException exception =
+ Assertions.assertThrows(
+ GravitinoRuntimeException.class, () -> ops.rename("db_name",
"old_table", "new_table"));
+
+ Assertions.assertSame(sqlException, exception.getCause());
+ }
+
//
---------------------------------------------------------------------------
// extractEngineParams
//
---------------------------------------------------------------------------
@@ -610,4 +691,38 @@ public class TestClickHouseTableOperationsUnit {
Assertions.assertTrue(exception.getCause() instanceof SQLException);
Mockito.verify(connection,
Mockito.times(2)).prepareStatement(Mockito.anyString());
}
+
+ private RenameMocks renameMocks(String storedComment, String engineFull)
throws Exception {
+ DataSource dataSource = Mockito.mock(DataSource.class);
+ Connection connection = Mockito.mock(Connection.class);
+ PreparedStatement metadataStatement =
Mockito.mock(PreparedStatement.class);
+ ResultSet metadataResult = Mockito.mock(ResultSet.class);
+ Statement updateStatement = Mockito.mock(Statement.class);
+
+ Mockito.when(dataSource.getConnection()).thenReturn(connection);
+
Mockito.when(connection.prepareStatement(Mockito.anyString())).thenReturn(metadataStatement);
+ Mockito.when(metadataStatement.executeQuery()).thenReturn(metadataResult);
+ Mockito.when(metadataResult.next()).thenReturn(true);
+
Mockito.when(metadataResult.getString("COMMENT")).thenReturn(storedComment);
+ Mockito.when(metadataResult.getString("ENGINE"))
+ .thenReturn(
+ engineFull.startsWith("Distributed")
+ ? ENGINE.DISTRIBUTED.getValue()
+ : ENGINE.MERGETREE.getValue());
+
Mockito.when(metadataResult.getString("engine_full")).thenReturn(engineFull);
+ Mockito.when(connection.createStatement()).thenReturn(updateStatement);
+ return new RenameMocks(dataSource, connection, updateStatement);
+ }
+
+ private static final class RenameMocks {
+ private final DataSource dataSource;
+ private final Connection connection;
+ private final Statement updateStatement;
+
+ private RenameMocks(DataSource dataSource, Connection connection,
Statement updateStatement) {
+ this.dataSource = dataSource;
+ this.connection = connection;
+ this.updateStatement = updateStatement;
+ }
+ }
}
diff --git
a/catalogs-contrib/catalog-jdbc-clickhouse/src/test/resources/remote_servers.xml
b/catalogs-contrib/catalog-jdbc-clickhouse/src/test/resources/remote_servers.xml
index 0b791ea3be..ce38abdc61 100644
---
a/catalogs-contrib/catalog-jdbc-clickhouse/src/test/resources/remote_servers.xml
+++
b/catalogs-contrib/catalog-jdbc-clickhouse/src/test/resources/remote_servers.xml
@@ -20,7 +20,15 @@
<gravitino_cluster>
<shard>
<replica>
- <host>gravitino-ci-clickhouse-cluster</host>
+ <host>gravitino-ci-clickhouse-cluster-1</host>
+ <port>9000</port>
+ </replica>
+ <replica>
+ <host>gravitino-ci-clickhouse-cluster-2</host>
+ <port>9000</port>
+ </replica>
+ <replica>
+ <host>gravitino-ci-clickhouse-cluster-3</host>
<port>9000</port>
</replica>
</shard>
diff --git
a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/ContainerSuite.java
b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/ContainerSuite.java
index 384243edf8..05ecac0186 100644
---
a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/ContainerSuite.java
+++
b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/ContainerSuite.java
@@ -35,6 +35,7 @@ import java.nio.file.FileSystems;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.nio.file.attribute.PosixFilePermission;
+import java.util.ArrayList;
import java.util.EnumMap;
import java.util.List;
import java.util.Map;
@@ -62,6 +63,8 @@ public class ContainerSuite implements Closeable {
private static final String CONTAINER_NETWORK_GATEWAY = "10.20.30.1";
private static final String CONTAINER_NETWORK_IPRANGE = "10.20.30.0/28";
private static final String NETWORK_NAME = "gravitino-ci-network";
+ private static final int CLICKHOUSE_CLUSTER_SIZE = 3;
+ private static final String CLICKHOUSE_CLUSTER_HOST_PREFIX =
"gravitino-ci-clickhouse-cluster-";
private static Network network = null;
private static volatile HiveContainer hiveContainer;
@@ -85,6 +88,7 @@ public class ContainerSuite implements Closeable {
private static volatile OceanBaseContainer oceanBaseContainer;
private static volatile ClickHouseContainer clickHouseContainer;
private static volatile ClickHouseContainer clickHouseClusterContainer;
+ private static volatile List<ClickHouseContainer>
clickHouseClusterContainers = List.of();
private static volatile ZooKeeperContainer zooKeeperContainer;
private static volatile GravitinoLocalStackContainer
gravitinoLocalStackContainer;
@@ -585,30 +589,35 @@ public class ContainerSuite implements Closeable {
public void startClickHouseClusterContainer(
TestDatabaseName testDatabaseName, String remoteServersTemplatePath) {
- if (clickHouseClusterContainer == null) {
+ if (clickHouseClusterContainers.isEmpty()) {
synchronized (ContainerSuite.class) {
- if (clickHouseClusterContainer == null) {
+ if (clickHouseClusterContainers.isEmpty()) {
initIfNecessary();
startZooKeeperContainer();
String zkHost = zooKeeperContainer.getContainerIpAddress();
String resolvedConfigPath =
prepareRemoteServersConfig(remoteServersTemplatePath, zkHost);
- ClickHouseContainer.Builder clickHouseBuilder =
- ClickHouseContainer.builder()
- .withHostName("gravitino-ci-clickhouse-cluster")
- .withEnvVars(
- ImmutableMap.<String, String>builder()
- .put("CLICKHOUSE_PASSWORD",
ClickHouseContainer.PASSWORD)
- .build())
- .withRemoteServersConfig(resolvedConfigPath)
- .withExposePorts(
- ImmutableSet.of(
- ClickHouseContainer.CLICKHOUSE_PORT,
- ClickHouseContainer.CLICKHOUSE_NATIVE_PORT))
- .withNetwork(network);
-
- ClickHouseContainer container =
closer.register(clickHouseBuilder.build());
- container.start();
- clickHouseClusterContainer = container;
+ List<ClickHouseContainer> containers = new
ArrayList<>(CLICKHOUSE_CLUSTER_SIZE);
+ for (int node = 1; node <= CLICKHOUSE_CLUSTER_SIZE; node++) {
+ ClickHouseContainer.Builder clickHouseBuilder =
+ ClickHouseContainer.builder()
+ .withHostName(CLICKHOUSE_CLUSTER_HOST_PREFIX + node)
+ .withEnvVars(
+ ImmutableMap.<String, String>builder()
+ .put("CLICKHOUSE_PASSWORD",
ClickHouseContainer.PASSWORD)
+ .build())
+ .withRemoteServersConfig(resolvedConfigPath)
+ .withExposePorts(
+ ImmutableSet.of(
+ ClickHouseContainer.CLICKHOUSE_PORT,
+ ClickHouseContainer.CLICKHOUSE_NATIVE_PORT))
+ .withNetwork(network);
+
+ ClickHouseContainer container =
closer.register(clickHouseBuilder.build());
+ container.start();
+ containers.add(container);
+ }
+ clickHouseClusterContainers = List.copyOf(containers);
+ clickHouseClusterContainer = clickHouseClusterContainers.get(0);
}
}
}
@@ -671,6 +680,16 @@ public class ContainerSuite implements Closeable {
return clickHouseClusterContainer;
}
+ /**
+ * Returns every ClickHouse node in the repository-managed cluster fixture.
+ *
+ * @return an immutable list whose first element is also returned by {@link
+ * #getClickHouseClusterContainer()}
+ */
+ public List<ClickHouseContainer> getClickHouseClusterContainers() {
+ return clickHouseClusterContainers;
+ }
+
public ZooKeeperContainer getZooKeeperContainer() {
return zooKeeperContainer;
}
