This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new aef812111e [flink] Support compact_chain_table procedure (#8508)
aef812111e is described below
commit aef812111e4e04079be624c52d895fcc27431e01
Author: Juntao Zhang <[email protected]>
AuthorDate: Fri Jul 10 08:37:23 2026 +0800
[flink] Support compact_chain_table procedure (#8508)
---
docs/docs/flink/procedures.md | 23 ++
.../procedure/CompactChainTableProcedure.java | 71 ++++
.../flink/action/CompactChainTableAction.java | 159 +++++++++
.../action/CompactChainTableActionFactory.java | 61 ++++
.../procedure/CompactChainTableProcedure.java | 72 ++++
.../paimon/flink/source/FlinkSourceBuilder.java | 10 +-
.../paimon/flink/source/StaticFileStoreSource.java | 20 +-
.../paimon/flink/source/SystemTableSource.java | 4 +-
.../services/org.apache.paimon.factories.Factory | 2 +
.../action/CompactChainTableActionITCase.java | 181 ++++++++++
.../CompactChainTableProcedureITCase.java | 395 +++++++++++++++++++++
11 files changed, 993 insertions(+), 5 deletions(-)
diff --git a/docs/docs/flink/procedures.md b/docs/docs/flink/procedures.md
index d264a58ca2..a8ab2abb65 100644
--- a/docs/docs/flink/procedures.md
+++ b/docs/docs/flink/procedures.md
@@ -139,6 +139,29 @@ All available procedures are listed below.
compat_strategy => 'full')
</td>
</tr>
+ <tr>
+ <td>compact_chain_table</td>
+ <td>
+ -- Use named argument<br/>
+ CALL [catalog.]sys.compact_chain_table(
+ `table` => 'table',
+ partition => 'partition',
+ overwrite => overwrite) <br/><br/>
+ -- Use indexed argument<br/>
+ CALL [catalog.]sys.compact_chain_table('table', 'partition') <br/>
+ CALL [catalog.]sys.compact_chain_table('table', 'partition',
overwrite) <br/><br/>
+ </td>
+ <td>
+ To compact chain table by merging snapshot and delta branches into
the snapshot branch. Arguments:
+ <li>table: the target chain table identifier. Cannot be empty.</li>
+ <li>partition: partition specification format (e.g.,
'dt=20250810,hour=22'). Cannot be empty.</li>
+ <li>overwrite: whether to overwrite if the partition already
exists in the snapshot branch. Default is false. Optional.</li>
+ </td>
+ <td>
+ CALL sys.compact_chain_table(`table` => 'default.T', partition =>
'dt=20250810,hour=22')<br/><br/>
+ CALL sys.compact_chain_table('default.T', 'dt=20250810,hour=22', true)
+ </td>
+ </tr>
<tr>
<td>create_tag</td>
<td>
diff --git
a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/CompactChainTableProcedure.java
b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/CompactChainTableProcedure.java
new file mode 100644
index 0000000000..57e4d4cb26
--- /dev/null
+++
b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/CompactChainTableProcedure.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.procedure;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.action.CompactChainTableAction;
+
+import org.apache.flink.table.procedure.ProcedureContext;
+
+import java.util.Map;
+
+/**
+ * Compact database procedure. Usage:
+ *
+ * <pre><code>
+ * -- NOTE: use '' as placeholder for optional arguments
+ *
+ * -- compact chain table (tableId should be 'database_name.table_name')
+ * CALL sys.compact_chain_table('tableId', 'partition')
+ *
+ * -- compact chain table with overwrite
+ * CALL sys.compact_chain_table('tableId', 'partition', true)
+ *
+ * </code></pre>
+ */
+public class CompactChainTableProcedure extends ProcedureBase {
+
+ public static final String IDENTIFIER = "compact_chain_table";
+
+ public String[] call(ProcedureContext procedureContext, String tableId,
String partition)
+ throws Exception {
+ return call(procedureContext, tableId, partition, null);
+ }
+
+ public String[] call(
+ ProcedureContext procedureContext, String tableId, String
partition, Boolean overwrite)
+ throws Exception {
+ Map<String, String> catalogOptions = catalog.options();
+ Identifier identifier = Identifier.fromString(tableId);
+ CompactChainTableAction action =
+ new CompactChainTableAction(
+ identifier.getDatabaseName(),
+ identifier.getObjectName(),
+ catalogOptions,
+ partition,
+ overwrite);
+ return execute(
+ procedureContext, action, "Compact chain table Job : " +
identifier.getFullName());
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactChainTableAction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactChainTableAction.java
new file mode 100644
index 0000000000..5dd78e9dab
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactChainTableAction.java
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.action;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.flink.sink.FlinkSinkBuilder;
+import org.apache.paimon.flink.source.FlinkSourceBuilder;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.table.ChainGroupReadTable;
+import org.apache.paimon.table.FallbackReadFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.ParameterUtils;
+import org.apache.paimon.utils.StringUtils;
+
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.configuration.ExecutionOptions;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
+import org.apache.flink.table.data.RowData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+import java.util.Map;
+
+import static
org.apache.paimon.flink.FlinkConnectorOptions.SCAN_DEDICATED_SPLIT_GENERATION;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/**
+ * Action to compact chain table by merging snapshot and delta branches into
the snapshot branch.
+ */
+public class CompactChainTableAction extends TableActionBase {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(CompactChainTableAction.class);
+
+ public static final String IDENTIFIER = "compact_chain_table";
+
+ protected String partition;
+
+ protected boolean overwrite;
+
+ public CompactChainTableAction(
+ String databaseName,
+ String tableName,
+ Map<String, String> catalogConfig,
+ String partition,
+ Boolean overwrite) {
+ super(databaseName, tableName, catalogConfig);
+ this.partition = partition;
+ this.overwrite = overwrite != null && overwrite;
+ }
+
+ public CompactChainTableAction withOverwrite(boolean overwrite) {
+ this.overwrite = overwrite;
+ return this;
+ }
+
+ public CompactChainTableAction withPartition(String partition) {
+ this.partition = partition;
+ return this;
+ }
+
+ @Override
+ public void build() throws Exception {
+ checkArgument(
+ StringUtils.isNotEmpty(partition),
+ "Partition string cannot be empty for compact_chain_table");
+ checkArgument(
+ !partition.contains(";"),
+ "compact_chain_table only supports a single partition, but
multiple partitions were provided: %s",
+ partition);
+ checkArgument(
+ new CoreOptions(table.options()).isChainTable(),
+ "compact_chain_table only supports chain table");
+ checkArgument(
+ table instanceof FallbackReadFileStoreTable,
+ "Table %s is not a chain table",
+ identifier.getFullName());
+ checkArgument(
+ env.getConfiguration().get(ExecutionOptions.RUNTIME_MODE)
+ == RuntimeExecutionMode.BATCH,
+ "compact_chain_table only supports batch execution mode");
+ checkArgument(
+
!Options.fromMap(table.options()).get(SCAN_DEDICATED_SPLIT_GENERATION),
+ "compact_chain_table does not support %s.",
+ SCAN_DEDICATED_SPLIT_GENERATION.key());
+
+ ChainGroupReadTable chainTable =
+ (ChainGroupReadTable) ((FallbackReadFileStoreTable)
table).other();
+ FileStoreTable snapshotTable = chainTable.wrapped();
+
+ RowType partitionType = snapshotTable.schema().logicalPartitionType();
+ String partitionDefaultName =
snapshotTable.coreOptions().partitionDefaultName();
+ Map<String, String> partitionSpec =
ParameterUtils.parseCommaSeparatedKeyValues(partition);
+ Predicate partitionPredicate =
+ PredicateBuilder.partition(partitionSpec, partitionType,
partitionDefaultName);
+ checkArgument(
+ partitionPredicate != null,
+ "Failed to build partition predicate for partition: %s",
+ partition);
+ PartitionPredicate predicate =
+ PartitionPredicate.fromPredicate(partitionType,
partitionPredicate);
+
+ boolean partitionExists =
+
!snapshotTable.newScan().withPartitionFilter(predicate).listPartitions().isEmpty();
+
+ if (partitionExists && !overwrite) {
+ LOG.info(
+ "Partition {} already exists in snapshot branch, skipping
compaction.",
+ partition);
+ buildEmptyPipeline();
+ return;
+ }
+ DataStream<RowData> source =
+ new FlinkSourceBuilder(chainTable)
+ .env(env)
+ .sourceBounded(true)
+ .partitionPredicate(predicate)
+ .withSkipPreloadTargetSnapshot(partitionExists &&
overwrite)
+ .sourceName(identifier.getFullName() +
"-chain-compact-source")
+ .build();
+
+ FlinkSinkBuilder sinkBuilder =
+ new FlinkSinkBuilder(
+ snapshotTable.copy(
+ Collections.singletonMap(
+
CoreOptions.DYNAMIC_PARTITION_OVERWRITE.key(),
+ "true")))
+ .forRowData(source);
+ if (partitionExists) {
+ sinkBuilder.overwrite(partitionSpec);
+ }
+ sinkBuilder.build();
+ }
+
+ private void buildEmptyPipeline() {
+ env.fromSequence(0, 0).sinkTo(new DiscardingSink<>());
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactChainTableActionFactory.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactChainTableActionFactory.java
new file mode 100644
index 0000000000..f777f9276d
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactChainTableActionFactory.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.action;
+
+import java.util.Map;
+import java.util.Optional;
+
+/** Factory to create {@link CompactChainTableAction}. */
+public class CompactChainTableActionFactory implements ActionFactory {
+
+ public static final String IDENTIFIER = "compact_chain_table";
+
+ private static final String PARTITION = "partition";
+ private static final String OVERWRITE = "overwrite";
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Optional<Action> create(MultipleParameterToolAdapter params) {
+ String database = params.getRequired(DATABASE);
+ String table = params.getRequired(TABLE);
+ Map<String, String> catalogConfig = catalogConfigMap(params);
+ String partition = params.getRequired(PARTITION);
+ boolean overwrite = params.getBoolean(OVERWRITE, false);
+ return Optional.of(
+ new CompactChainTableAction(database, table, catalogConfig,
partition, overwrite));
+ }
+
+ @Override
+ public void printHelp() {
+ System.out.println("Action \"compact_chain_table\" compacts a chain
table partition.");
+ System.out.println();
+ System.out.println("Syntax:");
+ System.out.println(
+ " compact_chain_table --warehouse <warehouse_path> --database
<db> --table <table> --partition <partition> [--overwrite true]");
+ System.out.println("Examples:");
+ System.out.println(
+ " compact_chain_table --warehouse hdfs:///path/to/warehouse
--database test_db --table test_table --partition dt=20250810");
+ System.out.println(
+ " compact_chain_table --warehouse hdfs:///path/to/warehouse
--database test_db --table test_table --partition dt=20250810,hour=22
--overwrite true");
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CompactChainTableProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CompactChainTableProcedure.java
new file mode 100644
index 0000000000..0e23d3322d
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CompactChainTableProcedure.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.procedure;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.action.CompactChainTableAction;
+
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+
+import java.util.Map;
+
+/**
+ * Procedure to compact chain table. Usage:
+ *
+ * <pre><code>
+ * -- Compact chain table, overwrite default is false
+ * CALL sys.compact_chain_table('db.table', 'dt=20250810,hour=22', [true])
+ * </code></pre>
+ */
+public class CompactChainTableProcedure extends ProcedureBase {
+
+ public static final String IDENTIFIER = "compact_chain_table";
+
+ @ProcedureHint(
+ argument = {
+ @ArgumentHint(name = "table", type = @DataTypeHint("STRING")),
+ @ArgumentHint(name = "partition", type =
@DataTypeHint("STRING")),
+ @ArgumentHint(
+ name = "overwrite",
+ type = @DataTypeHint("BOOLEAN"),
+ isOptional = true)
+ })
+ public String[] call(
+ ProcedureContext procedureContext, String tableId, String
partition, Boolean overwrite)
+ throws Exception {
+ Map<String, String> catalogOptions = catalog.options();
+ Identifier identifier = Identifier.fromString(tableId);
+ CompactChainTableAction action =
+ new CompactChainTableAction(
+ identifier.getDatabaseName(),
+ identifier.getObjectName(),
+ catalogOptions,
+ partition,
+ overwrite);
+ return execute(
+ procedureContext, action, "Compact chain table Job : " +
identifier.getFullName());
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
index d00d2843ff..b80b52be77 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
@@ -90,12 +90,14 @@ public class FlinkSourceBuilder {
@Nullable private Long limit;
@Nullable private WatermarkStrategy<RowData> watermarkStrategy;
@Nullable private DynamicPartitionFilteringInfo
dynamicPartitionFilteringInfo;
+ private boolean skipPreloadTargetSnapshot;
public FlinkSourceBuilder(Table table) {
this.table = table;
this.sourceName = table.name();
this.conf = Options.fromMap(table.options());
this.unordered = unordered(table);
+ this.skipPreloadTargetSnapshot = false;
}
private static boolean unordered(Table table) {
@@ -186,6 +188,11 @@ public class FlinkSourceBuilder {
return this;
}
+ public FlinkSourceBuilder withSkipPreloadTargetSnapshot(boolean skip) {
+ this.skipPreloadTargetSnapshot = skip;
+ return this;
+ }
+
private ReadBuilder createReadBuilder(@Nullable
org.apache.paimon.types.RowType readType) {
ReadBuilder readBuilder = table.newReadBuilder();
if (readType != null) {
@@ -213,7 +220,8 @@ public class FlinkSourceBuilder {
options.get(FlinkConnectorOptions.SCAN_SPLIT_ENUMERATOR_ASSIGN_MODE),
dynamicPartitionFilteringInfo,
outerProject(),
- options.get(CoreOptions.BLOB_AS_DESCRIPTOR)));
+ options.get(CoreOptions.BLOB_AS_DESCRIPTOR),
+ skipPreloadTargetSnapshot));
}
private DataStream<RowData> buildContinuousFileSource() {
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/StaticFileStoreSource.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/StaticFileStoreSource.java
index 775100aa81..940feb9d31 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/StaticFileStoreSource.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/StaticFileStoreSource.java
@@ -23,6 +23,7 @@ import org.apache.paimon.flink.metrics.FlinkMetricRegistry;
import org.apache.paimon.flink.source.assigners.FIFOSplitAssigner;
import org.apache.paimon.flink.source.assigners.PreAssignSplitAssigner;
import org.apache.paimon.flink.source.assigners.SplitAssigner;
+import org.apache.paimon.table.ChainGroupReadTable;
import org.apache.paimon.table.source.InnerTableScan;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.TableScan;
@@ -49,13 +50,23 @@ public class StaticFileStoreSource extends FlinkSource {
@Nullable private final DynamicPartitionFilteringInfo
dynamicPartitionFilteringInfo;
+ private final boolean skipPreloadTargetSnapshot;
+
public StaticFileStoreSource(
ReadBuilder readBuilder,
@Nullable Long limit,
int splitBatchSize,
SplitAssignMode splitAssignMode,
boolean blobAsDescriptor) {
- this(readBuilder, limit, splitBatchSize, splitAssignMode, null, null,
blobAsDescriptor);
+ this(
+ readBuilder,
+ limit,
+ splitBatchSize,
+ splitAssignMode,
+ null,
+ null,
+ blobAsDescriptor,
+ false);
}
public StaticFileStoreSource(
@@ -65,11 +76,13 @@ public class StaticFileStoreSource extends FlinkSource {
SplitAssignMode splitAssignMode,
@Nullable DynamicPartitionFilteringInfo
dynamicPartitionFilteringInfo,
@Nullable NestedProjectedRowData rowData,
- boolean blobAsDescriptor) {
+ boolean blobAsDescriptor,
+ boolean skipPreloadTargetSnapshot) {
super(readBuilder, limit, rowData, blobAsDescriptor);
this.splitBatchSize = splitBatchSize;
this.splitAssignMode = splitAssignMode;
this.dynamicPartitionFilteringInfo = dynamicPartitionFilteringInfo;
+ this.skipPreloadTargetSnapshot = skipPreloadTargetSnapshot;
}
@Override
@@ -92,6 +105,9 @@ public class StaticFileStoreSource extends FlinkSource {
private List<FileStoreSourceSplit> getSplits(SplitEnumeratorContext
context) {
FileStoreSourceSplitGenerator splitGenerator = new
FileStoreSourceSplitGenerator();
TableScan scan = readBuilder.newScan();
+ if (skipPreloadTargetSnapshot && scan instanceof
ChainGroupReadTable.ChainTableBatchScan) {
+ ((ChainGroupReadTable.ChainTableBatchScan)
scan).skipPreloadTargetSnapshot();
+ }
// register scan metrics
if (context.metricGroup() != null) {
((InnerTableScan) scan)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/SystemTableSource.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/SystemTableSource.java
index d9ede90a14..a78f44bc6f 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/SystemTableSource.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/SystemTableSource.java
@@ -114,8 +114,8 @@ public class SystemTableSource extends FlinkTableSource {
Boolean.parseBoolean(
table.options()
.getOrDefault(
-
CoreOptions.BLOB_AS_DESCRIPTOR.key(),
- "false")));
+
CoreOptions.BLOB_AS_DESCRIPTOR.key(), "false")),
+ false);
}
return new PaimonDataStreamScanProvider(
source.getBoundedness() == Boundedness.BOUNDED,
diff --git
a/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
b/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
index 26a1a2b9cb..b9696067ae 100644
---
a/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
+++
b/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
@@ -51,10 +51,12 @@ org.apache.paimon.flink.action.RescaleActionFactory
org.apache.paimon.flink.action.CloneActionFactory
org.apache.paimon.flink.action.DataEvolutionMergeIntoActionFactory
org.apache.paimon.flink.action.ReassignRowIdActionFactory
+org.apache.paimon.flink.action.CompactChainTableActionFactory
### procedure factories
org.apache.paimon.flink.procedure.CompactDatabaseProcedure
org.apache.paimon.flink.procedure.CompactProcedure
+org.apache.paimon.flink.procedure.CompactChainTableProcedure
org.apache.paimon.flink.procedure.RewriteFileIndexProcedure
org.apache.paimon.flink.procedure.CreateTagProcedure
org.apache.paimon.flink.procedure.CreateTagFromTimestampProcedure
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactChainTableActionITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactChainTableActionITCase.java
new file mode 100644
index 0000000000..c9bca5b6b7
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactChainTableActionITCase.java
@@ -0,0 +1,181 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.action;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableRead;
+
+import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableMap;
+
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.apache.paimon.flink.util.ReadWriteTableTestUtil.bEnv;
+import static org.apache.paimon.flink.util.ReadWriteTableTestUtil.buildDdl;
+import static org.apache.paimon.flink.util.ReadWriteTableTestUtil.init;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for {@link CompactChainTableAction}. */
+public class CompactChainTableActionITCase extends CompactActionITCaseBase {
+
+ private static final List<String> FIELDS_SPEC =
+ Arrays.asList("t1 INT", "t2 INT", "t3 INT", "dt STRING");
+ private static final List<String> PRIMARY_KEYS = Arrays.asList("dt", "t1");
+ private static final List<String> PARTITION_KEYS =
Collections.singletonList("dt");
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ init(warehouse);
+ prepareChainTable(Collections.emptyMap());
+ }
+
+ private void prepareChainTable(Map<String, String> extraOptions) throws
Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put("bucket", "1");
+ options.put("bucket-key", "t1");
+ options.put("sequence.field", "t2");
+ options.put("merge-engine", "deduplicate");
+ options.put("chain-table.enabled", "true");
+ options.put("partition.timestamp-pattern", "$dt");
+ options.put("partition.timestamp-formatter", "yyyyMMdd");
+ options.putAll(extraOptions);
+
+ String ddl = buildDdl(tableName, FIELDS_SPEC, PRIMARY_KEYS,
PARTITION_KEYS, options);
+ bEnv.executeSql(ddl).await();
+
+ // Create snapshot / delta branches and configure fallback branches.
+ bEnv.executeSql(
+ String.format(
+ "CALL sys.create_branch('%s.%s', 'snapshot')",
database, tableName))
+ .await();
+ bEnv.executeSql(
+ String.format(
+ "CALL sys.create_branch('%s.%s', 'delta')",
database, tableName))
+ .await();
+ bEnv.executeSql(
+ String.format(
+ "ALTER TABLE `%s`.`%s` SET
('scan.fallback-snapshot-branch' = 'snapshot', 'scan.fallback-delta-branch' =
'delta')",
+ database, tableName))
+ .await();
+ bEnv.executeSql(
+ String.format(
+ "ALTER TABLE `%s`.`%s$branch_snapshot` SET
('scan.fallback-snapshot-branch' = 'snapshot', 'scan.fallback-delta-branch' =
'delta')",
+ database, tableName))
+ .await();
+ bEnv.executeSql(
+ String.format(
+ "ALTER TABLE `%s`.`%s$branch_delta` SET
('scan.fallback-snapshot-branch' = 'snapshot', 'scan.fallback-delta-branch' =
'delta')",
+ database, tableName))
+ .await();
+ }
+
+ private void writeBranch(String branch, GenericRow... rows) throws
Exception {
+ FileStoreTable branchTable =
getFileStoreTable(tableName).switchToBranch(branch);
+ String commitUser = UUID.randomUUID().toString();
+ try (org.apache.paimon.table.sink.StreamTableWrite write =
+
branchTable.newStreamWriteBuilder().withCommitUser(commitUser).newWrite();
+ org.apache.paimon.table.sink.StreamTableCommit commit =
+ branchTable
+ .newStreamWriteBuilder()
+ .withCommitUser(commitUser)
+ .newCommit()) {
+ for (GenericRow row : rows) {
+ write.write(row);
+ }
+ commit.commit(0L, write.prepareCommit(true, 0L));
+ }
+ }
+
+ private List<String> readSnapshot(Map<String, String> partitionFilter)
throws Exception {
+ FileStoreTable snapshotTable =
getFileStoreTable(tableName).switchToBranch("snapshot");
+ ReadBuilder readBuilder =
+
snapshotTable.newReadBuilder().withPartitionFilter(partitionFilter);
+ TableRead read = readBuilder.newRead();
+ List<Split> splits = readBuilder.newScan().plan().splits();
+ return getResult(read, splits, ROW_TYPE);
+ }
+
+ private void runAction(String partition, boolean overwrite) throws
Exception {
+ CompactChainTableAction action =
+ createAction(
+ CompactChainTableAction.class,
+ "compact_chain_table",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName,
+ "--partition",
+ partition,
+ "--overwrite",
+ String.valueOf(overwrite));
+ StreamExecutionEnvironment env =
streamExecutionEnvironmentBuilder().batchMode().build();
+ action.withStreamExecutionEnvironment(env).run();
+ }
+
+ @Test
+ public void testCompactChainTableBasic() throws Exception {
+ // Compact empty partition
+ runAction("dt=20260708", false);
+
+ writeBranch(
+ "snapshot",
+ rowData(1, 1, 1, BinaryString.fromString("20260708")),
+ rowData(2, 1, 1, BinaryString.fromString("20260708")));
+ writeBranch("delta", rowData(3, 1, 1,
BinaryString.fromString("20260709")));
+ // Compact a partition that only exists in the delta branch.
+ runAction("dt=20260709", false);
+ // Compact empty partition
+ runAction("dt=20260710", false);
+
+ assertThat(readSnapshot(ImmutableMap.of("dt", "20260709")))
+ .containsExactlyInAnyOrder(
+ "+I[1, 1, 1, 20260709]", "+I[2, 1, 1, 20260709]",
"+I[3, 1, 1, 20260709]");
+ }
+
+ @Test
+ public void testCompactChainTableOverwrite() throws Exception {
+ writeBranch("snapshot", rowData(1, 1, 1,
BinaryString.fromString("20260708")));
+ writeBranch("snapshot", rowData(2, 1, 1,
BinaryString.fromString("20260709")));
+ writeBranch("delta", rowData(3, 1, 1,
BinaryString.fromString("20260709")));
+
+ // Without overwrite, the existing snapshot partition is skipped.
+ runAction("dt=20260709", false);
+ assertThat(readSnapshot(ImmutableMap.of("dt", "20260709")))
+ .containsExactlyInAnyOrder("+I[2, 1, 1, 20260709]");
+
+ // With overwrite, merge the snapshot anchor and the delta data.
+ runAction("dt=20260709", true);
+ assertThat(readSnapshot(ImmutableMap.of("dt", "20260709")))
+ .containsExactlyInAnyOrder("+I[1, 1, 1, 20260709]", "+I[3, 1,
1, 20260709]");
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactChainTableProcedureITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactChainTableProcedureITCase.java
new file mode 100644
index 0000000000..aaf2963e7f
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactChainTableProcedureITCase.java
@@ -0,0 +1,395 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.procedure;
+
+import org.apache.paimon.flink.CatalogITCaseBase;
+
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for {@link CompactChainTableProcedure}. */
+public class CompactChainTableProcedureITCase extends CatalogITCaseBase {
+
+ private List<String> collectResult(String query) throws Exception {
+ List<String> result = new ArrayList<>();
+ try (CloseableIterator<Row> it = tEnv.executeSql(query).collect()) {
+ while (it.hasNext()) {
+ result.add(it.next().toString());
+ }
+ }
+ return result;
+ }
+
+ private void createChainTableWithDate(String tableName) {
+ sql(
+ "CREATE TABLE %s ("
+ + " t1 BIGINT,"
+ + " t2 BIGINT,"
+ + " t3 STRING,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,t1',"
+ + " 'bucket-key' = 't1',"
+ + " 'bucket' = '1',"
+ + " 'sequence.field' = 't2',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
+ + ")",
+ tableName);
+ }
+
+ private void setupChainTableBranches(String tableName) {
+ String db = tEnv.getCurrentDatabase();
+ sql("CALL sys.create_branch('%s.%s', 'snapshot')", db, tableName);
+ sql("CALL sys.create_branch('%s.%s', 'delta')", db, tableName);
+
+ sql(
+ "ALTER TABLE %s SET ("
+ + " 'scan.fallback-snapshot-branch' = 'snapshot',"
+ + " 'scan.fallback-delta-branch' = 'delta'"
+ + ")",
+ tableName);
+ sql(
+ "ALTER TABLE `%s$branch_snapshot` SET ("
+ + " 'scan.fallback-snapshot-branch' = 'snapshot',"
+ + " 'scan.fallback-delta-branch' = 'delta'"
+ + ")",
+ tableName);
+ sql(
+ "ALTER TABLE `%s$branch_delta` SET ("
+ + " 'scan.fallback-snapshot-branch' = 'snapshot',"
+ + " 'scan.fallback-delta-branch' = 'delta'"
+ + ")",
+ tableName);
+ }
+
+ @Nullable
+ @Override
+ protected Boolean sqlSyncMode() {
+ return true;
+ }
+
+ @Test
+ public void testCompactChainTableBasic() throws Exception {
+ // Create chain table
+ createChainTableWithDate("chain_compact_t1");
+ setupChainTableBranches("chain_compact_t1");
+
+ // Compact empty partition
+ sql("CALL sys.compact_chain_table('default.chain_compact_t1',
'dt=20260224')");
+
+ // Insert snapshot data
+ sql(
+ "INSERT INTO `chain_compact_t1$branch_snapshot` PARTITION (dt
= '20260222') VALUES (0, 1, '0')");
+ sql(
+ "INSERT INTO `chain_compact_t1$branch_snapshot` PARTITION (dt
= '20260223') VALUES (1, 1, '1')");
+
+ // Insert delta data
+ sql(
+ "INSERT INTO `chain_compact_t1$branch_delta` PARTITION (dt =
'20260224') VALUES (2, 2, '2')");
+
+ // Before compaction: verify chain read shows snapshot + delta
+ assertThat(collectResult("SELECT * FROM chain_compact_t1 WHERE dt =
'20260224'"))
+ .containsExactlyInAnyOrder("+I[1, 1, 1, 20260224]", "+I[2, 2,
2, 20260224]");
+
+ sql("CALL sys.compact_chain_table('default.chain_compact_t1',
'dt=20260224')");
+
+ // After compaction: verify snapshot branch has merged data
+ assertThat(collectResult("SELECT * FROM chain_compact_t1 WHERE dt =
'20260224'"))
+ .containsExactlyInAnyOrder("+I[1, 1, 1, 20260224]", "+I[2, 2,
2, 20260224]");
+ assertThat(
+ collectResult(
+ "SELECT * FROM
chain_compact_t1$branch_snapshot WHERE dt = '20260223'"))
+ .containsExactlyInAnyOrder("+I[1, 1, 1, 20260223]");
+ assertThat(
+ collectResult(
+ "SELECT * FROM
`chain_compact_t1$branch_snapshot` WHERE dt = '20260224'"))
+ .containsExactlyInAnyOrder("+I[1, 1, 1, 20260224]", "+I[2, 2,
2, 20260224]");
+ }
+
+ @Test
+ public void testCompactChainTableOverwrite() throws Exception {
+ // Create chain table
+ createChainTableWithDate("chain_compact_t2");
+ setupChainTableBranches("chain_compact_t2");
+
+ sql("INSERT INTO `chain_compact_t2` PARTITION (dt = '20260222') VALUES
(1, 1, '1')");
+
+ // Insert snapshot data
+ sql(
+ "INSERT INTO `chain_compact_t2$branch_snapshot` PARTITION (dt
= '20260223') VALUES (2, 1, '2')");
+ sql(
+ "INSERT INTO `chain_compact_t2$branch_snapshot` PARTITION (dt
= '20260224') VALUES (3, 1, '3')");
+ sql(
+ "INSERT INTO `chain_compact_t2$branch_snapshot` PARTITION (dt
= '20260225') VALUES (3, 2, '3-1')");
+
+ // Insert delta data
+ sql(
+ "INSERT INTO `chain_compact_t2$branch_delta` PARTITION (dt =
'20260225') VALUES (4, 2, '4')");
+
+ // First call should skip because partition exists
+ sql("CALL sys.compact_chain_table('default.chain_compact_t2',
'dt=20260225')");
+
+ assertThat(collectResult("SELECT * FROM chain_compact_t2 WHERE dt =
'20260225'"))
+ .containsExactlyInAnyOrder("+I[3, 2, 3-1, 20260225]");
+
+ sql("CALL sys.compact_chain_table('default.chain_compact_t2',
'dt=20260225', true)");
+
+ // Check snapshots commit_kind
+ assertThat(
+ collectResult(
+ "SELECT snapshot_id, commit_kind FROM
`chain_compact_t2$branch_snapshot$snapshots`"))
+ .containsExactlyInAnyOrder(
+ "+I[1, APPEND]", "+I[2, APPEND]", "+I[3, APPEND]",
"+I[4, OVERWRITE]");
+
+ // Verify snapshot branch now has the data
+ assertThat(
+ collectResult(
+ "SELECT * FROM
`chain_compact_t2$branch_snapshot` WHERE dt = '20260225'"))
+ .containsExactlyInAnyOrder("+I[3, 1, 3, 20260225]", "+I[4, 2,
4, 20260225]");
+ assertThat(
+ collectResult(
+ "SELECT * FROM
`chain_compact_t2$branch_snapshot` WHERE dt <> '20260225'"))
+ .containsExactlyInAnyOrder("+I[2, 1, 2, 20260223]", "+I[3, 1,
3, 20260224]");
+ }
+
+ @Test
+ public void testCompactChainTableMultiplePartitions() throws Exception {
+ sql(
+ "CREATE TABLE chain_compact_t3 ("
+ + " t1 BIGINT,"
+ + " t2 BIGINT,"
+ + " t3 STRING,"
+ + " dt STRING,"
+ + " hr STRING"
+ + ") PARTITIONED BY (dt, hr) WITH ("
+ + " 'primary-key' = 'dt,hr,t1',"
+ + " 'bucket-key' = 't1',"
+ + " 'bucket' = '2',"
+ + " 'sequence.field' = 't2',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt $hr:00:00',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd
HH:mm:ss'"
+ + ")");
+ setupChainTableBranches("chain_compact_t3");
+
+ // Write snapshot branch data
+ sql(
+ "INSERT INTO `chain_compact_t3$branch_snapshot` PARTITION (dt
= '20250810', hr = '20') VALUES (0, 1, '0')");
+ sql(
+ "INSERT INTO `chain_compact_t3$branch_snapshot` PARTITION (dt
= '20250810', hr = '22') VALUES (1, 1, '1'),(2, 1, '1')");
+
+ // Write delta branch data
+ sql(
+ "INSERT INTO `chain_compact_t3$branch_delta` PARTITION (dt =
'20250810', hr = '21') VALUES (1, 1, '1'),(2, 1, '1')");
+ sql(
+ "INSERT INTO `chain_compact_t3$branch_delta` PARTITION (dt =
'20250810', hr = '22') VALUES (1, 2, '1-1'),(3, 1, '1')");
+ sql(
+ "INSERT INTO `chain_compact_t3$branch_delta` PARTITION (dt =
'20250810', hr = '23') VALUES (2, 2, '1-1'),(4, 1, '1')");
+
+ // Compact partition 20250810/hr=23
+ sql("CALL sys.compact_chain_table('default.chain_compact_t3', 'dt =
20250810, hr = 23')");
+
+ // Verify chain read still works correctly
+ assertThat(
+ collectResult(
+ "SELECT * FROM chain_compact_t3 WHERE dt =
'20250810' AND hr = '23'"))
+ .containsExactlyInAnyOrder(
+ "+I[1, 1, 1, 20250810, 23]",
+ "+I[2, 2, 1-1, 20250810, 23]",
+ "+I[4, 1, 1, 20250810, 23]");
+
+ // Write more snapshot and delta data
+ sql(
+ "INSERT INTO `chain_compact_t3$branch_snapshot` PARTITION (dt
= '20250811', hr = '00') VALUES (1, 2, '1-1'),(2, 2, '1-1'),(3, 2, '1-1'),(4,
2, '1-1')");
+ sql(
+ "INSERT INTO `chain_compact_t3$branch_snapshot` PARTITION (dt
= '20250811', hr = '02') VALUES (1, 2, '1-1'),(2, 2, '1-1'),(3, 2, '1-1'),(4,
2, '1-1'),(5, 1, '1'),(6, 1, '1')");
+ sql(
+ "INSERT INTO `chain_compact_t3$branch_delta` PARTITION (dt =
'20250811', hr = '00') VALUES (3, 2, '1-1'),(4, 2, '1-1')");
+ sql(
+ "INSERT INTO `chain_compact_t3$branch_delta` PARTITION (dt =
'20250811', hr = '01') VALUES (5, 1, '1'),(6, 1, '1')");
+ sql(
+ "INSERT INTO `chain_compact_t3$branch_delta` PARTITION (dt =
'20250811', hr = '02') VALUES (5, 2, '1-1'),(6, 2, '1-1')");
+
+ // Compact partition 20250811/hr=02 without overwrite
+ sql("CALL sys.compact_chain_table('default.chain_compact_t3',
'dt=20250811,hr=02')");
+
+ // Verify data is unchanged after skip
+ assertThat(
+ collectResult(
+ "SELECT * FROM chain_compact_t3 WHERE dt =
'20250811' AND hr = '02'"))
+ .containsExactlyInAnyOrder(
+ "+I[1, 2, 1-1, 20250811, 02]",
+ "+I[2, 2, 1-1, 20250811, 02]",
+ "+I[3, 2, 1-1, 20250811, 02]",
+ "+I[4, 2, 1-1, 20250811, 02]",
+ "+I[5, 1, 1, 20250811, 02]",
+ "+I[6, 1, 1, 20250811, 02]");
+
+ // Compact with overwrite to test overwrite path
+ sql("CALL sys.compact_chain_table('default.chain_compact_t3',
'dt=20250811,hr=02', true)");
+
+ // Check snapshots commit_kind
+ assertThat(
+ collectResult(
+ "SELECT snapshot_id, commit_kind FROM
`chain_compact_t3$branch_snapshot$snapshots`"))
+ .containsExactlyInAnyOrder(
+ "+I[1, APPEND]",
+ "+I[2, APPEND]",
+ "+I[3, APPEND]",
+ "+I[4, APPEND]",
+ "+I[5, APPEND]",
+ "+I[6, OVERWRITE]");
+
+ // Check all snapshot partition data
+ assertThat(
+ collectResult(
+ "SELECT * FROM chain_compact_t3 WHERE dt =
'20250811' AND hr = '02'"))
+ .containsExactlyInAnyOrder(
+ "+I[1, 2, 1-1, 20250811, 02]",
+ "+I[2, 2, 1-1, 20250811, 02]",
+ "+I[3, 2, 1-1, 20250811, 02]",
+ "+I[4, 2, 1-1, 20250811, 02]",
+ "+I[5, 2, 1-1, 20250811, 02]",
+ "+I[6, 2, 1-1, 20250811, 02]");
+ assertThat(
+ collectResult(
+ "SELECT * FROM chain_compact_t3 WHERE dt =
'20250811' AND hr = '00'"))
+ .containsExactlyInAnyOrder(
+ "+I[1, 2, 1-1, 20250811, 00]",
+ "+I[2, 2, 1-1, 20250811, 00]",
+ "+I[3, 2, 1-1, 20250811, 00]",
+ "+I[4, 2, 1-1, 20250811, 00]");
+ assertThat(
+ collectResult(
+ "SELECT * FROM
`chain_compact_t3$branch_snapshot` WHERE dt = '20250810' AND hr = '23'"))
+ .containsExactlyInAnyOrder(
+ "+I[1, 1, 1, 20250810, 23]",
+ "+I[2, 2, 1-1, 20250810, 23]",
+ "+I[4, 1, 1, 20250810, 23]");
+ assertThat(
+ collectResult(
+ "SELECT * FROM
`chain_compact_t3$branch_snapshot` WHERE dt = '20250810' AND hr = '22'"))
+ .containsExactlyInAnyOrder(
+ "+I[1, 1, 1, 20250810, 22]", "+I[2, 1, 1, 20250810,
22]");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testCompactChainTableWithGroupPartition(boolean overwrite)
throws Exception {
+ sql(
+ String.format(
+ "CREATE TABLE chain_compact_t4 ("
+ + " t1 BIGINT,"
+ + " t2 BIGINT,"
+ + " t3 STRING,"
+ + " region STRING,"
+ + " dt STRING,"
+ + " hr STRING"
+ + ") PARTITIONED BY (region, dt, hr) "
+ + " WITH ("
+ + " 'dynamic-partition-overwrite' = '%s',"
+ + " 'primary-key' = 'region,dt,hr,t1',"
+ + " 'bucket-key' = 't1',"
+ + " 'bucket' = '1',"
+ + " 'sequence.field' = 't2',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt
$hr:00:00',"
+ + " 'partition.timestamp-formatter' =
'yyyyMMdd HH:mm:ss',"
+ + " 'chain-table.chain-partition-keys' =
'dt,hr'"
+ + ")",
+ overwrite));
+ setupChainTableBranches("chain_compact_t4");
+
+ // Write snapshot branch data
+ sql(
+ "INSERT INTO `chain_compact_t4$branch_snapshot` PARTITION
(region='CN', dt = '20250810', hr = '20') VALUES (1, 1, '1')");
+ sql(
+ "INSERT INTO `chain_compact_t4$branch_snapshot` PARTITION
(region='CN', dt = '20250810', hr = '21') VALUES (2, 1, '1')");
+ sql(
+ "INSERT INTO `chain_compact_t4$branch_snapshot` PARTITION
(region='CN', dt = '20250810', hr = '22') VALUES (3, 1, '1')");
+ sql(
+ "INSERT INTO `chain_compact_t4$branch_snapshot` PARTITION
(region='UK', dt = '20250810', hr = '21') VALUES (21, 1, '1')");
+ sql(
+ "INSERT INTO `chain_compact_t4$branch_snapshot` PARTITION
(region='FR', dt = '20250810', hr = '22') VALUES (31, 1, '1')");
+ sql(
+ "INSERT INTO `chain_compact_t4$branch_snapshot` PARTITION
(region='CA', dt = '20250810', hr = '21') VALUES (41, 1, '1')");
+
+ // Write delta branch data
+ sql(
+ "INSERT INTO `chain_compact_t4$branch_delta` PARTITION
(region='CN', dt = '20250810', hr = '22') VALUES (4, 1, '1')");
+ sql(
+ "INSERT INTO `chain_compact_t4$branch_delta` PARTITION
(region='US', dt = '20250810', hr = '22') VALUES (11, 1, '1')");
+ sql(
+ "INSERT INTO `chain_compact_t4$branch_delta` PARTITION
(region='UK', dt = '20250810', hr = '22') VALUES (22, 1, '1')");
+
+ assertThat(
+ collectResult(
+ "SELECT * FROM
`chain_compact_t4$branch_snapshot` WHERE dt = '20250810' AND hr = '22'"))
+ .containsExactlyInAnyOrder(
+ "+I[3, 1, 1, CN, 20250810, 22]", "+I[31, 1, 1, FR,
20250810, 22]");
+
+ sql("CALL sys.compact_chain_table('default.chain_compact_t4',
'dt=20250810,hr=22', true)");
+
+ assertThat(
+ collectResult(
+ "select snapshot_id,commit_kind from
`chain_compact_t4$branch_snapshot$snapshots`"))
+ .containsExactlyInAnyOrder(
+ "+I[1, APPEND]",
+ "+I[2, APPEND]",
+ "+I[3, APPEND]",
+ "+I[4, APPEND]",
+ "+I[5, APPEND]",
+ "+I[6, APPEND]",
+ "+I[7, OVERWRITE]");
+
+ assertThat(
+ collectResult(
+ "SELECT * FROM
`chain_compact_t4$branch_snapshot` WHERE dt = '20250810' AND hr = '21'"))
+ .containsExactlyInAnyOrder(
+ "+I[2, 1, 1, CN, 20250810, 21]",
+ "+I[21, 1, 1, UK, 20250810, 21]",
+ "+I[41, 1, 1, CA, 20250810, 21]");
+
+ assertThat(
+ collectResult(
+ "SELECT * FROM
`chain_compact_t4$branch_snapshot` WHERE dt = '20250810' AND hr = '22'"))
+ .containsExactlyInAnyOrder(
+ "+I[2, 1, 1, CN, 20250810, 22]",
+ "+I[4, 1, 1, CN, 20250810, 22]",
+ "+I[11, 1, 1, US, 20250810, 22]",
+ "+I[21, 1, 1, UK, 20250810, 22]",
+ "+I[22, 1, 1, UK, 20250810, 22]",
+ "+I[31, 1, 1, FR, 20250810, 22]");
+ }
+}
