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..656dc550fa
--- /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}. When a newer and an older tiering job run against the same table across
+ * rounds, a build that reads a {@code version} higher than its own {@link #CURRENT_VERSION} cannot
+ * interpret it: {@link #fromJsonBytes} raises {@link IllegalArgumentException}, and the caller must
+ * pass the raw bytes through unchanged so the newer build's state is never dropped.
+ *
+ * @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/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..7594b96b5d
--- /dev/null
+++ b/fluss-common/src/main/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerde.java
@@ -0,0 +1,154 @@
+/*
+ * 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}}}.
+ *
+ * Validation is strict for known fields, tolerant for unknown ones. {@code version} is mandatory
+ * and positive; a corrupt known field fails fast. A higher-than-supported {@code version} is
+ * neither read nor written; the caller keeps the raw bytes and passes them through unchanged so a
+ * newer build's state is never dropped.
+ */
+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 {
+ // Defense: this build never constructs a higher-version state to write (it passes the raw
+ // bytes through instead), so this guards against misuse of the version-taking constructor.
+ if (state.getVersion() > LakeTieringTableState.CURRENT_VERSION) {
+ throw new IllegalStateException(
+ "Refusing to serialize unsupported tiering state version "
+ + state.getVersion()
+ + " > "
+ + LakeTieringTableState.CURRENT_VERSION
+ + ".");
+ }
+ 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) {
+ if (node == null || !node.isObject()) {
+ throw new IllegalArgumentException(
+ "Corrupt tiering state: expected a JSON object but got " + node);
+ }
+
+ // version is mandatory and a positive integer; it gates the rest of the parsing.
+ JsonNode versionNode = node.get(VERSION_KEY);
+ if (versionNode == null || !versionNode.isInt() || versionNode.asInt() <= 0) {
+ throw new IllegalArgumentException(
+ "Corrupt tiering state: '"
+ + VERSION_KEY
+ + "' must be a positive integer but got "
+ + versionNode);
+ }
+ int version = versionNode.asInt();
+
+ // A newer version cannot be interpreted here; the caller must pass the raw bytes through.
+ if (version > LakeTieringTableState.CURRENT_VERSION) {
+ throw new IllegalArgumentException(
+ "Unsupported tiering state version "
+ + version
+ + " > "
+ + LakeTieringTableState.CURRENT_VERSION
+ + "; pass the raw bytes through unchanged.");
+ }
+
+ // partition_done_initialized: optional, must be a boolean when present.
+ JsonNode initializedNode = node.get(PARTITION_DONE_INITIALIZED_KEY);
+ if (initializedNode != null && !initializedNode.isBoolean()) {
+ throw new IllegalArgumentException(
+ "Corrupt tiering state: '"
+ + PARTITION_DONE_INITIALIZED_KEY
+ + "' must be a boolean but got "
+ + initializedNode);
+ }
+ boolean partitionDoneInitialized = initializedNode != null && initializedNode.asBoolean();
+
+ // partition_update_times: optional object; positive-id keys, non-negative values.
+ Map partitionUpdateTimes = new HashMap<>();
+ JsonNode updateTimesNode = node.get(PARTITION_UPDATE_TIMES_KEY);
+ if (updateTimesNode != null && !updateTimesNode.isNull()) {
+ if (!updateTimesNode.isObject()) {
+ throw new IllegalArgumentException(
+ "Corrupt tiering state: '"
+ + PARTITION_UPDATE_TIMES_KEY
+ + "' must be a JSON object but got "
+ + updateTimesNode);
+ }
+ Iterator> fields = updateTimesNode.fields();
+ while (fields.hasNext()) {
+ Map.Entry field = fields.next();
+ long partitionId;
+ try {
+ partitionId = Long.parseLong(field.getKey());
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(
+ "Corrupt tiering state: invalid partitionId key '"
+ + field.getKey()
+ + "' in "
+ + PARTITION_UPDATE_TIMES_KEY);
+ }
+ if (partitionId <= 0) {
+ throw new IllegalArgumentException(
+ "Corrupt tiering state: partitionId must be positive but got "
+ + partitionId);
+ }
+ JsonNode valueNode = field.getValue();
+ if (!valueNode.canConvertToLong() || valueNode.asLong() < 0) {
+ throw new IllegalArgumentException(
+ "Corrupt tiering state: update time for partition "
+ + partitionId
+ + " must be a non-negative integer but got "
+ + valueNode);
+ }
+ partitionUpdateTimes.put(partitionId, valueNode.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..3eb15e8f8c
--- /dev/null
+++ b/fluss-common/src/test/java/org/apache/fluss/utils/json/LakeTieringTableStateJsonSerdeTest.java
@@ -0,0 +1,111 @@
+/*
+ * 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;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** 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);
+
+ // with a version present, missing fields fall back to defaults.
+ LakeTieringTableState defaults =
+ LakeTieringTableState.fromJsonBytes(
+ "{\"version\":1}".getBytes(StandardCharsets.UTF_8));
+ assertThat(defaults.getVersion()).isEqualTo(LakeTieringTableState.VERSION_1);
+ assertThat(defaults.isPartitionDoneInitialized()).isFalse();
+ assertThat(defaults.getPartitionUpdateTimes()).isEmpty();
+ }
+
+ @Test
+ void testStrictValidationRejectsCorruptFields() {
+ // version is mandatory and must be a positive integer; for a known version, corrupt known
+ // fields must fail fast rather than be silently coerced.
+ String[] corrupt = {
+ "{}",
+ "{\"version\":0}",
+ "{\"version\":\"x\"}",
+ "{\"version\":1,\"partition_done_initialized\":\"x\"}",
+ "{\"version\":1,\"partition_update_times\":5}",
+ "{\"version\":1,\"partition_update_times\":{\"x\":1}}",
+ "{\"version\":1,\"partition_update_times\":{\"0\":1}}",
+ "{\"version\":1,\"partition_update_times\":{\"5\":-1}}"
+ };
+ for (String bad : corrupt) {
+ assertThatThrownBy(
+ () ->
+ LakeTieringTableState.fromJsonBytes(
+ bad.getBytes(StandardCharsets.UTF_8)))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+ }
+
+ @Test
+ void testHigherVersionIsNeitherReadableNorWritable() {
+ // Reading a newer, unreadable version fails fast so the caller passes the raw bytes through
+ // instead of interpreting (and possibly dropping) it.
+ String json =
+ "{\"version\":99,\"partition_done_initialized\":true,"
+ + "\"partition_update_times\":{\"5\":1000},\"future_field\":\"x\"}";
+ assertThatThrownBy(
+ () ->
+ LakeTieringTableState.fromJsonBytes(
+ json.getBytes(StandardCharsets.UTF_8)))
+ .isInstanceOf(IllegalArgumentException.class);
+
+ // Writing a higher-version object is likewise refused (defense against misuse).
+ LakeTieringTableState higher = new LakeTieringTableState(99, true, new HashMap<>());
+ assertThatThrownBy(higher::toJsonBytes).isInstanceOf(IllegalStateException.class);
+ }
+}
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-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..d1fc649c70 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
@@ -67,7 +67,8 @@ public enum ApiKeys {
COMMIT_REMOTE_LOG_MANIFEST(1027, 0, 0, PRIVATE),
NOTIFY_REMOTE_LOG_OFFSETS(1028, 0, 0, PRIVATE),
NOTIFY_KV_SNAPSHOT_OFFSET(1029, 0, 0, PRIVATE),
- COMMIT_LAKE_TABLE_SNAPSHOT(1030, 0, 0, PRIVATE),
+ // Version 1: PbLakeTableSnapshotMetadata may carry tiering_epoch for fencing.
+ COMMIT_LAKE_TABLE_SNAPSHOT(1030, 0, 1, PRIVATE),
NOTIFY_LAKE_TABLE_OFFSET(1031, 0, 0, PRIVATE),
GET_LAKE_SNAPSHOT(1032, 0, 0, PUBLIC),
LIMIT_SCAN(1033, 0, 0, PUBLIC),
@@ -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..bcb1c74be7 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 {
@@ -1245,6 +1247,8 @@ message PbRebalancePlanForBucket {
message PbLakeTableSnapshotMetadata {
required int64 table_id = 1;
+ // The lake snapshot id; may repeat across entries when a state-only round (no new lake commit)
+ // reuses it. Reads resolve to the latest entry for a given id.
required int64 snapshot_id = 2;
required string tiered_bucket_offsets_file_path = 3;
optional string readable_bucket_offsets_file_path = 4;
@@ -1252,6 +1256,9 @@ message PbLakeTableSnapshotMetadata {
// 1. If set, the system will keep all snapshots in the range [earliest_snapshot_id_to_keep, current_snapshot_id].
// 2. If not set, the system defaults to a "Single Snapshot Retention" policy, keeping only the snapshot specified in this request.
optional int64 earliest_snapshot_id_to_keep = 5;
+ // Tiering assignment epoch for fencing (optional for compatibility). When present, the commit is
+ // rejected unless it matches the table's current epoch. Requires COMMIT api version >= 1.
+ optional int64 tiering_epoch = 6;
}
message PbLakeTableSnapshotInfo {
@@ -1284,6 +1291,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/coordinator/CoordinatorEventProcessor.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
index 18c4429aff..8d063d61f6 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
@@ -2273,6 +2273,11 @@ private void handleCommitLakeTableSnapshotV2(
throw new FlussRuntimeException(
"Lake snapshot metadata is null for table " + tableId);
}
+ // Fencing: reject a writer from a stale tiering assignment.
+ if (snapshot.getTieringEpoch() != null) {
+ lakeTableTieringManager.validateTieringEpoch(
+ tableId, snapshot.getTieringEpoch());
+ }
lakeTableHelper.registerLakeTableSnapshotV2(
tableId,
snapshot.getLakeSnapshotMetadata(),
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java
index e281381de5..355d6acecb 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/LakeTableTieringManager.java
@@ -511,6 +511,18 @@ public void renewTieringHeartbeat(long tableId, long tieringEpoch) {
});
}
+ /**
+ * Validates a lake commit's tiering epoch against the coordinator's current assignment epoch,
+ * fencing off a writer from a stale assignment. The epoch is in-memory (reset on restart), so
+ * this is an exact match; a mismatched writer must re-acquire the table via heartbeat first.
+ *
+ * @throws TableNotExistException if the table is not (or no longer) a lake table
+ * @throws FencedTieringEpochException if the epoch does not match the current epoch
+ */
+ public void validateTieringEpoch(long tableId, long tieringEpoch) {
+ inReadLock(lock, () -> validateTieringServiceRequest(tableId, tieringEpoch));
+ }
+
private void validateTieringServiceRequest(long tableId, long tieringEpoch) {
Long currentEpoch = tableTierEpoch.get(tableId);
// the table has been dropped, return false
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/entity/CommitLakeTableSnapshotsData.java b/fluss-server/src/main/java/org/apache/fluss/server/entity/CommitLakeTableSnapshotsData.java
index 9d88be9909..1d88552fd0 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/entity/CommitLakeTableSnapshotsData.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/entity/CommitLakeTableSnapshotsData.java
@@ -64,7 +64,8 @@ public void addTableSnapshot(
@Nullable LakeTableSnapshot lakeTableSnapshot,
@Nullable Map tableMaxTieredTimestamps,
@Nullable LakeTable.LakeSnapshotMetadata lakeSnapshotMetadata,
- @Nullable Long earliestSnapshotIDToKeep) {
+ @Nullable Long earliestSnapshotIDToKeep,
+ @Nullable Long tieringEpoch) {
snapshotMap.put(
tableId,
new CommitLakeTableSnapshot(
@@ -73,7 +74,8 @@ public void addTableSnapshot(
? tableMaxTieredTimestamps
: Collections.emptyMap(),
lakeSnapshotMetadata,
- earliestSnapshotIDToKeep));
+ earliestSnapshotIDToKeep,
+ tieringEpoch));
}
/**
@@ -149,15 +151,20 @@ public static class CommitLakeTableSnapshot {
// The earliest snapshot ID to keep for Paimon DV tables. Null for non-Paimon-DV tables.
@Nullable private final Long earliestSnapshotIDToKeep;
+ // Tiering assignment epoch for fencing; null when the committer did not send it (legacy).
+ @Nullable private final Long tieringEpoch;
+
public CommitLakeTableSnapshot(
@Nullable LakeTableSnapshot lakeTableSnapshot,
@Nullable Map tableMaxTieredTimestamps,
@Nullable LakeTable.LakeSnapshotMetadata lakeSnapshotMetadata,
- @Nullable Long earliestSnapshotIDToKeep) {
+ @Nullable Long earliestSnapshotIDToKeep,
+ @Nullable Long tieringEpoch) {
this.lakeTableSnapshot = lakeTableSnapshot;
this.tableMaxTieredTimestamps = tableMaxTieredTimestamps;
this.lakeSnapshotMetadata = lakeSnapshotMetadata;
this.earliestSnapshotIDToKeep = earliestSnapshotIDToKeep;
+ this.tieringEpoch = tieringEpoch;
}
@Nullable
@@ -174,5 +181,10 @@ public LakeTable.LakeSnapshotMetadata getLakeSnapshotMetadata() {
public Long getEarliestSnapshotIDToKeep() {
return earliestSnapshotIDToKeep;
}
+
+ @Nullable
+ public Long getTieringEpoch() {
+ return tieringEpoch;
+ }
}
}
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..c0d292dd17 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
@@ -1775,7 +1775,8 @@ public static CommitLakeTableSnapshotsData getCommitLakeTableSnapshotData(
entry.getValue(),
tableBucketsMaxTimestamp.get(tableId),
null, // no metadata for V1
- LakeCommitResult.KEEP_LATEST); // V1: keep only latest snapshot
+ LakeCommitResult.KEEP_LATEST, // V1: keep only latest snapshot
+ null); // V1: no tiering epoch
}
// Add V2 format snapshots (current)
@@ -1796,6 +1797,10 @@ public static CommitLakeTableSnapshotsData getCommitLakeTableSnapshotData(
pbLakeTableSnapshotMetadata.hasEarliestSnapshotIdToKeep()
? pbLakeTableSnapshotMetadata.getEarliestSnapshotIdToKeep()
: null;
+ Long tieringEpoch =
+ pbLakeTableSnapshotMetadata.hasTieringEpoch()
+ ? pbLakeTableSnapshotMetadata.getTieringEpoch()
+ : null;
// If this table already exists in builder (from V1), update it; otherwise add new
builder.addTableSnapshot(
@@ -1803,7 +1808,8 @@ public static CommitLakeTableSnapshotsData getCommitLakeTableSnapshotData(
lakeTableInfoByTableId.get(tableId), // may be null for V2-only
tableBucketsMaxTimestamp.get(tableId), // may be null
lakeSnapshotMetadata,
- earliestSnapshotIDToKeep);
+ earliestSnapshotIDToKeep,
+ tieringEpoch);
}
return builder.build();
@@ -1822,7 +1828,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 +1970,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..19500a2f06 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;
@@ -53,8 +51,9 @@
*/
public class LakeTable {
- // Version 2 (current):
- // a list of lake snapshot metadata, record the metadata for different lake snapshots
+ // Version 2 (current): an append-only log of lake snapshot metadata. A lake snapshot id may
+ // repeat across entries (a state-only round appends a new entry reusing it); entries are
+ // immutable once written and reads resolve to the latest matching one.
@Nullable private final List lakeSnapshotMetadatas;
// Version 1 (legacy): the full lake table snapshot info stored in ZK, will be null in version2
@@ -108,7 +107,9 @@ public LakeSnapshotMetadata getLatestLakeSnapshotMetadata() {
@Nullable
private LakeSnapshotMetadata getLakeSnapshotMetadata(long snapshotId) {
if (lakeSnapshotMetadatas != null) {
- for (LakeSnapshotMetadata lakeSnapshotMetadata : lakeSnapshotMetadatas) {
+ // Resolve to the latest (last) entry for the id: it may repeat across entries.
+ for (int i = lakeSnapshotMetadatas.size() - 1; i >= 0; i--) {
+ LakeSnapshotMetadata lakeSnapshotMetadata = lakeSnapshotMetadatas.get(i);
if (lakeSnapshotMetadata.snapshotId == snapshotId) {
return lakeSnapshotMetadata;
}
@@ -203,9 +204,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..a6c69f556c 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
@@ -63,12 +63,21 @@ public void registerLakeTableSnapshotV1(long tableId, LakeTableSnapshot lakeTabl
Optional optPreviousLakeTable = zkClient.getLakeTable(tableId);
// Merge with previous snapshot if exists
if (optPreviousLakeTable.isPresent()) {
- TableBucketOffsets tableBucketOffsets =
- mergeTableBucketOffsets(
- optPreviousLakeTable.get(),
- new TableBucketOffsets(
- tableId, lakeTableSnapshot.getBucketLogEndOffset()));
- lakeTableSnapshot = new LakeTableSnapshot(tableId, tableBucketOffsets.getOffsets());
+ LakeTableSnapshot previousSnapshot =
+ optPreviousLakeTable.get().getOrReadLatestTableSnapshot();
+ // The legacy (v1) format cannot carry tiering state. This path is not expected to run
+ // on a state-bearing table; if it does (e.g. an old committer), drop the state with a
+ // warning instead of failing the commit.
+ if (previousSnapshot.getTieringStateJson() != null) {
+ LOG.warn(
+ "Dropping tiering state for table {} on a legacy (v1) lake commit; "
+ + "the v1 format cannot store it.",
+ tableId);
+ }
+ Map bucketLogEndOffset =
+ new HashMap<>(previousSnapshot.getBucketLogEndOffset());
+ bucketLogEndOffset.putAll(lakeTableSnapshot.getBucketLogEndOffset());
+ lakeTableSnapshot = new LakeTableSnapshot(tableId, bucketLogEndOffset);
}
zkClient.upsertLakeTable(
tableId, new LakeTable(lakeTableSnapshot), optPreviousLakeTable.isPresent());
@@ -171,13 +180,20 @@ 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);
+
+ // The tiering state is opaque and cannot be field-merged; the committer always sends the
+ // complete state, so it is overwritten wholesale (absent means "no state"). Byte pass-
+ // through on the committer side keeps a newer, unreadable state from being dropped.
+ byte[] tieringStateJson = newTableBucketOffsets.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/main/java/org/apache/fluss/server/zk/data/lake/LakeTableSnapshotLegacyJsonSerde.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableSnapshotLegacyJsonSerde.java
index d1e5a52589..7d2700efb5 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableSnapshotLegacyJsonSerde.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableSnapshotLegacyJsonSerde.java
@@ -71,6 +71,12 @@ public class LakeTableSnapshotLegacyJsonSerde
@Override
public void serialize(LakeTableSnapshot lakeTableSnapshot, JsonGenerator generator)
throws IOException {
+ // Tiering state is a v2-only feature; the legacy v1 format cannot carry it (fail fast
+ // rather than silently drop it).
+ if (lakeTableSnapshot.getTieringStateJson() != null) {
+ throw new IllegalStateException(
+ "The legacy (v1) lake table format does not support tiering state.");
+ }
generator.writeStartObject();
generator.writeNumberField(VERSION_KEY, VERSION_1);
generator.writeNumberField(SNAPSHOT_ID, lakeTableSnapshot.getSnapshotId());
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/CommitLakeTableSnapshotITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/CommitLakeTableSnapshotITCase.java
index e5788f32b6..8bad9e6b8d 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/CommitLakeTableSnapshotITCase.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/CommitLakeTableSnapshotITCase.java
@@ -19,14 +19,23 @@
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.lake.committer.LakeTieringTableState;
import org.apache.fluss.metadata.DataLakeFormat;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.rpc.gateway.CoordinatorGateway;
import org.apache.fluss.rpc.gateway.TabletServerGateway;
import org.apache.fluss.rpc.messages.CommitLakeTableSnapshotRequest;
+import org.apache.fluss.rpc.messages.CommitLakeTableSnapshotResponse;
+import org.apache.fluss.rpc.messages.GetLakeSnapshotRequest;
+import org.apache.fluss.rpc.messages.GetLakeSnapshotResponse;
+import org.apache.fluss.rpc.messages.PbBucketOffset;
import org.apache.fluss.rpc.messages.PbLakeTableOffsetForBucket;
import org.apache.fluss.rpc.messages.PbLakeTableSnapshotInfo;
+import org.apache.fluss.rpc.messages.PbLakeTableSnapshotMetadata;
+import org.apache.fluss.rpc.messages.PbTableOffsets;
+import org.apache.fluss.rpc.messages.PrepareLakeTableSnapshotRequest;
+import org.apache.fluss.rpc.messages.PrepareLakeTableSnapshotResponse;
import org.apache.fluss.server.log.LogTablet;
import org.apache.fluss.server.testutils.FlussClusterExtension;
import org.apache.fluss.server.testutils.RpcMessageTestUtils;
@@ -38,6 +47,7 @@
import org.junit.jupiter.api.extension.RegisterExtension;
import java.time.Duration;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
@@ -136,6 +146,131 @@ private void checkLakeTableDataInZk(long tableId, LakeTableSnapshot expected) th
assertThat(lakeTableSnapshot).isEqualTo(expected);
}
+ /**
+ * Real-cluster RPC coverage: PREPARE embeds the state in the offsets file, COMMIT registers it
+ * (fencing a stale epoch), and GetLakeSnapshot reads it back. A state-only round reuses the
+ * previous lake snapshot id, and latest/by-id/readable resolve to the latest entry for that id.
+ */
+ @Test
+ void testCommitTieringStateAndReadBackViaRpc() throws Exception {
+ long tableId = createLogTable();
+ CoordinatorGateway coordinatorGateway = FLUSS_CLUSTER_EXTENSION.newCoordinatorClient();
+
+ Map offsets = new HashMap<>();
+ for (int bucket = 0; bucket < BUCKET_NUM; bucket++) {
+ offsets.put(new TableBucket(tableId, bucket), (bucket + 1) * 10L);
+ }
+
+ // Round 1 (normal round): PREPARE carries the state into the offsets file; COMMIT registers
+ // snapshot 1 with the current tiering epoch (0 for a freshly created lake table).
+ long snapshotId = 1L;
+ LakeTieringTableState state1 =
+ new LakeTieringTableState(true, Collections.singletonMap(1L, 1000L));
+ String path1 = prepareOffsetsFile(coordinatorGateway, tableId, offsets, state1);
+ assertThat(commit(coordinatorGateway, tableId, snapshotId, path1, 0L).hasErrorCode())
+ .isFalse();
+
+ // GetLakeSnapshot latest / by-id / readable all report state1.
+ assertThat(readState(coordinatorGateway, latestRequest())).isEqualTo(state1);
+ assertThat(readState(coordinatorGateway, byIdRequest(snapshotId))).isEqualTo(state1);
+ assertThat(readState(coordinatorGateway, readableRequest())).isEqualTo(state1);
+
+ // Fencing: a commit carrying a stale/wrong epoch is rejected per-table.
+ assertThat(commit(coordinatorGateway, tableId, snapshotId, path1, 999L).hasErrorCode())
+ .isTrue();
+
+ // Round 2 (state-only round): no new lake commit, reuse snapshot id 1, only the state
+ // advances. It appends another entry with the same id; reads resolve to the latest.
+ LakeTieringTableState state2 =
+ new LakeTieringTableState(true, Collections.singletonMap(1L, 2000L));
+ String path2 = prepareOffsetsFile(coordinatorGateway, tableId, offsets, state2);
+ assertThat(commit(coordinatorGateway, tableId, snapshotId, path2, 0L).hasErrorCode())
+ .isFalse();
+
+ GetLakeSnapshotResponse latest = coordinatorGateway.getLakeSnapshot(latestRequest()).get();
+ // the lake snapshot id is unchanged (state-only round reused it)
+ assertThat(latest.getSnapshotId()).isEqualTo(snapshotId);
+ // latest and by-id both resolve deterministically to the newest state for that id
+ assertThat(LakeTieringTableState.fromJsonBytes(latest.getTieringStateJson()))
+ .isEqualTo(state2);
+ assertThat(readState(coordinatorGateway, byIdRequest(snapshotId))).isEqualTo(state2);
+ }
+
+ private static String prepareOffsetsFile(
+ CoordinatorGateway gateway,
+ long tableId,
+ Map offsets,
+ LakeTieringTableState state)
+ throws Exception {
+ PrepareLakeTableSnapshotRequest request = new PrepareLakeTableSnapshotRequest();
+ PbTableOffsets pbTableOffsets = request.addBucketOffset();
+ pbTableOffsets.setTableId(tableId);
+ pbTableOffsets
+ .setTablePath()
+ .setDatabaseName(DATA1_TABLE_PATH.getDatabaseName())
+ .setTableName(DATA1_TABLE_PATH.getTableName());
+ pbTableOffsets.setTieringStateJson(state.toJsonBytes());
+ for (Map.Entry entry : offsets.entrySet()) {
+ PbBucketOffset pbBucketOffset = pbTableOffsets.addBucketOffset();
+ pbBucketOffset.setBucketId(entry.getKey().getBucket());
+ pbBucketOffset.setLogEndOffset(entry.getValue());
+ }
+ PrepareLakeTableSnapshotResponse response = gateway.prepareLakeTableSnapshot(request).get();
+ return response.getPrepareLakeTableRespsList().get(0).getLakeTableOffsetsPath();
+ }
+
+ private static org.apache.fluss.rpc.messages.PbCommitLakeTableSnapshotRespForTable commit(
+ CoordinatorGateway gateway,
+ long tableId,
+ long snapshotId,
+ String offsetsPath,
+ long tieringEpoch)
+ throws Exception {
+ CommitLakeTableSnapshotRequest request = new CommitLakeTableSnapshotRequest();
+ PbLakeTableSnapshotMetadata metadata = request.addLakeTableSnapshotMetadata();
+ metadata.setTableId(tableId);
+ metadata.setSnapshotId(snapshotId);
+ metadata.setTieredBucketOffsetsFilePath(offsetsPath);
+ // make it readable so getReadableLakeSnapshot returns it
+ metadata.setReadableBucketOffsetsFilePath(offsetsPath);
+ // keep all previous entries so a reused snapshot id yields multiple entries
+ metadata.setEarliestSnapshotIdToKeep(-1L);
+ metadata.setTieringEpoch(tieringEpoch);
+ CommitLakeTableSnapshotResponse response = gateway.commitLakeTableSnapshot(request).get();
+ return response.getTableRespsList().get(0);
+ }
+
+ private static LakeTieringTableState readState(
+ CoordinatorGateway gateway, GetLakeSnapshotRequest request) throws Exception {
+ GetLakeSnapshotResponse response = gateway.getLakeSnapshot(request).get();
+ assertThat(response.hasTieringStateJson()).isTrue();
+ return LakeTieringTableState.fromJsonBytes(response.getTieringStateJson());
+ }
+
+ private static GetLakeSnapshotRequest latestRequest() {
+ return newGetLakeSnapshotRequest();
+ }
+
+ private static GetLakeSnapshotRequest byIdRequest(long snapshotId) {
+ GetLakeSnapshotRequest request = newGetLakeSnapshotRequest();
+ request.setSnapshotId(snapshotId);
+ return request;
+ }
+
+ private static GetLakeSnapshotRequest readableRequest() {
+ GetLakeSnapshotRequest request = newGetLakeSnapshotRequest();
+ request.setReadable(true);
+ return request;
+ }
+
+ private static GetLakeSnapshotRequest newGetLakeSnapshotRequest() {
+ GetLakeSnapshotRequest request = new GetLakeSnapshotRequest();
+ request.setTablePath()
+ .setDatabaseName(DATA1_TABLE_PATH.getDatabaseName())
+ .setTableName(DATA1_TABLE_PATH.getTableName());
+ return request;
+ }
+
private static CommitLakeTableSnapshotRequest genCommitLakeTableSnapshotRequest(
long tableId, int buckets, long snapshotId, long logEndOffset, long maxTimestamp) {
CommitLakeTableSnapshotRequest commitLakeTableSnapshotRequest =
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..07af40e449
--- /dev/null
+++ b/fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTieringStateTest.java
@@ -0,0 +1,58 @@
+/*
+ * 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.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.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests that {@link ServerRpcMessageUtils} omits the tiering state when absent (PREPARE read yields
+ * {@code null}, GET fill leaves the field unset). The present-state passthrough is covered
+ * end-to-end by {@code CommitLakeTableSnapshotITCase}.
+ */
+class ServerRpcMessageUtilsTieringStateTest {
+
+ @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/LakeTableSnapshotLegacyJsonSerdeTest.java b/fluss-server/src/test/java/org/apache/fluss/server/zk/data/LakeTableSnapshotLegacyJsonSerdeTest.java
index 369a53062d..e7fca60b5d 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/zk/data/LakeTableSnapshotLegacyJsonSerdeTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/zk/data/LakeTableSnapshotLegacyJsonSerdeTest.java
@@ -30,6 +30,7 @@
import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test for {@link LakeTableSnapshotLegacyJsonSerde}. */
class LakeTableSnapshotLegacyJsonSerdeTest extends JsonSerdeTestBase {
@@ -104,4 +105,22 @@ void testBackwardCompatibility() {
LakeTableSnapshotLegacyJsonSerde.INSTANCE);
assertThat(snapshot3.getSnapshotId()).isEqualTo(3);
}
+
+ @Test
+ void testSerializeRejectsTieringState() {
+ // Tiering state is a version-2 only feature; the legacy v1 format must reject it rather
+ // than
+ // silently drop it.
+ Map offsets = new HashMap<>();
+ offsets.put(new TableBucket(1L, 0), 100L);
+ LakeTableSnapshot withState =
+ new LakeTableSnapshot(
+ 1L, offsets, "{\"version\":1}".getBytes(StandardCharsets.UTF_8));
+ assertThatThrownBy(
+ () ->
+ JsonSerdeUtils.writeValueAsBytes(
+ withState, LakeTableSnapshotLegacyJsonSerde.INSTANCE))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("does not support tiering state");
+ }
}
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..83434b38c8 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,126 @@ 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 the opaque tiering state uses pure overwrite semantics: a new request without
+ * tiering state drops it (the committer sends the complete state each round), while bucket
+ * offsets are still merged.
+ */
+ @Test
+ void testMergeDropsTieringStateWhenAbsent(@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, 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) -> dropped, offsets still merged.
+ TableBucketOffsets merged =
+ lakeTableHelper.mergeTableBucketOffsets(
+ previousLakeTable, new TableBucketOffsets(tableId, newOffsets, null));
+
+ assertThat(merged.getTieringStateJson()).isNull();
+ assertThat(merged.getOffsets()).containsEntry(new TableBucket(tableId, 1L, 0), 150L);
+ }
+
+ /**
+ * A legacy (v1) commit on a state-bearing table drops the state (with a warning), not fails.
+ */
+ @Test
+ void testRegisterV1DropsExistingTieringState(@TempDir Path tempDir) throws Exception {
+ LakeTableHelper lakeTableHelper = new LakeTableHelper(zookeeperClient, tempDir.toString());
+ long tableId = 102L;
+ TablePath tablePath = TablePath.of("test_db", "v1_drop_state_test");
+ zookeeperClient.registerTable(tablePath, createTableReg(tableId));
+
+ // previous v2 snapshot carrying tiering state
+ Map offsets = new HashMap<>();
+ offsets.put(new TableBucket(tableId, 0), 100L);
+ byte[] state =
+ new LakeTieringTableState(true, Collections.singletonMap(1L, 1000L)).toJsonBytes();
+ FsPath path =
+ lakeTableHelper.storeLakeTableOffsetsFile(
+ tablePath, new TableBucketOffsets(tableId, offsets, state));
+ lakeTableHelper.registerLakeTableSnapshotV2(
+ tableId, new LakeTable.LakeSnapshotMetadata(1L, path, path));
+
+ // a v1 commit drops the state without throwing
+ Map newOffsets = new HashMap<>();
+ newOffsets.put(new TableBucket(tableId, 0), 150L);
+ lakeTableHelper.registerLakeTableSnapshotV1(tableId, new LakeTableSnapshot(2L, newOffsets));
+
+ LakeTableSnapshot stored =
+ zookeeperClient.getLakeTable(tableId).get().getOrReadLatestTableSnapshot();
+ assertThat(stored.getTieringStateJson()).isNull();
+ assertThat(stored.getBucketLogEndOffset()).containsEntry(new TableBucket(tableId, 0), 150L);
+ }
+
/** Helper to store offset files and return the FsPath. */
private FsPath storeOffsetFile(
LakeTableHelper helper, TablePath path, long tableId, long offset) throws Exception {