diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java index 23c50097ba6..ec81f33470b 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java @@ -2520,6 +2520,15 @@ public class ConfigOptions { + ConfigOptions.TABLE_DATALAKE_AUTO_EXPIRE_SNAPSHOT + " is false."); + public static final ConfigOption LAKE_TIERING_PARTITION_MARK_DONE_ENABLED = + key("lake.tiering.partition.mark-done.enabled") + .booleanType() + .defaultValue(false) + .withDescription( + "Whether the tiering service marks idle partitions of tiered partitioned tables as done. " + + "Disabled by default. When enabled, a table opts in via its lake-format prefixed " + + "mark-done custom properties (e.g. 'paimon.partition.idle-time-to-done' for Paimon)."); + public static final ConfigOption LAKE_TIERING_IO_TMP_DIRS = key("lake.tiering.io.tmp.dirs") .stringType() diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/committer/PartitionMarkDoneMaintainer.java b/fluss-common/src/main/java/org/apache/fluss/lake/committer/PartitionMarkDoneMaintainer.java new file mode 100644 index 00000000000..ece13daacfc --- /dev/null +++ b/fluss-common/src/main/java/org/apache/fluss/lake/committer/PartitionMarkDoneMaintainer.java @@ -0,0 +1,48 @@ +/* + * 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.fluss.lake.committer; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.utils.function.SupplierWithException; + +import javax.annotation.Nullable; + +import java.io.IOException; + +/** + * An optional capability that a {@link LakeCommitter} may implement to persist the partition + * mark-done state for a tiering round without any data to commit. + */ +@Internal +public interface PartitionMarkDoneMaintainer { + + /** + * Performs partition mark-done maintenance for a tiering round without any data to commit. + * + * @param offsetsFileProvider provides a freshly prepared bucket offsets file for the + * maintenance snapshot; it is only invoked when a snapshot will actually be created. Every + * snapshot must carry its own offsets file since offsets files are deleted along with their + * snapshot metadata and thus must not be shared across snapshots. + * @return the properties-only lake snapshot created to persist the mark-done state, or null if + * no snapshot was created (feature disabled or state unchanged) + * @throws IOException if an I/O error occurs + */ + @Nullable + CommittedLakeSnapshot commitMarkDoneMaintenance( + SupplierWithException offsetsFileProvider) throws IOException; +} diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/writer/PartitionMarkDoneEnabler.java b/fluss-common/src/main/java/org/apache/fluss/lake/writer/PartitionMarkDoneEnabler.java new file mode 100644 index 00000000000..59c5af2acb3 --- /dev/null +++ b/fluss-common/src/main/java/org/apache/fluss/lake/writer/PartitionMarkDoneEnabler.java @@ -0,0 +1,40 @@ +/* + * 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.fluss.lake.writer; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.metadata.TableInfo; + +/** + * An optional capability that a {@link LakeTieringFactory} may implement to support marking idle + * partitions of a tiered table as done. Lake formats that do not implement this interface never + * trigger mark-done maintenance. + */ +@Internal +public interface PartitionMarkDoneEnabler { + + /** + * Whether partition mark-done is enabled for the given table, used to decide whether an empty + * tiering round should still reach the committer to run mark-done maintenance. + * + *

The switch only recognizes the Fluss table metadata (e.g. custom properties): mark-done + * options configured on the lake table directly are deliberately not honored, so the check is + * cheap (no lake access) and there is a single source of truth. + */ + boolean isPartitionMarkDoneEnabled(TableInfo tableInfo); +} diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/PartitionUtils.java b/fluss-common/src/main/java/org/apache/fluss/utils/PartitionUtils.java index fcc584047e0..b0dd1cd484e 100644 --- a/fluss-common/src/main/java/org/apache/fluss/utils/PartitionUtils.java +++ b/fluss-common/src/main/java/org/apache/fluss/utils/PartitionUtils.java @@ -439,7 +439,7 @@ public String toString() { } /** Returns the time string format pattern for the given time unit. */ - private static String getPartitionTimeFormat( + public static String getPartitionTimeFormat( AutoPartitionTimeUnit timeUnit, AutoPartitionStrategy autoPartitionStrategy) { String timeFormat = autoPartitionStrategy.timeFormat(); if (timeFormat != null) { diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperator.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperator.java index 477beea1134..63db4bf3b27 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperator.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperator.java @@ -21,6 +21,7 @@ import org.apache.fluss.client.ConnectionFactory; import org.apache.fluss.client.admin.Admin; import org.apache.fluss.client.metadata.LakeSnapshot; +import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; import org.apache.fluss.exception.LakeTableSnapshotNotExistException; import org.apache.fluss.flink.tiering.event.FailedTieringEvent; @@ -30,9 +31,11 @@ import org.apache.fluss.lake.committer.CommittedLakeSnapshot; import org.apache.fluss.lake.committer.LakeCommitResult; import org.apache.fluss.lake.committer.LakeCommitter; +import org.apache.fluss.lake.committer.PartitionMarkDoneMaintainer; import org.apache.fluss.lake.committer.TieringStats; import org.apache.fluss.lake.writer.LakeTieringFactory; import org.apache.fluss.lake.writer.LakeWriter; +import org.apache.fluss.lake.writer.PartitionMarkDoneEnabler; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TableInfo; import org.apache.fluss.metadata.TablePath; @@ -219,6 +222,8 @@ private CommitResult commitWriteResults( "Commit tiering write results is empty for table {}, table path {}", tableId, tablePath); + // still run partition mark-done maintenance for the empty round + maybeCommitMarkDoneMaintenance(tableId, tablePath); return new CommitResult(null, null); } @@ -290,6 +295,66 @@ private CommitResult commitWriteResults( } } + /** + * Runs partition mark-done maintenance for an empty tiering round, and commits the resulting + * properties-only lake snapshot (if any) to Fluss. + */ + private void maybeCommitMarkDoneMaintenance(long tableId, TablePath tablePath) + throws Exception { + if (!lakeTieringConfig.get(ConfigOptions.LAKE_TIERING_PARTITION_MARK_DONE_ENABLED) + || !(lakeTieringFactory instanceof PartitionMarkDoneEnabler)) { + return; + } + TableInfo tableInfo = admin.getTableInfo(tablePath).get(); + if (tableInfo.getTableId() != tableId + || !((PartitionMarkDoneEnabler) lakeTieringFactory) + .isPartitionMarkDoneEnabled(tableInfo)) { + return; + } + try (LakeCommitter lakeCommitter = + lakeTieringFactory.createLakeCommitter( + new TieringCommitterInitContext( + tablePath, tableInfo, lakeTieringConfig, flussConfig))) { + if (!(lakeCommitter instanceof PartitionMarkDoneMaintainer)) { + return; + } + // first bring Fluss up to date in case a previous round committed to the lake but + // failed to commit to Fluss; otherwise the maintenance would see an unchanged + // state, keep returning null and Fluss would stay on the old snapshot forever + LakeSnapshot flussCurrentLakeSnapshot = getLatestLakeSnapshot(tablePath); + CommittedLakeSnapshot missingCommittedSnapshot = + lakeCommitter.getMissingLakeSnapshot( + flussCurrentLakeSnapshot == null + ? null + : flussCurrentLakeSnapshot.getSnapshotId()); + if (missingCommittedSnapshot != null) { + commitMissingLakeSnapshotToFluss(tablePath, tableId, missingCommittedSnapshot); + } + CommittedLakeSnapshot maintenanceSnapshot = + ((PartitionMarkDoneMaintainer) lakeCommitter) + .commitMarkDoneMaintenance( + // a fresh offsets file for the maintenance snapshot, since + // offsets files are deleted along with their snapshot + // metadata and must not be shared across snapshots + () -> + flussTableLakeSnapshotCommitter.prepareLakeSnapshot( + tableId, tablePath, Collections.emptyMap())); + if (maintenanceSnapshot != null) { + flussTableLakeSnapshotCommitter.commit( + tableId, + maintenanceSnapshot.getLakeSnapshotId(), + maintenanceSnapshot + .getSnapshotProperties() + .get(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY), + null, + // no data was written in this round + Collections.emptyMap(), + Collections.emptyMap(), + LakeCommitResult.KEEP_ALL_PREVIOUS); + } + } + } + @Nullable private LakeSnapshot getLatestLakeSnapshot(TablePath tablePath) throws Exception { LakeSnapshot flussCurrentLakeSnapshot; @@ -323,48 +388,8 @@ private void checkFlussNotMissingLakeSnapshot( // known lake snapshot, which means the data already has been committed to lake, // not to commit to lake to avoid data duplicated if (missingCommittedSnapshot != null) { - String lakeSnapshotOffsetPath = - missingCommittedSnapshot - .getSnapshotProperties() - .get(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY); - - // should only will happen in v0.7 which won't put offsets info - // to properties - if (lakeSnapshotOffsetPath == null) { - throw new IllegalStateException( - String.format( - "Can't find %s field from snapshot property.", - FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY)); - } - - // the fluss-offsets will be a json string if it's tiered by v0.8, - // since this code path should be rare, we do not consider backward compatibility - // and throw IllegalStateException directly - String trimmedPath = lakeSnapshotOffsetPath.trim(); - if (trimmedPath.contains("{")) { - throw new IllegalStateException( - String.format( - "The %s field in snapshot property is a JSON string (tiered by v0.8), " - + "which is not supported to restore. Snapshot ID: %d, Table: {tablePath=%s, tableId=%d}.", - FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY, - missingCommittedSnapshot.getLakeSnapshotId(), - tablePath, - tableId)); - } - // commit this missing snapshot to fluss - flussTableLakeSnapshotCommitter.commit( - tableId, - missingCommittedSnapshot.getLakeSnapshotId(), - lakeSnapshotOffsetPath, - // don't care readable snapshot and offsets, - null, - // use empty log offsets, log max timestamp, since we can't know that - // in last tiering, it doesn't matter for they are just used to - // report metrics - Collections.emptyMap(), - Collections.emptyMap(), - LakeCommitResult.KEEP_ALL_PREVIOUS); + commitMissingLakeSnapshotToFluss(tablePath, tableId, missingCommittedSnapshot); // abort this committable to delete the written files lakeCommitter.abort(committable); throw new IllegalStateException( @@ -380,6 +405,53 @@ private void checkFlussNotMissingLakeSnapshot( } } + /** Commits a lake snapshot committed by Fluss but unknown to Fluss back to Fluss. */ + private void commitMissingLakeSnapshotToFluss( + TablePath tablePath, long tableId, CommittedLakeSnapshot missingCommittedSnapshot) + throws Exception { + String lakeSnapshotOffsetPath = + missingCommittedSnapshot + .getSnapshotProperties() + .get(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY); + + // should only will happen in v0.7 which won't put offsets info + // to properties + if (lakeSnapshotOffsetPath == null) { + throw new IllegalStateException( + String.format( + "Can't find %s field from snapshot property.", + FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY)); + } + + // the fluss-offsets will be a json string if it's tiered by v0.8, + // since this code path should be rare, we do not consider backward compatibility + // and throw IllegalStateException directly + String trimmedPath = lakeSnapshotOffsetPath.trim(); + if (trimmedPath.contains("{")) { + throw new IllegalStateException( + String.format( + "The %s field in snapshot property is a JSON string (tiered by v0.8), " + + "which is not supported to restore. Snapshot ID: %d, Table: {tablePath=%s, tableId=%d}.", + FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY, + missingCommittedSnapshot.getLakeSnapshotId(), + tablePath, + tableId)); + } + + flussTableLakeSnapshotCommitter.commit( + tableId, + missingCommittedSnapshot.getLakeSnapshotId(), + lakeSnapshotOffsetPath, + // don't care readable snapshot and offsets, + null, + // use empty log offsets, log max timestamp, since we can't know that + // in last tiering, it doesn't matter for they are just used to + // report metrics + Collections.emptyMap(), + Collections.emptyMap(), + LakeCommitResult.KEEP_ALL_PREVIOUS); + } + private void registerTableBucketWriteResult( long tableId, TableBucketWriteResult tableBucketWriteResult) { collectedTableBucketWriteResults diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSource.java index 696d4722b8a..2fff5eed91e 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSource.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSource.java @@ -88,7 +88,11 @@ public Boundedness getBoundedness() { public SplitEnumerator createEnumerator( SplitEnumeratorContext splitEnumeratorContext) { return new TieringSourceEnumerator( - flussConf, splitEnumeratorContext, lakeTieringFactory, pollTieringTableIntervalMs); + flussConf, + lakeTieringConfig, + splitEnumeratorContext, + lakeTieringFactory, + pollTieringTableIntervalMs); } @Override @@ -97,7 +101,11 @@ public SplitEnumerator restoreEnumer TieringSourceEnumeratorState tieringSourceEnumeratorState) { // stateless operator return new TieringSourceEnumerator( - flussConf, splitEnumeratorContext, lakeTieringFactory, pollTieringTableIntervalMs); + flussConf, + lakeTieringConfig, + splitEnumeratorContext, + lakeTieringFactory, + pollTieringTableIntervalMs); } @Override diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java index f1bf98527af..20e85a59ec0 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java @@ -22,17 +22,21 @@ import org.apache.fluss.client.ConnectionFactory; import org.apache.fluss.client.admin.Admin; import org.apache.fluss.client.metadata.MetadataUpdater; +import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; import org.apache.fluss.flink.metrics.FlinkMetricRegistry; import org.apache.fluss.flink.tiering.event.FailedTieringEvent; import org.apache.fluss.flink.tiering.event.FinishedTieringEvent; import org.apache.fluss.flink.tiering.event.TieringReachMaxDurationEvent; +import org.apache.fluss.flink.tiering.source.split.TieringLogSplit; import org.apache.fluss.flink.tiering.source.split.TieringSplit; import org.apache.fluss.flink.tiering.source.split.TieringSplitGenerator; import org.apache.fluss.flink.tiering.source.state.TieringSourceEnumeratorState; import org.apache.fluss.lake.committer.TieringStats; import org.apache.fluss.lake.writer.LakeTieringFactory; +import org.apache.fluss.lake.writer.PartitionMarkDoneEnabler; import org.apache.fluss.lake.writer.TieringTableValidator; +import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TableInfo; import org.apache.fluss.metadata.TablePath; import org.apache.fluss.rpc.GatewayClientProxy; @@ -100,6 +104,7 @@ public class TieringSourceEnumerator private static final Logger LOG = LoggerFactory.getLogger(TieringSourceEnumerator.class); private final Configuration flussConf; + private final Configuration lakeTieringConfig; private final SplitEnumeratorContext context; private final LakeTieringFactory lakeTieringFactory; private final ScheduledExecutorService timerService; @@ -130,7 +135,22 @@ public TieringSourceEnumerator( SplitEnumeratorContext context, LakeTieringFactory lakeTieringFactory, long pollTieringTableIntervalMs) { + this( + flussConf, + new Configuration(), + context, + lakeTieringFactory, + pollTieringTableIntervalMs); + } + + public TieringSourceEnumerator( + Configuration flussConf, + Configuration lakeTieringConfig, + SplitEnumeratorContext context, + LakeTieringFactory lakeTieringFactory, + long pollTieringTableIntervalMs) { this.flussConf = flussConf; + this.lakeTieringConfig = lakeTieringConfig; this.context = context; this.lakeTieringFactory = lakeTieringFactory; this.timerService = @@ -459,6 +479,24 @@ private void generateTieringSplits(Tuple3 tieringTable) ((TieringTableValidator) lakeTieringFactory).validateTable(tableInfo); } List tieringSplits = splitGenerator.generateTableSplits(tableInfo); + if (tieringSplits.isEmpty() + && lakeTieringConfig.get(ConfigOptions.LAKE_TIERING_PARTITION_MARK_DONE_ENABLED) + && lakeTieringFactory instanceof PartitionMarkDoneEnabler + && ((PartitionMarkDoneEnabler) lakeTieringFactory) + .isPartitionMarkDoneEnabled(tableInfo)) { + // fully caught up but mark-done enabled: emit one skip-round split so the + // commit operator can run mark-done maintenance for the empty round + tieringSplits = new ArrayList<>(); + tieringSplits.add( + new TieringLogSplit( + tablePath, + new TableBucket(tableInfo.getTableId(), 0), + null, + 0L, + 0L, + 1, + true)); + } // shuffle tiering split to avoid splits tiering skew // after introduce tiering max duration Collections.shuffle(tieringSplits); diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/TestingLakeTieringFactory.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/TestingLakeTieringFactory.java index 1400651a0d5..cf2db925a89 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/TestingLakeTieringFactory.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/TestingLakeTieringFactory.java @@ -23,13 +23,16 @@ import org.apache.fluss.lake.committer.CommitterInitContext; import org.apache.fluss.lake.committer.LakeCommitResult; import org.apache.fluss.lake.committer.LakeCommitter; +import org.apache.fluss.lake.committer.PartitionMarkDoneMaintainer; import org.apache.fluss.lake.serializer.SimpleVersionedSerializer; import org.apache.fluss.lake.writer.LakeTieringFactory; import org.apache.fluss.lake.writer.LakeWriter; +import org.apache.fluss.lake.writer.PartitionMarkDoneEnabler; import org.apache.fluss.lake.writer.TieringTableValidator; import org.apache.fluss.lake.writer.WriterInitContext; import org.apache.fluss.metadata.TableInfo; import org.apache.fluss.record.LogRecord; +import org.apache.fluss.utils.function.SupplierWithException; import javax.annotation.Nullable; @@ -41,7 +44,8 @@ /** An implementation of {@link LakeTieringFactory} for testing purpose. */ public class TestingLakeTieringFactory implements LakeTieringFactory, - TieringTableValidator { + TieringTableValidator, + PartitionMarkDoneEnabler { @Nullable private TestingLakeCommitter testingLakeCommitter; @@ -50,6 +54,9 @@ public class TestingLakeTieringFactory private final List createdLakeWriters = new ArrayList<>(); + // whether partition mark-done is enabled for all tables of this factory + private boolean partitionMarkDoneEnabled; + public TestingLakeTieringFactory(@Nullable TestingLakeCommitter testingLakeCommitter) { this(testingLakeCommitter, null); } @@ -100,6 +107,15 @@ public SimpleVersionedSerializer getCommittableSerializer() "method getCommittableSerializer is not supported."); } + public void enablePartitionMarkDone() { + this.partitionMarkDoneEnabled = true; + } + + @Override + public boolean isPartitionMarkDoneEnabled(TableInfo tableInfo) { + return partitionMarkDoneEnabled; + } + /** A lake writer for testing purpose which tracks the closed state. */ public static final class TestingLakeWriter implements LakeWriter { @@ -142,12 +158,15 @@ public boolean isClosed() { /** A lake committer for testing purpose. */ public static final class TestingLakeCommitter - implements LakeCommitter { + implements LakeCommitter, + PartitionMarkDoneMaintainer { private long currentSnapshot; @Nullable private final CommittedLakeSnapshot mockMissingCommittedLakeSnapshot; + private int maintenanceInvocations; + public TestingLakeCommitter() { this(null); } @@ -187,6 +206,18 @@ public void abort(TestingCommittable committable) throws IOException { return null; } + @Nullable + @Override + public CommittedLakeSnapshot commitMarkDoneMaintenance( + SupplierWithException offsetsFileProvider) throws IOException { + maintenanceInvocations++; + return null; + } + + public int getMaintenanceInvocations() { + return maintenanceInvocations; + } + @Override public void close() throws Exception {} } diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperatorTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperatorTest.java index 3eda8f5984c..873b025f522 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperatorTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperatorTest.java @@ -547,6 +547,49 @@ void testCommitFailsWhenTableRecreated() throws Exception { .contains("dropped and recreated during tiering"); } + @Test + void testEmptyRoundRecoversMissingLakeSnapshot() throws Exception { + TablePath tablePath = + TablePath.of("fluss", "test_empty_round_recovers_missing_lake_snapshot"); + long tableId = createTable(tablePath, DATA1_PARTITIONED_TABLE_DESCRIPTOR); + + // mimic a previous round that committed snapshot 5 to the lake but failed to commit + // to Fluss + Map expectedLogEndOffsets = new HashMap<>(); + expectedLogEndOffsets.put(new TableBucket(tableId, 0), 3L); + CommittedLakeSnapshot mockMissingCommittedLakeSnapshot = + mockCommittedLakeSnapshot(tableId, tablePath, 5, expectedLogEndOffsets); + TestingLakeTieringFactory.TestingLakeCommitter testingLakeCommitter = + new TestingLakeTieringFactory.TestingLakeCommitter( + mockMissingCommittedLakeSnapshot); + TestingLakeTieringFactory lakeTieringFactory = + new TestingLakeTieringFactory(testingLakeCommitter); + lakeTieringFactory.enablePartitionMarkDone(); + // mark-done must also be enabled at the job level (disabled by default) + org.apache.fluss.config.Configuration lakeTieringConfig = + new org.apache.fluss.config.Configuration(); + lakeTieringConfig.set( + org.apache.fluss.config.ConfigOptions.LAKE_TIERING_PARTITION_MARK_DONE_ENABLED, + true); + committerOperator = + new TieringCommitOperator<>( + parameters, + FLUSS_CLUSTER_EXTENSION.getClientConfig(), + lakeTieringConfig, + lakeTieringFactory); + committerOperator.open(); + + // an empty round runs mark-done maintenance and first brings Fluss up to date with + // the missing lake snapshot + committerOperator.processElement( + createTableBucketWriteResultStreamRecord( + tablePath, new TableBucket(tableId, 0), null, null, -1, -1, 1)); + assertThat(testingLakeCommitter.getMaintenanceInvocations()).isEqualTo(1); + LakeSnapshot lakeSnapshot = admin.getLatestLakeSnapshot(tablePath).get(); + assertThat(lakeSnapshot.getSnapshotId()).isEqualTo(5); + assertThat(lakeSnapshot.getTableBucketsOffset()).isEqualTo(expectedLogEndOffsets); + } + private CommittedLakeSnapshot mockCommittedLakeSnapshot( long tableId, TablePath tablePath, int snapshotId, Map logEndOffsets) throws Exception { diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/MarkDoneState.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/MarkDoneState.java new file mode 100644 index 00000000000..2f2a30f9d08 --- /dev/null +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/MarkDoneState.java @@ -0,0 +1,83 @@ +/* + * 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.fluss.lake.paimon.tiering; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; + +/** + * The partition mark-done state persisted as JSON in the lake snapshot properties committed by the + * tiering service. It only keeps a table-level {@code initialized} cold-start flag and the pending + * (not yet done) partitions mapped to their last update time. The done fact itself is not stored: + * done partitions are removed (done-is-delete), it lives in the lake via the idempotent mark-done + * actions. + */ +public class MarkDoneState { + + private final boolean initialized; + // partition name -> last time the tiering service wrote data into the partition + private final Map pendingPartitions; + + public MarkDoneState(boolean initialized, Map pendingPartitions) { + this.initialized = initialized; + this.pendingPartitions = new HashMap<>(pendingPartitions); + } + + /** Creates an empty state: not initialized, no pending partitions. */ + public static MarkDoneState empty() { + return new MarkDoneState(false, new HashMap<>()); + } + + public boolean isInitialized() { + return initialized; + } + + public Map getPendingPartitions() { + return Collections.unmodifiableMap(pendingPartitions); + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + MarkDoneState that = (MarkDoneState) o; + return initialized == that.initialized + && Objects.equals(pendingPartitions, that.pendingPartitions); + } + + @Override + public int hashCode() { + return Objects.hash(initialized, pendingPartitions); + } + + @Override + public String toString() { + return "MarkDoneState{" + + "initialized=" + + initialized + + ", pendingPartitions=" + + pendingPartitions + + '}'; + } +} diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/MarkDoneStateJsonSerde.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/MarkDoneStateJsonSerde.java new file mode 100644 index 00000000000..3208c4b0ec1 --- /dev/null +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/MarkDoneStateJsonSerde.java @@ -0,0 +1,96 @@ +/* + * 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.fluss.lake.paimon.tiering; + +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator; +import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode; +import org.apache.fluss.utils.json.JsonDeserializer; +import org.apache.fluss.utils.json.JsonSerdeUtils; +import org.apache.fluss.utils.json.JsonSerializer; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.Iterator; +import java.util.Map; + +import static org.apache.fluss.utils.Preconditions.checkArgument; + +/** + * Json serde for {@link MarkDoneState}: {@code {"initialized": true, "pending": {"": + * }}}. Evolves via field-level compatibility (unknown fields ignored, missing + * fields defaulted), no version gating. Wrongly typed fields are rejected instead of being silently + * coerced, so a corrupt state is detected and healed by the caller. + */ +public class MarkDoneStateJsonSerde + implements JsonSerializer, JsonDeserializer { + + public static final MarkDoneStateJsonSerde INSTANCE = new MarkDoneStateJsonSerde(); + + private static final String INITIALIZED_FIELD = "initialized"; + private static final String PENDING_FIELD = "pending"; + + @Override + public void serialize(MarkDoneState state, JsonGenerator generator) throws IOException { + generator.writeStartObject(); + generator.writeBooleanField(INITIALIZED_FIELD, state.isInitialized()); + generator.writeObjectFieldStart(PENDING_FIELD); + for (Map.Entry entry : state.getPendingPartitions().entrySet()) { + generator.writeNumberField(entry.getKey(), entry.getValue()); + } + generator.writeEndObject(); + generator.writeEndObject(); + } + + @Override + public MarkDoneState deserialize(JsonNode node) { + boolean initialized = false; + JsonNode initializedNode = node.get(INITIALIZED_FIELD); + if (initializedNode != null) { + checkArgument( + initializedNode.isBoolean(), "Field %s must be a boolean.", INITIALIZED_FIELD); + initialized = initializedNode.asBoolean(); + } + Map pendingPartitions = new HashMap<>(); + JsonNode pendingNode = node.get(PENDING_FIELD); + if (pendingNode != null) { + checkArgument(pendingNode.isObject(), "Field %s must be an object.", PENDING_FIELD); + Iterator> fields = pendingNode.fields(); + while (fields.hasNext()) { + Map.Entry field = fields.next(); + checkArgument( + field.getValue().canConvertToLong(), + "Time of pending partition %s must be a long.", + field.getKey()); + pendingPartitions.put(field.getKey(), field.getValue().asLong()); + } + } + return new MarkDoneState(initialized, pendingPartitions); + } + + /** Serializes the given state to a JSON string. */ + public static String toJson(MarkDoneState state) { + return new String( + JsonSerdeUtils.writeValueAsBytes(state, INSTANCE), StandardCharsets.UTF_8); + } + + /** Deserializes the state from a JSON string. */ + public static MarkDoneState fromJson(String json) { + return JsonSerdeUtils.readValue(json.getBytes(StandardCharsets.UTF_8), INSTANCE); + } +} diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeCommitter.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeCommitter.java index 287df30e28e..0a8bc11b2b7 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeCommitter.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeCommitter.java @@ -23,9 +23,11 @@ import org.apache.fluss.lake.committer.CommitterInitContext; import org.apache.fluss.lake.committer.LakeCommitResult; import org.apache.fluss.lake.committer.LakeCommitter; +import org.apache.fluss.lake.committer.PartitionMarkDoneMaintainer; import org.apache.fluss.lake.committer.TieringStats; import org.apache.fluss.lake.paimon.utils.DvTableReadableSnapshotRetriever; import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.utils.function.SupplierWithException; import org.apache.paimon.CoreOptions; import org.apache.paimon.Snapshot; @@ -46,10 +48,12 @@ import javax.annotation.Nullable; import java.io.IOException; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.Set; import static org.apache.fluss.lake.paimon.tiering.PaimonLakeTieringFactory.FLUSS_LAKE_TIERING_COMMIT_USER; import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon; @@ -57,7 +61,9 @@ import static org.apache.paimon.table.sink.BatchWriteBuilder.COMMIT_IDENTIFIER; /** Implementation of {@link LakeCommitter} for Paimon. */ -public class PaimonLakeCommitter implements LakeCommitter { +public class PaimonLakeCommitter + implements LakeCommitter, + PartitionMarkDoneMaintainer { private static final Logger LOG = LoggerFactory.getLogger(PaimonLakeCommitter.class); @@ -66,6 +72,7 @@ public class PaimonLakeCommitter implements LakeCommitter currentCommitSnapshotId = new ThreadLocal<>(); @@ -87,6 +94,26 @@ public PaimonLakeCommitter( || committerInitContext .lakeTieringConfig() .get(ConfigOptions.LAKE_TIERING_AUTO_EXPIRE_SNAPSHOT)); + PaimonPartitionMarkDone partitionMarkDone = null; + if (committerInitContext + .lakeTieringConfig() + .get(ConfigOptions.LAKE_TIERING_PARTITION_MARK_DONE_ENABLED) + && PaimonPartitionMarkDone.isEnabled(committerInitContext.tableInfo())) { + try { + partitionMarkDone = + new PaimonPartitionMarkDone( + fileStoreTable, committerInitContext.tableInfo()); + } catch (Exception e) { + // an invalid mark-done configuration only disables mark-done, never fails + // the committer + LOG.warn( + "Invalid partition mark-done configuration for table {}, " + + "partition mark-done is disabled.", + tablePath, + e); + } + } + this.partitionMarkDone = partitionMarkDone; } @Override @@ -107,17 +134,11 @@ public LakeCommitResult commit( snapshotProperties.forEach(manifestCommittable::addProperty); try { - tableCommit = fileStoreTable.newCommit(FLUSS_LAKE_TIERING_COMMIT_USER); - // don't skip empty commits: tiering relies on empty snapshots to persist bucket - // offsets when only empty WAL batches were consumed - tableCommit.ignoreEmptyCommit(false); - tableCommit.commit(manifestCommittable); + if (partitionMarkDone != null) { + runPartitionMarkDone(manifestCommittable); + } - long committedSnapshotId = - checkNotNull( - currentCommitSnapshotId.get(), - "Paimon committed snapshot id must be non-null."); - currentCommitSnapshotId.remove(); + long committedSnapshotId = commitManifest(manifestCommittable); // Collect cumulative table stats from the exact snapshot that was just committed. TieringStats stats = computeTableStats(); @@ -160,6 +181,136 @@ public LakeCommitResult commit( } } + /** + * Runs partition mark-done for a data commit and attaches the state (re-attached even if + * unchanged) so the latest Fluss-committed snapshot always holds the full state. Known runtime + * mark-done failures are only logged and don't fail the data commit: the previous state is + * re-attached if readable, otherwise the next round re-initializes via cold start — a lossy + * last resort that can't recover zero-file pending partitions. + */ + private void runPartitionMarkDone(ManifestCommittable manifestCommittable) { + checkNotNull(partitionMarkDone); + String stateJson = null; + try { + String previousStateJson = getLatestMarkDoneStateJson(); + stateJson = previousStateJson; + Set tieredPartitions = + partitionMarkDone.extractTieredPartitions( + manifestCommittable.fileCommittables()); + String newStateJson = partitionMarkDone.run(previousStateJson, tieredPartitions); + if (newStateJson != null) { + stateJson = newStateJson; + } + } catch (Exception e) { + LOG.warn( + "Failed to run partition mark-done for table {}, " + + "the data commit continues without it.", + tablePath, + e); + } + if (stateJson != null) { + manifestCommittable.addProperty( + PaimonPartitionMarkDone.MARK_DONE_STATE_PROPERTY, stateJson); + } + } + + @Nullable + private String getLatestMarkDoneStateJson() throws IOException { + Snapshot latestFlussSnapshot = + getCommittedLatestSnapshotOfLake(FLUSS_LAKE_TIERING_COMMIT_USER); + if (latestFlussSnapshot == null) { + return null; + } + // null when the properties snapshot can't be found (e.g. expired): re-initialize via + // cold start, the mark-done actions are idempotent + Snapshot propertiesSnapshot = findRoundPropertiesSnapshot(latestFlussSnapshot); + return propertiesSnapshot == null + ? null + : propertiesSnapshot + .properties() + .get(PaimonPartitionMarkDone.MARK_DONE_STATE_PROPERTY); + } + + @Nullable + @Override + public CommittedLakeSnapshot commitMarkDoneMaintenance( + SupplierWithException offsetsFileProvider) throws IOException { + if (partitionMarkDone == null) { + return null; + } + // mark-done itself is best-effort: a failure only skips this round and is retried in + // a later round; failures of the snapshot commit below still propagate + String newStateJson; + try { + Snapshot latestFlussSnapshot = + getCommittedLatestSnapshotOfLake(FLUSS_LAKE_TIERING_COMMIT_USER); + if (latestFlussSnapshot == null) { + // never tiered by Fluss, cold start happens with the first data commit + return null; + } + // null when the properties snapshot can't be found (e.g. expired): re-initialize + // via cold start, the mark-done actions are idempotent + Snapshot propertiesSnapshot = findRoundPropertiesSnapshot(latestFlussSnapshot); + String previousStateJson = + propertiesSnapshot == null + ? null + : propertiesSnapshot + .properties() + .get(PaimonPartitionMarkDone.MARK_DONE_STATE_PROPERTY); + newStateJson = partitionMarkDone.run(previousStateJson, Collections.emptySet()); + } catch (Exception e) { + LOG.warn( + "Failed to run partition mark-done maintenance for table {}, " + + "will retry in a later round.", + tablePath, + e); + return null; + } + if (newStateJson == null) { + return null; + } + + try { + // persist the new state with a properties-only snapshot carrying a freshly + // prepared offsets file: offsets files are deleted along with their snapshot + // metadata, so they must never be shared across snapshots + String offsetsFilePath = offsetsFileProvider.get(); + Map snapshotProperties = new HashMap<>(); + snapshotProperties.put(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY, offsetsFilePath); + snapshotProperties.put(PaimonPartitionMarkDone.MARK_DONE_STATE_PROPERTY, newStateJson); + + ManifestCommittable manifestCommittable = new ManifestCommittable(COMMIT_IDENTIFIER); + snapshotProperties.forEach(manifestCommittable::addProperty); + long committedSnapshotId = commitManifest(manifestCommittable); + return new CommittedLakeSnapshot(committedSnapshotId, snapshotProperties); + } catch (Throwable t) { + throw new IOException(t); + } + } + + /** + * Commits the manifest committable and returns the id of the snapshot created for it, as + * recorded by {@link PaimonCommitCallback}. Shared by the data commit and the mark-done + * maintenance commit which both carry the bucket offsets property of the round. + */ + private long commitManifest(ManifestCommittable manifestCommittable) throws Exception { + // clear any residue left by a previous failed commit on the same thread + currentCommitSnapshotId.remove(); + try { + tableCommit = fileStoreTable.newCommit(FLUSS_LAKE_TIERING_COMMIT_USER); + // don't skip empty commits: tiering relies on empty snapshots to persist bucket + // offsets when only empty WAL batches were consumed, and mark-done maintenance + // commits properties-only snapshots + tableCommit.ignoreEmptyCommit(false); + tableCommit.commit(manifestCommittable); + return checkNotNull( + currentCommitSnapshotId.get(), + "Paimon committed snapshot id must be non-null."); + } finally { + currentCommitSnapshotId.remove(); + } + } + /** Computes cumulative table stats from the latest snapshot by REST API. */ @Nullable private TieringStats computeTableStats() { @@ -210,12 +361,15 @@ public CommittedLakeSnapshot getMissingLakeSnapshot(@Nullable Long latestLakeSna return null; } - if (latestLakeSnapshotOfLake.properties() == null) { + // the round may end with a maintenance tail (e.g. partition expiration), its offsets + // and state live on the nearest properties-carrying snapshot of the same round + Snapshot propertiesSnapshot = findRoundPropertiesSnapshot(latestLakeSnapshotOfLake); + if (propertiesSnapshot == null) { throw new IOException("Failed to load committed lake snapshot properties from Paimon."); } return new CommittedLakeSnapshot( - latestLakeSnapshotOfLake.id(), latestLakeSnapshotOfLake.properties()); + latestLakeSnapshotOfLake.id(), propertiesSnapshot.properties()); } @Nullable @@ -241,9 +395,45 @@ private Snapshot getCommittedLatestSnapshotOfLake(String commitUser) throws IOEx return snapshot; } + /** + * Finds the snapshot carrying the offsets (and mark-done state) properties of the tiering round + * the given latest Fluss snapshot belongs to: Paimon may append a tail of maintenance snapshots + * after ours within the same commit call (e.g. batched partition expiration OVERWRITE snapshots + * without properties), so walk back over such a tail. Returns null when it can't be found (e.g. + * a legacy v0.7 commit, or the properties snapshot was expired). + */ + @Nullable + private Snapshot findRoundPropertiesSnapshot(Snapshot latestFlussSnapshot) { + SnapshotManager snapshotManager = fileStoreTable.snapshotManager(); + Snapshot snapshot = latestFlussSnapshot; + while (true) { + if (FLUSS_LAKE_TIERING_COMMIT_USER.equals(snapshot.commitUser()) + && snapshot.properties() != null + && snapshot.properties().containsKey(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY)) { + return snapshot; + } + // only walk over the maintenance tail of the round + if (snapshot.properties() != null + || snapshot.commitKind() != Snapshot.CommitKind.OVERWRITE + || !FLUSS_LAKE_TIERING_COMMIT_USER.equals(snapshot.commitUser()) + || snapshot.id() == Snapshot.FIRST_SNAPSHOT_ID) { + return null; + } + try { + snapshot = snapshotManager.tryGetSnapshot(snapshot.id() - 1); + } catch (Exception e) { + // the previous snapshot has been expired + return null; + } + } + } + @Override public void close() throws Exception { try { + if (partitionMarkDone != null) { + partitionMarkDone.close(); + } if (tableCommit != null) { tableCommit.close(); } diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeTieringFactory.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeTieringFactory.java index 432620efb30..f414cd401cf 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeTieringFactory.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonLakeTieringFactory.java @@ -23,13 +23,16 @@ import org.apache.fluss.lake.serializer.SimpleVersionedSerializer; import org.apache.fluss.lake.writer.LakeTieringFactory; import org.apache.fluss.lake.writer.LakeWriter; +import org.apache.fluss.lake.writer.PartitionMarkDoneEnabler; import org.apache.fluss.lake.writer.WriterInitContext; +import org.apache.fluss.metadata.TableInfo; import java.io.IOException; /** Implementation of {@link LakeTieringFactory} for Paimon . */ public class PaimonLakeTieringFactory - implements LakeTieringFactory { + implements LakeTieringFactory, + PartitionMarkDoneEnabler { private static final long serialVersionUID = 1L; @@ -60,4 +63,9 @@ public LakeCommitter createLakeCommitter( public SimpleVersionedSerializer getCommittableSerializer() { return new PaimonCommittableSerializer(); } + + @Override + public boolean isPartitionMarkDoneEnabled(TableInfo tableInfo) { + return PaimonPartitionMarkDone.isEnabled(tableInfo); + } } diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonPartitionMarkDone.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonPartitionMarkDone.java new file mode 100644 index 00000000000..89542f7538e --- /dev/null +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonPartitionMarkDone.java @@ -0,0 +1,483 @@ +/* + * 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.fluss.lake.paimon.tiering; + +import org.apache.fluss.config.AutoPartitionTimeUnit; +import org.apache.fluss.metadata.ResolvedPartitionSpec; +import org.apache.fluss.metadata.TableInfo; +import org.apache.fluss.utils.AutoPartitionStrategy; +import org.apache.fluss.utils.IOUtils; +import org.apache.fluss.utils.PartitionUtils; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.manifest.PartitionEntry; +import org.apache.paimon.options.ConfigOption; +import org.apache.paimon.options.ConfigOptions; +import org.apache.paimon.options.Options; +import org.apache.paimon.partition.PartitionTimeExtractor; +import org.apache.paimon.partition.actions.PartitionMarkDoneAction; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.utils.InternalRowPartitionComputer; +import org.apache.paimon.utils.PartitionPathUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.time.Duration; +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.time.format.DateTimeFormatter; +import java.time.format.DateTimeFormatterBuilder; +import java.time.temporal.ChronoField; +import java.time.temporal.IsoFields; +import java.time.temporal.TemporalAccessor; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; +import static org.apache.fluss.utils.Preconditions.checkState; + +/** + * Judges and marks idle partitions of a Paimon lake table as done during tiering. The judgment + * ({@code done <=> now - max(lastUpdateTime, partitionEndTime) > partition.idle-time-to-done}) is + * delegated to {@link PartitionMarkDoneTrigger} copied from Paimon. It reuses the semantics of + * Paimon's options ({@code partition.idle-time-to-done}, {@code partition.time-interval}, {@code + * partition.timestamp-pattern/formatter}) but reads them exclusively from the {@code paimon.} + * prefixed Fluss table custom properties as the single source of truth: options configured on the + * Paimon table directly are deliberately not honored. It executes the idempotent Paimon mark-done + * actions ({@code partition.mark-done-action}). For auto-partitioned tables the partition end time + * is derived from the auto-partition time unit instead of {@code partition.time-interval}. See + * {@link MarkDoneState} for the state model. + */ +public class PaimonPartitionMarkDone implements AutoCloseable { + + private static final Logger LOG = LoggerFactory.getLogger(PaimonPartitionMarkDone.class); + + /** Snapshot property key storing the mark-done state JSON. */ + public static final String MARK_DONE_STATE_PROPERTY = "fluss.tiering.mark-done-state"; + + /** The prefix of Paimon options in the Fluss table custom properties. */ + private static final String PAIMON_PROPERTY_PREFIX = "paimon."; + + /** Same option as Paimon's FlinkConnectorOptions#PARTITION_IDLE_TIME_TO_DONE. */ + private static final ConfigOption PARTITION_IDLE_TIME_TO_DONE = + ConfigOptions.key("partition.idle-time-to-done").durationType().noDefaultValue(); + + /** Same option as Paimon's FlinkConnectorOptions#PARTITION_TIME_INTERVAL. */ + private static final ConfigOption PARTITION_TIME_INTERVAL = + ConfigOptions.key("partition.time-interval").durationType().noDefaultValue(); + + /** Same option as Paimon's FlinkConnectorOptions#PARTITION_MARK_DONE_MODE. */ + private static final ConfigOption PARTITION_MARK_DONE_MODE = + ConfigOptions.key("partition.mark-done-action.mode") + .stringType() + .defaultValue("process-time"); + + /** Same option as Paimon's CoreOptions#PARTITION_TIMESTAMP_PATTERN. */ + private static final ConfigOption PARTITION_TIMESTAMP_PATTERN = + ConfigOptions.key("partition.timestamp-pattern").stringType().noDefaultValue(); + + /** Same option as Paimon's CoreOptions#PARTITION_TIMESTAMP_FORMATTER. */ + private static final ConfigOption PARTITION_TIMESTAMP_FORMATTER = + ConfigOptions.key("partition.timestamp-formatter").stringType().noDefaultValue(); + + private final FileStoreTable fileStoreTable; + private final TableInfo tableInfo; + private final List partitionKeys; + private final long idleTimeToDoneMillis; + // partition end time interval for non auto-partitioned tables (required by isEnabled, + // same as Paimon), null for auto-partitioned tables whose interval derives from the + // auto-partition time unit + @Nullable private final Long timeIntervalMillis; + private final AutoPartitionStrategy autoPartitionStrategy; + private final PartitionTimeExtractor partitionTimeExtractor; + private final InternalRowPartitionComputer partitionComputer; + private final List markDoneActions; + + public PaimonPartitionMarkDone(FileStoreTable fileStoreTable, TableInfo tableInfo) { + checkState( + isEnabled(tableInfo), + "Partition mark-done is not enabled for table %s.", + tableInfo.getTablePath()); + this.fileStoreTable = fileStoreTable; + this.tableInfo = tableInfo; + this.partitionKeys = tableInfo.getPartitionKeys(); + Options options = paimonOptionsInCustomProperties(tableInfo); + this.idleTimeToDoneMillis = options.get(PARTITION_IDLE_TIME_TO_DONE).toMillis(); + Duration timeInterval = options.get(PARTITION_TIME_INTERVAL); + this.timeIntervalMillis = timeInterval == null ? null : timeInterval.toMillis(); + this.autoPartitionStrategy = tableInfo.getTableConfig().getAutoPartitionStrategy(); + // pattern/formatter may be null, then Paimon's default extraction rule applies + String timestampFormatter = options.get(PARTITION_TIMESTAMP_FORMATTER); + if (timestampFormatter != null) { + // an invalid formatter syntax must disable mark-done as a whole: otherwise every + // partition would be treated as illegal and drained from the pending set + DateTimeFormatter.ofPattern(timestampFormatter); + } + this.partitionTimeExtractor = + new PartitionTimeExtractor( + options.get(PARTITION_TIMESTAMP_PATTERN), timestampFormatter); + this.partitionComputer = + new InternalRowPartitionComputer( + fileStoreTable.coreOptions().partitionDefaultName(), + fileStoreTable.schema().logicalPartitionType(), + fileStoreTable.partitionKeys().toArray(new String[0]), + fileStoreTable.coreOptions().legacyPartitionName()); + this.markDoneActions = + PartitionMarkDoneAction.createActions( + PaimonPartitionMarkDone.class.getClassLoader(), + fileStoreTable, + // the action configuration also comes from the Fluss custom properties + // (the single source of truth), the table only provides the environment + new CoreOptions(options.toMap())); + } + + /** + * Whether partition mark-done is enabled for the given table according to the {@code paimon.} + * prefixed Fluss table custom properties, the single source of truth (options configured on the + * Paimon table directly are deliberately not honored). Besides {@code + * partition.idle-time-to-done}, a derivable partition end time is required (auto-partitioning + * or {@code partition.time-interval}, same as Paimon): otherwise no partition could ever be + * judged done while the pending state would grow unbounded and keep triggering useless + * maintenance rounds. + */ + public static boolean isEnabled(TableInfo tableInfo) { + if (!tableInfo.isPartitioned()) { + return false; + } + Options options = paimonOptionsInCustomProperties(tableInfo); + if (!options.containsKey(PARTITION_IDLE_TIME_TO_DONE.key())) { + return false; + } + try { + options.get(PARTITION_IDLE_TIME_TO_DONE); + options.get(PARTITION_TIME_INTERVAL); + } catch (Exception e) { + LOG.warn( + "Invalid mark-done duration option for table {}, " + + "partition mark-done is disabled.", + tableInfo.getTablePath(), + e); + return false; + } + if (!tableInfo.getTableConfig().getAutoPartitionStrategy().isAutoPartitionEnabled() + && !options.containsKey(PARTITION_TIME_INTERVAL.key())) { + LOG.warn( + "Option {} is set for table {} but the partition end time can't be derived " + + "(neither auto-partitioning nor option {} is set), " + + "partition mark-done is disabled.", + PARTITION_IDLE_TIME_TO_DONE.key(), + tableInfo.getTablePath(), + PARTITION_TIME_INTERVAL.key()); + return false; + } + // only the process-time mode is supported: the tiering service can't derive the + // watermark of the table yet, silently degrading the watermark mode to process-time + // would trigger the done actions ahead of the user-configured watermark boundary + String markDoneMode = options.get(PARTITION_MARK_DONE_MODE); + if (!"process-time".equalsIgnoreCase(markDoneMode)) { + LOG.warn( + "Option {} is set to {} for table {} but only the process-time mode is " + + "supported, partition mark-done is disabled.", + PARTITION_MARK_DONE_MODE.key(), + markDoneMode, + tableInfo.getTablePath()); + return false; + } + return true; + } + + /** Extracts the Paimon options carried in the Fluss table custom properties. */ + private static Options paimonOptionsInCustomProperties(TableInfo tableInfo) { + Map paimonOptions = new HashMap<>(); + for (Map.Entry entry : tableInfo.getCustomProperties().toMap().entrySet()) { + if (entry.getKey().startsWith(PAIMON_PROPERTY_PREFIX)) { + paimonOptions.put( + entry.getKey().substring(PAIMON_PROPERTY_PREFIX.length()), + entry.getValue()); + } + } + return Options.fromMap(paimonOptions); + } + + /** Extracts the Fluss partition names of the given commit messages. */ + public Set extractTieredPartitions(List commitMessages) { + Set tieredPartitions = new HashSet<>(); + for (CommitMessage commitMessage : commitMessages) { + tieredPartitions.add(toPartitionName(commitMessage.partition())); + } + return tieredPartitions; + } + + /** + * Runs one round: cold-start backfill, tracks tiered partitions and marks idle partitions done + * (done partitions are removed from the pending set). Dropped or expired partitions are not + * pruned specially: same as Paimon's native listener, they stay pending until the idle time + * elapses and are then marked done. Returns the new state JSON, or null if unchanged. + */ + @Nullable + public String run(@Nullable String previousStateJson, Set tieredPartitions) + throws Exception { + MarkDoneState previousState = parsePreviousState(previousStateJson); + + long now = System.currentTimeMillis(); + boolean initialized = previousState.isInitialized(); + PartitionMarkDoneTrigger trigger = + new PartitionMarkDoneTrigger( + previousState.getPendingPartitions(), + this::extractPartitionEndTime, + idleTimeToDoneMillis); + + // cold start: backfill all existing lake partitions into the pending set; on failure + // keep initialized=false so the next round retries the backfill, while the tiered + // partitions of this round are still tracked and persisted below + if (!initialized) { + try { + for (Map.Entry entry : listLivePartitions().entrySet()) { + if (!trigger.pendingPartitions().containsKey(entry.getKey())) { + trigger.notifyPartition( + entry.getKey(), entry.getValue().lastFileCreationTime()); + } + } + initialized = true; + } catch (Exception e) { + LOG.warn( + "Failed to backfill lake partitions of table {}, " + + "will retry the cold start in the next round.", + tableInfo.getTablePath(), + e); + } + } + + // track tiered partitions; this also re-adds a done partition on late data + for (String tieredPartition : tieredPartitions) { + trigger.notifyPartition(tieredPartition, now); + } + + // done judgment: idle partitions are marked done and removed from the pending set + Map lastUpdateTimes = new HashMap<>(trigger.pendingPartitions()); + List donePartitions = trigger.donePartitions(now); + + // done-is-delete: done partitions are already removed from the trigger's pending set. + // Execute the idempotent actions unconditionally (same as Paimon's native listener): a + // done partition may legitimately hold zero files (e.g. a PK partition whose data was + // fully deleted and compacted) and thus be invisible in the partition entries, but it + // still deserves the done signal + for (String partitionName : donePartitions) { + try { + markPartitionDone(partitionName); + } catch (Exception e) { + LOG.warn( + "Failed to mark partition {} of table {} as done, " + + "will retry in the next round.", + partitionName, + tableInfo.getTablePath(), + e); + // keep the original last update time so the partition is judged done again + trigger.notifyPartition(partitionName, lastUpdateTimes.get(partitionName)); + } + } + + MarkDoneState newState = new MarkDoneState(initialized, trigger.pendingPartitions()); + if (newState.equals(previousState)) { + return null; + } + return MarkDoneStateJsonSerde.toJson(newState); + } + + /** Parses the restored state; a corrupt state falls back to a cold-start re-initialization. */ + private MarkDoneState parsePreviousState(@Nullable String previousStateJson) { + if (previousStateJson == null) { + return MarkDoneState.empty(); + } + try { + return MarkDoneStateJsonSerde.fromJson(previousStateJson); + } catch (Exception e) { + LOG.warn( + "Corrupt mark-done state of table {}, re-initializing via cold start.", + tableInfo.getTablePath(), + e); + return MarkDoneState.empty(); + } + } + + /** Executes the configured idempotent Paimon mark-done actions. */ + private void markPartitionDone(String partitionName) throws Exception { + LinkedHashMap partitionSpec = toPartitionSpec(partitionName); + String partitionPath = PartitionPathUtils.generatePartitionPath(partitionSpec); + LOG.info("Mark partition {} of table {} as done.", partitionPath, tableInfo.getTablePath()); + for (PartitionMarkDoneAction action : markDoneActions) { + action.markDone(partitionPath); + } + } + + /** Lists the live partitions of the lake table, keyed by Fluss partition name. */ + private Map listLivePartitions() { + Map livePartitions = new HashMap<>(); + if (fileStoreTable.snapshotManager().latestSnapshotId() == null) { + return livePartitions; + } + for (PartitionEntry partitionEntry : + fileStoreTable.newSnapshotReader().partitionEntries()) { + livePartitions.put(toPartitionName(partitionEntry.partition()), partitionEntry); + } + return livePartitions; + } + + private String toPartitionName(BinaryRow partition) { + return String.join( + ResolvedPartitionSpec.PARTITION_SPEC_SEPARATOR, + partitionComputer.generatePartValues(partition).values()); + } + + private LinkedHashMap toPartitionSpec(String partitionName) { + ResolvedPartitionSpec resolvedPartitionSpec = + ResolvedPartitionSpec.fromPartitionName(partitionKeys, partitionName); + LinkedHashMap partitionSpec = new LinkedHashMap<>(); + List partitionValues = resolvedPartitionSpec.getPartitionValues(); + for (int i = 0; i < partitionKeys.size(); i++) { + partitionSpec.put(partitionKeys.get(i), partitionValues.get(i)); + } + return partitionSpec; + } + + /** + * Extracts the partition end time in epoch millis: partition start time (from the + * auto-partition time unit, or Paimon's partition.timestamp-pattern/formatter) plus one + * partition interval; null if it cannot be derived. Times are resolved in the zone that + * generated the partition name: the table's auto-partition time zone for auto-partitioned + * tables, or the JVM default zone otherwise (same as Paimon). + */ + @Nullable + private Long extractPartitionEndTime(String partitionName) { + try { + List partitionValues = + ResolvedPartitionSpec.fromPartitionName(partitionKeys, partitionName) + .getPartitionValues(); + if (autoPartitionStrategy.isAutoPartitionEnabled()) { + String timeValue = partitionValues.get(autoPartitionTimeKeyIndex()); + LocalDateTime startTime = parseAutoPartitionTime(timeValue, autoPartitionStrategy); + LocalDateTime endTime = plusOneTimeUnit(startTime, autoPartitionStrategy); + return endTime.atZone(autoPartitionStrategy.timeZone().toZoneId()) + .toInstant() + .toEpochMilli(); + } else { + LocalDateTime startTime = + partitionTimeExtractor.extract(partitionKeys, partitionValues); + return startTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() + + checkNotNull(timeIntervalMillis); + } + } catch (Exception e) { + LOG.debug( + "Fail to extract partition end time from partition {} of table {}.", + partitionName, + tableInfo.getTablePath(), + e); + return null; + } + } + + private int autoPartitionTimeKeyIndex() { + if (partitionKeys.size() == 1) { + return 0; + } + int index = partitionKeys.indexOf(autoPartitionStrategy.key()); + if (index < 0) { + throw new IllegalStateException( + String.format( + "Auto partition time key %s is not found in partition keys %s.", + autoPartitionStrategy.key(), partitionKeys)); + } + return index; + } + + private static LocalDateTime parseAutoPartitionTime( + String timeValue, AutoPartitionStrategy strategy) { + AutoPartitionTimeUnit timeUnit = strategy.timeUnit(); + if (timeUnit == AutoPartitionTimeUnit.QUARTER) { + return parseQuarterPartitionTime(timeValue, strategy); + } + String format = PartitionUtils.getPartitionTimeFormat(timeUnit, strategy); + DateTimeFormatter formatter = + new DateTimeFormatterBuilder() + .appendPattern(format) + .parseDefaulting(ChronoField.MONTH_OF_YEAR, 1) + .parseDefaulting(ChronoField.DAY_OF_MONTH, 1) + .parseDefaulting(ChronoField.HOUR_OF_DAY, 0) + .parseDefaulting(ChronoField.MINUTE_OF_HOUR, 0) + .parseDefaulting(ChronoField.SECOND_OF_MINUTE, 0) + .toFormatter(); + return LocalDateTime.parse(timeValue, formatter); + } + + private static LocalDateTime parseQuarterPartitionTime( + String timeValue, AutoPartitionStrategy strategy) { + int year; + int quarter; + if (strategy.timeFormat() == null) { + // default quarter format is 'yyyyQ' which can't be parsed field by field + year = Integer.parseInt(timeValue.substring(0, 4)); + quarter = Integer.parseInt(timeValue.substring(4)); + } else { + // resolving a custom quarter format (e.g. yyyy-'Q'Q) into a date would conflict + // with month/day defaults for Q2-Q4, so extract the year and quarter directly + // (via getLong since TemporalAccessor#get can't range-check the quarter field) + TemporalAccessor accessor = + DateTimeFormatter.ofPattern(strategy.timeFormat()).parse(timeValue); + year = + (int) + (accessor.isSupported(ChronoField.YEAR) + ? accessor.getLong(ChronoField.YEAR) + : accessor.getLong(ChronoField.YEAR_OF_ERA)); + quarter = (int) accessor.getLong(IsoFields.QUARTER_OF_YEAR); + } + return LocalDateTime.of(year, (quarter - 1) * 3 + 1, 1, 0, 0); + } + + private static LocalDateTime plusOneTimeUnit( + LocalDateTime startTime, AutoPartitionStrategy strategy) { + switch (strategy.timeUnit()) { + case YEAR: + return startTime.plusYears(1); + case QUARTER: + return startTime.plusMonths(3); + case MONTH: + return startTime.plusMonths(1); + case DAY: + return startTime.plusDays(1); + case HOUR: + return startTime.plusHours(1); + default: + throw new IllegalArgumentException("Unsupported time unit: " + strategy.timeUnit()); + } + } + + @Override + public void close() { + for (PartitionMarkDoneAction action : markDoneActions) { + IOUtils.closeQuietly(action); + } + } +} diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PartitionMarkDoneTrigger.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PartitionMarkDoneTrigger.java new file mode 100644 index 00000000000..93f39a330e7 --- /dev/null +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PartitionMarkDoneTrigger.java @@ -0,0 +1,108 @@ +/* + * 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.fluss.lake.paimon.tiering; + +import org.apache.paimon.utils.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; + +/* This file is based on source code of Apache Paimon Project (https://paimon.apache.org/), licensed by the Apache + * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for + * additional information regarding copyright ownership. */ + +/** + * Trigger to mark partitions done, copied from Paimon's {@code + * org.apache.paimon.flink.sink.listener.PartitionMarkDoneTrigger} (release 1.3) with the following + * modifications: + * + *

    + *
  • the restored state carries the last update time per partition instead of resetting it to + * the current time, since the tiering service recreates the trigger for every round; + *
  • the partition end time is extracted by an injected {@link PartitionEndTimeExtractor} to + * also support Fluss auto-partitioned tables besides Paimon's timestamp-pattern/formatter + * plus time-interval rule; + *
  • Flink operator state, end-input and watermark related code is removed. + *
+ */ +public class PartitionMarkDoneTrigger { + + private static final Logger LOG = LoggerFactory.getLogger(PartitionMarkDoneTrigger.class); + + private final PartitionEndTimeExtractor endTimeExtractor; + private final long idleTime; + private final Map pendingPartitions; + + public PartitionMarkDoneTrigger( + Map restoredPendingPartitions, + PartitionEndTimeExtractor endTimeExtractor, + long idleTime) { + this.pendingPartitions = new HashMap<>(restoredPendingPartitions); + this.endTimeExtractor = endTimeExtractor; + this.idleTime = idleTime; + } + + public void notifyPartition(String partition, long currentTimeMillis) { + if (!StringUtils.isNullOrWhitespaceOnly(partition)) { + this.pendingPartitions.put(partition, currentTimeMillis); + } + } + + public List donePartitions(long currentTimeMillis) { + List needDone = new ArrayList<>(); + Iterator> iter = pendingPartitions.entrySet().iterator(); + while (iter.hasNext()) { + Map.Entry entry = iter.next(); + String partition = entry.getKey(); + long lastUpdateTime = entry.getValue(); + + Long partitionEndTime = endTimeExtractor.extract(partition); + // skip illegal partition + if (partitionEndTime == null) { + LOG.warn("Can't extract partition end time from partition {}, skip it.", partition); + iter.remove(); + continue; + } + lastUpdateTime = Math.max(lastUpdateTime, partitionEndTime); + + if (currentTimeMillis - lastUpdateTime > idleTime) { + needDone.add(partition); + iter.remove(); + } + } + return needDone; + } + + /** Returns the pending (not yet done) partitions to be persisted as state. */ + public Map pendingPartitions() { + return pendingPartitions; + } + + /** Extracts the partition end time in epoch millis, null if it cannot be derived. */ + public interface PartitionEndTimeExtractor { + @Nullable + Long extract(String partition); + } +} diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/testutils/FlinkPaimonTieringTestBase.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/testutils/FlinkPaimonTieringTestBase.java index 90ab136fde9..507435d3cfb 100644 --- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/testutils/FlinkPaimonTieringTestBase.java +++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/testutils/FlinkPaimonTieringTestBase.java @@ -124,13 +124,18 @@ public void beforeEach() { } protected JobClient buildTieringJob(StreamExecutionEnvironment execEnv) throws Exception { + return buildTieringJob(execEnv, new Configuration()); + } + + protected JobClient buildTieringJob( + StreamExecutionEnvironment execEnv, Configuration lakeTieringConfig) throws Exception { Configuration flussConfig = new Configuration(clientConf); flussConfig.set(POLL_TIERING_TABLE_INTERVAL, Duration.ofMillis(500L)); return LakeTieringJobBuilder.newBuilder( execEnv, flussConfig, Configuration.fromMap(getPaimonCatalogConf()), - new Configuration(), + lakeTieringConfig, DataLakeFormat.PAIMON.toString()) .build(); } diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonPartitionMarkDoneTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonPartitionMarkDoneTest.java new file mode 100644 index 00000000000..77918dae79e --- /dev/null +++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonPartitionMarkDoneTest.java @@ -0,0 +1,977 @@ +/* + * 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.fluss.lake.paimon.tiering; + +import org.apache.fluss.config.AutoPartitionTimeUnit; +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.lake.committer.CommittedLakeSnapshot; +import org.apache.fluss.lake.committer.CommitterInitContext; +import org.apache.fluss.lake.committer.LakeCommitter; +import org.apache.fluss.lake.committer.PartitionMarkDoneMaintainer; +import org.apache.fluss.lake.writer.LakeWriter; +import org.apache.fluss.lake.writer.WriterInitContext; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TableDescriptor; +import org.apache.fluss.metadata.TableInfo; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.record.ChangeType; +import org.apache.fluss.record.GenericRecord; +import org.apache.fluss.row.BinaryString; +import org.apache.fluss.row.GenericRow; + +import org.apache.paimon.catalog.Catalog; +import org.apache.paimon.catalog.CatalogContext; +import org.apache.paimon.catalog.CatalogFactory; +import org.apache.paimon.manifest.ManifestCommittable; +import org.apache.paimon.options.Options; +import org.apache.paimon.partition.actions.PartitionMarkDoneAction; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.sink.TableCommitImpl; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import javax.annotation.Nullable; + +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.apache.fluss.lake.committer.LakeCommitter.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY; +import static org.apache.fluss.lake.paimon.tiering.PaimonLakeTieringFactory.FLUSS_LAKE_TIERING_COMMIT_USER; +import static org.apache.fluss.lake.paimon.tiering.PaimonPartitionMarkDone.MARK_DONE_STATE_PROPERTY; +import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon; +import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME; +import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME; +import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME; +import static org.apache.fluss.record.TestData.DEFAULT_REMOTE_DATA_DIR; +import static org.apache.paimon.table.sink.BatchWriteBuilder.COMMIT_IDENTIFIER; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** The UT for partition mark-done during tiering to Paimon. */ +class PaimonPartitionMarkDoneTest { + + private static final String DATABASE = "paimon"; + private static final String IDLE_TIME_KEY = "partition.idle-time-to-done"; + private static final String TIME_INTERVAL_KEY = "partition.time-interval"; + + private @TempDir File tempWarehouseDir; + private PaimonLakeTieringFactory paimonLakeTieringFactory; + private Catalog paimonCatalog; + + @BeforeEach + void beforeEach() { + Configuration configuration = new Configuration(); + configuration.setString("warehouse", tempWarehouseDir.toString()); + paimonLakeTieringFactory = new PaimonLakeTieringFactory(configuration); + paimonCatalog = + CatalogFactory.createCatalog( + CatalogContext.create(Options.fromMap(configuration.toMap()))); + } + + @Test + void testMarkDoneLifecycle() throws Exception { + TablePath tablePath = TablePath.of(DATABASE, "test_mark_done_lifecycle"); + // the mark-done options are carried in the Fluss custom properties only + createPaimonTable(tablePath, Collections.emptyMap()); + TableInfo tableInfo = markDoneTableInfo(tablePath, false); + + // first data commit: cold start along with the commit, all time-parsable partitions + // pending; the illegal partition 'px' (no time can be extracted) is dropped without + // marking done, same as Paimon + long snapshot1 = writeAndCommit(tablePath, tableInfo, "2024-01-01", "2024-01-02", "px"); + MarkDoneState state = getMarkDoneState(tablePath, snapshot1); + assertThat(state.isInitialized()).isTrue(); + assertThat(state.getPendingPartitions()).containsOnlyKeys("2024-01-01", "2024-01-02"); + + // empty round: idle partitions are marked done via a properties-only snapshot which + // carries a freshly prepared offsets file + Thread.sleep(50); + try (LakeCommitter lakeCommitter = + createLakeCommitter(tablePath, tableInfo)) { + CommittedLakeSnapshot maintenanceSnapshot = + commitMarkDoneMaintenance(lakeCommitter, "offsets-2"); + assertThat(maintenanceSnapshot).isNotNull(); + assertThat(maintenanceSnapshot.getLakeSnapshotId()).isEqualTo(snapshot1 + 1); + // the maintenance snapshot carries its own offsets file, not the previous one + assertThat(maintenanceSnapshot.getSnapshotProperties()) + .containsEntry(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY, "offsets-2"); + assertThat(getSnapshotProperties(tablePath, snapshot1 + 1)) + .isEqualTo(maintenanceSnapshot.getSnapshotProperties()); + } + MarkDoneState state2 = getMarkDoneState(tablePath, snapshot1 + 1); + assertThat(state2.isInitialized()).isTrue(); + assertThat(state2.getPendingPartitions()).isEmpty(); + // the default success-file action wrote _SUCCESS files + assertThat(successFile(tablePath, "2024-01-01")).exists(); + assertThat(successFile(tablePath, "2024-01-02")).exists(); + assertThat(successFile(tablePath, "px")).doesNotExist(); + + // another empty round: state unchanged, no snapshot created + try (LakeCommitter lakeCommitter = + createLakeCommitter(tablePath, tableInfo)) { + assertThat(commitMarkDoneMaintenance(lakeCommitter, "offsets-3")).isNull(); + } + + // late data for 2024-01-01: re-added to pending and marked done again later + long snapshot3 = writeAndCommit(tablePath, tableInfo, "2024-01-01"); + assertThat(getMarkDoneState(tablePath, snapshot3).getPendingPartitions()) + .containsOnlyKeys("2024-01-01"); + + Thread.sleep(50); + try (LakeCommitter lakeCommitter = + createLakeCommitter(tablePath, tableInfo)) { + assertThat(commitMarkDoneMaintenance(lakeCommitter, "offsets-4")).isNotNull(); + } + assertThat(getMarkDoneState(tablePath, snapshot3 + 1).getPendingPartitions()).isEmpty(); + assertThat(successFile(tablePath, "2024-01-01")).exists(); + } + + @Test + void testColdStartBackfill() throws Exception { + TablePath tablePath = TablePath.of(DATABASE, "test_mark_done_cold_start"); + // first tier data without the mark-done custom properties + createPaimonTable(tablePath, Collections.emptyMap()); + TableInfo tableInfo = tableInfo(tablePath, false); + long snapshot1 = writeAndCommit(tablePath, tableInfo, "2024-01-01", "2024-01-02"); + assertThat(getSnapshotProperties(tablePath, snapshot1)) + .doesNotContainKey(MARK_DONE_STATE_PROPERTY); + + // then enable mark-done via the Fluss custom properties: the cold start backfills the + // existing idle partitions + Thread.sleep(50); + TableInfo enabledTableInfo = markDoneTableInfo(tablePath, false); + try (LakeCommitter lakeCommitter = + createLakeCommitter(tablePath, enabledTableInfo)) { + assertThat(commitMarkDoneMaintenance(lakeCommitter, "offsets-2")).isNotNull(); + } + MarkDoneState state = getMarkDoneState(tablePath, snapshot1 + 1); + assertThat(state.isInitialized()).isTrue(); + assertThat(state.getPendingPartitions()).isEmpty(); + assertThat(successFile(tablePath, "2024-01-01")).exists(); + assertThat(successFile(tablePath, "2024-01-02")).exists(); + } + + @Test + void testPartitionEndTimeGuardsFuturePartition() throws Exception { + TablePath tablePath = TablePath.of(DATABASE, "test_mark_done_partition_end_time"); + createPaimonTable(tablePath, Collections.emptyMap()); + // auto-partitioned by day: partition end time is derived from the partition name + TableInfo tableInfo = markDoneTableInfo(tablePath, true); + + long snapshot1 = writeAndCommit(tablePath, tableInfo, "20200101", "99991231"); + Thread.sleep(50); + try (LakeCommitter lakeCommitter = + createLakeCommitter(tablePath, tableInfo)) { + assertThat(commitMarkDoneMaintenance(lakeCommitter, "offsets-2")).isNotNull(); + } + // the ancient partition is done, the future partition is guarded by its end time + MarkDoneState state = getMarkDoneState(tablePath, snapshot1 + 1); + assertThat(state.getPendingPartitions()).containsOnlyKeys("99991231"); + assertThat(successFile(tablePath, "20200101")).exists(); + assertThat(successFile(tablePath, "99991231")).doesNotExist(); + } + + @Test + void testCustomQuarterFormat() throws Exception { + TablePath tablePath = TablePath.of(DATABASE, "test_mark_done_custom_quarter_format"); + createPaimonTable(tablePath, Collections.emptyMap()); + // auto-partitioned by quarter with the custom time format yyyy-'Q'Q + TableInfo tableInfo = quarterTableInfo(tablePath); + + long snapshot1 = writeAndCommit(tablePath, tableInfo, "2024-Q3", "9999-Q4"); + Thread.sleep(50); + try (LakeCommitter lakeCommitter = + createLakeCommitter(tablePath, tableInfo)) { + assertThat(commitMarkDoneMaintenance(lakeCommitter, "offsets-2")).isNotNull(); + } + // the elapsed quarter is marked done; the future quarter must stay pending instead of + // being dropped as unparsable (regression: Q2-Q4 used to conflict with the month + // default while resolving the custom quarter format) + MarkDoneState state = getMarkDoneState(tablePath, snapshot1 + 1); + assertThat(state.getPendingPartitions()).containsOnlyKeys("9999-Q4"); + assertThat(successFile(tablePath, "2024-Q3")).exists(); + assertThat(successFile(tablePath, "9999-Q4")).doesNotExist(); + } + + @Test + void testDisabledWithoutTimeInterval() throws Exception { + TablePath tablePath = TablePath.of(DATABASE, "test_mark_done_disabled_without_interval"); + // non auto-partitioned table with only the idle custom property: the partition end + // time can never be derived, so mark-done must be disabled instead of accumulating + // unbounded pending state and triggering useless maintenance rounds + createPaimonTable(tablePath, Collections.emptyMap()); + TableInfo tableInfo = + TableInfo.of( + tablePath, + 0, + 1, + newTableBuilder(false) + .customProperty("paimon." + IDLE_TIME_KEY, "1 ms") + .build(), + DEFAULT_REMOTE_DATA_DIR, + 1L, + 1L); + + assertThat(paimonLakeTieringFactory.isPartitionMarkDoneEnabled(tableInfo)).isFalse(); + + long snapshot1 = writeAndCommit(tablePath, tableInfo, "2024-01-01"); + assertThat(getSnapshotProperties(tablePath, snapshot1)) + .doesNotContainKey(MARK_DONE_STATE_PROPERTY); + try (LakeCommitter lakeCommitter = + createLakeCommitter(tablePath, tableInfo)) { + assertThat(commitMarkDoneMaintenance(lakeCommitter, "offsets-2")).isNull(); + } + } + + @Test + void testPaimonSideOptionsNotHonored() throws Exception { + TablePath tablePath = TablePath.of(DATABASE, "test_mark_done_paimon_side_options"); + // mark-done options configured on the Paimon table directly are deliberately not + // honored: the Fluss custom properties are the single source of truth + createPaimonTable(tablePath, markDoneOptions()); + TableInfo tableInfo = tableInfo(tablePath, false); + + assertThat(paimonLakeTieringFactory.isPartitionMarkDoneEnabled(tableInfo)).isFalse(); + long snapshot1 = writeAndCommit(tablePath, tableInfo, "2024-01-01"); + assertThat(getSnapshotProperties(tablePath, snapshot1)) + .doesNotContainKey(MARK_DONE_STATE_PROPERTY); + + // while the switch in the Fluss custom properties works without any lake access + // (the Paimon table of the path doesn't even exist yet) + TablePath notCreatedTablePath = TablePath.of(DATABASE, "test_mark_done_enabler_props"); + assertThat( + paimonLakeTieringFactory.isPartitionMarkDoneEnabled( + markDoneTableInfo(notCreatedTablePath, false))) + .isTrue(); + } + + @Test + void testPartitionExpirationDoesNotBreakMarkDoneState() throws Exception { + TablePath tablePath = TablePath.of(DATABASE, "test_mark_done_partition_expiration"); + // partition expiration configured on the Paimon table: Paimon appends an OVERWRITE + // snapshot with the same commit user and null properties right after our commit + Map paimonOptions = new HashMap<>(); + paimonOptions.put("partition.expiration-time", "1 d"); + paimonOptions.put("partition.expiration-check-interval", "10 min"); + paimonOptions.put("partition.timestamp-formatter", "yyyy-MM-dd"); + createPaimonTable(tablePath, paimonOptions); + TableInfo tableInfo = + TableInfo.of( + tablePath, + 0, + 1, + newTableBuilder(false) + .customProperty("paimon." + IDLE_TIME_KEY, "1 ms") + .customProperty("paimon." + TIME_INTERVAL_KEY, "1 d") + // enable snapshot auto-expiration so the committer runs the + // partition expiration check on every commit + .property(ConfigOptions.TABLE_DATALAKE_AUTO_EXPIRE_SNAPSHOT, true) + .build(), + DEFAULT_REMOTE_DATA_DIR, + 1L, + 1L); + + // '2020-01-01' is long expired for partition expiration, '9999-12-31' is alive + long snapshot1 = writeAndCommit(tablePath, tableInfo, "2020-01-01", "9999-12-31"); + + // the returned snapshot is the latest physical one, i.e. the expiration OVERWRITE + // snapshot appended within the same commit call, so Fluss reads don't resurrect the + // expired partitions; the offsets & state live on the data snapshot before it + FileStoreTable fileStoreTable = + (FileStoreTable) paimonCatalog.getTable(toPaimon(tablePath)); + assertThat(fileStoreTable.snapshotManager().latestSnapshotId()).isEqualTo(snapshot1); + assertThat(getSnapshotProperties(tablePath, snapshot1)).isNull(); + + Thread.sleep(50); + try (LakeCommitter lakeCommitter = + createLakeCommitter(tablePath, tableInfo)) { + // missing recovery pairs the latest snapshot id with the properties of the round + CommittedLakeSnapshot missing = lakeCommitter.getMissingLakeSnapshot(null); + assertThat(missing).isNotNull(); + assertThat(missing.getLakeSnapshotId()).isEqualTo(snapshot1); + assertThat(missing.getSnapshotProperties()) + .containsKey(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY) + .containsKey(MARK_DONE_STATE_PROPERTY); + assertThat(lakeCommitter.getMissingLakeSnapshot(snapshot1)).isNull(); + + // the mark-done state must still be found instead of restarting from cold start + CommittedLakeSnapshot maintenanceSnapshot = + commitMarkDoneMaintenance(lakeCommitter, "offsets-2"); + assertThat(maintenanceSnapshot).isNotNull(); + MarkDoneState state = + getMarkDoneState(tablePath, maintenanceSnapshot.getLakeSnapshotId()); + assertThat(state.isInitialized()).isTrue(); + assertThat(state.getPendingPartitions()).containsOnlyKeys("9999-12-31"); + assertThat(successFile(tablePath, "2020-01-01")).exists(); + } + } + + @Test + void testMaintenanceTailLookback() throws Exception { + TablePath tablePath = TablePath.of(DATABASE, "test_mark_done_maintenance_tail"); + createPaimonTable(tablePath, Collections.emptyMap()); + TableInfo tableInfo = markDoneTableInfo(tablePath, false); + + long snapshot1 = writeAndCommit(tablePath, tableInfo, "2024-01-01", "9999-12-31"); + + // simulate a batched partition expiration tail (e.g. partition.expiration-max-num=100 + // with batch size 1): 100 OVERWRITE snapshots with the same commit user and no + // properties, crossed by the unbounded walk-back + FileStoreTable fileStoreTable = + (FileStoreTable) paimonCatalog.getTable(toPaimon(tablePath)); + for (int i = 0; i < 100; i++) { + try (TableCommitImpl truncateCommit = + fileStoreTable.newCommit(FLUSS_LAKE_TIERING_COMMIT_USER)) { + truncateCommit.truncatePartitions( + Collections.singletonList(Collections.singletonMap("c3", "2024-01-01"))); + } + } + + Thread.sleep(50); + try (LakeCommitter lakeCommitter = + createLakeCommitter(tablePath, tableInfo)) { + // the properties of the round are found behind the full-length tail + CommittedLakeSnapshot missing = lakeCommitter.getMissingLakeSnapshot(snapshot1); + assertThat(missing).isNotNull(); + assertThat(missing.getLakeSnapshotId()).isEqualTo(snapshot1 + 100); + assertThat(missing.getSnapshotProperties()) + .containsKey(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY) + .containsKey(MARK_DONE_STATE_PROPERTY); + + // so is the mark-done state + CommittedLakeSnapshot maintenanceSnapshot = + commitMarkDoneMaintenance(lakeCommitter, "offsets-2"); + assertThat(maintenanceSnapshot).isNotNull(); + MarkDoneState state = + getMarkDoneState(tablePath, maintenanceSnapshot.getLakeSnapshotId()); + assertThat(state.isInitialized()).isTrue(); + assertThat(state.getPendingPartitions()).containsOnlyKeys("9999-12-31"); + } + } + + @Test + void testZeroFileDonePartitionStillMarkedDone() throws Exception { + TablePath tablePath = TablePath.of(DATABASE, "test_mark_done_zero_file_partition"); + createPaimonTable(tablePath, Collections.emptyMap()); + TableInfo tableInfo = markDoneTableInfo(tablePath, false); + + long snapshot1 = writeAndCommit(tablePath, tableInfo, "2024-01-01"); + + // empty the partition (like a PK partition whose data was fully deleted and + // compacted): it disappears from the partition entries though it legitimately existed + FileStoreTable fileStoreTable = + (FileStoreTable) paimonCatalog.getTable(toPaimon(tablePath)); + try (TableCommitImpl truncateCommit = fileStoreTable.newCommit("test-truncate")) { + truncateCommit.truncatePartitions( + Collections.singletonList(Collections.singletonMap("c3", "2024-01-01"))); + } + assertThat(fileStoreTable.newSnapshotReader().partitionEntries()).isEmpty(); + + // the partition is still marked done although it holds zero files + Thread.sleep(50); + try (LakeCommitter lakeCommitter = + createLakeCommitter(tablePath, tableInfo)) { + CommittedLakeSnapshot maintenanceSnapshot = + commitMarkDoneMaintenance(lakeCommitter, "offsets-2"); + assertThat(maintenanceSnapshot).isNotNull(); + assertThat( + getMarkDoneState(tablePath, maintenanceSnapshot.getLakeSnapshotId()) + .getPendingPartitions()) + .isEmpty(); + } + assertThat(successFile(tablePath, "2024-01-01")).exists(); + } + + @Test + void testActionConfigFromCustomProperties() throws Exception { + TablePath tablePath = TablePath.of(DATABASE, "test_mark_done_action_from_props"); + // the Paimon table itself carries a broken action config (custom action without the + // class): it must be ignored since the action config is also read from the Fluss + // custom properties only, falling back to the default success-file action + createPaimonTable( + tablePath, Collections.singletonMap("partition.mark-done-action", "custom")); + TableInfo tableInfo = markDoneTableInfo(tablePath, false); + + writeAndCommit(tablePath, tableInfo, "2024-01-01"); + Thread.sleep(50); + try (LakeCommitter lakeCommitter = + createLakeCommitter(tablePath, tableInfo)) { + assertThat(commitMarkDoneMaintenance(lakeCommitter, "offsets-2")).isNotNull(); + } + assertThat(successFile(tablePath, "2024-01-01")).exists(); + } + + @Test + void testWatermarkModeRejected() throws Exception { + TablePath tablePath = TablePath.of(DATABASE, "test_mark_done_watermark_mode"); + createPaimonTable(tablePath, Collections.emptyMap()); + // the watermark mode is not supported yet: it must be rejected with a warning instead + // of being silently degraded to the process-time judgment, which would trigger the + // done actions ahead of the user-configured watermark boundary + TableInfo tableInfo = + TableInfo.of( + tablePath, + 0, + 1, + newTableBuilder(false) + .customProperty("paimon." + IDLE_TIME_KEY, "1 ms") + .customProperty("paimon." + TIME_INTERVAL_KEY, "1 d") + .customProperty( + "paimon.partition.mark-done-action.mode", "watermark") + .build(), + DEFAULT_REMOTE_DATA_DIR, + 1L, + 1L); + + assertThat(paimonLakeTieringFactory.isPartitionMarkDoneEnabled(tableInfo)).isFalse(); + + long snapshot1 = writeAndCommit(tablePath, tableInfo, "2024-01-01"); + assertThat(getSnapshotProperties(tablePath, snapshot1)) + .doesNotContainKey(MARK_DONE_STATE_PROPERTY); + try (LakeCommitter lakeCommitter = + createLakeCommitter(tablePath, tableInfo)) { + assertThat(commitMarkDoneMaintenance(lakeCommitter, "offsets-2")).isNull(); + } + assertThat(successFile(tablePath, "2024-01-01")).doesNotExist(); + } + + @Test + void testLegacySnapshotWithoutPropertiesFailsFast() throws Exception { + TablePath tablePath = TablePath.of(DATABASE, "test_mark_done_legacy_snapshot"); + createPaimonTable(tablePath, Collections.emptyMap()); + TableInfo tableInfo = tableInfo(tablePath, false); + + // simulate a legacy (v0.7) Fluss data commit: same commit user, APPEND kind, no + // properties at all + FileStoreTable fileStoreTable = + (FileStoreTable) paimonCatalog.getTable(toPaimon(tablePath)); + try (TableCommitImpl legacyCommit = + fileStoreTable.newCommit(FLUSS_LAKE_TIERING_COMMIT_USER)) { + legacyCommit.ignoreEmptyCommit(false); + legacyCommit.commit(new ManifestCommittable(COMMIT_IDENTIFIER)); + } + + // the legacy snapshot can't be registered to Fluss (no offsets recorded), the + // missing-snapshot check must fail fast instead of silently re-tiering the data + // that the legacy snapshot already holds + try (LakeCommitter lakeCommitter = + createLakeCommitter(tablePath, tableInfo)) { + assertThatThrownBy(() -> lakeCommitter.getMissingLakeSnapshot(null)) + .isInstanceOf(IOException.class) + .hasMessageContaining("Failed to load committed lake snapshot properties"); + } + } + + @Test + void testDisabledByJobLevelSwitch() throws Exception { + TablePath tablePath = TablePath.of(DATABASE, "test_mark_done_job_level_switch"); + createPaimonTable(tablePath, Collections.emptyMap()); + // the table opts in via its custom properties but the job-level switch stays off + // (the default): no state is written and no maintenance happens + TableInfo tableInfo = markDoneTableInfo(tablePath, false); + + long snapshot1 = writeAndCommit(tablePath, tableInfo, new Configuration(), "2024-01-01"); + assertThat(getSnapshotProperties(tablePath, snapshot1)) + .doesNotContainKey(MARK_DONE_STATE_PROPERTY); + Thread.sleep(50); + try (LakeCommitter lakeCommitter = + createLakeCommitter(tablePath, tableInfo, new Configuration())) { + assertThat(commitMarkDoneMaintenance(lakeCommitter, "offsets-2")).isNull(); + } + assertThat(successFile(tablePath, "2024-01-01")).doesNotExist(); + } + + @Test + void testBadRestoredStateDoesNotFailDataCommit() throws Exception { + // a structurally illegal partition name ('a$b' can't be resolved against the single + // partition key) is dropped by the trigger; a syntactically or type-wise corrupt + // state JSON falls back to a cold-start re-initialization + Map badStates = new LinkedHashMap<>(); + badStates.put( + "test_mark_done_illegal_state", + MarkDoneStateJsonSerde.toJson( + new MarkDoneState(true, Collections.singletonMap("a$b", 1L)))); + badStates.put("test_mark_done_corrupt_state", "corrupt-json"); + badStates.put( + "test_mark_done_bad_time_state", + "{\"initialized\":true,\"pending\":{\"p\":\"bad\"}}"); + badStates.put("test_mark_done_bad_pending_state", "{\"initialized\":true,\"pending\":[]}"); + + for (Map.Entry badState : badStates.entrySet()) { + TablePath tablePath = TablePath.of(DATABASE, badState.getKey()); + createPaimonTable(tablePath, Collections.emptyMap()); + TableInfo tableInfo = markDoneTableInfo(tablePath, false); + + FileStoreTable fileStoreTable = + (FileStoreTable) paimonCatalog.getTable(toPaimon(tablePath)); + try (TableCommitImpl stateCommit = + fileStoreTable.newCommit(FLUSS_LAKE_TIERING_COMMIT_USER)) { + stateCommit.ignoreEmptyCommit(false); + ManifestCommittable committable = new ManifestCommittable(COMMIT_IDENTIFIER); + committable.addProperty(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY, "offsets"); + committable.addProperty(MARK_DONE_STATE_PROPERTY, badState.getValue()); + stateCommit.commit(committable); + } + + // the data commit succeeds, the bad state is healed and the tiered partition + // is tracked in the new state + long snapshot2 = writeAndCommit(tablePath, tableInfo, "2024-01-01"); + MarkDoneState state = getMarkDoneState(tablePath, snapshot2); + assertThat(state.isInitialized()).isTrue(); + assertThat(state.getPendingPartitions()).containsOnlyKeys("2024-01-01"); + + // the healed state works: the next round marks the partition done + Thread.sleep(50); + try (LakeCommitter lakeCommitter = + createLakeCommitter(tablePath, tableInfo)) { + assertThat(commitMarkDoneMaintenance(lakeCommitter, "offsets-2")).isNotNull(); + } + assertThat(successFile(tablePath, "2024-01-01")).exists(); + } + } + + @Test + void testInvalidConfigDisablesMarkDone() throws Exception { + TablePath tablePath = TablePath.of(DATABASE, "test_mark_done_invalid_config"); + createPaimonTable(tablePath, Collections.emptyMap()); + + // an invalid idle duration is rejected by the cheap switch + TableInfo invalidDuration = + TableInfo.of( + tablePath, + 0, + 1, + newTableBuilder(false) + .customProperty("paimon." + IDLE_TIME_KEY, "not-a-duration") + .customProperty("paimon." + TIME_INTERVAL_KEY, "1 d") + .build(), + DEFAULT_REMOTE_DATA_DIR, + 1L, + 1L); + assertThat(paimonLakeTieringFactory.isPartitionMarkDoneEnabled(invalidDuration)).isFalse(); + + // a custom action without its class passes the switch but only disables mark-done + // in the committer instead of failing the committer creation + TableInfo invalidAction = + TableInfo.of( + tablePath, + 0, + 1, + newTableBuilder(false) + .customProperty("paimon." + IDLE_TIME_KEY, "1 ms") + .customProperty("paimon." + TIME_INTERVAL_KEY, "1 d") + .customProperty("paimon.partition.mark-done-action", "custom") + .build(), + DEFAULT_REMOTE_DATA_DIR, + 1L, + 1L); + assertThat(paimonLakeTieringFactory.isPartitionMarkDoneEnabled(invalidAction)).isTrue(); + long snapshot1 = writeAndCommit(tablePath, invalidAction, "2024-01-01"); + assertThat(getSnapshotProperties(tablePath, snapshot1)) + .doesNotContainKey(MARK_DONE_STATE_PROPERTY); + } + + @Test + void testInvalidFormatterDisablesMarkDoneAndRecoversByColdStart() throws Exception { + TablePath tablePath = TablePath.of(DATABASE, "test_mark_done_invalid_formatter"); + createPaimonTable(tablePath, Collections.emptyMap()); + TableInfo tableInfo = markDoneTableInfo(tablePath, false); + + writeAndCommit(tablePath, tableInfo, "2024-01-01"); + + // an invalid formatter syntax disables mark-done as a whole instead of draining the + // pending set partition by partition; the data commit still succeeds without state + TableInfo invalidFormatter = + TableInfo.of( + tablePath, + 0, + 1, + newTableBuilder(false) + .customProperty("paimon." + IDLE_TIME_KEY, "1 ms") + .customProperty("paimon." + TIME_INTERVAL_KEY, "1 d") + .customProperty("paimon.partition.timestamp-formatter", "{invalid}") + .build(), + DEFAULT_REMOTE_DATA_DIR, + 1L, + 1L); + long snapshot2 = writeAndCommit(tablePath, invalidFormatter, "2024-01-02"); + assertThat(getSnapshotProperties(tablePath, snapshot2)) + .doesNotContainKey(MARK_DONE_STATE_PROPERTY); + + // once the formatter is fixed, cold start recovers all live partitions + Thread.sleep(50); + try (LakeCommitter lakeCommitter = + createLakeCommitter(tablePath, tableInfo)) { + assertThat(commitMarkDoneMaintenance(lakeCommitter, "offsets-3")).isNotNull(); + } + assertThat(successFile(tablePath, "2024-01-01")).exists(); + assertThat(successFile(tablePath, "2024-01-02")).exists(); + } + + @Test + void testFailedActionRetriedNextRound() throws Exception { + TablePath tablePath = TablePath.of(DATABASE, "test_mark_done_flaky_action"); + createPaimonTable(tablePath, Collections.emptyMap()); + TableInfo tableInfo = + TableInfo.of( + tablePath, + 0, + 1, + newTableBuilder(false) + .customProperty("paimon." + IDLE_TIME_KEY, "1 ms") + .customProperty("paimon." + TIME_INTERVAL_KEY, "1 d") + .customProperty("paimon.partition.mark-done-action", "custom") + .customProperty( + "paimon.partition.mark-done-action.custom.class", + FlakyMarkDoneAction.class.getName()) + .build(), + DEFAULT_REMOTE_DATA_DIR, + 1L, + 1L); + + long snapshot1 = writeAndCommit(tablePath, tableInfo, "2024-01-01"); + MarkDoneState state1 = getMarkDoneState(tablePath, snapshot1); + assertThat(state1.getPendingPartitions()).containsOnlyKeys("2024-01-01"); + + // the action fails in the data round tiering a new partition: the round doesn't + // fail, the failed partition stays pending with its original last update time and + // the new partition is tracked + FlakyMarkDoneAction.remainingFailures.set(1); + FlakyMarkDoneAction.invocations.set(0); + Thread.sleep(50); + long snapshot2 = writeAndCommit(tablePath, tableInfo, "2024-01-02"); + assertThat(FlakyMarkDoneAction.invocations.get()).isEqualTo(1); + assertThat(getMarkDoneState(tablePath, snapshot2).getPendingPartitions()) + .containsOnlyKeys("2024-01-01", "2024-01-02") + .containsEntry("2024-01-01", state1.getPendingPartitions().get("2024-01-01")); + + // the next round retries the action and marks both idle partitions done + Thread.sleep(50); + try (LakeCommitter lakeCommitter = + createLakeCommitter(tablePath, tableInfo)) { + CommittedLakeSnapshot maintenanceSnapshot = + commitMarkDoneMaintenance(lakeCommitter, "offsets-3"); + assertThat(maintenanceSnapshot).isNotNull(); + assertThat( + getMarkDoneState(tablePath, maintenanceSnapshot.getLakeSnapshotId()) + .getPendingPartitions()) + .isEmpty(); + } + assertThat(FlakyMarkDoneAction.invocations.get()).isEqualTo(3); + } + + @Test + void testMaintenanceOffsetsFailurePropagates() throws Exception { + TablePath tablePath = TablePath.of(DATABASE, "test_mark_done_offsets_failure"); + createPaimonTable(tablePath, Collections.emptyMap()); + TableInfo tableInfo = markDoneTableInfo(tablePath, false); + + long snapshot1 = writeAndCommit(tablePath, tableInfo, "2024-01-01"); + Thread.sleep(50); + + // a failure preparing the offsets file involves snapshot consistency and must + // propagate instead of being swallowed as a mark-done failure + try (LakeCommitter lakeCommitter = + createLakeCommitter(tablePath, tableInfo)) { + assertThatThrownBy( + () -> + ((PartitionMarkDoneMaintainer) lakeCommitter) + .commitMarkDoneMaintenance( + () -> { + throw new IOException("injected"); + })) + .isInstanceOf(IOException.class) + .hasMessageContaining("injected"); + } + // no maintenance snapshot was committed + FileStoreTable fileStoreTable = + (FileStoreTable) paimonCatalog.getTable(toPaimon(tablePath)); + assertThat(fileStoreTable.snapshotManager().latestSnapshotId()).isEqualTo(snapshot1); + } + + /** A custom mark-done action failing on demand to verify the next-round retry. */ + public static class FlakyMarkDoneAction implements PartitionMarkDoneAction { + + private static final AtomicInteger remainingFailures = new AtomicInteger(); + private static final AtomicInteger invocations = new AtomicInteger(); + + @Override + public void markDone(String partition) { + invocations.incrementAndGet(); + if (remainingFailures.getAndUpdate(n -> Math.max(0, n - 1)) > 0) { + throw new RuntimeException("injected mark-done failure"); + } + } + + @Override + public void close() {} + } + + @Test + void testStateJsonSerde() { + // round trip + Map pending = new HashMap<>(); + pending.put("20240101", 1234L); + pending.put("2024-01-02", -1L); + MarkDoneState state = new MarkDoneState(true, pending); + int stateHashCode = state.hashCode(); + String stateJson = MarkDoneStateJsonSerde.toJson(state); + pending.clear(); + assertThat(state.getPendingPartitions()).hasSize(2); + assertThat(state.hashCode()).isEqualTo(stateHashCode); + assertThat(MarkDoneStateJsonSerde.toJson(state)).isEqualTo(stateJson); + assertThat(MarkDoneStateJsonSerde.fromJson(stateJson)).isEqualTo(state); + + // missing fields fall back to defaults, unknown fields are ignored + assertThat(MarkDoneStateJsonSerde.fromJson("{}")).isEqualTo(MarkDoneState.empty()); + state = + MarkDoneStateJsonSerde.fromJson( + "{\"initialized\":true,\"pending\":{\"p1\":100},\"unknown\":\"x\"}"); + assertThat(state.isInitialized()).isTrue(); + assertThat(state.getPendingPartitions()).containsEntry("p1", 100L); + + // wrongly typed fields are rejected as corrupt instead of silently coerced + assertThatThrownBy(() -> MarkDoneStateJsonSerde.fromJson("{\"initialized\":\"x\"}")) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> MarkDoneStateJsonSerde.fromJson("{\"pending\":[]}")) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> MarkDoneStateJsonSerde.fromJson("{\"pending\":{\"p\":\"bad\"}}")) + .isInstanceOf(IllegalArgumentException.class); + } + + private static CommittedLakeSnapshot commitMarkDoneMaintenance( + LakeCommitter lakeCommitter, String offsetsPath) + throws IOException { + return ((PartitionMarkDoneMaintainer) lakeCommitter) + .commitMarkDoneMaintenance(() -> offsetsPath); + } + + private static Map markDoneOptions() { + Map options = new HashMap<>(); + options.put(IDLE_TIME_KEY, "1 ms"); + options.put(TIME_INTERVAL_KEY, "1 d"); + return options; + } + + private TableInfo tableInfo(TablePath tablePath, boolean autoPartition) { + return TableInfo.of( + tablePath, + 0, + 1, + newTableBuilder(autoPartition).build(), + DEFAULT_REMOTE_DATA_DIR, + 1L, + 1L); + } + + /** A table info carrying the mark-done options in the Fluss custom properties. */ + private TableInfo markDoneTableInfo(TablePath tablePath, boolean autoPartition) { + TableDescriptor.Builder builder = + newTableBuilder(autoPartition).customProperty("paimon." + IDLE_TIME_KEY, "1 ms"); + if (!autoPartition) { + builder.customProperty("paimon." + TIME_INTERVAL_KEY, "1 d"); + } + return TableInfo.of(tablePath, 0, 1, builder.build(), DEFAULT_REMOTE_DATA_DIR, 1L, 1L); + } + + private TableInfo quarterTableInfo(TablePath tablePath) { + TableDescriptor.Builder builder = + newTableBuilder(true) + .customProperty("paimon." + IDLE_TIME_KEY, "1 ms") + .property( + ConfigOptions.TABLE_AUTO_PARTITION_TIME_UNIT, + AutoPartitionTimeUnit.QUARTER) + .property(ConfigOptions.TABLE_AUTO_PARTITION_TIME_FORMAT, "yyyy-'Q'Q"); + return TableInfo.of(tablePath, 0, 1, builder.build(), DEFAULT_REMOTE_DATA_DIR, 1L, 1L); + } + + private TableDescriptor.Builder newTableBuilder(boolean autoPartition) { + TableDescriptor.Builder builder = + TableDescriptor.builder() + .schema( + org.apache.fluss.metadata.Schema.newBuilder() + .column("c1", org.apache.fluss.types.DataTypes.INT()) + .column("c2", org.apache.fluss.types.DataTypes.STRING()) + .column("c3", org.apache.fluss.types.DataTypes.STRING()) + .build()) + .partitionedBy("c3") + .distributedBy(1) + .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true); + if (autoPartition) { + builder.property(ConfigOptions.TABLE_AUTO_PARTITION_ENABLED, true) + .property( + ConfigOptions.TABLE_AUTO_PARTITION_TIME_UNIT, + AutoPartitionTimeUnit.DAY); + } + return builder; + } + + /** Writes one record to each given partition and commits, returns the snapshot id. */ + private long writeAndCommit(TablePath tablePath, TableInfo tableInfo, String... partitions) + throws Exception { + return writeAndCommit(tablePath, tableInfo, enabledLakeTieringConfig(), partitions); + } + + private long writeAndCommit( + TablePath tablePath, + TableInfo tableInfo, + Configuration lakeTieringConfig, + String... partitions) + throws Exception { + List writeResults = new ArrayList<>(); + long partitionId = 1; + for (String partition : partitions) { + try (LakeWriter lakeWriter = + createLakeWriter(tablePath, partition, partitionId++, tableInfo)) { + GenericRow row = new GenericRow(3); + row.setField(0, 1); + row.setField(1, BinaryString.fromString("v1")); + row.setField(2, BinaryString.fromString(partition)); + lakeWriter.write( + new GenericRecord( + 0, System.currentTimeMillis(), ChangeType.APPEND_ONLY, row)); + writeResults.add(lakeWriter.complete()); + } + } + try (LakeCommitter lakeCommitter = + createLakeCommitter(tablePath, tableInfo, lakeTieringConfig)) { + PaimonCommittable committable = lakeCommitter.toCommittable(writeResults); + return lakeCommitter + .commit( + committable, + Collections.singletonMap( + FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY, "offsets")) + .getCommittedSnapshotId(); + } + } + + private MarkDoneState getMarkDoneState(TablePath tablePath, long snapshotId) throws Exception { + Map properties = getSnapshotProperties(tablePath, snapshotId); + assertThat(properties).containsKey(MARK_DONE_STATE_PROPERTY); + return MarkDoneStateJsonSerde.fromJson(properties.get(MARK_DONE_STATE_PROPERTY)); + } + + private Map getSnapshotProperties(TablePath tablePath, long snapshotId) + throws Exception { + FileStoreTable fileStoreTable = + (FileStoreTable) paimonCatalog.getTable(toPaimon(tablePath)); + return fileStoreTable.snapshotManager().snapshot(snapshotId).properties(); + } + + private File successFile(TablePath tablePath, String partition) { + return new File( + tempWarehouseDir, + String.format( + "%s.db/%s/c3=%s/_SUCCESS", + tablePath.getDatabaseName(), tablePath.getTableName(), partition)); + } + + private void createPaimonTable(TablePath tablePath, Map options) + throws Exception { + Schema.Builder builder = + Schema.newBuilder() + .column("c1", org.apache.paimon.types.DataTypes.INT()) + .column("c2", org.apache.paimon.types.DataTypes.STRING()) + .column("c3", org.apache.paimon.types.DataTypes.STRING()) + .partitionKeys("c3") + .options(options); + builder.column(BUCKET_COLUMN_NAME, org.apache.paimon.types.DataTypes.INT()); + builder.column(OFFSET_COLUMN_NAME, org.apache.paimon.types.DataTypes.BIGINT()); + builder.column( + TIMESTAMP_COLUMN_NAME, org.apache.paimon.types.DataTypes.TIMESTAMP_LTZ_MILLIS()); + paimonCatalog.createDatabase(tablePath.getDatabaseName(), true); + paimonCatalog.createTable(toPaimon(tablePath), builder.build(), true); + } + + private LakeWriter createLakeWriter( + TablePath tablePath, @Nullable String partition, Long partitionId, TableInfo tableInfo) + throws IOException { + return paimonLakeTieringFactory.createLakeWriter( + new WriterInitContext() { + @Override + public TablePath tablePath() { + return tablePath; + } + + @Override + public TableBucket tableBucket() { + return new TableBucket(0, partitionId, 0); + } + + @Nullable + @Override + public String partition() { + return partition; + } + + @Override + public TableInfo tableInfo() { + return tableInfo; + } + }); + } + + /** A job-level tiering config with the mark-done switch (disabled by default) enabled. */ + private static Configuration enabledLakeTieringConfig() { + Configuration lakeTieringConfig = new Configuration(); + lakeTieringConfig.set(ConfigOptions.LAKE_TIERING_PARTITION_MARK_DONE_ENABLED, true); + return lakeTieringConfig; + } + + private LakeCommitter createLakeCommitter( + TablePath tablePath, TableInfo tableInfo) throws IOException { + return createLakeCommitter(tablePath, tableInfo, enabledLakeTieringConfig()); + } + + private LakeCommitter createLakeCommitter( + TablePath tablePath, TableInfo tableInfo, Configuration lakeTieringConfig) + throws IOException { + return paimonLakeTieringFactory.createLakeCommitter( + new CommitterInitContext() { + @Override + public TablePath tablePath() { + return tablePath; + } + + @Override + public TableInfo tableInfo() { + return tableInfo; + } + + @Override + public Configuration lakeTieringConfig() { + return lakeTieringConfig; + } + + @Override + public Configuration flussClientConfig() { + return new Configuration(); + } + }); + } +} diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringITCase.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringITCase.java index 3d4da4fe50c..27dbe1122ea 100644 --- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringITCase.java +++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringITCase.java @@ -20,8 +20,10 @@ import org.apache.fluss.client.table.getter.PartitionGetter; import org.apache.fluss.config.AutoPartitionTimeUnit; import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; import org.apache.fluss.lake.paimon.testutils.FlinkPaimonTieringTestBase; import org.apache.fluss.metadata.PartitionInfo; +import org.apache.fluss.metadata.PartitionSpec; import org.apache.fluss.metadata.Schema; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TableChange; @@ -39,6 +41,7 @@ import org.apache.flink.core.execution.JobClient; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.paimon.Snapshot; import org.apache.paimon.catalog.Catalog; import org.apache.paimon.catalog.Identifier; import org.apache.paimon.reader.RecordReader; @@ -67,7 +70,10 @@ import java.util.Map; import java.util.stream.Stream; +import static org.apache.fluss.lake.committer.LakeCommitter.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY; +import static org.apache.fluss.lake.paimon.tiering.PaimonPartitionMarkDone.MARK_DONE_STATE_PROPERTY; import static org.apache.fluss.testutils.DataTestUtils.row; +import static org.apache.fluss.testutils.common.CommonTestUtils.retry; import static org.assertj.core.api.Assertions.assertThat; /** IT case for tiering tables to paimon. */ @@ -622,4 +628,111 @@ void testTieringWithAddColumn() throws Exception { protected FlussClusterExtension getFlussClusterExtension() { return FLUSS_CLUSTER_EXTENSION; } + + @Test + void testPartitionMarkDone() throws Exception { + TablePath tablePath = TablePath.of(DEFAULT_DB, "markDoneTable"); + Map customProperties = new HashMap<>(); + customProperties.put("paimon.partition.idle-time-to-done", "1 s"); + customProperties.put("paimon.partition.time-interval", "1 d"); + TableDescriptor descriptor = + TableDescriptor.builder() + .schema( + Schema.newBuilder() + .column("a", DataTypes.INT()) + .column("b", DataTypes.STRING()) + .column("c", DataTypes.STRING()) + .build()) + .partitionedBy("c") + .distributedBy(1, "a") + .property(ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true") + .property(ConfigOptions.TABLE_DATALAKE_FRESHNESS, Duration.ofMillis(500)) + .customProperties(customProperties) + .build(); + long tableId = createTable(tablePath, descriptor); + String partition = "2024-01-01"; + admin.createPartition( + tablePath, + new PartitionSpec(Collections.singletonMap("c", partition)), + false) + .get(); + long partitionId = + waitUntilPartitions(FLUSS_CLUSTER_EXTENSION.getZooKeeperClient(), tablePath, 1) + .keySet() + .iterator() + .next(); + writeRows( + tablePath, + Arrays.asList( + row(1, "v1", partition), row(2, "v2", partition), row(3, "v3", partition)), + true); + + // mark-done must also be enabled at the job level (disabled by default) + Configuration lakeTieringConfig = new Configuration(); + lakeTieringConfig.set(ConfigOptions.LAKE_TIERING_PARTITION_MARK_DONE_ENABLED, true); + JobClient jobClient = buildTieringJob(execEnv, lakeTieringConfig); + try { + // the partition data is tiered, then the empty tiering round marks the idle + // partition done with a properties-only snapshot writing the _SUCCESS file + java.io.File successFile = + new java.io.File( + warehousePath, + String.format( + "%s.db/%s/c=%s/_SUCCESS", + DEFAULT_DB, tablePath.getTableName(), partition)); + retry(Duration.ofMinutes(2), () -> assertThat(successFile).exists()); + + FileStoreTable table = + (FileStoreTable) + paimonCatalog.getTable( + Identifier.create(DEFAULT_DB, tablePath.getTableName())); + retry( + Duration.ofMinutes(1), + () -> { + Snapshot snapshot = table.snapshotManager().latestSnapshot(); + assertThat(snapshot).isNotNull(); + Map properties = snapshot.properties(); + // the properties-only snapshot carries over the bucket offsets and + // holds the mark-done state: initialized and nothing pending + assertThat(properties) + .containsKey(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY) + .containsKey(MARK_DONE_STATE_PROPERTY); + MarkDoneState state = + MarkDoneStateJsonSerde.fromJson( + properties.get(MARK_DONE_STATE_PROPERTY)); + assertThat(state.isInitialized()).isTrue(); + assertThat(state.getPendingPartitions()).isEmpty(); + // the properties-only snapshot is committed back to Fluss + assertThat(admin.getLatestLakeSnapshot(tablePath).get().getSnapshotId()) + .isEqualTo(snapshot.id()); + }); + + // late data: tiering still works after the properties-only snapshot, and the + // partition is re-tracked then marked done again + long snapshotBeforeLateData = table.snapshotManager().latestSnapshot().id(); + assertThat(successFile.delete()).isTrue(); + writeRows( + tablePath, + Arrays.asList(row(4, "v4", partition), row(5, "v5", partition)), + true); + assertReplicaStatus(new TableBucket(tableId, partitionId, 0), 5); + retry( + Duration.ofMinutes(2), + () -> { + Snapshot snapshot = table.snapshotManager().latestSnapshot(); + assertThat(snapshot).isNotNull(); + assertThat(snapshot.id()).isGreaterThan(snapshotBeforeLateData); + assertThat(snapshot.properties()).containsKey(MARK_DONE_STATE_PROPERTY); + MarkDoneState state = + MarkDoneStateJsonSerde.fromJson( + snapshot.properties().get(MARK_DONE_STATE_PROPERTY)); + assertThat(state.getPendingPartitions()).isEmpty(); + assertThat(successFile).exists(); + assertThat(admin.getLatestLakeSnapshot(tablePath).get().getSnapshotId()) + .isEqualTo(snapshot.id()); + }); + } finally { + jobClient.cancel().get(); + } + } } diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java index a433af06c82..e8531875001 100644 --- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java +++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java @@ -939,9 +939,6 @@ private void doCreatePaimonTable(TablePath tablePath, Schema.Builder paimonSchem paimonSchemaBuilder.column(BUCKET_COLUMN_NAME, DataTypes.INT()); paimonSchemaBuilder.column(OFFSET_COLUMN_NAME, DataTypes.BIGINT()); paimonSchemaBuilder.column(TIMESTAMP_COLUMN_NAME, DataTypes.TIMESTAMP_LTZ_MILLIS()); - paimonSchemaBuilder.option( - CoreOptions.COMMIT_CALLBACKS.key(), - PaimonLakeCommitter.PaimonCommitCallback.class.getName()); paimonCatalog.createDatabase(tablePath.getDatabaseName(), true); paimonCatalog.createTable(toPaimon(tablePath), paimonSchemaBuilder.build(), true); } diff --git a/website/docs/maintenance/tiered-storage/lakehouse-storage.md b/website/docs/maintenance/tiered-storage/lakehouse-storage.md index 51b4a72db6e..544db283e78 100644 --- a/website/docs/maintenance/tiered-storage/lakehouse-storage.md +++ b/website/docs/maintenance/tiered-storage/lakehouse-storage.md @@ -75,6 +75,7 @@ CREATE TABLE my_table ( | Option | Type | Default | Description | |--------|------|---------|-------------| | `lake.tiering.auto-expire-snapshot` | Boolean | false | Auto-trigger snapshot expiration on commit, even if `table.datalake.auto-expire-snapshot` is false | +| `lake.tiering.partition.mark-done.enabled` | Boolean | false | Whether the tiering service marks idle partitions of tiered partitioned tables as done. When enabled, a table opts in via its lake-format prefixed mark-done custom properties (e.g. `paimon.partition.idle-time-to-done` for Paimon) | | `lake.tiering.io.tmp.dirs` | String | Flink temporary directories | Local directories used for temporary I/O files. If not configured, a `fluss` child directory under each Flink temporary directory is used. Separate multiple directories with commas or the system path separator | ## Data Retention diff --git a/website/docs/streaming-lakehouse/tiering-service.md b/website/docs/streaming-lakehouse/tiering-service.md index 93ad1914bba..ddea8fb6e01 100644 --- a/website/docs/streaming-lakehouse/tiering-service.md +++ b/website/docs/streaming-lakehouse/tiering-service.md @@ -77,6 +77,7 @@ The following `--lake.tiering.*` options are set when starting the tiering job: | Option | Type | Default | Description | |--------|------|---------|-------------| | `lake.tiering.auto-expire-snapshot` | Boolean | false | Auto-trigger snapshot expiration on commit | +| `lake.tiering.partition.mark-done.enabled` | Boolean | false | Whether the tiering service marks idle partitions of tiered partitioned tables as done. When enabled, a table opts in via its lake-format prefixed mark-done custom properties (e.g. `paimon.partition.idle-time-to-done` for Paimon) | | `lake.tiering.io.tmp.dirs` | String | Flink temporary directories | Local directories used for temporary I/O files. If not configured, a `fluss` child directory under each Flink temporary directory is used. Separate multiple directories with commas or the system path separator | ### Table-Level Options