This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 77f3e91344 [Cherry-pick to branch-1.3] [#12748] fix(clickhouse):
preserve composite primary key on table load (#12765) (#12863)
77f3e91344 is described below
commit 77f3e913441914ab3e44372364fdc968418ce6f7
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Sep 3 12:03:01 2026 +0800
[Cherry-pick to branch-1.3] [#12748] fix(clickhouse): preserve composite
primary key on table load (#12765) (#12863)
**Cherry-pick Information:**
- Original commit: d8066df8de91e4303bf7d366c614f6109db451e9
- Target branch: `branch-1.3`
- Status: ✅ Clean cherry-pick (no conflicts)
Signed-off-by: jiangxt2 <[email protected]>
Co-authored-by: StormSpirit <[email protected]>
---
.../operations/ClickHouseTableOperations.java | 41 +++++-
.../integration/test/CatalogClickHouseIT.java | 45 ++++++
.../TestClickHouseTableOperationsUnit.java | 156 +++++++++++++++++++++
3 files changed, 238 insertions(+), 4 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 e0e67c73bd..835b6d9fa6 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
@@ -40,8 +40,11 @@ import java.sql.Statement;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.Comparator;
import java.util.EnumSet;
import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
@@ -64,6 +67,7 @@ import
org.apache.gravitino.catalog.clickhouse.ClickHouseTablePropertiesMetadata
import
org.apache.gravitino.catalog.clickhouse.ClickHouseTablePropertiesMetadata.ENGINE;
import org.apache.gravitino.catalog.jdbc.JdbcColumn;
import org.apache.gravitino.catalog.jdbc.JdbcTable;
+import org.apache.gravitino.catalog.jdbc.bean.JdbcIndexBean;
import org.apache.gravitino.catalog.jdbc.converter.JdbcTypeConverter;
import org.apache.gravitino.catalog.jdbc.operation.JdbcTableOperations;
import org.apache.gravitino.catalog.jdbc.utils.JdbcConnectorUtils;
@@ -126,7 +130,7 @@ public class ClickHouseTableOperations extends
JdbcTableOperations {
WHERE system.tables.primary_key <> ''
AND system.tables.database = '%s'
AND system.tables.name = '%s'
- ORDER BY COLUMN_NAME
+ ORDER BY PK_NAME, KEY_SEQ
""";
@Override
@@ -243,12 +247,41 @@ public class ClickHouseTableOperations extends
JdbcTableOperations {
try (PreparedStatement preparedStatement =
connection.prepareStatement(sql);
ResultSet resultSet = preparedStatement.executeQuery()) {
- List<Index> indexes = new ArrayList<>();
+ Map<String, List<JdbcIndexBean>> primaryKeysByName = new
LinkedHashMap<>();
while (resultSet.next()) {
+ // ClickHouse exposes one primary-key expression without a constraint
name; the query
+ // synthesizes PRIMARY to match Gravitino's default primary-key name.
String indexName = resultSet.getString("PK_NAME");
String columnName = resultSet.getString("COLUMN_NAME");
- indexes.add(
- Indexes.of(Index.IndexType.PRIMARY_KEY, indexName, new String[][]
{{columnName}}));
+ int keySequence = resultSet.getInt("KEY_SEQ");
+ Preconditions.checkArgument(
+ !resultSet.wasNull() && keySequence > 0,
+ "Primary key %s column %s has invalid KEY_SEQ %s",
+ indexName,
+ columnName,
+ keySequence);
+ primaryKeysByName
+ .computeIfAbsent(indexName, ignored -> new ArrayList<>())
+ .add(
+ new JdbcIndexBean(Index.IndexType.PRIMARY_KEY, columnName,
indexName, keySequence));
+ }
+
+ List<Index> indexes = new ArrayList<>();
+ for (Map.Entry<String, List<JdbcIndexBean>> entry :
primaryKeysByName.entrySet()) {
+ Set<Integer> keySequences = new HashSet<>();
+ for (JdbcIndexBean primaryKeyColumn : entry.getValue()) {
+ Preconditions.checkArgument(
+ keySequences.add(primaryKeyColumn.getOrder()),
+ "Primary key %s has duplicate KEY_SEQ %s",
+ entry.getKey(),
+ primaryKeyColumn.getOrder());
+ }
+ List<String> columnNames =
+ entry.getValue().stream()
+ .sorted(Comparator.comparingInt(JdbcIndexBean::getOrder))
+ .map(JdbcIndexBean::getColName)
+ .collect(Collectors.toList());
+ indexes.add(Indexes.primary(entry.getKey(),
convertIndexFieldNames(columnNames)));
}
indexes.addAll(getSecondaryIndexes(connection, databaseName, tableName));
return indexes;
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 d715af3a25..45057b5eff 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
@@ -485,6 +485,51 @@ public class CatalogClickHouseIT extends BaseIT {
&& Arrays.deepEquals(idx.fieldNames(), new String[][]
{{"amount"}})));
}
+ @Test
+ void testCreateAndLoadCompositePrimaryKey() {
+ String table = GravitinoITUtils.genRandomName("composite_primary_key");
+ NameIdentifier ident = NameIdentifier.of(schemaName, table);
+ Column[] columns =
+ new Column[] {
+ Column.of("id", Types.LongType.get(), "id", false, false,
DEFAULT_VALUE_NOT_SET),
+ Column.of(
+ "ts",
+ Types.TimestampType.withoutTimeZone(),
+ "timestamp",
+ false,
+ false,
+ DEFAULT_VALUE_NOT_SET),
+ Column.of("value", Types.StringType.get(), "value")
+ };
+ SortOrder[] sortOrders =
+ new SortOrder[] {
+ SortOrders.of(NamedReference.field("id"), SortDirection.ASCENDING),
+ SortOrders.of(NamedReference.field("ts"), SortDirection.ASCENDING)
+ };
+ Index[] indexes =
+ new Index[] {
+ Indexes.primary(Indexes.DEFAULT_PRIMARY_KEY_NAME, new String[][]
{{"id"}, {"ts"}})
+ };
+
+ catalog
+ .asTableCatalog()
+ .createTable(
+ ident,
+ columns,
+ "composite primary key roundtrip",
+ createProperties(),
+ Transforms.EMPTY_TRANSFORM,
+ Distributions.NONE,
+ sortOrders,
+ indexes);
+
+ Index[] loadedIndexes = catalog.asTableCatalog().loadTable(ident).index();
+ Assertions.assertEquals(1, loadedIndexes.length);
+ Assertions.assertEquals(Index.IndexType.PRIMARY_KEY,
loadedIndexes[0].type());
+ Assertions.assertEquals(Indexes.DEFAULT_PRIMARY_KEY_NAME,
loadedIndexes[0].name());
+ Assertions.assertArrayEquals(new String[][] {{"id"}, {"ts"}},
loadedIndexes[0].fieldNames());
+ }
+
@Test
void testCreateAndLoadWithPartitionSortAndIndexes() {
String table = GravitinoITUtils.genRandomName("meta_roundtrip");
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 da3b533de5..357d2783ee 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
@@ -119,6 +119,37 @@ public class TestClickHouseTableOperationsUnit {
return newOps().callGetTableProperties(connection, "test_table");
}
+ private Connection mockGetIndexesConnection(
+ ResultSet primaryKeyResultSet, ResultSet secondaryIndexResultSet) throws
SQLException {
+ PreparedStatement primaryKeyStatement =
Mockito.mock(PreparedStatement.class);
+ PreparedStatement secondaryIndexStatement =
Mockito.mock(PreparedStatement.class);
+
Mockito.when(primaryKeyStatement.executeQuery()).thenReturn(primaryKeyResultSet);
+
Mockito.when(secondaryIndexStatement.executeQuery()).thenReturn(secondaryIndexResultSet);
+
+ Connection connection = Mockito.mock(Connection.class);
+ Mockito.when(connection.prepareStatement(Mockito.anyString()))
+ .thenReturn(primaryKeyStatement)
+ .thenReturn(secondaryIndexStatement);
+ return connection;
+ }
+
+ private void assertInvalidPrimaryKeySequence(int keySequence, boolean
wasNull) throws Exception {
+ ResultSet primaryKeyResultSet = Mockito.mock(ResultSet.class);
+ ResultSet secondaryIndexResultSet = Mockito.mock(ResultSet.class);
+ Mockito.when(primaryKeyResultSet.next()).thenReturn(true, false);
+
Mockito.when(primaryKeyResultSet.getString("PK_NAME")).thenReturn("PRIMARY");
+
Mockito.when(primaryKeyResultSet.getString("COLUMN_NAME")).thenReturn("id");
+
Mockito.when(primaryKeyResultSet.getInt("KEY_SEQ")).thenReturn(keySequence);
+ Mockito.when(primaryKeyResultSet.wasNull()).thenReturn(wasNull);
+ Mockito.when(secondaryIndexResultSet.next()).thenReturn(false);
+
+ Connection connection = mockGetIndexesConnection(primaryKeyResultSet,
secondaryIndexResultSet);
+ IllegalArgumentException exception =
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
newOps().callGetIndexes(connection, "db", "tbl"));
+ Assertions.assertTrue(exception.getMessage().contains("invalid KEY_SEQ"));
+ }
+
//
---------------------------------------------------------------------------
// getIndexes — SQL injection escape
//
---------------------------------------------------------------------------
@@ -150,6 +181,131 @@ public class TestClickHouseTableOperationsUnit {
Assertions.assertTrue(
primaryKeySql.contains("db''1"), "database single quote should be
doubled");
Assertions.assertTrue(primaryKeySql.contains("t''1"), "table single quote
should be doubled");
+ Assertions.assertTrue(primaryKeySql.contains("ORDER BY PK_NAME, KEY_SEQ"));
+ }
+
+ @Test
+ void testGetIndexesAggregatesCompositePrimaryKeyInSequenceOrder() throws
Exception {
+ ResultSet primaryKeyResultSet = Mockito.mock(ResultSet.class);
+ ResultSet secondaryIndexResultSet = Mockito.mock(ResultSet.class);
+ Mockito.when(primaryKeyResultSet.next()).thenReturn(true, true, false);
+
Mockito.when(primaryKeyResultSet.getString("PK_NAME")).thenReturn("PRIMARY",
"PRIMARY");
+
Mockito.when(primaryKeyResultSet.getString("COLUMN_NAME")).thenReturn("ts",
"id");
+ Mockito.when(primaryKeyResultSet.getInt("KEY_SEQ")).thenReturn(2, 1);
+ Mockito.when(secondaryIndexResultSet.next()).thenReturn(true, false);
+
Mockito.when(secondaryIndexResultSet.getString("name")).thenReturn("idx_value");
+
Mockito.when(secondaryIndexResultSet.getString("type")).thenReturn("minmax");
+
Mockito.when(secondaryIndexResultSet.getString("type_full")).thenReturn("minmax");
+
Mockito.when(secondaryIndexResultSet.getString("expr")).thenReturn("value");
+
Mockito.when(secondaryIndexResultSet.getLong("granularity")).thenReturn(1L);
+
+ List<Index> indexes =
+ newOps()
+ .callGetIndexes(
+ mockGetIndexesConnection(primaryKeyResultSet,
secondaryIndexResultSet),
+ "db",
+ "tbl");
+
+ Assertions.assertEquals(2, indexes.size());
+ Assertions.assertEquals(Index.IndexType.PRIMARY_KEY,
indexes.get(0).type());
+ Assertions.assertEquals("PRIMARY", indexes.get(0).name());
+ Assertions.assertArrayEquals(new String[][] {{"id"}, {"ts"}},
indexes.get(0).fieldNames());
+ Assertions.assertEquals(Index.IndexType.DATA_SKIPPING_MINMAX,
indexes.get(1).type());
+ Assertions.assertEquals("idx_value", indexes.get(1).name());
+ Assertions.assertArrayEquals(new String[][] {{"value"}},
indexes.get(1).fieldNames());
+ }
+
+ @Test
+ void testGetIndexesPreservesSingleColumnPrimaryKey() throws Exception {
+ ResultSet primaryKeyResultSet = Mockito.mock(ResultSet.class);
+ ResultSet secondaryIndexResultSet = Mockito.mock(ResultSet.class);
+ Mockito.when(primaryKeyResultSet.next()).thenReturn(true, false);
+
Mockito.when(primaryKeyResultSet.getString("PK_NAME")).thenReturn("PRIMARY");
+
Mockito.when(primaryKeyResultSet.getString("COLUMN_NAME")).thenReturn("id");
+ Mockito.when(primaryKeyResultSet.getInt("KEY_SEQ")).thenReturn(1);
+ Mockito.when(secondaryIndexResultSet.next()).thenReturn(false);
+
+ List<Index> indexes =
+ newOps()
+ .callGetIndexes(
+ mockGetIndexesConnection(primaryKeyResultSet,
secondaryIndexResultSet),
+ "db",
+ "tbl");
+
+ Assertions.assertEquals(1, indexes.size());
+ Assertions.assertEquals(Index.IndexType.PRIMARY_KEY,
indexes.get(0).type());
+ Assertions.assertEquals("PRIMARY", indexes.get(0).name());
+ Assertions.assertArrayEquals(new String[][] {{"id"}},
indexes.get(0).fieldNames());
+ }
+
+ @Test
+ void testGetIndexesKeepsDifferentPrimaryKeyNamesSeparate() throws Exception {
+ ResultSet primaryKeyResultSet = Mockito.mock(ResultSet.class);
+ ResultSet secondaryIndexResultSet = Mockito.mock(ResultSet.class);
+ Mockito.when(primaryKeyResultSet.next()).thenReturn(true, true, false);
+
Mockito.when(primaryKeyResultSet.getString("PK_NAME")).thenReturn("PRIMARY_A",
"PRIMARY_B");
+
Mockito.when(primaryKeyResultSet.getString("COLUMN_NAME")).thenReturn("id",
"ts");
+ Mockito.when(primaryKeyResultSet.getInt("KEY_SEQ")).thenReturn(1, 1);
+ Mockito.when(secondaryIndexResultSet.next()).thenReturn(false);
+
+ List<Index> indexes =
+ newOps()
+ .callGetIndexes(
+ mockGetIndexesConnection(primaryKeyResultSet,
secondaryIndexResultSet),
+ "db",
+ "tbl");
+
+ Assertions.assertEquals(2, indexes.size());
+ Assertions.assertEquals("PRIMARY_A", indexes.get(0).name());
+ Assertions.assertArrayEquals(new String[][] {{"id"}},
indexes.get(0).fieldNames());
+ Assertions.assertEquals("PRIMARY_B", indexes.get(1).name());
+ Assertions.assertArrayEquals(new String[][] {{"ts"}},
indexes.get(1).fieldNames());
+ }
+
+ @Test
+ void testGetIndexesAllowsNonContiguousPositiveKeySequence() throws Exception
{
+ ResultSet primaryKeyResultSet = Mockito.mock(ResultSet.class);
+ ResultSet secondaryIndexResultSet = Mockito.mock(ResultSet.class);
+ Mockito.when(primaryKeyResultSet.next()).thenReturn(true, true, false);
+
Mockito.when(primaryKeyResultSet.getString("PK_NAME")).thenReturn("PRIMARY",
"PRIMARY");
+
Mockito.when(primaryKeyResultSet.getString("COLUMN_NAME")).thenReturn("ts",
"id");
+ Mockito.when(primaryKeyResultSet.getInt("KEY_SEQ")).thenReturn(3, 1);
+ Mockito.when(secondaryIndexResultSet.next()).thenReturn(false);
+
+ List<Index> indexes =
+ newOps()
+ .callGetIndexes(
+ mockGetIndexesConnection(primaryKeyResultSet,
secondaryIndexResultSet),
+ "db",
+ "tbl");
+
+ Assertions.assertEquals(1, indexes.size());
+ Assertions.assertArrayEquals(new String[][] {{"id"}, {"ts"}},
indexes.get(0).fieldNames());
+ }
+
+ @Test
+ void testGetIndexesRejectsMissingOrNonPositiveKeySequence() throws Exception
{
+ assertInvalidPrimaryKeySequence(0, true);
+ assertInvalidPrimaryKeySequence(0, false);
+ assertInvalidPrimaryKeySequence(-1, false);
+ }
+
+ @Test
+ void testGetIndexesRejectsDuplicateKeySequence() throws Exception {
+ ResultSet primaryKeyResultSet = Mockito.mock(ResultSet.class);
+ ResultSet secondaryIndexResultSet = Mockito.mock(ResultSet.class);
+ Mockito.when(primaryKeyResultSet.next()).thenReturn(true, true, false);
+
Mockito.when(primaryKeyResultSet.getString("PK_NAME")).thenReturn("PRIMARY",
"PRIMARY");
+
Mockito.when(primaryKeyResultSet.getString("COLUMN_NAME")).thenReturn("id",
"ts");
+ Mockito.when(primaryKeyResultSet.getInt("KEY_SEQ")).thenReturn(1, 1);
+ Mockito.when(secondaryIndexResultSet.next()).thenReturn(false);
+
+ Connection connection = mockGetIndexesConnection(primaryKeyResultSet,
secondaryIndexResultSet);
+ IllegalArgumentException exception =
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
newOps().callGetIndexes(connection, "db", "tbl"));
+
+ Assertions.assertTrue(exception.getMessage().contains("duplicate KEY_SEQ
1"));
}
@Test
