offsets = Collections.singletonMap(bucket, 100L);
+ LakeSnapshot snapshot =
+ new LakeSnapshot(1L, offsets, "{not-json".getBytes(StandardCharsets.UTF_8));
+
+ // bucket offsets remain accessible even though the tiering state is corrupt.
+ assertThat(snapshot.getTableBucketsOffset()).containsEntry(bucket, 100L);
+ // only the explicit state accessor surfaces the parse failure.
+ assertThatThrownBy(snapshot::getLakeTieringTableState).isInstanceOf(RuntimeException.class);
+ }
+}
diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/committer/LakeTieringTableState.java b/fluss-common/src/main/java/org/apache/fluss/lake/committer/LakeTieringTableState.java
new file mode 100644
index 0000000000..1528cdbdce
--- /dev/null
+++ b/fluss-common/src/main/java/org/apache/fluss/lake/committer/LakeTieringTableState.java
@@ -0,0 +1,127 @@
+/*
+ * 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.PublicEvolving;
+import org.apache.fluss.utils.json.JsonSerdeUtils;
+import org.apache.fluss.utils.json.LakeTieringTableStateJsonSerde;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Table-level tiering state carried by the lake offsets file as an opaque {@code tiering_state}
+ * payload (transported as {@code bytes tiering_state_json} in RPC); owned by the lake side and
+ * passed through by the offsets serde without parsing.
+ *
+ * Partition mark-done is the first consumer (delete-on-done model):
+ *
+ *
+ * - {@code partitionDoneInitialized}: table-level one-shot flag, {@code true} once the
+ * full-table cold-start back-fill has completed; disambiguates "all done" from "first run".
+ *
- {@code partitionUpdateTimes}: last update time of only the not-yet-done partitions, keyed
+ * by partitionId. A partition is removed once marked done; no {@code doneTime} is persisted.
+ *
+ *
+ * This is a flat, versioned model. Evolve it by adding fields and bumping {@link
+ * #CURRENT_VERSION} (e.g. a future watermark mode). To stay safe when a newer and an older tiering
+ * job run against the same table across rounds, a consumer that reads a {@code version} higher than
+ * its own {@link #CURRENT_VERSION} MUST NOT overwrite the persisted state (leave it untouched, i.e.
+ * do not send {@code tiering_state} back), so it never drops fields written by a newer build.
+ *
+ * @since 0.9
+ */
+@PublicEvolving
+public class LakeTieringTableState {
+
+ public static final int VERSION_1 = 1;
+
+ public static final int CURRENT_VERSION = VERSION_1;
+
+ private final int version;
+ private final boolean partitionDoneInitialized;
+ private final Map partitionUpdateTimes;
+
+ public LakeTieringTableState(
+ boolean partitionDoneInitialized, Map partitionUpdateTimes) {
+ this(CURRENT_VERSION, partitionDoneInitialized, partitionUpdateTimes);
+ }
+
+ public LakeTieringTableState(
+ int version, boolean partitionDoneInitialized, Map partitionUpdateTimes) {
+ this.version = version;
+ this.partitionDoneInitialized = partitionDoneInitialized;
+ this.partitionUpdateTimes =
+ partitionUpdateTimes == null
+ ? Collections.emptyMap()
+ : new HashMap<>(partitionUpdateTimes);
+ }
+
+ public int getVersion() {
+ return version;
+ }
+
+ public boolean isPartitionDoneInitialized() {
+ return partitionDoneInitialized;
+ }
+
+ public Map getPartitionUpdateTimes() {
+ return Collections.unmodifiableMap(partitionUpdateTimes);
+ }
+
+ public byte[] toJsonBytes() {
+ return JsonSerdeUtils.writeValueAsBytes(this, LakeTieringTableStateJsonSerde.INSTANCE);
+ }
+
+ public static LakeTieringTableState fromJsonBytes(byte[] json) {
+ return JsonSerdeUtils.readValue(json, LakeTieringTableStateJsonSerde.INSTANCE);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ LakeTieringTableState that = (LakeTieringTableState) o;
+ return version == that.version
+ && partitionDoneInitialized == that.partitionDoneInitialized
+ && Objects.equals(partitionUpdateTimes, that.partitionUpdateTimes);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(version, partitionDoneInitialized, partitionUpdateTimes);
+ }
+
+ @Override
+ public String toString() {
+ return "LakeTieringTableState{"
+ + "version="
+ + version
+ + ", partitionDoneInitialized="
+ + partitionDoneInitialized
+ + ", partitionUpdateTimes="
+ + partitionUpdateTimes
+ + '}';
+ }
+}
diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/committer/PartitionDoneCandidate.java b/fluss-common/src/main/java/org/apache/fluss/lake/committer/PartitionDoneCandidate.java
new file mode 100644
index 0000000000..4e58c5dd63
--- /dev/null
+++ b/fluss-common/src/main/java/org/apache/fluss/lake/committer/PartitionDoneCandidate.java
@@ -0,0 +1,84 @@
+/*
+ * 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.PublicEvolving;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+/**
+ * A candidate partition passed to {@link PartitionDoneHandler} for mark-done judgement. Transient,
+ * runtime-only input keyed by {@code partitionName} (not to be confused with the persisted {@link
+ * LakeTieringTableState} keyed by partitionId).
+ *
+ * {@code lastUpdateTime} is the aggregated MAX of bucket max-timestamps (the partition's last
+ * write time in processing-time mode); -1 means unknown.
+ *
+ * @since 0.9
+ */
+@PublicEvolving
+public class PartitionDoneCandidate implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String partitionName;
+ private final long lastUpdateTime;
+
+ public PartitionDoneCandidate(String partitionName, long lastUpdateTime) {
+ this.partitionName = partitionName;
+ this.lastUpdateTime = lastUpdateTime;
+ }
+
+ public String partitionName() {
+ return partitionName;
+ }
+
+ public long lastUpdateTime() {
+ return lastUpdateTime;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ PartitionDoneCandidate that = (PartitionDoneCandidate) o;
+ return lastUpdateTime == that.lastUpdateTime
+ && Objects.equals(partitionName, that.partitionName);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(partitionName, lastUpdateTime);
+ }
+
+ @Override
+ public String toString() {
+ return "PartitionDoneCandidate{"
+ + "partitionName='"
+ + partitionName
+ + '\''
+ + ", lastUpdateTime="
+ + lastUpdateTime
+ + '}';
+ }
+}
diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/committer/PartitionDoneHandler.java b/fluss-common/src/main/java/org/apache/fluss/lake/committer/PartitionDoneHandler.java
new file mode 100644
index 0000000000..4abcb7cdf8
--- /dev/null
+++ b/fluss-common/src/main/java/org/apache/fluss/lake/committer/PartitionDoneHandler.java
@@ -0,0 +1,57 @@
+/*
+ * 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.PublicEvolving;
+
+import java.util.List;
+
+/**
+ * The handler that decides which partitions can be marked done and executes the lake-specific
+ * mark-done action (e.g. writing a {@code _SUCCESS} file for Paimon).
+ *
+ *
This is a lake-specific SPI created by {@link
+ * org.apache.fluss.lake.writer.LakeTieringFactory#createPartitionDoneHandler}. The tiering commit
+ * operator only feeds partitions that are not yet done (those still tracked in the tiering state,
+ * since done partitions are removed under delete-on-done) and relies on this handler to perform the
+ * idle judgement and the actual action.
+ *
+ *
Implementations MUST be idempotent: the same partition may be marked done more than once in
+ * rare failure windows (e.g. action executed but the new tiering state not yet persisted before a
+ * crash), so repeatedly executing the action for an already-done partition must be safe.
+ *
+ * @since 0.9
+ */
+@PublicEvolving
+public interface PartitionDoneHandler extends AutoCloseable {
+
+ /**
+ * Judges which partitions can be marked done and executes the mark-done action (idempotent),
+ * returning the names of the partitions that were newly marked done in this round.
+ *
+ * @param candidates the candidate partitions that are not yet done
+ * @param currentTime the current time (System.currentTimeMillis() in processing-time mode)
+ * @return the names of partitions newly marked done in this round
+ * @throws Exception if the mark-done action fails
+ */
+ List markDoneIfReady(List candidates, long currentTime)
+ throws Exception;
+
+ @Override
+ default void close() throws Exception {}
+}
diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/committer/PartitionDoneInitContext.java b/fluss-common/src/main/java/org/apache/fluss/lake/committer/PartitionDoneInitContext.java
new file mode 100644
index 0000000000..bcc6f55495
--- /dev/null
+++ b/fluss-common/src/main/java/org/apache/fluss/lake/committer/PartitionDoneInitContext.java
@@ -0,0 +1,60 @@
+/*
+ * 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.PublicEvolving;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.metadata.TablePath;
+
+/**
+ * The context needed to create a {@link PartitionDoneHandler}.
+ *
+ * It provides the table path, the table info (which carries partition keys, table config
+ * including the {@code table.datalake.partition.*} mark-done options, and auto-partition strategy)
+ * and the Fluss client configuration. Lake implementations read the mark-done related options from
+ * {@link TableInfo#getTableConfig()}.
+ *
+ * @since 0.9
+ */
+@PublicEvolving
+public interface PartitionDoneInitContext {
+
+ /**
+ * Returns the table path.
+ *
+ * @return the table path
+ */
+ TablePath tablePath();
+
+ /**
+ * Returns the table info, which carries partition keys, table config (including the {@code
+ * table.datalake.partition.*} mark-done options) and auto-partition strategy.
+ *
+ * @return the table info
+ */
+ TableInfo tableInfo();
+
+ /**
+ * Returns the Fluss client configuration. This configuration can be used to build a Fluss
+ * client, such as a Connection.
+ *
+ * @return the Fluss client configuration
+ */
+ Configuration flussClientConfig();
+}
diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/writer/LakeTieringFactory.java b/fluss-common/src/main/java/org/apache/fluss/lake/writer/LakeTieringFactory.java
index 955e8b7581..2aa407cf48 100644
--- a/fluss-common/src/main/java/org/apache/fluss/lake/writer/LakeTieringFactory.java
+++ b/fluss-common/src/main/java/org/apache/fluss/lake/writer/LakeTieringFactory.java
@@ -20,10 +20,13 @@
import org.apache.fluss.annotation.PublicEvolving;
import org.apache.fluss.lake.committer.CommitterInitContext;
import org.apache.fluss.lake.committer.LakeCommitter;
+import org.apache.fluss.lake.committer.PartitionDoneHandler;
+import org.apache.fluss.lake.committer.PartitionDoneInitContext;
import org.apache.fluss.lake.serializer.SimpleVersionedSerializer;
import java.io.IOException;
import java.io.Serializable;
+import java.util.Optional;
/**
* The LakeTieringFactory interface defines how to create lake writers and committers. It provides
@@ -71,4 +74,18 @@ LakeCommitter createLakeCommitter(
* @return the serializer for committable objects
*/
SimpleVersionedSerializer getCommittableSerializer();
+
+ /**
+ * Creates a {@link PartitionDoneHandler} for lake partition mark-done, if the lake supports it.
+ *
+ * The default implementation returns {@link Optional#empty()}, meaning the lake does not
+ * support partition mark-done. Lakes that support it (e.g. Paimon) should override this method.
+ *
+ * @param partitionDoneInitContext the context for initializing the handler
+ * @return an optional partition done handler; empty if not supported
+ */
+ default Optional createPartitionDoneHandler(
+ PartitionDoneInitContext partitionDoneInitContext) {
+ return Optional.empty();
+ }
}
diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/json/JsonSerdeUtils.java b/fluss-common/src/main/java/org/apache/fluss/utils/json/JsonSerdeUtils.java
index b6e0794bef..21afdc7d58 100644
--- a/fluss-common/src/main/java/org/apache/fluss/utils/json/JsonSerdeUtils.java
+++ b/fluss-common/src/main/java/org/apache/fluss/utils/json/JsonSerdeUtils.java
@@ -73,5 +73,13 @@ public static T readValue(byte[] json, JsonDeserializer deserializer) {
}
}
+ /**
+ * Parse the given JSON bytes into a tree node using the shared object mapper. Useful for
+ * opaque/pass-through JSON payloads whose concrete type is not known to the caller.
+ */
+ public static JsonNode readTree(byte[] json) throws IOException {
+ return OBJECT_MAPPER_INSTANCE.readTree(json);
+ }
+
private JsonSerdeUtils() {}
}
diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerde.java b/fluss-common/src/main/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerde.java
new file mode 100644
index 0000000000..ea9be581c3
--- /dev/null
+++ b/fluss-common/src/main/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerde.java
@@ -0,0 +1,90 @@
+/*
+ * 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.utils.json;
+
+import org.apache.fluss.lake.committer.LakeTieringTableState;
+import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
+import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+
+/**
+ * Json serializer and deserializer for {@link LakeTieringTableState}, e.g. {@code
+ * {"version":1,"partition_done_initialized":true,"partition_update_times":{"5":1704153550000}}}.
+ *
+ * Field-level compatibility: unknown fields are ignored and missing fields fall back to
+ * defaults, and {@code version} is NOT strictly gated on read, so a higher-version payload still
+ * degrades to the fields this build understands. Writers must honor the version rule documented on
+ * {@link LakeTieringTableState} (do not overwrite a state whose version is higher than {@link
+ * LakeTieringTableState#CURRENT_VERSION}).
+ */
+public class LakeTieringTableStateJsonSerde
+ implements JsonSerializer, JsonDeserializer {
+
+ public static final LakeTieringTableStateJsonSerde INSTANCE =
+ new LakeTieringTableStateJsonSerde();
+
+ private static final String VERSION_KEY = "version";
+ private static final String PARTITION_DONE_INITIALIZED_KEY = "partition_done_initialized";
+ private static final String PARTITION_UPDATE_TIMES_KEY = "partition_update_times";
+
+ @Override
+ public void serialize(LakeTieringTableState state, JsonGenerator generator) throws IOException {
+ generator.writeStartObject();
+ generator.writeNumberField(VERSION_KEY, state.getVersion());
+ generator.writeBooleanField(
+ PARTITION_DONE_INITIALIZED_KEY, state.isPartitionDoneInitialized());
+ generator.writeObjectFieldStart(PARTITION_UPDATE_TIMES_KEY);
+ for (Map.Entry entry : state.getPartitionUpdateTimes().entrySet()) {
+ generator.writeNumberField(String.valueOf(entry.getKey()), entry.getValue());
+ }
+ generator.writeEndObject();
+ generator.writeEndObject();
+ }
+
+ @Override
+ public LakeTieringTableState deserialize(JsonNode node) {
+ // no strict version gating on read: read best-effort and ignore unknown fields.
+ int version =
+ node.has(VERSION_KEY)
+ ? node.get(VERSION_KEY).asInt()
+ : LakeTieringTableState.VERSION_1;
+
+ // VERSION_1 fields, read for any version so a higher (unsupported) version degrades to the
+ // fields this build knows. When evolving, bump CURRENT_VERSION and read the new fields here
+ // guarded by `if (version >= VERSION_2)`.
+ boolean partitionDoneInitialized =
+ node.has(PARTITION_DONE_INITIALIZED_KEY)
+ && node.get(PARTITION_DONE_INITIALIZED_KEY).asBoolean();
+
+ Map partitionUpdateTimes = new HashMap<>();
+ JsonNode updateTimesNode = node.get(PARTITION_UPDATE_TIMES_KEY);
+ if (updateTimesNode != null && updateTimesNode.isObject()) {
+ Iterator> fields = updateTimesNode.fields();
+ while (fields.hasNext()) {
+ Map.Entry field = fields.next();
+ partitionUpdateTimes.put(Long.valueOf(field.getKey()), field.getValue().asLong());
+ }
+ }
+
+ return new LakeTieringTableState(version, partitionDoneInitialized, partitionUpdateTimes);
+ }
+}
diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/json/TableBucketOffsets.java b/fluss-common/src/main/java/org/apache/fluss/utils/json/TableBucketOffsets.java
index 030d3da180..5aea2a84ae 100644
--- a/fluss-common/src/main/java/org/apache/fluss/utils/json/TableBucketOffsets.java
+++ b/fluss-common/src/main/java/org/apache/fluss/utils/json/TableBucketOffsets.java
@@ -20,17 +20,24 @@
import org.apache.fluss.metadata.TableBucket;
+import javax.annotation.Nullable;
+
+import java.util.Arrays;
import java.util.Map;
import java.util.Objects;
/**
* Represents the offsets for all buckets of a table. This class stores the mapping from {@link
- * TableBucket} to their corresponding offsets.
+ * TableBucket} to their corresponding offsets, as well as an opaque table-level tiering state.
*
* This class is used to track the log end offsets for each bucket in a table. It supports both
* non-partitioned tables (where buckets are identified only by bucket id) and partitioned tables
* (where buckets are identified by partition id and bucket id).
*
+ *
It also carries an opaque table-level tiering-state payload (owned by the lake side, e.g.
+ * {@code LakeTieringTableState}), passed through without parsing and used by e.g. partition
+ * mark-done.
+ *
*
The offsets map contains entries for each bucket that has a valid offset. Missing buckets are
* not included in the map.
*
@@ -48,14 +55,34 @@ public class TableBucketOffsets {
private final Map offsets;
/**
- * Creates a new {@link TableBucketOffsets} instance.
+ * Opaque table-level tiering state (owned by the lake side, e.g. {@code
+ * LakeTieringTableState}), serialized as JSON bytes and passed through by the serde without
+ * parsing; {@code null} when absent.
+ */
+ @Nullable private final byte[] tieringStateJson;
+
+ /**
+ * Creates a new {@link TableBucketOffsets} instance without tiering state.
*
* @param tableId the table ID that all buckets belong to
* @param offsets the mapping from {@link TableBucket} to their offsets
*/
public TableBucketOffsets(long tableId, Map offsets) {
+ this(tableId, offsets, null);
+ }
+
+ /**
+ * Creates a new {@link TableBucketOffsets} instance with an opaque tiering-state payload.
+ *
+ * @param tableId the table ID that all buckets belong to
+ * @param offsets the mapping from {@link TableBucket} to their offsets
+ * @param tieringStateJson the opaque tiering-state JSON bytes (nullable, passed through)
+ */
+ public TableBucketOffsets(
+ long tableId, Map offsets, @Nullable byte[] tieringStateJson) {
this.tableId = tableId;
this.offsets = offsets;
+ this.tieringStateJson = tieringStateJson;
}
/**
@@ -76,6 +103,12 @@ public Map getOffsets() {
return offsets;
}
+ /** Returns the opaque tiering-state JSON bytes, or {@code null} if absent. */
+ @Nullable
+ public byte[] getTieringStateJson() {
+ return tieringStateJson;
+ }
+
/**
* Serialize to a JSON byte array.
*
@@ -103,16 +136,25 @@ public boolean equals(Object o) {
return false;
}
TableBucketOffsets that = (TableBucketOffsets) o;
- return tableId == that.tableId && Objects.equals(offsets, that.offsets);
+ return tableId == that.tableId
+ && Objects.equals(offsets, that.offsets)
+ && Arrays.equals(tieringStateJson, that.tieringStateJson);
}
@Override
public int hashCode() {
- return Objects.hash(tableId, offsets);
+ return Objects.hash(tableId, offsets, Arrays.hashCode(tieringStateJson));
}
@Override
public String toString() {
- return "TableBucketOffsets{" + "tableId=" + tableId + ", offsets=" + offsets + '}';
+ return "TableBucketOffsets{"
+ + "tableId="
+ + tableId
+ + ", offsets="
+ + offsets
+ + ", tieringStateJson="
+ + Arrays.toString(tieringStateJson)
+ + '}';
}
}
diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/json/TableBucketOffsetsJsonSerde.java b/fluss-common/src/main/java/org/apache/fluss/utils/json/TableBucketOffsetsJsonSerde.java
index a177c53f66..c72861fd1b 100644
--- a/fluss-common/src/main/java/org/apache/fluss/utils/json/TableBucketOffsetsJsonSerde.java
+++ b/fluss-common/src/main/java/org/apache/fluss/utils/json/TableBucketOffsetsJsonSerde.java
@@ -23,6 +23,7 @@
import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
@@ -70,8 +71,10 @@ public class TableBucketOffsetsJsonSerde
private static final String BUCKET_OFFSETS_KEY = "bucket_offsets";
private static final String PARTITION_OFFSETS_KEY = "partition_offsets";
private static final String PARTITION_ID_KEY = "partition_id";
+ private static final String TIERING_STATE_KEY = "tiering_state";
- private static final int VERSION = 1;
+ private static final int VERSION_1 = 1;
+ private static final int CURRENT_VERSION = VERSION_1;
private static final long UNKNOWN_OFFSET = -1;
/**
@@ -94,7 +97,7 @@ public void serialize(TableBucketOffsets tableBucketOffsets, JsonGenerator gener
throws IOException {
generator.writeStartObject();
long expectedTableId = tableBucketOffsets.getTableId();
- generator.writeNumberField(VERSION_KEY, VERSION);
+ generator.writeNumberField(VERSION_KEY, CURRENT_VERSION);
generator.writeNumberField(TABLE_ID_KEY, expectedTableId);
Map offsets = tableBucketOffsets.getOffsets();
@@ -149,6 +152,23 @@ public void serialize(TableBucketOffsets tableBucketOffsets, JsonGenerator gener
}
}
+ // embed the opaque tiering-state payload (optional); require a JSON object, content not
+ // parsed.
+ byte[] tieringStateJson = tableBucketOffsets.getTieringStateJson();
+ if (tieringStateJson != null) {
+ JsonNode tieringStateNode;
+ try {
+ tieringStateNode = JsonSerdeUtils.readTree(tieringStateJson);
+ } catch (IOException e) {
+ throw new IllegalArgumentException("tiering_state payload is not valid JSON", e);
+ }
+ if (tieringStateNode == null || !tieringStateNode.isObject()) {
+ throw new IllegalArgumentException("tiering_state payload must be a JSON object");
+ }
+ generator.writeFieldName(TIERING_STATE_KEY);
+ generator.writeTree(tieringStateNode);
+ }
+
generator.writeEndObject();
}
@@ -165,7 +185,7 @@ public void serialize(TableBucketOffsets tableBucketOffsets, JsonGenerator gener
@Override
public TableBucketOffsets deserialize(JsonNode node) {
int version = node.get(VERSION_KEY).asInt();
- if (version != VERSION) {
+ if (version != CURRENT_VERSION) {
throw new IllegalArgumentException("Unsupported version: " + version);
}
@@ -211,7 +231,14 @@ public TableBucketOffsets deserialize(JsonNode node) {
}
}
- return new TableBucketOffsets(tableId, offsets);
+ // read the opaque tiering-state payload (optional) as raw JSON bytes; absent yields null.
+ byte[] tieringStateJson = null;
+ JsonNode tieringStateNode = node.get(TIERING_STATE_KEY);
+ if (tieringStateNode != null && !tieringStateNode.isNull()) {
+ tieringStateJson = tieringStateNode.toString().getBytes(StandardCharsets.UTF_8);
+ }
+
+ return new TableBucketOffsets(tableId, offsets, tieringStateJson);
}
private void serializeBucketLogEndOffset(
diff --git a/fluss-common/src/test/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerdeTest.java b/fluss-common/src/test/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerdeTest.java
new file mode 100644
index 0000000000..4e4c00f3bc
--- /dev/null
+++ b/fluss-common/src/test/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerdeTest.java
@@ -0,0 +1,82 @@
+/*
+ * 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.utils.json;
+
+import org.apache.fluss.lake.committer.LakeTieringTableState;
+
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link LakeTieringTableStateJsonSerde}. */
+class LakeTieringTableStateJsonSerdeTest {
+
+ @Test
+ void testSerializeFormat() {
+ Map updateTimes = new HashMap<>();
+ updateTimes.put(5L, 1704153550000L);
+ LakeTieringTableState state = new LakeTieringTableState(true, updateTimes);
+
+ String json = new String(state.toJsonBytes(), StandardCharsets.UTF_8);
+ assertThat(json)
+ .isEqualTo(
+ "{\"version\":1,\"partition_done_initialized\":true,"
+ + "\"partition_update_times\":{\"5\":1704153550000}}");
+ }
+
+ @Test
+ void testRoundTripAndDefaults() {
+ // populated state round-trips exactly.
+ Map updateTimes = new HashMap<>();
+ updateTimes.put(5L, 1704153550000L);
+ updateTimes.put(8L, 1704157200000L);
+ LakeTieringTableState state = new LakeTieringTableState(true, updateTimes);
+ LakeTieringTableState result = LakeTieringTableState.fromJsonBytes(state.toJsonBytes());
+ assertThat(result).isEqualTo(state);
+ assertThat(result.getVersion()).isEqualTo(LakeTieringTableState.VERSION_1);
+ assertThat(result.isPartitionDoneInitialized()).isTrue();
+ assertThat(result.getPartitionUpdateTimes())
+ .containsEntry(5L, 1704153550000L)
+ .containsEntry(8L, 1704157200000L);
+
+ // missing fields (e.g. "{}") fall back to defaults.
+ LakeTieringTableState defaults =
+ LakeTieringTableState.fromJsonBytes("{}".getBytes(StandardCharsets.UTF_8));
+ assertThat(defaults.getVersion()).isEqualTo(LakeTieringTableState.VERSION_1);
+ assertThat(defaults.isPartitionDoneInitialized()).isFalse();
+ assertThat(defaults.getPartitionUpdateTimes()).isEmpty();
+ }
+
+ @Test
+ void testHigherVersionToleratedIgnoringUnknownFields() {
+ // a higher (unsupported) version and unknown fields must not raise; the version is kept and
+ // the known fields degrade cleanly.
+ String json =
+ "{\"version\":99,\"partition_done_initialized\":true,"
+ + "\"partition_update_times\":{\"5\":1000},\"future_field\":\"x\"}";
+ LakeTieringTableState state =
+ LakeTieringTableState.fromJsonBytes(json.getBytes(StandardCharsets.UTF_8));
+ assertThat(state.getVersion()).isEqualTo(99);
+ assertThat(state.isPartitionDoneInitialized()).isTrue();
+ assertThat(state.getPartitionUpdateTimes()).containsEntry(5L, 1000L);
+ }
+}
diff --git a/fluss-common/src/test/java/org/apache/fluss/utils/json/TableBucketOffsetsJsonSerdeTest.java b/fluss-common/src/test/java/org/apache/fluss/utils/json/TableBucketOffsetsJsonSerdeTest.java
index 831171af2f..c984ab98d5 100644
--- a/fluss-common/src/test/java/org/apache/fluss/utils/json/TableBucketOffsetsJsonSerdeTest.java
+++ b/fluss-common/src/test/java/org/apache/fluss/utils/json/TableBucketOffsetsJsonSerdeTest.java
@@ -18,12 +18,18 @@
package org.apache.fluss.utils.json;
+import org.apache.fluss.lake.committer.LakeTieringTableState;
import org.apache.fluss.metadata.TableBucket;
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
/** Test for {@link TableBucketOffsetsJsonSerde}. */
class TableBucketOffsetsJsonSerdeTest extends JsonSerdeTestBase {
@@ -74,12 +80,26 @@ protected TableBucketOffsets[] createObjects() {
TableBucketOffsets tableBucketOffsets5 =
new TableBucketOffsets(tableId, bucketLogEndOffset);
+ // Test case 6: Partition table with an opaque tiering-state payload
+ tableId = 8;
+ bucketLogEndOffset = new HashMap<>();
+ bucketLogEndOffset.put(new TableBucket(tableId, 5L, 0), 100L);
+ bucketLogEndOffset.put(new TableBucket(tableId, 5L, 1), 230L);
+ bucketLogEndOffset.put(new TableBucket(tableId, 4L, 0), 420L);
+ Map partitionUpdateTimes = new HashMap<>();
+ partitionUpdateTimes.put(5L, 1704153550000L);
+ byte[] tieringStateJson =
+ new LakeTieringTableState(true, partitionUpdateTimes).toJsonBytes();
+ TableBucketOffsets tableBucketOffsets6 =
+ new TableBucketOffsets(tableId, bucketLogEndOffset, tieringStateJson);
+
return new TableBucketOffsets[] {
tableBucketOffsets1,
tableBucketOffsets2,
tableBucketOffsets3,
tableBucketOffsets4,
tableBucketOffsets5,
+ tableBucketOffsets6,
};
}
@@ -102,7 +122,41 @@ protected String[] expectedJsons() {
// Test case 4: Partition table with consecutive bucket ids
"{\"version\":1,\"table_id\":6,\"partition_offsets\":[{\"partition_id\":1,\"bucket_offsets\":[100,200]},{\"partition_id\":2,\"bucket_offsets\":[300,400]}]}",
// Test case 5: Partition table with missing bucket ids
- "{\"version\":1,\"table_id\":7,\"partition_offsets\":[{\"partition_id\":1,\"bucket_offsets\":[100,-1,300]},{\"partition_id\":2,\"bucket_offsets\":[-1,400,-1,600]}]}"
+ "{\"version\":1,\"table_id\":7,\"partition_offsets\":[{\"partition_id\":1,\"bucket_offsets\":[100,-1,300]},{\"partition_id\":2,\"bucket_offsets\":[-1,400,-1,600]}]}",
+ // Test case 6: Partition table with an opaque tiering-state payload
+ "{\"version\":1,\"table_id\":8,\"partition_offsets\":[{\"partition_id\":4,\"bucket_offsets\":[420]},{\"partition_id\":5,\"bucket_offsets\":[100,230]}],\"tiering_state\":{\"version\":1,\"partition_done_initialized\":true,\"partition_update_times\":{\"5\":1704153550000}}}"
};
}
+
+ /**
+ * Test that a malformed (non-JSON) or non-object tiering-state payload is rejected on
+ * serialization, so a bad payload never corrupts the persisted offsets file. The happy path
+ * (with/without tiering state, exact JSON and round-trip) is covered by {@link
+ * #createObjects()} / {@link #expectedJsons()}.
+ */
+ @Test
+ void testSerializeRejectsBadTieringStateJson() {
+ Map offsets = new HashMap<>();
+ offsets.put(new TableBucket(1L, 1L, 0), 100L);
+
+ assertThatThrownBy(
+ () ->
+ new TableBucketOffsets(
+ 1L,
+ offsets,
+ "{not-json".getBytes(StandardCharsets.UTF_8))
+ .toJsonBytes())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("not valid JSON");
+
+ assertThatThrownBy(
+ () ->
+ new TableBucketOffsets(
+ 1L,
+ offsets,
+ "[1,2,3]".getBytes(StandardCharsets.UTF_8))
+ .toJsonBytes())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("must be a JSON object");
+ }
}
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/FlussTableLakeSnapshotCommitter.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/FlussTableLakeSnapshotCommitter.java
index 76474a5b26..b9ffa31a9c 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/FlussTableLakeSnapshotCommitter.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/FlussTableLakeSnapshotCommitter.java
@@ -94,10 +94,24 @@ public void open() {
public String prepareLakeSnapshot(
long tableId, TablePath tablePath, Map logEndOffsets)
throws IOException {
+ return prepareLakeSnapshot(tableId, tablePath, logEndOffsets, null);
+ }
+
+ /**
+ * Prepares a lake snapshot carrying the opaque table-level tiering state. The {@code
+ * tieringStateJson} is passed through as-is (JSON bytes) and only written when non-null.
+ */
+ public String prepareLakeSnapshot(
+ long tableId,
+ TablePath tablePath,
+ Map logEndOffsets,
+ @Nullable byte[] tieringStateJson)
+ throws IOException {
PbPrepareLakeTableRespForTable prepareResp;
try {
PrepareLakeTableSnapshotRequest prepareLakeTableSnapshotRequest =
- toPrepareLakeTableSnapshotRequest(tableId, tablePath, logEndOffsets);
+ toPrepareLakeTableSnapshotRequest(
+ tableId, tablePath, logEndOffsets, tieringStateJson);
PrepareLakeTableSnapshotResponse prepareLakeTableSnapshotResponse =
coordinatorGateway
.prepareLakeTableSnapshot(prepareLakeTableSnapshotRequest)
@@ -125,6 +139,30 @@ public String prepareLakeSnapshot(
}
}
+ /**
+ * Persists only the table-level tiering state (e.g. mark-done progress) in an empty tiering
+ * round where no lake data is written. It reuses the previous lake snapshot id: PREPARE writes
+ * a new offsets file (bucket offsets unchanged, only tiering_state updated) and COMMIT
+ * registers it under the previous snapshot id with KEEP_ALL_PREVIOUS retention.
+ */
+ public void commitPartitionMarkDoneOnly(
+ long tableId,
+ TablePath tablePath,
+ long previousSnapshotId,
+ @Nullable byte[] tieringStateJson)
+ throws IOException {
+ String offsetsFile =
+ prepareLakeSnapshot(tableId, tablePath, Collections.emptyMap(), tieringStateJson);
+ commit(
+ tableId,
+ previousSnapshotId,
+ offsetsFile,
+ offsetsFile,
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ LakeCommitResult.KEEP_ALL_PREVIOUS);
+ }
+
public void commit(
long tableId,
TablePath tablePath,
@@ -242,7 +280,10 @@ void commit(
* @return the prepared commit request
*/
private PrepareLakeTableSnapshotRequest toPrepareLakeTableSnapshotRequest(
- long tableId, TablePath tablePath, Map logEndOffsets) {
+ long tableId,
+ TablePath tablePath,
+ Map logEndOffsets,
+ @Nullable byte[] tieringStateJson) {
PrepareLakeTableSnapshotRequest prepareLakeTableSnapshotRequest =
new PrepareLakeTableSnapshotRequest();
PbTableOffsets pbTableOffsets = prepareLakeTableSnapshotRequest.addBucketOffset();
@@ -261,6 +302,11 @@ private PrepareLakeTableSnapshotRequest toPrepareLakeTableSnapshotRequest(
pbBucketOffset.setBucketId(tableBucket.getBucket());
pbBucketOffset.setLogEndOffset(logEndOffsetEntry.getValue());
}
+
+ // pass through the opaque tiering state (JSON bytes), only when present.
+ if (tieringStateJson != null) {
+ pbTableOffsets.setTieringStateJson(tieringStateJson);
+ }
return prepareLakeTableSnapshotRequest;
}
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 2f3a69a7ea..549602dc4c 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
@@ -30,9 +30,13 @@
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.LakeTieringTableState;
+import org.apache.fluss.lake.committer.PartitionDoneCandidate;
+import org.apache.fluss.lake.committer.PartitionDoneHandler;
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.metadata.PartitionInfo;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
@@ -95,6 +99,14 @@ public class TieringCommitOperator
private final Map>>
collectedTableBucketWriteResults;
+ // tableId -> cached PartitionDoneHandler; NO_OP_HANDLER means the table does not support
+ // mark-done (checked once to avoid recreating the handler every round).
+ private final Map partitionDoneHandlers = new HashMap<>();
+
+ // sentinel meaning the table has been checked and does not support partition mark-done
+ private static final PartitionDoneHandler NO_OP_HANDLER =
+ (candidates, currentTime) -> Collections.emptyList();
+
/**
* The result of one table's commit round, holding the lake committable (nullable for empty
* commits where no data was written) and the associated tiering statistics.
@@ -199,20 +211,100 @@ private CommitResult commitWriteResults(
.filter(r -> r.writeResult() != null)
.collect(Collectors.toList());
+ // fetch current table info once; used both for the drop/recreate check and for deciding
+ // whether partition mark-done is enabled.
+ TableInfo currentTableInfo = admin.getTableInfo(tablePath).get();
+ boolean tableIdMatches = currentTableInfo.getTableId() == tableId;
+
+ // ========== Mark Done: read (getLakeSnapshot) -> judge -> execute action ==========
+ // null handler means the lake has no partition mark-done for this table.
+ PartitionDoneHandler handler =
+ tableIdMatches ? getOrCreateHandler(tableId, tablePath, currentTableInfo) : null;
+ boolean markDoneEnabled = handler != null;
+
+ LakeSnapshot prevSnapshot = null;
+ byte[] newTieringStateJson = null;
+ boolean tieringStateChanged = false;
+ if (markDoneEnabled) {
+ prevSnapshot = getLatestLakeSnapshot(tablePath);
+ long now = System.currentTimeMillis();
+ Map partitionIdToName = listPartitionIdToName(tablePath);
+ LakeTieringTableState prevState =
+ prevSnapshot != null ? prevSnapshot.getLakeTieringTableState() : null;
+
+ // not-yet-done partitions' last update time: previous state MAX this round's per-bucket
+ // maxTimestamp. Done partitions are absent (delete-on-done).
+ Map partitionUpdateTimes =
+ aggregatePartitionUpdateTimes(committableWriteResults, prevState);
+
+ // Cold start (uninitialized): judge every existing partition by partitionEndTime;
+ // otherwise only the not-yet-done partitions in the state.
+ boolean initialized = prevState != null && prevState.isPartitionDoneInitialized();
+ Set candidateIds =
+ initialized ? partitionUpdateTimes.keySet() : partitionIdToName.keySet();
+ List candidates =
+ buildCandidates(candidateIds, partitionUpdateTimes, partitionIdToName);
+
+ // Safe to mark done before the data commit below: candidates exclude this round's
+ // active partitions (their lastUpdateTime is ~now, never idle), so the two sets are
+ // disjoint.
+ List newlyDone;
+ boolean judged = true;
+ try {
+ newlyDone = handler.markDoneIfReady(candidates, now);
+ } catch (Exception e) {
+ // mark-done action failure must not fail the tiering main path; retry next round.
+ LOG.warn(
+ "Mark done action failed for table {}, will retry next round.",
+ tablePath,
+ e);
+ newlyDone = Collections.emptyList();
+ judged = false;
+ }
+
+ // delete-on-done: drop the newly-done partitions from the state.
+ if (!newlyDone.isEmpty()) {
+ Map nameToId = new HashMap<>();
+ for (Map.Entry e : partitionIdToName.entrySet()) {
+ nameToId.put(e.getValue(), e.getKey());
+ }
+ for (String name : newlyDone) {
+ Long id = nameToId.get(name);
+ if (id != null) {
+ partitionUpdateTimes.remove(id);
+ }
+ }
+ }
+
+ // Keep initialized once set; on cold start only flip to true when the judgement ran
+ // (not when the action threw), so a failed cold start retries next round.
+ boolean newInitialized = initialized || judged;
+ LakeTieringTableState newState =
+ new LakeTieringTableState(newInitialized, partitionUpdateTimes);
+ tieringStateChanged = !newState.equals(prevState);
+ newTieringStateJson = newState.toJsonBytes();
+ }
+
// all buckets were empty — nothing to commit to the lake
if (nonEmptyResults.isEmpty()) {
- LOG.info(
- "Commit tiering write results is empty for table {}, table path {}",
- tableId,
- tablePath);
+ // empty round with a changed tiering state: persist it via a light-weight commit that
+ // reuses the previous lake snapshot id (no lake data written).
+ if (markDoneEnabled && prevSnapshot != null && tieringStateChanged) {
+ flussTableLakeSnapshotCommitter.commitPartitionMarkDoneOnly(
+ tableId, tablePath, prevSnapshot.getSnapshotId(), newTieringStateJson);
+ } else {
+ LOG.info(
+ "Commit tiering write results is empty for table {}, table path {}",
+ tableId,
+ tablePath);
+ }
return new CommitResult(null, null);
}
// Check if the table was dropped and recreated during tiering.
// If the current table id differs from the committable's table id, fail this commit
// to avoid dirty commit to a newly created table.
- TableInfo currentTableInfo = admin.getTableInfo(tablePath).get();
- if (currentTableInfo.getTableId() != tableId) {
+ if (!tableIdMatches) {
throw new IllegalStateException(
String.format(
"The current table id %s for table path %s is different from the table id %s in the committable. "
@@ -241,7 +333,9 @@ private CommitResult commitWriteResults(
// to committable
Committable committable = lakeCommitter.toCommittable(writeResults);
// before commit to lake, check fluss not missing any lake snapshot committed by fluss
- LakeSnapshot flussCurrentLakeSnapshot = getLatestLakeSnapshot(tablePath);
+ // reuse prevSnapshot fetched for mark-done to avoid an extra RPC when possible.
+ LakeSnapshot flussCurrentLakeSnapshot =
+ markDoneEnabled ? prevSnapshot : getLatestLakeSnapshot(tablePath);
checkFlussNotMissingLakeSnapshot(
tablePath,
tableId,
@@ -251,10 +345,11 @@ private CommitResult commitWriteResults(
? null
: flussCurrentLakeSnapshot.getSnapshotId());
- // get the lake bucket offsets file storing the log end offsets
+ // get the lake bucket offsets file storing the log end offsets, carrying the opaque
+ // tiering state so that it is persisted together with the offsets in this same commit.
String lakeBucketTieredOffsetsFile =
flussTableLakeSnapshotCommitter.prepareLakeSnapshot(
- tableId, tablePath, logEndOffsets);
+ tableId, tablePath, logEndOffsets, newTieringStateJson);
// record the lake snapshot bucket offsets file to snapshot property
Map snapshotProperties =
@@ -274,6 +369,81 @@ private CommitResult commitWriteResults(
}
}
+ /**
+ * Gets or creates the {@link PartitionDoneHandler} for the table, caching the result. Returns
+ * {@code null} if the lake does not support partition mark-done for this table.
+ */
+ private PartitionDoneHandler getOrCreateHandler(
+ long tableId, TablePath tablePath, TableInfo tableInfo) {
+ PartitionDoneHandler cached = partitionDoneHandlers.get(tableId);
+ if (cached != null) {
+ return cached == NO_OP_HANDLER ? null : cached;
+ }
+ PartitionDoneHandler handler =
+ lakeTieringFactory
+ .createPartitionDoneHandler(
+ new TieringCommitterInitContext(
+ tablePath, tableInfo, lakeTieringConfig, flussConfig))
+ .orElse(null);
+ partitionDoneHandlers.put(tableId, handler == null ? NO_OP_HANDLER : handler);
+ return handler;
+ }
+
+ /** Lists the partitionId -> partitionName mapping for a partitioned table at runtime. */
+ private Map listPartitionIdToName(TablePath tablePath) throws Exception {
+ Map result = new HashMap<>();
+ for (PartitionInfo partitionInfo : admin.listPartitionInfos(tablePath).get()) {
+ result.put(partitionInfo.getPartitionId(), partitionInfo.getPartitionName());
+ }
+ return result;
+ }
+
+ /**
+ * Aggregates the partition-level last update time (processing-time mode): starts from the
+ * previous persisted tiering state and merges this round's per-bucket maxTimestamp by MAX per
+ * partition. Buckets with maxTimestamp <= 0 (empty / snapshot splits) are ignored.
+ */
+ private Map aggregatePartitionUpdateTimes(
+ List> writeResults,
+ @Nullable LakeTieringTableState prevState) {
+ Map result = new HashMap<>();
+ if (prevState != null) {
+ result.putAll(prevState.getPartitionUpdateTimes());
+ }
+ for (TableBucketWriteResult wr : writeResults) {
+ Long partitionId = wr.tableBucket().getPartitionId();
+ if (wr.maxTimestamp() > 0 && partitionId != null) {
+ result.merge(partitionId, wr.maxTimestamp(), Math::max);
+ }
+ }
+ return result;
+ }
+
+ /**
+ * Builds the candidate partitions to be judged by the handler for the given partition ids. Each
+ * candidate carries its last update time from {@code partitionUpdateTimes} (or {@code 0} when
+ * absent, e.g. a cold-start partition that has no data this round, so the handler judges it
+ * purely by its partitionEndTime). Partitions whose name cannot be resolved are skipped.
+ */
+ private List buildCandidates(
+ Set partitionIds,
+ Map partitionUpdateTimes,
+ Map partitionIdToName) {
+ List candidates = new ArrayList<>();
+ for (Long partitionId : partitionIds) {
+ String partitionName = partitionIdToName.get(partitionId);
+ if (partitionName == null) {
+ LOG.warn(
+ "Cannot find partition name for partition id {}, skip mark-done check.",
+ partitionId);
+ continue;
+ }
+ long lastUpdateTime = partitionUpdateTimes.getOrDefault(partitionId, 0L);
+ candidates.add(new PartitionDoneCandidate(partitionName, lastUpdateTime));
+ }
+ return candidates;
+ }
+
@Nullable
private LakeSnapshot getLatestLakeSnapshot(TablePath tablePath) throws Exception {
LakeSnapshot flussCurrentLakeSnapshot;
@@ -411,6 +581,16 @@ private List> collectTableAllBucketWriteResu
@Override
public void close() throws Exception {
+ for (PartitionDoneHandler handler : partitionDoneHandlers.values()) {
+ if (handler != NO_OP_HANDLER) {
+ try {
+ handler.close();
+ } catch (Exception e) {
+ LOG.warn("Failed to close partition done handler.", e);
+ }
+ }
+ }
+ partitionDoneHandlers.clear();
flussTableLakeSnapshotCommitter.close();
if (admin != null) {
admin.close();
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitterInitContext.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitterInitContext.java
index 79b7aaeb4f..f8806f1a0b 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitterInitContext.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitterInitContext.java
@@ -20,11 +20,15 @@
import org.apache.fluss.config.Configuration;
import org.apache.fluss.lake.committer.CommitterInitContext;
import org.apache.fluss.lake.committer.LakeCommitter;
+import org.apache.fluss.lake.committer.PartitionDoneInitContext;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
-/** The {@link CommitterInitContext} implementation for {@link LakeCommitter}. */
-public class TieringCommitterInitContext implements CommitterInitContext {
+/**
+ * The {@link CommitterInitContext} implementation for {@link LakeCommitter}. It also implements
+ * {@link PartitionDoneInitContext} so it can be reused to create a partition done handler.
+ */
+public class TieringCommitterInitContext implements CommitterInitContext, PartitionDoneInitContext {
private final TablePath tablePath;
private final TableInfo tableInfo;
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 ab902e1941..25617439f8 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
@@ -21,16 +21,21 @@
import org.apache.fluss.client.Connection;
import org.apache.fluss.client.ConnectionFactory;
import org.apache.fluss.client.admin.Admin;
+import org.apache.fluss.client.metadata.LakeSnapshot;
import org.apache.fluss.client.metadata.MetadataUpdater;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.LakeTableSnapshotNotExistException;
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.LakeTieringTableState;
import org.apache.fluss.lake.committer.TieringStats;
+import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.rpc.GatewayClientProxy;
@@ -97,6 +102,11 @@ public class TieringSourceEnumerator
private static final Logger LOG = LoggerFactory.getLogger(TieringSourceEnumerator.class);
+ // Paimon-native property enabling lake partition mark-done (currently only Paimon). Checked
+ // inline to avoid depending on the concrete lake module.
+ private static final String PARTITION_IDLE_TIME_TO_DONE_KEY =
+ "paimon.partition.idle-time-to-done";
+
private final Configuration flussConf;
private final SplitEnumeratorContext context;
private final ScheduledExecutorService timerService;
@@ -461,11 +471,35 @@ private void generateTieringSplits(Tuple3 tieringTable)
tieringTable.f2,
System.currentTimeMillis() - start);
if (tieringSplits.isEmpty()) {
- LOG.info(
- "Generate Tiering splits for table {} is empty, no need to tier data.",
- tieringTable.f2.getTableName());
- tieringTableEpochs.remove(tieringTable.f0);
- finishedTables.put(tieringTable.f0, TieringFinishInfo.from(tieringTable.f1));
+ if (tableInfo.isPartitioned()
+ && tableInfo
+ .getCustomProperties()
+ .containsKey(PARTITION_IDLE_TIME_TO_DONE_KEY)
+ && hasPendingMarkDonePartitions(tablePath)) {
+ // No data this round, but mark-done is enabled and there is pending work (or a
+ // cold-start back-fill). Emit a heartbeat split so the commit operator runs one
+ // mark-done round instead of finishing the table.
+ TieringSplit heartbeatSplit =
+ new TieringLogSplit(
+ tablePath,
+ new TableBucket(tieringTable.f0, 0),
+ null,
+ 0,
+ 0,
+ 1,
+ true);
+ // epoch was already registered when the table was requested; keep it active.
+ pendingSplits.add(heartbeatSplit);
+ LOG.info(
+ "Generate a heartbeat split for idle mark-done table {}.",
+ tieringTable.f2.getTableName());
+ } else {
+ LOG.info(
+ "Generate Tiering splits for table {} is empty, no need to tier data.",
+ tieringTable.f2.getTableName());
+ tieringTableEpochs.remove(tieringTable.f0);
+ finishedTables.put(tieringTable.f0, TieringFinishInfo.from(tieringTable.f1));
+ }
} else {
pendingSplits.addAll(tieringSplits);
@@ -489,6 +523,31 @@ private void generateTieringSplits(Tuple3 tieringTable)
}
}
+ /**
+ * Returns whether a mark-done table still has pending work this round: cold start not yet run
+ * ({@code !partition_done_initialized}, or no snapshot), or not-yet-done partitions remain.
+ * When all partitions are done, an empty round is a no-op, so no heartbeat is needed.
+ */
+ private boolean hasPendingMarkDonePartitions(TablePath tablePath) {
+ try {
+ LakeSnapshot lakeSnapshot = flussAdmin.getLatestLakeSnapshot(tablePath).get();
+ LakeTieringTableState state = lakeSnapshot.getLakeTieringTableState();
+ if (state == null || !state.isPartitionDoneInitialized()) {
+ // cold-start back-fill still needed
+ return true;
+ }
+ return !state.getPartitionUpdateTimes().isEmpty();
+ } catch (Exception e) {
+ if (e.getCause() instanceof LakeTableSnapshotNotExistException) {
+ // no snapshot yet -> first (cold-start) round still needed
+ return true;
+ }
+ // on any other error, be conservative and emit the heartbeat
+ LOG.warn("Failed to read tiering state for table {}, emit a heartbeat.", tablePath, e);
+ return true;
+ }
+ }
+
private List populateTieringRoundMetadata(List tieringSplits) {
int numberOfSplits = tieringSplits.size();
if (numberOfSplits == 0) {
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 92f7c562a6..7be2644c74 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,6 +23,8 @@
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.PartitionDoneHandler;
+import org.apache.fluss.lake.committer.PartitionDoneInitContext;
import org.apache.fluss.lake.serializer.SimpleVersionedSerializer;
import org.apache.fluss.lake.writer.LakeTieringFactory;
import org.apache.fluss.lake.writer.LakeWriter;
@@ -35,19 +37,34 @@
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
/** An implementation of {@link LakeTieringFactory} for testing purpose. */
public class TestingLakeTieringFactory
implements LakeTieringFactory {
@Nullable private TestingLakeCommitter testingLakeCommitter;
+ @Nullable private final PartitionDoneHandler partitionDoneHandler;
public TestingLakeTieringFactory(@Nullable TestingLakeCommitter testingLakeCommitter) {
+ this(testingLakeCommitter, null);
+ }
+
+ public TestingLakeTieringFactory(
+ @Nullable TestingLakeCommitter testingLakeCommitter,
+ @Nullable PartitionDoneHandler partitionDoneHandler) {
this.testingLakeCommitter = testingLakeCommitter;
+ this.partitionDoneHandler = partitionDoneHandler;
}
public TestingLakeTieringFactory() {
- this(null);
+ this(null, null);
+ }
+
+ @Override
+ public Optional createPartitionDoneHandler(
+ PartitionDoneInitContext partitionDoneInitContext) {
+ return Optional.ofNullable(partitionDoneHandler);
}
@Override
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/committer/FlussTableLakeSnapshotCommitterTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/committer/FlussTableLakeSnapshotCommitterTest.java
index 875190f4d7..dd984069fc 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/committer/FlussTableLakeSnapshotCommitterTest.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/committer/FlussTableLakeSnapshotCommitterTest.java
@@ -23,6 +23,7 @@
import org.apache.fluss.flink.utils.FlinkTestBase;
import org.apache.fluss.fs.FsPath;
import org.apache.fluss.lake.committer.LakeCommitResult;
+import org.apache.fluss.lake.committer.LakeTieringTableState;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.rpc.messages.CommitLakeTableSnapshotRequest;
@@ -243,6 +244,69 @@ void testPrepareLakeSnapshotSurfacesServerError() throws Exception {
.isInstanceOf(ApiException.class);
}
+ /**
+ * Verifies the empty-round light-weight commit ({@link
+ * FlussTableLakeSnapshotCommitter#commitPartitionMarkDoneOnly}) at the coordinator level:
+ * reusing the previous snapshot id with empty bucket offsets and only a new tiering_state must
+ * (a) not create a new snapshot, (b) not overwrite the existing bucket offsets, and it must
+ * update+persist the tiering_state so the client can read it back.
+ */
+ @Test
+ void testCommitPartitionMarkDoneOnlyReusesSnapshot() throws Exception {
+ TablePath tablePath = TablePath.of("fluss", "test_mark_done_only_reuse");
+ Tuple2> tableIdAndPartitions = createTable(tablePath, true);
+ long tableId = tableIdAndPartitions.f0;
+ Collection partitions = tableIdAndPartitions.f1;
+ long partitionId = partitions.iterator().next();
+
+ Map expectedOffsets = mockLogEndOffsets(tableId, partitions);
+
+ // ---- Round 1: a normal commit carrying bucket offsets + a tiering state ----
+ long snapshotId = 1L;
+ byte[] state1 =
+ new LakeTieringTableState(true, Collections.singletonMap(partitionId, 1000L))
+ .toJsonBytes();
+ String file1 =
+ flussTableLakeSnapshotCommitter.prepareLakeSnapshot(
+ tableId, tablePath, expectedOffsets, state1);
+ flussTableLakeSnapshotCommitter.commit(
+ tableId,
+ snapshotId,
+ file1,
+ null,
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ LakeCommitResult.KEEP_ALL_PREVIOUS);
+
+ LakeSnapshot afterRound1 = admin.getLatestLakeSnapshot(tablePath).get();
+ assertThat(afterRound1.getSnapshotId()).isEqualTo(snapshotId);
+ assertThat(afterRound1.getTableBucketsOffset()).isEqualTo(expectedOffsets);
+ assertThat(afterRound1.getLakeTieringTableState()).isNotNull();
+ assertThat(afterRound1.getLakeTieringTableState().getPartitionUpdateTimes())
+ .containsEntry(partitionId, 1000L);
+
+ // ---- Round 2: empty round (no bucket data), reuse the previous snapshot id and only
+ // update the tiering state (delete-on-done: the partition is now done and dropped) ----
+ byte[] state2 = new LakeTieringTableState(true, Collections.emptyMap()).toJsonBytes();
+ flussTableLakeSnapshotCommitter.commitPartitionMarkDoneOnly(
+ tableId, tablePath, snapshotId, state2);
+
+ LakeSnapshot afterRound2 = admin.getLatestLakeSnapshot(tablePath).get();
+ // (a) no new snapshot created: the snapshot id is reused
+ assertThat(afterRound2.getSnapshotId()).isEqualTo(snapshotId);
+ // (b) existing bucket offsets preserved (not overwritten by the empty round)
+ assertThat(afterRound2.getTableBucketsOffset()).isEqualTo(expectedOffsets);
+ // the tiering state is updated to the new (empty) one and persisted
+ assertThat(afterRound2.getLakeTieringTableState()).isNotNull();
+ assertThat(afterRound2.getLakeTieringTableState().isPartitionDoneInitialized()).isTrue();
+ assertThat(afterRound2.getLakeTieringTableState().getPartitionUpdateTimes()).isEmpty();
+
+ // no snapshot with id 2 was ever created
+ assertThatThrownBy(() -> admin.getLakeSnapshot(tablePath, snapshotId + 1).get())
+ .rootCause()
+ .isInstanceOf(LakeTableSnapshotNotExistException.class);
+ }
+
private Map mockLogEndOffsets(long tableId, Collection partitionsIds) {
Map logEndOffsets = new HashMap<>();
for (int bucket = 0; bucket < 3; bucket++) {
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 70d86bb2c8..2693da36b8 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
@@ -18,6 +18,8 @@
package org.apache.fluss.flink.tiering.committer;
import org.apache.fluss.client.metadata.LakeSnapshot;
+import org.apache.fluss.config.AutoPartitionTimeUnit;
+import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.exception.LakeTableSnapshotNotExistException;
import org.apache.fluss.flink.adapter.StreamOperatorParametersAdapter;
import org.apache.fluss.flink.tiering.TestingLakeTieringFactory;
@@ -27,7 +29,11 @@
import org.apache.fluss.flink.tiering.source.TableBucketWriteResult;
import org.apache.fluss.flink.utils.FlinkTestBase;
import org.apache.fluss.lake.committer.CommittedLakeSnapshot;
+import org.apache.fluss.lake.committer.LakeTieringTableState;
+import org.apache.fluss.lake.committer.PartitionDoneCandidate;
+import org.apache.fluss.lake.committer.PartitionDoneHandler;
import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TablePath;
import org.apache.flink.configuration.Configuration;
@@ -58,6 +64,7 @@
import static org.apache.fluss.lake.committer.LakeCommitter.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
import static org.apache.fluss.record.TestData.DATA1_PARTITIONED_TABLE_DESCRIPTOR;
+import static org.apache.fluss.record.TestData.DATA1_SCHEMA;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -207,6 +214,156 @@ void testCommitPartitionedTable() throws Exception {
}
}
+ @Test
+ void testPartitionMarkDoneDeletesDoneFromState() throws Exception {
+ TablePath tablePath = TablePath.of("fluss", "mark_done_persist");
+ long tableId = createTable(tablePath, markDoneEnabledDescriptor());
+ Map partitionIdByNames =
+ FLUSS_CLUSTER_EXTENSION.waitUntilPartitionAllReady(tablePath);
+
+ // a handler that marks all candidate partitions done and records the calls
+ RecordingPartitionDoneHandler handler = new RecordingPartitionDoneHandler(false);
+ TieringCommitOperator operator =
+ createOperatorWithHandler(handler);
+ try {
+ // one data round: all buckets of all partitions carry data
+ int numberOfWriteResults = 3 * partitionIdByNames.size();
+ long offset = 0;
+ long timestamp = System.currentTimeMillis();
+ for (int bucket = 0; bucket < 3; bucket++) {
+ for (Map.Entry entry : partitionIdByNames.entrySet()) {
+ TableBucket tableBucket = new TableBucket(tableId, entry.getValue(), bucket);
+ operator.processElement(
+ createTableBucketWriteResultStreamRecord(
+ tablePath,
+ tableBucket,
+ entry.getKey(),
+ 1,
+ offset++,
+ timestamp++,
+ numberOfWriteResults));
+ }
+ }
+
+ // handler should have been invoked with the active partitions as candidates
+ assertThat(handler.getMarkedPartitions())
+ .containsExactlyInAnyOrderElementsOf(partitionIdByNames.keySet());
+
+ // delete-on-done: all done partitions are removed from the state, which is initialized.
+ LakeSnapshot lakeSnapshot = admin.getLatestLakeSnapshot(tablePath).get();
+ LakeTieringTableState state = lakeSnapshot.getLakeTieringTableState();
+ assertThat(state).isNotNull();
+ assertThat(state.isPartitionDoneInitialized()).isTrue();
+ assertThat(state.getPartitionUpdateTimes()).isEmpty();
+ } finally {
+ operator.close();
+ }
+ }
+
+ @Test
+ void testMarkDoneActionFailureDoesNotFailMainPath() throws Exception {
+ TablePath tablePath = TablePath.of("fluss", "mark_done_action_failure");
+ long tableId = createTable(tablePath, markDoneEnabledDescriptor());
+ Map partitionIdByNames =
+ FLUSS_CLUSTER_EXTENSION.waitUntilPartitionAllReady(tablePath);
+
+ // a handler that always throws
+ RecordingPartitionDoneHandler handler = new RecordingPartitionDoneHandler(true);
+ TieringCommitOperator operator =
+ createOperatorWithHandler(handler);
+ try {
+ int numberOfWriteResults = 3 * partitionIdByNames.size();
+ long offset = 0;
+ long timestamp = System.currentTimeMillis();
+ for (int bucket = 0; bucket < 3; bucket++) {
+ for (Map.Entry entry : partitionIdByNames.entrySet()) {
+ TableBucket tableBucket = new TableBucket(tableId, entry.getValue(), bucket);
+ operator.processElement(
+ createTableBucketWriteResultStreamRecord(
+ tablePath,
+ tableBucket,
+ entry.getKey(),
+ 1,
+ offset++,
+ timestamp++,
+ numberOfWriteResults));
+ }
+ }
+
+ // the lake snapshot (main tiering path) must still be committed despite action failure
+ LakeSnapshot lakeSnapshot = admin.getLatestLakeSnapshot(tablePath).get();
+ assertThat(lakeSnapshot.getSnapshotId()).isGreaterThanOrEqualTo(0L);
+ // action failed -> cold start not initialized, and no partition dropped from the state
+ LakeTieringTableState state = lakeSnapshot.getLakeTieringTableState();
+ assertThat(state).isNotNull();
+ assertThat(state.isPartitionDoneInitialized()).isFalse();
+ assertThat(state.getPartitionUpdateTimes().keySet())
+ .containsExactlyInAnyOrderElementsOf(partitionIdByNames.values());
+ } finally {
+ operator.close();
+ }
+ }
+
+ private TableDescriptor markDoneEnabledDescriptor() {
+ return TableDescriptor.builder()
+ .schema(DATA1_SCHEMA)
+ .distributedBy(3)
+ .partitionedBy("b")
+ .property(ConfigOptions.TABLE_AUTO_PARTITION_ENABLED, true)
+ .property(ConfigOptions.TABLE_AUTO_PARTITION_TIME_UNIT, AutoPartitionTimeUnit.YEAR)
+ .customProperty("paimon.partition.idle-time-to-done", "60 s")
+ .build();
+ }
+
+ private TieringCommitOperator createOperatorWithHandler(
+ PartitionDoneHandler handler) throws Exception {
+ MockOperatorEventGateway gateway = new MockOperatorEventGateway();
+ StreamOperatorParameters> params =
+ StreamOperatorParametersAdapter.create(
+ new SourceOperatorStreamTask(new DummyEnvironment()),
+ new MockStreamConfig(new Configuration(), 1),
+ new MockOutput<>(new ArrayList<>()),
+ null,
+ new MockOperatorEventDispatcher(gateway),
+ null);
+ TieringCommitOperator operator =
+ new TieringCommitOperator<>(
+ params,
+ FLUSS_CLUSTER_EXTENSION.getClientConfig(),
+ new org.apache.fluss.config.Configuration(),
+ new TestingLakeTieringFactory(null, handler));
+ operator.open();
+ return operator;
+ }
+
+ /** A testing handler that marks all candidate partitions done, or always throws. */
+ private static final class RecordingPartitionDoneHandler implements PartitionDoneHandler {
+ private final boolean alwaysThrow;
+ private final List markedPartitions = new ArrayList<>();
+
+ RecordingPartitionDoneHandler(boolean alwaysThrow) {
+ this.alwaysThrow = alwaysThrow;
+ }
+
+ @Override
+ public List markDoneIfReady(
+ List candidates, long currentTime) throws Exception {
+ if (alwaysThrow) {
+ throw new RuntimeException("testing mark-done action failure");
+ }
+ List done = new ArrayList<>();
+ for (PartitionDoneCandidate state : candidates) {
+ done.add(state.partitionName());
+ }
+ markedPartitions.addAll(done);
+ return done;
+ }
+
+ List getMarkedPartitions() {
+ return markedPartitions;
+ }
+ }
+
@Test
void testCommitMeetsEmptyWriteResult() throws Exception {
TablePath tablePath1 = TablePath.of("fluss", "test_commit_empty_write_result");
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 432620efb3..86e8a28c12 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
@@ -17,15 +17,31 @@
package org.apache.fluss.lake.paimon.tiering;
+import org.apache.fluss.config.ConfigBuilder;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.lake.committer.CommitterInitContext;
import org.apache.fluss.lake.committer.LakeCommitter;
+import org.apache.fluss.lake.committer.PartitionDoneHandler;
+import org.apache.fluss.lake.committer.PartitionDoneInitContext;
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.WriterInitContext;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.utils.AutoPartitionStrategy;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.partition.PartitionTimeExtractor;
+import org.apache.paimon.partition.actions.PartitionMarkDoneAction;
+import org.apache.paimon.table.FileStoreTable;
import java.io.IOException;
+import java.time.ZoneId;
+import java.util.List;
+import java.util.Optional;
+
+import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon;
/** Implementation of {@link LakeTieringFactory} for Paimon . */
public class PaimonLakeTieringFactory
@@ -33,6 +49,13 @@ public class PaimonLakeTieringFactory
private static final long serialVersionUID = 1L;
+ /**
+ * Paimon-native property ({@code paimon.}-prefixed) that both enables lake partition mark-done
+ * and configures the idle time (Paimon's {@code partition.idle-time-to-done}).
+ */
+ public static final String PARTITION_IDLE_TIME_TO_DONE_KEY =
+ "paimon.partition.idle-time-to-done";
+
private final PaimonCatalogProvider paimonCatalogProvider;
public PaimonLakeTieringFactory(Configuration paimonConfig) {
@@ -60,4 +83,96 @@ public LakeCommitter createLakeCommitter(
public SimpleVersionedSerializer getCommittableSerializer() {
return new PaimonCommittableSerializer();
}
+
+ @Override
+ public Optional createPartitionDoneHandler(
+ PartitionDoneInitContext partitionDoneInitContext) {
+ TableInfo tableInfo = partitionDoneInitContext.tableInfo();
+ // only partitioned tables with the mark-done feature enabled are supported. It is enabled
+ // when Paimon's native partition.idle-time-to-done (paimon.-prefixed) property is set.
+ if (!tableInfo.isPartitioned()
+ || !tableInfo.getCustomProperties().containsKey(PARTITION_IDLE_TIME_TO_DONE_KEY)) {
+ return Optional.empty();
+ }
+
+ Configuration customProps = tableInfo.getCustomProperties();
+ // Fluss reads Paimon's native partition.idle-time-to-done (paimon.-prefixed in the Fluss
+ // table properties) to judge idle partitions. Its presence is guaranteed by the enabled
+ // check above.
+ long idleTimeMs =
+ customProps
+ .getOptional(
+ ConfigBuilder.key(PARTITION_IDLE_TIME_TO_DONE_KEY)
+ .durationType()
+ .noDefaultValue())
+ .orElseThrow(
+ () ->
+ new IllegalStateException(
+ "Missing " + PARTITION_IDLE_TIME_TO_DONE_KEY))
+ .toMillis();
+
+ Catalog catalog = paimonCatalogProvider.get();
+ try {
+ FileStoreTable fileStoreTable =
+ (FileStoreTable)
+ catalog.getTable(toPaimon(partitionDoneInitContext.tablePath()));
+ // The Paimon table already carries the mark-done options transferred from the Fluss
+ // table properties (paimon.partition.mark-done-action* -> partition.mark-done-action*)
+ // at table creation, so we can directly reuse Paimon's native createActions.
+ CoreOptions coreOptions = fileStoreTable.coreOptions();
+ List actions =
+ PartitionMarkDoneAction.createActions(
+ Thread.currentThread().getContextClassLoader(),
+ fileStoreTable,
+ coreOptions);
+
+ AutoPartitionStrategy autoPartitionStrategy =
+ tableInfo.getTableConfig().getAutoPartitionStrategy();
+ if (autoPartitionStrategy.isAutoPartitionEnabled()) {
+ // auto-partition table: use Fluss auto-partition format and the time unit to
+ // compute the exact partition end time.
+ return Optional.of(
+ new PaimonPartitionDoneHandler(
+ actions,
+ tableInfo.getPartitionKeys(),
+ idleTimeMs,
+ autoPartitionStrategy.timeZone().toZoneId(),
+ autoPartitionStrategy.timeUnit(),
+ null));
+ } else {
+ // non-auto-partition table: use Paimon's PartitionTimeExtractor with the native
+ // timestamp-pattern / timestamp-formatter (paimon.-prefixed) to get the partition
+ // time used in the idle judgement.
+ PartitionTimeExtractor timeExtractor =
+ new PartitionTimeExtractor(
+ customProps
+ .getOptional(
+ ConfigBuilder.key(
+ "paimon.partition.timestamp-pattern")
+ .stringType()
+ .noDefaultValue())
+ .orElse(null),
+ customProps
+ .getOptional(
+ ConfigBuilder.key(
+ "paimon.partition.timestamp-formatter")
+ .stringType()
+ .noDefaultValue())
+ .orElse(null));
+ return Optional.of(
+ new PaimonPartitionDoneHandler(
+ actions,
+ tableInfo.getPartitionKeys(),
+ idleTimeMs,
+ ZoneId.systemDefault(),
+ null,
+ timeExtractor));
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to create PaimonPartitionDoneHandler for table "
+ + partitionDoneInitContext.tablePath(),
+ e);
+ }
+ }
}
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonPartitionDoneHandler.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonPartitionDoneHandler.java
new file mode 100644
index 0000000000..be0f8d5b03
--- /dev/null
+++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/PaimonPartitionDoneHandler.java
@@ -0,0 +1,208 @@
+/*
+ * 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.lake.committer.PartitionDoneCandidate;
+import org.apache.fluss.lake.committer.PartitionDoneHandler;
+import org.apache.fluss.metadata.ResolvedPartitionSpec;
+
+import org.apache.paimon.partition.PartitionTimeExtractor;
+import org.apache.paimon.partition.actions.PartitionMarkDoneAction;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Paimon implementation of {@link PartitionDoneHandler}.
+ *
+ * It reuses Paimon's native {@link PartitionMarkDoneAction} (e.g. success-file) to perform the
+ * actual mark-done, and follows Paimon's mark-done judgement:
+ *
+ *
+ * done <=> currentTime - MAX(lastUpdateTime, partitionEndTime) > idleTime
+ *
+ *
+ * The {@code partitionEndTime} is computed differently for auto-partitioned and
+ * non-auto-partitioned tables:
+ *
+ *
+ * - Auto-partitioned tables: the partition value uses Fluss auto-partition format (e.g. {@code
+ * yyyyMMdd} for DAY). We parse it by the {@link AutoPartitionTimeUnit} and add exactly one
+ * calendar unit to get the (exact) partition end time.
+ *
- Non-auto-partitioned tables: we use Paimon's {@link PartitionTimeExtractor} (configured via
+ * timestamp-pattern / timestamp-formatter) to get the partition time, used directly in the
+ * idle judgement.
+ *
+ *
+ * The action is idempotent (Paimon success-file keeps creationTime and only updates
+ * modificationTime), so repeated execution is safe.
+ */
+class PaimonPartitionDoneHandler implements PartitionDoneHandler {
+
+ private static final Logger LOG = LoggerFactory.getLogger(PaimonPartitionDoneHandler.class);
+
+ private final List actions;
+ private final List partitionKeys;
+ private final long idleTimeMs;
+ private final ZoneId zoneId;
+
+ // auto-partition mode: non-null time unit, timeExtractor unused
+ @Nullable private final AutoPartitionTimeUnit autoPartitionTimeUnit;
+
+ // non-auto-partition mode: non-null extractor
+ @Nullable private final PartitionTimeExtractor timeExtractor;
+
+ PaimonPartitionDoneHandler(
+ List actions,
+ List partitionKeys,
+ long idleTimeMs,
+ ZoneId zoneId,
+ @Nullable AutoPartitionTimeUnit autoPartitionTimeUnit,
+ @Nullable PartitionTimeExtractor timeExtractor) {
+ this.actions = actions;
+ this.partitionKeys = partitionKeys;
+ this.idleTimeMs = idleTimeMs;
+ this.zoneId = zoneId;
+ this.autoPartitionTimeUnit = autoPartitionTimeUnit;
+ this.timeExtractor = timeExtractor;
+ }
+
+ @Override
+ public List markDoneIfReady(List candidates, long currentTime)
+ throws Exception {
+ List done = new ArrayList<>();
+ for (PartitionDoneCandidate state : candidates) {
+ ResolvedPartitionSpec spec =
+ ResolvedPartitionSpec.fromPartitionName(partitionKeys, state.partitionName());
+ long partitionEndTime = partitionEndTime(spec);
+ long lastUpdateTime = state.lastUpdateTime();
+
+ // Cannot judge if both the last update time and the partition end time are unavailable.
+ if (lastUpdateTime <= 0 && partitionEndTime <= 0) {
+ LOG.warn(
+ "Skip mark-done for partition {}: both lastUpdateTime and partitionEndTime "
+ + "are unavailable.",
+ state.partitionName());
+ continue;
+ }
+
+ long threshold = Math.max(lastUpdateTime, partitionEndTime);
+ if (currentTime - threshold > idleTimeMs) {
+ String paimonPartition = spec.getPartitionQualifiedName();
+ for (PartitionMarkDoneAction action : actions) {
+ action.markDone(paimonPartition);
+ }
+ done.add(state.partitionName());
+ }
+ }
+ return done;
+ }
+
+ /**
+ * Computes the partition end time in epoch millis. Returns 0 when the time cannot be extracted,
+ * which degrades the judgement to a pure idle check on {@code lastUpdateTime}.
+ */
+ private long partitionEndTime(ResolvedPartitionSpec spec) {
+ try {
+ if (autoPartitionTimeUnit != null) {
+ return autoPartitionEndTimeMillis(spec.getPartitionValues().get(0));
+ } else {
+ LocalDateTime start =
+ timeExtractor.extract(spec.getPartitionKeys(), spec.getPartitionValues());
+ return start.atZone(zoneId).toInstant().toEpochMilli();
+ }
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to extract partition time for partition {}, degrade to pure idle "
+ + "judgement.",
+ spec.getPartitionQualifiedName(),
+ e);
+ return 0;
+ }
+ }
+
+ /**
+ * Computes the exact partition end time for an auto-partitioned table by parsing the Fluss
+ * auto-partition value and adding exactly one calendar unit.
+ */
+ private long autoPartitionEndTimeMillis(String value) {
+ LocalDateTime end;
+ switch (autoPartitionTimeUnit) {
+ case YEAR:
+ {
+ // yyyy
+ int year = Integer.parseInt(value);
+ end = LocalDateTime.of(year, 1, 1, 0, 0).plusYears(1);
+ break;
+ }
+ case QUARTER:
+ {
+ // yyyyQ
+ int year = Integer.parseInt(value.substring(0, 4));
+ int quarter = Integer.parseInt(value.substring(4));
+ end = LocalDateTime.of(year, (quarter - 1) * 3 + 1, 1, 0, 0).plusMonths(3);
+ break;
+ }
+ case MONTH:
+ {
+ // yyyyMM
+ int year = Integer.parseInt(value.substring(0, 4));
+ int month = Integer.parseInt(value.substring(4, 6));
+ end = LocalDateTime.of(year, month, 1, 0, 0).plusMonths(1);
+ break;
+ }
+ case DAY:
+ {
+ // yyyyMMdd
+ int year = Integer.parseInt(value.substring(0, 4));
+ int month = Integer.parseInt(value.substring(4, 6));
+ int day = Integer.parseInt(value.substring(6, 8));
+ end = LocalDateTime.of(year, month, day, 0, 0).plusDays(1);
+ break;
+ }
+ case HOUR:
+ {
+ // yyyyMMddHH
+ int year = Integer.parseInt(value.substring(0, 4));
+ int month = Integer.parseInt(value.substring(4, 6));
+ int day = Integer.parseInt(value.substring(6, 8));
+ int hour = Integer.parseInt(value.substring(8, 10));
+ end = LocalDateTime.of(year, month, day, hour, 0).plusHours(1);
+ break;
+ }
+ default:
+ throw new IllegalArgumentException(
+ "Unsupported auto partition time unit: " + autoPartitionTimeUnit);
+ }
+ return end.atZone(zoneId).toInstant().toEpochMilli();
+ }
+
+ @Override
+ public void close() throws Exception {
+ for (PartitionMarkDoneAction action : actions) {
+ action.close();
+ }
+ }
+}
diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonPartitionDoneHandlerTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonPartitionDoneHandlerTest.java
new file mode 100644
index 0000000000..c527d345dc
--- /dev/null
+++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonPartitionDoneHandlerTest.java
@@ -0,0 +1,234 @@
+/*
+ * 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.lake.committer.PartitionDoneCandidate;
+
+import org.apache.paimon.partition.PartitionTimeExtractor;
+import org.apache.paimon.partition.actions.PartitionMarkDoneAction;
+import org.junit.jupiter.api.Test;
+
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Unit tests for {@link PaimonPartitionDoneHandler} mark-done judgement logic. */
+class PaimonPartitionDoneHandlerTest {
+
+ private static final ZoneId UTC = ZoneId.of("UTC");
+ private static final long ONE_DAY_MS = 24L * 60 * 60 * 1000;
+ private static final long ONE_HOUR_MS = 60L * 60 * 1000;
+
+ // 2024-01-01T00:00:00 UTC
+ private static final long DAY_20240101_START = 1704067200000L;
+ // 2024-01-02T00:00:00 UTC
+ private static final long DAY_20240101_END = DAY_20240101_START + ONE_DAY_MS;
+
+ /** A fake action recording the partitions it was asked to mark done. */
+ private static class RecordingAction implements PartitionMarkDoneAction {
+ final List marked = new ArrayList<>();
+ boolean closed = false;
+
+ @Override
+ public void markDone(String partition) {
+ marked.add(partition);
+ }
+
+ @Override
+ public void close() {
+ closed = true;
+ }
+ }
+
+ private PaimonPartitionDoneHandler nonAutoHandler(RecordingAction action, long idleMs) {
+ // non-auto-partition table: default PartitionTimeExtractor parses "yyyy-MM-dd".
+ return new PaimonPartitionDoneHandler(
+ Collections.singletonList(action),
+ Collections.singletonList("dt"),
+ idleMs,
+ UTC,
+ null,
+ new PartitionTimeExtractor(null, null));
+ }
+
+ private PaimonPartitionDoneHandler autoDayHandler(RecordingAction action, long idleMs) {
+ return new PaimonPartitionDoneHandler(
+ Collections.singletonList(action),
+ Collections.singletonList("dt"),
+ idleMs,
+ UTC,
+ AutoPartitionTimeUnit.DAY,
+ null);
+ }
+
+ @Test
+ void testMarkDoneWhenIdleExceeded() throws Exception {
+ RecordingAction action = new RecordingAction();
+ PaimonPartitionDoneHandler handler = nonAutoHandler(action, ONE_HOUR_MS);
+
+ // non-auto table: threshold = MAX(lastUpdate, partitionStart) = partition start.
+ PartitionDoneCandidate state = new PartitionDoneCandidate("2024-01-01", DAY_20240101_START);
+ long now = DAY_20240101_END + ONE_HOUR_MS + 1; // strictly greater than idle
+
+ List done = handler.markDoneIfReady(Collections.singletonList(state), now);
+
+ assertThat(done).containsExactly("2024-01-01");
+ assertThat(action.marked).containsExactly("dt=2024-01-01");
+ }
+
+ @Test
+ void testNotMarkDoneAtIdleBoundary() throws Exception {
+ RecordingAction action = new RecordingAction();
+ PaimonPartitionDoneHandler handler = nonAutoHandler(action, ONE_HOUR_MS);
+
+ PartitionDoneCandidate state = new PartitionDoneCandidate("2024-01-01", DAY_20240101_START);
+ long now =
+ DAY_20240101_START
+ + ONE_HOUR_MS; // exactly one idle after partition start -> not done
+
+ List done = handler.markDoneIfReady(Collections.singletonList(state), now);
+
+ assertThat(done).isEmpty();
+ assertThat(action.marked).isEmpty();
+ }
+
+ @Test
+ void testNotMarkDoneWhenPartitionStartInFuture() throws Exception {
+ RecordingAction action = new RecordingAction();
+ PaimonPartitionDoneHandler handler = nonAutoHandler(action, ONE_HOUR_MS);
+
+ // lastUpdateTime is very old, but the partition's own start time is in the future relative
+ // to now, so threshold = MAX(lastUpdateTime, partitionStart) = partitionStart and it must
+ // not be done.
+ PartitionDoneCandidate state = new PartitionDoneCandidate("2024-01-02", 1L);
+ long now = DAY_20240101_START + ONE_HOUR_MS; // before the 2024-01-02 partition start
+
+ List done = handler.markDoneIfReady(Collections.singletonList(state), now);
+
+ assertThat(done).isEmpty();
+ }
+
+ @Test
+ void testAutoPartitionDayEndTimeComputation() throws Exception {
+ RecordingAction action = new RecordingAction();
+ PaimonPartitionDoneHandler handler = autoDayHandler(action, ONE_HOUR_MS);
+
+ // auto-partition value uses Fluss format yyyyMMdd.
+ PartitionDoneCandidate state = new PartitionDoneCandidate("20240101", DAY_20240101_START);
+
+ // at boundary -> not done
+ assertThat(
+ handler.markDoneIfReady(
+ Collections.singletonList(state), DAY_20240101_END + ONE_HOUR_MS))
+ .isEmpty();
+ // just past boundary -> done
+ assertThat(
+ handler.markDoneIfReady(
+ Collections.singletonList(state),
+ DAY_20240101_END + ONE_HOUR_MS + 1))
+ .containsExactly("20240101");
+ assertThat(action.marked).containsExactly("dt=20240101");
+ }
+
+ @Test
+ void testPartialDoneAmongMultiplePartitions() throws Exception {
+ RecordingAction action = new RecordingAction();
+ PaimonPartitionDoneHandler handler = autoDayHandler(action, ONE_HOUR_MS);
+
+ // 2024-01-01 ended long ago -> done; 2024-01-03 end in the future -> not done.
+ PartitionDoneCandidate oldPartition =
+ new PartitionDoneCandidate("20240101", DAY_20240101_START);
+ long day03Start = DAY_20240101_START + 2 * ONE_DAY_MS;
+ PartitionDoneCandidate currentPartition =
+ new PartitionDoneCandidate("20240103", day03Start);
+ long now = day03Start + ONE_HOUR_MS; // 2024-01-03 not ended yet
+
+ List done =
+ handler.markDoneIfReady(Arrays.asList(oldPartition, currentPartition), now);
+
+ assertThat(done).containsExactly("20240101");
+ }
+
+ private PaimonPartitionDoneHandler autoHandler(
+ RecordingAction action, long idleMs, AutoPartitionTimeUnit unit) {
+ return new PaimonPartitionDoneHandler(
+ Collections.singletonList(action),
+ Collections.singletonList("dt"),
+ idleMs,
+ UTC,
+ unit,
+ null);
+ }
+
+ @Test
+ void testAutoPartitionEndTimeForAllTimeUnits() throws Exception {
+ // now far in the future so every partition below is idle enough to be done, exercising
+ // each auto-partition-value parsing branch (YEAR/QUARTER/MONTH/HOUR; DAY covered above).
+ long farFuture = 1893456000000L; // 2030-01-01T00:00:00 UTC
+ assertAutoDone(AutoPartitionTimeUnit.YEAR, "2024", farFuture);
+ assertAutoDone(AutoPartitionTimeUnit.QUARTER, "20241", farFuture);
+ assertAutoDone(AutoPartitionTimeUnit.MONTH, "202401", farFuture);
+ assertAutoDone(AutoPartitionTimeUnit.HOUR, "2024010100", farFuture);
+ }
+
+ private void assertAutoDone(AutoPartitionTimeUnit unit, String value, long now)
+ throws Exception {
+ RecordingAction action = new RecordingAction();
+ PaimonPartitionDoneHandler handler = autoHandler(action, ONE_HOUR_MS, unit);
+ PartitionDoneCandidate state = new PartitionDoneCandidate(value, 0L);
+ assertThat(handler.markDoneIfReady(Collections.singletonList(state), now))
+ .containsExactly(value);
+ assertThat(action.marked).containsExactly("dt=" + value);
+ }
+
+ @Test
+ void testDegradeToIdleWhenPartitionTimeUnparseable() throws Exception {
+ // a non-date partition value makes PartitionTimeExtractor throw, so partitionEndTime
+ // degrades to 0 and the judgement falls back to lastUpdateTime alone.
+ RecordingAction action = new RecordingAction();
+ PaimonPartitionDoneHandler handler = nonAutoHandler(action, ONE_HOUR_MS);
+ PartitionDoneCandidate state = new PartitionDoneCandidate("not-a-date", 1000L);
+ long now = 1000L + ONE_HOUR_MS + 1;
+ assertThat(handler.markDoneIfReady(Collections.singletonList(state), now))
+ .containsExactly("not-a-date");
+ }
+
+ @Test
+ void testSkipWhenNeitherTimeAvailable() throws Exception {
+ // both lastUpdateTime and partitionEndTime unavailable -> skip.
+ RecordingAction action = new RecordingAction();
+ PaimonPartitionDoneHandler handler = nonAutoHandler(action, ONE_HOUR_MS);
+ PartitionDoneCandidate state = new PartitionDoneCandidate("not-a-date", 0L);
+ assertThat(handler.markDoneIfReady(Collections.singletonList(state), Long.MAX_VALUE))
+ .isEmpty();
+ assertThat(action.marked).isEmpty();
+ }
+
+ @Test
+ void testCloseClosesActions() throws Exception {
+ RecordingAction action = new RecordingAction();
+ PaimonPartitionDoneHandler handler = autoDayHandler(action, ONE_HOUR_MS);
+ handler.close();
+ assertThat(action.closed).isTrue();
+ }
+}
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 3d4da4fe50..343ab5db4c 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
@@ -22,6 +22,7 @@
import org.apache.fluss.config.ConfigOptions;
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;
@@ -188,6 +189,73 @@ void testTiering() throws Exception {
}
}
+ @Test
+ void testPartitionMarkDone() throws Exception {
+ // A non-auto partitioned table with a past-date partition so that, once the table becomes
+ // fully idle, the heartbeat rounds drive the partition to be marked done (a _SUCCESS file
+ // is written to the Paimon partition directory).
+ TablePath tablePath = TablePath.of(DEFAULT_DB, "markDonePartitionedTable");
+ Schema schema =
+ Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .column("b", DataTypes.STRING())
+ .column("dt", DataTypes.STRING())
+ .build();
+ TableDescriptor descriptor =
+ TableDescriptor.builder()
+ .schema(schema)
+ .distributedBy(1, "a")
+ .partitionedBy("dt")
+ .property(ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true")
+ .property(ConfigOptions.TABLE_DATALAKE_FRESHNESS, Duration.ofMillis(500))
+ .customProperty("paimon.partition.idle-time-to-done", "3 s")
+ .build();
+ createTable(tablePath, descriptor);
+
+ String partition = "2020-01-01";
+ admin.createPartition(
+ tablePath,
+ new PartitionSpec(Collections.singletonMap("dt", partition)),
+ false)
+ .get();
+
+ // write some rows into the partition, then stop writing so the table becomes idle
+ writeRows(
+ tablePath,
+ Arrays.asList(
+ row(1, "v1", partition), row(2, "v2", partition), row(3, "v3", partition)),
+ true);
+
+ JobClient jobClient = buildTieringJob(execEnv);
+ try {
+ FileStoreTable paimonTable =
+ (FileStoreTable)
+ paimonCatalog.getTable(
+ Identifier.create(
+ tablePath.getDatabaseName(), tablePath.getTableName()));
+ org.apache.paimon.fs.Path successFile =
+ new org.apache.paimon.fs.Path(
+ paimonTable.location(), "dt=" + partition + "/_SUCCESS");
+
+ // wait until the _SUCCESS file appears (idle 3s + heartbeat rounds drive mark-done
+ // once the table becomes fully idle)
+ long deadline = System.currentTimeMillis() + 60_000L;
+ boolean marked = false;
+ while (System.currentTimeMillis() < deadline) {
+ if (paimonTable.fileIO().exists(successFile)) {
+ marked = true;
+ break;
+ }
+ Thread.sleep(1000L);
+ }
+ assertThat(marked)
+ .as("the _SUCCESS mark-done file should be created for partition " + partition)
+ .isTrue();
+ } finally {
+ jobClient.cancel().get();
+ }
+ }
+
private static Stream tieringAllTypesWriteArgs() {
return Stream.of(Arguments.of(true), Arguments.of(false));
}
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java
index b03e1ec63a..abf43fdd57 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java
@@ -69,7 +69,8 @@ public enum ApiKeys {
NOTIFY_KV_SNAPSHOT_OFFSET(1029, 0, 0, PRIVATE),
COMMIT_LAKE_TABLE_SNAPSHOT(1030, 0, 0, PRIVATE),
NOTIFY_LAKE_TABLE_OFFSET(1031, 0, 0, PRIVATE),
- GET_LAKE_SNAPSHOT(1032, 0, 0, PUBLIC),
+ // Version 1: response may carry the opaque tiering_state_json (for capability detection).
+ GET_LAKE_SNAPSHOT(1032, 0, 1, PUBLIC),
LIMIT_SCAN(1033, 0, 0, PUBLIC),
// Version 0: Uses lake's encoder for prefix key encoding (legacy behavior).
@@ -94,7 +95,8 @@ public enum ApiKeys {
REBALANCE(1049, 0, 0, PUBLIC),
LIST_REBALANCE_PROGRESS(1050, 0, 0, PUBLIC),
CANCEL_REBALANCE(1051, 0, 0, PUBLIC),
- PREPARE_LAKE_TABLE_SNAPSHOT(1052, 0, 0, PRIVATE),
+ // Version 1: request may carry the opaque tiering_state_json (for capability detection).
+ PREPARE_LAKE_TABLE_SNAPSHOT(1052, 0, 1, PRIVATE),
REGISTER_PRODUCER_OFFSETS(1053, 0, 0, PUBLIC),
GET_PRODUCER_OFFSETS(1054, 0, 0, PUBLIC),
DELETE_PRODUCER_OFFSETS(1055, 0, 0, PUBLIC),
diff --git a/fluss-rpc/src/main/proto/FlussApi.proto b/fluss-rpc/src/main/proto/FlussApi.proto
index 57e96092d4..fd76044dec 100644
--- a/fluss-rpc/src/main/proto/FlussApi.proto
+++ b/fluss-rpc/src/main/proto/FlussApi.proto
@@ -476,6 +476,8 @@ message GetLakeSnapshotResponse {
required int64 table_id = 1;
required int64 snapshotId = 2;
repeated PbLakeSnapshotForBucket bucket_snapshots = 3;
+ // Opaque table-level tiering state (see LakeTieringTableState), passed through as JSON bytes.
+ optional bytes tiering_state_json = 4;
}
message GetFileSystemSecurityTokenRequest {
@@ -1284,6 +1286,8 @@ message PbTableOffsets {
required int64 table_id = 1;
required PbTablePath table_path = 2;
repeated PbBucketOffset bucket_offsets = 3;
+ // Opaque table-level tiering state (see LakeTieringTableState), passed through as JSON bytes.
+ optional bytes tiering_state_json = 4;
}
message PbBucketOffset {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
index 1e5e914c0d..38ea6f2f72 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
@@ -1822,7 +1822,11 @@ public static TableBucketOffsets toTableBucketOffsets(PbTableOffsets pbTableOffs
pbBucketOffset.getBucketId());
bucketOffsets.put(tableBucket, pbBucketOffset.getLogEndOffset());
}
- return new TableBucketOffsets(tableId, bucketOffsets);
+
+ // pass through the opaque tiering-state payload
+ byte[] tieringStateJson =
+ pbTableOffsets.hasTieringStateJson() ? pbTableOffsets.getTieringStateJson() : null;
+ return new TableBucketOffsets(tableId, bucketOffsets, tieringStateJson);
}
/**
@@ -1960,6 +1964,12 @@ public static GetLakeSnapshotResponse makeGetLakeSnapshotResponse(
}
}
+ // pass through the opaque tiering-state payload
+ if (lakeTableSnapshot.getTieringStateJson() != null) {
+ getLakeTableSnapshotResponse.setTieringStateJson(
+ lakeTableSnapshot.getTieringStateJson());
+ }
+
return getLakeTableSnapshotResponse;
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTable.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTable.java
index 9e3e183861..d2659862b1 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTable.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTable.java
@@ -21,7 +21,6 @@
import org.apache.fluss.fs.FSDataInputStream;
import org.apache.fluss.fs.FileSystem;
import org.apache.fluss.fs.FsPath;
-import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.server.zk.data.ZkData;
import org.apache.fluss.utils.IOUtils;
import org.apache.fluss.utils.json.TableBucketOffsets;
@@ -32,7 +31,6 @@
import java.io.IOException;
import java.util.Collections;
import java.util.List;
-import java.util.Map;
import java.util.Objects;
import static org.apache.fluss.metrics.registry.MetricRegistry.LOG;
@@ -203,9 +201,12 @@ private LakeTableSnapshot toLakeTableSnapshot(long snapshotId, FsPath offsetFile
FSDataInputStream inputStream = offsetFilePath.getFileSystem().open(offsetFilePath);
try (ByteArrayOutputStream outputStream = new ByteArrayOutputStream()) {
IOUtils.copyBytes(inputStream, outputStream, true);
- Map logOffsets =
- TableBucketOffsets.fromJsonBytes(outputStream.toByteArray()).getOffsets();
- return new LakeTableSnapshot(snapshotId, logOffsets);
+ TableBucketOffsets tableBucketOffsets =
+ TableBucketOffsets.fromJsonBytes(outputStream.toByteArray());
+ return new LakeTableSnapshot(
+ snapshotId,
+ tableBucketOffsets.getOffsets(),
+ tableBucketOffsets.getTieringStateJson());
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableHelper.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableHelper.java
index 0e70d80767..60c328e1c1 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableHelper.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableHelper.java
@@ -171,13 +171,25 @@ public TableBucketOffsets mergeTableBucketOffsets(
// Merge current with previous one since the current request
// may not carry all buckets for the table. It typically only carries buckets
// that were written after the previous commit.
+ LakeTableSnapshot previousSnapshot = previousLakeTable.getOrReadLatestTableSnapshot();
// merge log end offsets, current will override the previous
Map bucketLogEndOffset =
- new HashMap<>(
- previousLakeTable.getOrReadLatestTableSnapshot().getBucketLogEndOffset());
+ new HashMap<>(previousSnapshot.getBucketLogEndOffset());
bucketLogEndOffset.putAll(newTableBucketOffsets.getOffsets());
- return new TableBucketOffsets(newTableBucketOffsets.getTableId(), bucketLogEndOffset);
+
+ // Unlike the per-bucket offsets above (partial merge), the tiering state is opaque here and
+ // cannot be field-merged, so it uses whole-value snapshot (PUT) semantics:
+ // - present (including an empty object) overwrites the whole state;
+ // - absent keeps the previous state (a legacy/no-op writer never sends it, so keeping the
+ // previous value is required for rolling-upgrade safety).
+ byte[] tieringStateJson =
+ newTableBucketOffsets.getTieringStateJson() != null
+ ? newTableBucketOffsets.getTieringStateJson()
+ : previousSnapshot.getTieringStateJson();
+
+ return new TableBucketOffsets(
+ newTableBucketOffsets.getTableId(), bucketLogEndOffset, tieringStateJson);
}
public FsPath storeLakeTableOffsetsFile(
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableSnapshot.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableSnapshot.java
index f567a19d36..9fc6894574 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableSnapshot.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableSnapshot.java
@@ -20,6 +20,9 @@
import org.apache.fluss.metadata.TableBucket;
+import javax.annotation.Nullable;
+
+import java.util.Arrays;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
@@ -35,9 +38,20 @@ public class LakeTableSnapshot {
// will be null if log offset is unknown such as reading the snapshot of primary key table
private final Map bucketLogEndOffset;
+ // opaque tiering-state JSON bytes; null when absent.
+ @Nullable private final byte[] tieringStateJson;
+
public LakeTableSnapshot(long snapshotId, Map bucketLogEndOffset) {
+ this(snapshotId, bucketLogEndOffset, null);
+ }
+
+ public LakeTableSnapshot(
+ long snapshotId,
+ Map bucketLogEndOffset,
+ @Nullable byte[] tieringStateJson) {
this.snapshotId = snapshotId;
this.bucketLogEndOffset = bucketLogEndOffset;
+ this.tieringStateJson = tieringStateJson;
}
public long getSnapshotId() {
@@ -52,6 +66,12 @@ public Map getBucketLogEndOffset() {
return bucketLogEndOffset;
}
+ /** Returns the opaque tiering-state JSON bytes, or {@code null} if absent. */
+ @Nullable
+ public byte[] getTieringStateJson() {
+ return tieringStateJson;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -62,12 +82,13 @@ public boolean equals(Object o) {
}
LakeTableSnapshot that = (LakeTableSnapshot) o;
return snapshotId == that.snapshotId
- && Objects.equals(bucketLogEndOffset, that.bucketLogEndOffset);
+ && Objects.equals(bucketLogEndOffset, that.bucketLogEndOffset)
+ && Arrays.equals(tieringStateJson, that.tieringStateJson);
}
@Override
public int hashCode() {
- return Objects.hash(snapshotId, bucketLogEndOffset);
+ return Objects.hash(snapshotId, bucketLogEndOffset, Arrays.hashCode(tieringStateJson));
}
@Override
@@ -77,6 +98,8 @@ public String toString() {
+ snapshotId
+ ", bucketLogEndOffset="
+ bucketLogEndOffset
+ + ", tieringStateJson="
+ + Arrays.toString(tieringStateJson)
+ '}';
}
}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTieringStateTest.java b/fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTieringStateTest.java
new file mode 100644
index 0000000000..06a297b5d5
--- /dev/null
+++ b/fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTieringStateTest.java
@@ -0,0 +1,89 @@
+/*
+ * 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.server.utils;
+
+import org.apache.fluss.lake.committer.LakeTieringTableState;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.rpc.messages.GetLakeSnapshotResponse;
+import org.apache.fluss.rpc.messages.PbTableOffsets;
+import org.apache.fluss.server.zk.data.lake.LakeTableSnapshot;
+import org.apache.fluss.utils.json.TableBucketOffsets;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for the opaque tiering-state passthrough in {@link ServerRpcMessageUtils}, covering the
+ * PREPARE read path ({@link ServerRpcMessageUtils#toTableBucketOffsets}) and the GET fill path
+ * ({@link ServerRpcMessageUtils#makeGetLakeSnapshotResponse}).
+ */
+class ServerRpcMessageUtilsTieringStateTest {
+
+ @Test
+ void testTieringStatePassthroughRoundTrip() {
+ // PREPARE read (toTableBucketOffsets) -> persist -> GET fill (makeGetLakeSnapshotResponse).
+ long tableId = 1L;
+ byte[] tieringStateJson =
+ new LakeTieringTableState(true, Collections.singletonMap(5L, 1704153550000L))
+ .toJsonBytes();
+
+ PbTableOffsets pbTableOffsets = new PbTableOffsets();
+ pbTableOffsets.setTableId(tableId);
+ pbTableOffsets.setTablePath().setDatabaseName("db").setTableName("t");
+ pbTableOffsets.addBucketOffset().setPartitionId(5L).setBucketId(0).setLogEndOffset(100L);
+ pbTableOffsets.setTieringStateJson(tieringStateJson);
+
+ TableBucketOffsets offsets = ServerRpcMessageUtils.toTableBucketOffsets(pbTableOffsets);
+ assertThat(offsets.getOffsets()).containsEntry(new TableBucket(tableId, 5L, 0), 100L);
+ assertThat(offsets.getTieringStateJson()).isEqualTo(tieringStateJson);
+
+ LakeTableSnapshot snapshot =
+ new LakeTableSnapshot(9L, offsets.getOffsets(), offsets.getTieringStateJson());
+ GetLakeSnapshotResponse response =
+ ServerRpcMessageUtils.makeGetLakeSnapshotResponse(tableId, snapshot);
+ assertThat(response.getTableId()).isEqualTo(tableId);
+ assertThat(response.getSnapshotId()).isEqualTo(9L);
+ assertThat(response.hasTieringStateJson()).isTrue();
+ assertThat(response.getTieringStateJson()).isEqualTo(tieringStateJson);
+ }
+
+ @Test
+ void testAbsentTieringState() {
+ // PREPARE read without tiering state -> null.
+ PbTableOffsets pbTableOffsets = new PbTableOffsets();
+ pbTableOffsets.setTableId(2L);
+ pbTableOffsets.setTablePath().setDatabaseName("db").setTableName("t");
+ pbTableOffsets.addBucketOffset().setBucketId(0).setLogEndOffset(100L);
+ assertThat(ServerRpcMessageUtils.toTableBucketOffsets(pbTableOffsets).getTieringStateJson())
+ .isNull();
+
+ // GET fill with a null tiering state -> response omits it.
+ Map bucketOffsets = new HashMap<>();
+ bucketOffsets.put(new TableBucket(2L, 0), 100L);
+ GetLakeSnapshotResponse response =
+ ServerRpcMessageUtils.makeGetLakeSnapshotResponse(
+ 2L, new LakeTableSnapshot(9L, bucketOffsets, null));
+ assertThat(response.hasTieringStateJson()).isFalse();
+ }
+}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/zk/data/lake/LakeTableHelperTest.java b/fluss-server/src/test/java/org/apache/fluss/server/zk/data/lake/LakeTableHelperTest.java
index e3b0f78694..0d5ec3ea59 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/zk/data/lake/LakeTableHelperTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/zk/data/lake/LakeTableHelperTest.java
@@ -24,6 +24,7 @@
import org.apache.fluss.fs.FsPath;
import org.apache.fluss.fs.local.LocalFileSystem;
import org.apache.fluss.lake.committer.LakeCommitResult;
+import org.apache.fluss.lake.committer.LakeTieringTableState;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TablePath;
@@ -253,6 +254,91 @@ void testRegisterLakeTableSnapshotWithRetention(@TempDir Path tempDir) throws Ex
.containsExactly(4L, 5L, 6L);
}
+ /**
+ * Verifies the opaque tiering state uses overwrite semantics on merge: a new request carrying
+ * tiering_state replaces the previous one, while bucket offsets are still merged.
+ */
+ @Test
+ void testMergeTieringStateOverwrite(@TempDir Path tempDir) throws Exception {
+ LakeTableHelper lakeTableHelper = new LakeTableHelper(zookeeperClient, tempDir.toString());
+ long tableId = 100L;
+ TablePath tablePath = TablePath.of("test_db", "markdone_merge_test");
+ zookeeperClient.registerTable(tablePath, createTableReg(tableId));
+
+ // --- Previous snapshot: partitions 1, 2, 3 with a tiering state ---
+ Map prevOffsets = new HashMap<>();
+ prevOffsets.put(new TableBucket(tableId, 1L, 0), 100L);
+ prevOffsets.put(new TableBucket(tableId, 2L, 0), 200L);
+ prevOffsets.put(new TableBucket(tableId, 3L, 0), 300L);
+ Map prevUpdateTimes = new HashMap<>();
+ prevUpdateTimes.put(1L, 1000L);
+ prevUpdateTimes.put(2L, 2000L);
+ prevUpdateTimes.put(3L, 3000L);
+ byte[] prevState = new LakeTieringTableState(true, prevUpdateTimes).toJsonBytes();
+ FsPath prevPath =
+ lakeTableHelper.storeLakeTableOffsetsFile(
+ tablePath, new TableBucketOffsets(tableId, prevOffsets, prevState));
+ lakeTableHelper.registerLakeTableSnapshotV2(
+ tableId, new LakeTable.LakeSnapshotMetadata(1L, prevPath, prevPath));
+ LakeTable previousLakeTable = zookeeperClient.getLakeTable(tableId).get();
+
+ // --- New offsets: advance partitions 1 & 3, carry a full new tiering state (partition 2
+ // done -> dropped). ---
+ Map newOffsets = new HashMap<>();
+ newOffsets.put(new TableBucket(tableId, 1L, 0), 150L);
+ newOffsets.put(new TableBucket(tableId, 3L, 0), 350L);
+ Map newUpdateTimes = new HashMap<>();
+ newUpdateTimes.put(1L, 1500L);
+ newUpdateTimes.put(3L, 3500L);
+ byte[] newState = new LakeTieringTableState(true, newUpdateTimes).toJsonBytes();
+ TableBucketOffsets merged =
+ lakeTableHelper.mergeTableBucketOffsets(
+ previousLakeTable, new TableBucketOffsets(tableId, newOffsets, newState));
+
+ // tiering state overwritten by the new value
+ assertThat(merged.getTieringStateJson()).isEqualTo(newState);
+ LakeTieringTableState mergedState =
+ LakeTieringTableState.fromJsonBytes(merged.getTieringStateJson());
+ assertThat(mergedState.getPartitionUpdateTimes()).containsOnlyKeys(1L, 3L);
+ assertThat(mergedState.getPartitionUpdateTimes()).containsEntry(1L, 1500L);
+ assertThat(mergedState.getPartitionUpdateTimes()).containsEntry(3L, 3500L);
+ // bucket offsets merged (partition 2 kept from previous)
+ assertThat(merged.getOffsets()).containsEntry(new TableBucket(tableId, 1L, 0), 150L);
+ assertThat(merged.getOffsets()).containsEntry(new TableBucket(tableId, 2L, 0), 200L);
+ assertThat(merged.getOffsets()).containsEntry(new TableBucket(tableId, 3L, 0), 350L);
+ }
+
+ /** Verifies a new request without tiering state keeps the previous one on merge. */
+ @Test
+ void testMergeKeepsPreviousTieringStateWhenAbsent(@TempDir Path tempDir) throws Exception {
+ LakeTableHelper lakeTableHelper = new LakeTableHelper(zookeeperClient, tempDir.toString());
+ long tableId = 101L;
+ TablePath tablePath = TablePath.of("test_db", "markdone_keep_test");
+ zookeeperClient.registerTable(tablePath, createTableReg(tableId));
+
+ Map prevOffsets = new HashMap<>();
+ prevOffsets.put(new TableBucket(tableId, 1L, 0), 100L);
+ byte[] prevState =
+ new LakeTieringTableState(true, java.util.Collections.singletonMap(1L, 1000L))
+ .toJsonBytes();
+ FsPath prevPath =
+ lakeTableHelper.storeLakeTableOffsetsFile(
+ tablePath, new TableBucketOffsets(tableId, prevOffsets, prevState));
+ lakeTableHelper.registerLakeTableSnapshotV2(
+ tableId, new LakeTable.LakeSnapshotMetadata(1L, prevPath, prevPath));
+ LakeTable previousLakeTable = zookeeperClient.getLakeTable(tableId).get();
+
+ Map newOffsets = new HashMap<>();
+ newOffsets.put(new TableBucket(tableId, 1L, 0), 150L);
+ // new request carries no tiering state (null)
+ TableBucketOffsets merged =
+ lakeTableHelper.mergeTableBucketOffsets(
+ previousLakeTable, new TableBucketOffsets(tableId, newOffsets, null));
+
+ assertThat(merged.getTieringStateJson()).isEqualTo(prevState);
+ assertThat(merged.getOffsets()).containsEntry(new TableBucket(tableId, 1L, 0), 150L);
+ }
+
/** Helper to store offset files and return the FsPath. */
private FsPath storeOffsetFile(
LakeTableHelper helper, TablePath path, long tableId, long offset) throws Exception {
diff --git a/fluss-test-coverage/pom.xml b/fluss-test-coverage/pom.xml
index 357b3e98a2..ed503b490b 100644
--- a/fluss-test-coverage/pom.xml
+++ b/fluss-test-coverage/pom.xml
@@ -496,6 +496,9 @@
org.apache.fluss.flink.tiering.LakeTieringJobBuilder
org.apache.fluss.lake.committer.LakeCommitResult*
+
+ org.apache.fluss.lake.committer.PartitionDoneCandidate
+ org.apache.fluss.lake.committer.PartitionDoneHandler
org.apache.fluss.flink.tiering.FlussLakeTieringEntrypoint
org.apache.fluss.flink.tiering.FlussLakeTiering