[flink] Support compact_chain_table procedure

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>

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;
+    }
+}

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<>());
+    }
+}

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");
+    }
+}

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;
+    }
+}