wzhero1 commented on code in PR #7027:
URL: https://github.com/apache/paimon/pull/7027#discussion_r2973183685
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/ExpireSnapshotsAction.java:
##########
@@ -50,12 +93,135 @@ public ExpireSnapshotsAction(
this.olderThan = olderThan;
this.maxDeletes = maxDeletes;
this.options = options;
+ this.parallelism = resolveParallelism(parallelism);
+ }
+
+ private int resolveParallelism(Integer parallelism) {
+ if (parallelism != null) {
+ return parallelism;
+ }
+ int envParallelism = env.getParallelism();
+ return envParallelism > 0 ? envParallelism : 1;
}
+ @Override
+ public void run() throws Exception {
+ if (forceStartFlinkJob) {
+ // Parallel mode: build custom multi-parallelism Flink pipeline
+ build();
+ execute(this.getClass().getSimpleName());
+ } else {
+ // Default: ActionBase handles LocalAction → executeLocally()
+ super.run();
+ }
+ }
+
+ @Override
public void executeLocally() throws Exception {
ExpireSnapshotsProcedure expireSnapshotsProcedure = new
ExpireSnapshotsProcedure();
expireSnapshotsProcedure.withCatalog(catalog);
expireSnapshotsProcedure.call(
null, database + "." + table, retainMax, retainMin, olderThan,
maxDeletes, options);
}
+
+ @Override
+ public void build() throws Exception {
+ Identifier identifier = new Identifier(database, table);
+
+ // Prepare table with dynamic options
+ HashMap<String, String> dynamicOptions = new HashMap<>();
+ ProcedureUtils.putAllOptions(dynamicOptions, options);
+ FileStoreTable fileStoreTable =
+ (FileStoreTable)
catalog.getTable(identifier).copy(dynamicOptions);
+
+ // Build expire config
+ CoreOptions tableOptions = fileStoreTable.store().options();
+ ExpireConfig expireConfig =
+ ProcedureUtils.fillInSnapshotOptions(
+ tableOptions, retainMax, retainMin, olderThan,
maxDeletes)
+ .build();
+
+ // Create planner using factory method
+ ExpireSnapshotsPlanner planner =
ExpireSnapshotsPlanner.create(fileStoreTable);
+
+ // Plan the expiration
+ ExpireSnapshotsPlan plan = planner.plan(expireConfig);
+ if (plan.isEmpty()) {
+ LOG.info("No snapshots to expire");
+ return;
+ }
+
+ LOG.info(
+ "Planning to expire {} snapshots, range=[{}, {})",
+ plan.endExclusiveId() - plan.beginInclusiveId(),
+ plan.beginInclusiveId(),
+ plan.endExclusiveId());
+
+ // Build worker phase
+ DataStream<DeletionReport> reports = buildWorkerPhase(plan,
identifier, expireConfig);
+
+ // Build sink phase
+ buildSinkPhase(reports, plan, identifier, expireConfig);
+ }
+
+ /**
+ * Build the worker phase of the Flink job.
+ *
+ * <p>Workers process data file and changelog file deletion tasks in
parallel. Tasks are
+ * partitioned by snapshot range to ensure:
+ *
+ * <ul>
+ * <li>Same snapshot range tasks are processed by the same worker
+ * <li>Within each worker, data files are deleted before changelog files
+ * <li>Cache locality is maximized (adjacent snapshots often share
manifest files)
+ * </ul>
+ */
+ private DataStream<DeletionReport> buildWorkerPhase(
+ ExpireSnapshotsPlan plan, Identifier identifier, ExpireConfig
expireConfig) {
+ // Partition by snapshot range: each worker gets a contiguous range of
snapshots
+ // with dataFileTasks first, then changelogFileTasks
+ List<List<SnapshotExpireTask>> partitionedGroups =
+ plan.partitionTasksBySnapshotRange(parallelism);
+
+ DataStreamSource<List<SnapshotExpireTask>> source =
+ env.fromCollection(partitionedGroups).setParallelism(1);
+
+ return source.rebalance()
+ .flatMap(
+ new RangePartitionedExpireFunction(
+ catalogOptions.toMap(),
+ identifier,
+ plan.protectionSet().taggedSnapshots(),
+ expireConfig.isChangelogDecoupled()))
+ // Use JavaTypeInfo to ensure proper Java serialization of
DeletionReport,
+ // avoiding Kryo's FieldSerializer which cannot handle
BinaryRow correctly.
+ // This approach is compatible with both Flink 1.x and 2.x.
+ .returns(new JavaTypeInfo<>(DeletionReport.class))
+ .setParallelism(parallelism)
+ .name("RangePartitionedExpire");
+ }
+
+ /**
+ * Build the sink phase of the Flink job.
+ *
+ * <p>The sink collects deletion reports from workers, then serially
deletes manifests and
+ * snapshot metadata files to avoid concurrent deletion issues.
+ */
+ private void buildSinkPhase(
+ DataStream<DeletionReport> reports,
+ ExpireSnapshotsPlan plan,
+ Identifier identifier,
+ ExpireConfig expireConfig) {
+ reports.sinkTo(
+ new SnapshotExpireSink(
+ catalogOptions.toMap(),
+ identifier,
+ plan.endExclusiveId(),
+ plan.protectionSet().manifestSkippingSet(),
+ plan.manifestTasks(),
+ plan.snapshotFileTasks(),
+ expireConfig.isChangelogDecoupled()))
+ .setParallelism(1)
+ .name("SnapshotExpire");
Review Comment:
Renamed to "SnapshotExpireCommit" to distinguish from the upstream worker
phase.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/expire/RangePartitionedExpireFunction.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.expire;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.operation.SnapshotDeletion;
+import org.apache.paimon.operation.expire.DeletionReport;
+import org.apache.paimon.operation.expire.ExpireSnapshotsExecutor;
+import org.apache.paimon.operation.expire.SnapshotExpireTask;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.FileStoreTable;
+
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.functions.RichFlatMapFunction;
+import org.apache.flink.util.Collector;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Flink flatMap function for range-partitioned snapshot expiration (worker
phase).
+ *
+ * <p>This function processes a batch of {@link SnapshotExpireTask}s that
belong to the same
+ * contiguous range. Each subtask receives a list of tasks (e.g., subtask 0
gets snap 1-4, subtask 1
+ * gets snap 5-8) and processes them sequentially in order.
+ *
+ * <p>Processing tasks in order within each subtask maximizes cache locality
since adjacent
+ * snapshots often share manifest files.
+ *
+ * <p>In worker phase, this function only deletes data files and changelog
data files. Manifest and
+ * snapshot metadata deletion is deferred to the sink phase to avoid
concurrent deletion issues.
+ *
+ * <p>This function uses {@link ExpireSnapshotsExecutor#execute} which loads
tag data files
+ * on-demand with internal caching.
+ */
+public class RangePartitionedExpireFunction
+ extends RichFlatMapFunction<List<SnapshotExpireTask>, DeletionReport> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final Map<String, String> catalogConfig;
+ private final Identifier identifier;
+ private final List<Snapshot> taggedSnapshots;
+ private final boolean changelogDecoupled;
+
+ private transient ExpireSnapshotsExecutor executor;
+
+ public RangePartitionedExpireFunction(
+ Map<String, String> catalogConfig,
+ Identifier identifier,
+ List<Snapshot> taggedSnapshots,
+ boolean changelogDecoupled) {
+ this.catalogConfig = catalogConfig;
+ this.identifier = identifier;
+ this.taggedSnapshots = taggedSnapshots;
+ this.changelogDecoupled = changelogDecoupled;
+ }
+
+ @Override
+ public void open(OpenContext openContext) throws Exception {
+ this.executor = initExecutor();
+ }
+
+ /**
+ * Initializes and returns the executor for processing expire tasks.
Subclasses can override
+ * this method to provide a custom executor for testing without catalog
access.
+ *
+ * <p>Default implementation creates executor from catalog using {@link
#catalogConfig} and
+ * {@link #identifier}.
+ */
+ protected ExpireSnapshotsExecutor initExecutor() throws Exception {
+ Options options = Options.fromMap(catalogConfig);
+ Catalog catalog = FlinkCatalogFactory.createPaimonCatalog(options);
+ FileStoreTable table = (FileStoreTable) catalog.getTable(identifier);
+ SnapshotDeletion deletion = table.store().newSnapshotDeletion();
+ deletion.setChangelogDecoupled(changelogDecoupled);
+ return new ExpireSnapshotsExecutor(table.snapshotManager(), deletion);
+ }
+
+ @Override
+ public void flatMap(List<SnapshotExpireTask> tasks,
Collector<DeletionReport> out)
+ throws Exception {
+ // Process tasks sequentially in order to maximize cache locality
+ for (SnapshotExpireTask task : tasks) {
+ DeletionReport report = processTask(task);
+ out.collect(report);
+ }
+ }
+
+ private DeletionReport processTask(SnapshotExpireTask task) {
Review Comment:
Done. Inlined processTask into flatMap.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
