From 5604c4bd9aff08166b145bf6e394b9320ab8dbc7 Mon Sep 17 00:00:00 2001 From: Junbo Wang Date: Fri, 31 Jul 2026 12:32:59 +0800 Subject: [PATCH] [lake] Carry keyed tiering-state entries in lake offsets file and RPC Introduce TieringStateEntry, a keyed and versioned envelope with an opaque JSON payload, carried by the lake offsets file alongside bucket offsets and transported via PREPARE/GET (repeated PbTieringStateEntry). Partition mark-done is the first state (key fluss.partition-mark-done): each tracked partition records its last update time and done time (-1 until first marked done), so an absent partition unambiguously means never tracked and no table-level flag is needed. Server merges states by key (upsert): entries not carried by a request are inherited from the previous snapshot, so a writer only touches the keys it owns and unrecognized or newer-version entries are passed through unchanged. The legacy v1 zk format rejects/drops states, and COMMIT keeps tiering-epoch fencing with deterministic latest-entry reads for reused snapshot ids. --- .../fluss/client/metadata/LakeSnapshot.java | 44 +++++ .../client/utils/ClientRpcMessageUtils.java | 15 +- .../client/metadata/LakeSnapshotTest.java | 96 ++++++++++ .../committer/PartitionMarkDoneState.java | 171 ++++++++++++++++++ .../lake/committer/TieringStateEntry.java | 101 +++++++++++ .../fluss/utils/json/JsonSerdeUtils.java | 8 + .../json/PartitionMarkDoneStateJsonSerde.java | 139 ++++++++++++++ .../fluss/utils/json/TableBucketOffsets.java | 46 ++++- .../json/TableBucketOffsetsJsonSerde.java | 72 +++++++- .../committer/PartitionMarkDoneStateTest.java | 149 +++++++++++++++ .../json/TableBucketOffsetsJsonSerdeTest.java | 90 ++++++++- .../apache/fluss/rpc/protocol/ApiKeys.java | 6 +- fluss-rpc/src/main/proto/FlussApi.proto | 17 ++ .../CoordinatorEventProcessor.java | 5 + .../coordinator/LakeTableTieringManager.java | 12 ++ .../entity/CommitLakeTableSnapshotsData.java | 18 +- .../server/utils/ServerRpcMessageUtils.java | 33 +++- .../fluss/server/zk/data/lake/LakeTable.java | 20 +- .../server/zk/data/lake/LakeTableHelper.java | 44 ++++- .../zk/data/lake/LakeTableSnapshot.java | 26 ++- .../LakeTableSnapshotLegacyJsonSerde.java | 6 + .../CommitLakeTableSnapshotITCase.java | 157 ++++++++++++++++ ...ServerRpcMessageUtilsTieringStateTest.java | 58 ++++++ .../LakeTableSnapshotLegacyJsonSerdeTest.java | 24 +++ .../zk/data/lake/LakeTableHelperTest.java | 146 +++++++++++++++ 25 files changed, 1465 insertions(+), 38 deletions(-) create mode 100644 fluss-client/src/test/java/org/apache/fluss/client/metadata/LakeSnapshotTest.java create mode 100644 fluss-common/src/main/java/org/apache/fluss/lake/committer/PartitionMarkDoneState.java create mode 100644 fluss-common/src/main/java/org/apache/fluss/lake/committer/TieringStateEntry.java create mode 100644 fluss-common/src/main/java/org/apache/fluss/utils/json/PartitionMarkDoneStateJsonSerde.java create mode 100644 fluss-common/src/test/java/org/apache/fluss/lake/committer/PartitionMarkDoneStateTest.java create mode 100644 fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTieringStateTest.java diff --git a/fluss-client/src/main/java/org/apache/fluss/client/metadata/LakeSnapshot.java b/fluss-client/src/main/java/org/apache/fluss/client/metadata/LakeSnapshot.java index 89c7864106..d929df75b6 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/metadata/LakeSnapshot.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/metadata/LakeSnapshot.java @@ -18,14 +18,22 @@ package org.apache.fluss.client.metadata; import org.apache.fluss.annotation.PublicEvolving; +import org.apache.fluss.lake.committer.PartitionMarkDoneState; +import org.apache.fluss.lake.committer.TieringStateEntry; import org.apache.fluss.metadata.TableBucket; +import javax.annotation.Nullable; + import java.util.Collections; +import java.util.List; import java.util.Map; /** * A class representing the lake snapshot information of a table. It contains: *
  • The snapshot id and the log offset for each bucket. + *
  • The keyed tiering-state entries, exposed raw via {@link #getTieringState(String)} or parsed + * via typed accessors like {@link #getPartitionMarkDoneState()}. Empty when talking to an old + * coordinator that does not report them. * * @since 0.3 */ @@ -37,9 +45,20 @@ public class LakeSnapshot { // the specific log offset of the snapshot private final Map tableBucketsOffset; + // the keyed tiering-state entries; payloads parsed lazily by the typed accessors. + private final List tieringStates; + public LakeSnapshot(long snapshotId, Map tableBucketsOffset) { + this(snapshotId, tableBucketsOffset, Collections.emptyList()); + } + + public LakeSnapshot( + long snapshotId, + Map tableBucketsOffset, + List tieringStates) { this.snapshotId = snapshotId; this.tableBucketsOffset = tableBucketsOffset; + this.tieringStates = tieringStates; } public long getSnapshotId() { @@ -50,6 +69,29 @@ public Map getTableBucketsOffset() { return Collections.unmodifiableMap(tableBucketsOffset); } + /** Returns the raw tiering-state entry for the given key, or {@code null} if absent. */ + @Nullable + public TieringStateEntry getTieringState(String stateKey) { + for (TieringStateEntry entry : tieringStates) { + if (entry.getStateKey().equals(stateKey)) { + return entry; + } + } + return null; + } + + /** + * Parses and returns the partition mark-done state, or {@code null} if absent. + * + * @throws IllegalArgumentException if the entry has an unsupported (newer) version or a corrupt + * payload; the raw entry stays available via {@link #getTieringState(String)} + */ + @Nullable + public PartitionMarkDoneState getPartitionMarkDoneState() { + TieringStateEntry entry = getTieringState(PartitionMarkDoneState.STATE_KEY); + return entry == null ? null : PartitionMarkDoneState.fromStateEntry(entry); + } + @Override public String toString() { return "LakeSnapshot{" @@ -57,6 +99,8 @@ public String toString() { + snapshotId + ", tableBucketsOffset=" + tableBucketsOffset + + ", tieringStates=" + + tieringStates + '}'; } } diff --git a/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java b/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java index 70f940c3cd..1290ff1113 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java @@ -41,6 +41,7 @@ import org.apache.fluss.fs.FsPath; import org.apache.fluss.fs.FsPathAndFileName; import org.apache.fluss.fs.token.ObtainedSecurityToken; +import org.apache.fluss.lake.committer.TieringStateEntry; import org.apache.fluss.metadata.AggFunction; import org.apache.fluss.metadata.DatabaseChange; import org.apache.fluss.metadata.DatabaseSummary; @@ -97,6 +98,7 @@ import org.apache.fluss.rpc.messages.PbRenameColumn; import org.apache.fluss.rpc.messages.PbTableBucket; import org.apache.fluss.rpc.messages.PbTableStatsReqForBucket; +import org.apache.fluss.rpc.messages.PbTieringStateEntry; import org.apache.fluss.rpc.messages.PrefixLookupRequest; import org.apache.fluss.rpc.messages.ProduceLogRequest; import org.apache.fluss.rpc.messages.PutKvRequest; @@ -286,7 +288,18 @@ public static LakeSnapshot toLakeTableSnapshotInfo(GetLakeSnapshotResponse respo new TableBucket(tableId, partitionId, pbLakeSnapshotForBucket.getBucketId()); tableBucketsOffset.put(tableBucket, pbLakeSnapshotForBucket.getLogOffset()); } - return new LakeSnapshot(snapshotId, tableBucketsOffset); + + // pass through the tiering-state entries unparsed (typed parsing happens lazily in + // LakeSnapshot). + List tieringStates = new ArrayList<>(); + for (PbTieringStateEntry pbEntry : response.getTieringStatesList()) { + tieringStates.add( + new TieringStateEntry( + pbEntry.getStateKey(), + pbEntry.getStateVersion(), + pbEntry.getPayload())); + } + return new LakeSnapshot(snapshotId, tableBucketsOffset, tieringStates); } public static List toFsPathAndFileName( diff --git a/fluss-client/src/test/java/org/apache/fluss/client/metadata/LakeSnapshotTest.java b/fluss-client/src/test/java/org/apache/fluss/client/metadata/LakeSnapshotTest.java new file mode 100644 index 0000000000..1b2d13627a --- /dev/null +++ b/fluss-client/src/test/java/org/apache/fluss/client/metadata/LakeSnapshotTest.java @@ -0,0 +1,96 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.client.metadata; + +import org.apache.fluss.lake.committer.PartitionMarkDoneState; +import org.apache.fluss.lake.committer.PartitionMarkDoneState.PartitionState; +import org.apache.fluss.lake.committer.TieringStateEntry; +import org.apache.fluss.metadata.TableBucket; + +import org.junit.jupiter.api.Test; + +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Collections; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Test for {@link LakeSnapshot}, in particular the keyed tiering-state accessors. */ +class LakeSnapshotTest { + + @Test + void testKeyedStateAccess() { + // absent -> null (raw and typed). + LakeSnapshot absent = new LakeSnapshot(1L, Collections.emptyMap()); + assertThat(absent.getTieringState(PartitionMarkDoneState.STATE_KEY)).isNull(); + assertThat(absent.getPartitionMarkDoneState()).isNull(); + + // present -> raw entry exposed, typed accessor parses lazily; unrelated keys untouched. + PartitionMarkDoneState state = + new PartitionMarkDoneState( + Collections.singletonMap( + 5L, new PartitionState(1000L, PartitionMarkDoneState.NOT_DONE))); + TieringStateEntry otherEntry = + new TieringStateEntry("other-key", 3, "{}".getBytes(StandardCharsets.UTF_8)); + LakeSnapshot present = + new LakeSnapshot( + 1L, + Collections.emptyMap(), + Arrays.asList(otherEntry, state.toStateEntry())); + assertThat(present.getPartitionMarkDoneState()).isEqualTo(state); + assertThat(present.getTieringState("other-key")).isEqualTo(otherEntry); + assertThat(present.getTieringState("unknown-key")).isNull(); + } + + @Test + void testNewerVersionExposesRawEntryForPassthrough() { + TieringStateEntry newerEntry = + new TieringStateEntry( + PartitionMarkDoneState.STATE_KEY, + PartitionMarkDoneState.CURRENT_VERSION + 1, + "{}".getBytes(StandardCharsets.UTF_8)); + LakeSnapshot snapshot = + new LakeSnapshot(1L, Collections.emptyMap(), Collections.singletonList(newerEntry)); + // unreadable here: parsing fails so the caller passes the raw entry through unchanged. + assertThatThrownBy(snapshot::getPartitionMarkDoneState) + .isInstanceOf(IllegalArgumentException.class); + assertThat(snapshot.getTieringState(PartitionMarkDoneState.STATE_KEY)) + .isEqualTo(newerEntry); + } + + @Test + void testCorruptStateDoesNotBlockBucketOffsets() { + TableBucket bucket = new TableBucket(1L, 0); + Map offsets = Collections.singletonMap(bucket, 100L); + TieringStateEntry corruptEntry = + new TieringStateEntry( + PartitionMarkDoneState.STATE_KEY, + 1, + "{not-json".getBytes(StandardCharsets.UTF_8)); + LakeSnapshot snapshot = + new LakeSnapshot(1L, offsets, Collections.singletonList(corruptEntry)); + + // bucket offsets remain accessible even though the state payload is corrupt. + assertThat(snapshot.getTableBucketsOffset()).containsEntry(bucket, 100L); + // only the typed accessor surfaces the parse failure. + assertThatThrownBy(snapshot::getPartitionMarkDoneState) + .isInstanceOf(RuntimeException.class); + } +} diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/committer/PartitionMarkDoneState.java b/fluss-common/src/main/java/org/apache/fluss/lake/committer/PartitionMarkDoneState.java new file mode 100644 index 0000000000..3760e7726c --- /dev/null +++ b/fluss-common/src/main/java/org/apache/fluss/lake/committer/PartitionMarkDoneState.java @@ -0,0 +1,171 @@ +/* + * 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.PartitionMarkDoneStateJsonSerde; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; + +import static org.apache.fluss.utils.Preconditions.checkArgument; + +/** + * The partition mark-done state, persisted as the {@link TieringStateEntry} keyed by {@link + * #STATE_KEY} in the lake offsets file. Each tracked partition records its last update time and + * done time ({@link #NOT_DONE} until first marked done); a partition absent from the map has never + * been tracked. Entries of dropped partitions are removed by omitting them from the next payload. + * + *

    The payload schema version lives in the entry envelope. Evolve the payload by bumping {@link + * #CURRENT_VERSION}; {@link #fromStateEntry} rejects a higher version, and the caller must pass the + * entry through unchanged so a newer build's state is never dropped. + * + * @since 0.9 + */ +@PublicEvolving +public class PartitionMarkDoneState { + + /** The tiering-state key of the partition mark-done state. */ + public static final String STATE_KEY = "fluss.partition-mark-done"; + + public static final int CURRENT_VERSION = 1; + + /** Sentinel done time of a partition that has not been marked done yet. */ + public static final long NOT_DONE = -1L; + + private final Map partitionStates; + + public PartitionMarkDoneState(Map partitionStates) { + this.partitionStates = + partitionStates == null ? Collections.emptyMap() : new HashMap<>(partitionStates); + } + + public Map getPartitionStates() { + return Collections.unmodifiableMap(partitionStates); + } + + /** Wraps this state into its {@link TieringStateEntry} at {@link #CURRENT_VERSION}. */ + public TieringStateEntry toStateEntry() { + return new TieringStateEntry( + STATE_KEY, + CURRENT_VERSION, + JsonSerdeUtils.writeValueAsBytes(this, PartitionMarkDoneStateJsonSerde.INSTANCE)); + } + + /** + * Parses the state from its {@link TieringStateEntry}. + * + * @throws IllegalArgumentException if the entry has a different key, an unsupported (newer) + * version, or a corrupt payload + */ + public static PartitionMarkDoneState fromStateEntry(TieringStateEntry entry) { + checkArgument( + STATE_KEY.equals(entry.getStateKey()), + "Expected state key %s but got %s.", + STATE_KEY, + entry.getStateKey()); + if (entry.getStateVersion() > CURRENT_VERSION) { + throw new IllegalArgumentException( + "Unsupported partition mark-done state version " + + entry.getStateVersion() + + " > " + + CURRENT_VERSION + + "; pass the entry through unchanged."); + } + return JsonSerdeUtils.readValue( + entry.getPayload(), PartitionMarkDoneStateJsonSerde.INSTANCE); + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + PartitionMarkDoneState that = (PartitionMarkDoneState) o; + return Objects.equals(partitionStates, that.partitionStates); + } + + @Override + public int hashCode() { + return Objects.hash(partitionStates); + } + + @Override + public String toString() { + return "PartitionMarkDoneState{" + "partitionStates=" + partitionStates + '}'; + } + + /** The mark-done state of a single partition. */ + public static class PartitionState { + + private final long updateTime; + private final long doneTime; + + public PartitionState(long updateTime, long doneTime) { + checkArgument( + updateTime >= 0, "updateTime must be non-negative but got %s.", updateTime); + checkArgument( + doneTime == NOT_DONE || doneTime > 0, + "doneTime must be %s (not done) or positive but got %s.", + NOT_DONE, + doneTime); + this.updateTime = updateTime; + this.doneTime = doneTime; + } + + public long getUpdateTime() { + return updateTime; + } + + public long getDoneTime() { + return doneTime; + } + + public boolean isDone() { + return doneTime != NOT_DONE; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + PartitionState that = (PartitionState) o; + return updateTime == that.updateTime && doneTime == that.doneTime; + } + + @Override + public int hashCode() { + return Objects.hash(updateTime, doneTime); + } + + @Override + public String toString() { + return "PartitionState{" + "updateTime=" + updateTime + ", doneTime=" + doneTime + '}'; + } + } +} diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/committer/TieringStateEntry.java b/fluss-common/src/main/java/org/apache/fluss/lake/committer/TieringStateEntry.java new file mode 100644 index 0000000000..69558606d0 --- /dev/null +++ b/fluss-common/src/main/java/org/apache/fluss/lake/committer/TieringStateEntry.java @@ -0,0 +1,101 @@ +/* + * 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.Arrays; +import java.util.Objects; + +import static org.apache.fluss.utils.Preconditions.checkArgument; +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** + * A keyed, versioned tiering-state entry carried by the lake offsets file alongside the bucket + * offsets. + * + *

    The entry is an opaque envelope: {@code stateKey} identifies the state owner (e.g. {@link + * PartitionMarkDoneState#STATE_KEY}), {@code stateVersion} is the schema version of the payload, + * and {@code payload} is a JSON object serialized as bytes. The transport and storage layers never + * interpret the payload; the offsets-file serde may re-encode it as a JSON tree (all fields + * preserved, byte-level layout not guaranteed). Entries with an unrecognized key or a + * higher-than-supported version must be passed through unchanged so state written by a newer build + * is never dropped. + * + * @since 0.9 + */ +@PublicEvolving +public class TieringStateEntry { + + private final String stateKey; + private final int stateVersion; + private final byte[] payload; + + public TieringStateEntry(String stateKey, int stateVersion, byte[] payload) { + checkArgument( + stateKey != null && !stateKey.isEmpty(), "stateKey must be a non-empty string."); + checkArgument(stateVersion > 0, "stateVersion must be positive but got %s.", stateVersion); + this.stateKey = stateKey; + this.stateVersion = stateVersion; + this.payload = checkNotNull(payload, "payload must not be null."); + } + + public String getStateKey() { + return stateKey; + } + + public int getStateVersion() { + return stateVersion; + } + + public byte[] getPayload() { + return payload; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + TieringStateEntry that = (TieringStateEntry) o; + return stateVersion == that.stateVersion + && stateKey.equals(that.stateKey) + && Arrays.equals(payload, that.payload); + } + + @Override + public int hashCode() { + return Objects.hash(stateKey, stateVersion, Arrays.hashCode(payload)); + } + + @Override + public String toString() { + return "TieringStateEntry{" + + "stateKey='" + + stateKey + + '\'' + + ", stateVersion=" + + stateVersion + + ", payload=" + + Arrays.toString(payload) + + '}'; + } +} 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/PartitionMarkDoneStateJsonSerde.java b/fluss-common/src/main/java/org/apache/fluss/utils/json/PartitionMarkDoneStateJsonSerde.java new file mode 100644 index 0000000000..19c7475674 --- /dev/null +++ b/fluss-common/src/main/java/org/apache/fluss/utils/json/PartitionMarkDoneStateJsonSerde.java @@ -0,0 +1,139 @@ +/* + * 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.PartitionMarkDoneState; +import org.apache.fluss.lake.committer.PartitionMarkDoneState.PartitionState; +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; +import java.util.TreeMap; + +/** + * Json serializer and deserializer for the {@link PartitionMarkDoneState} payload, e.g. {@code + * {"partitions":{"5":{"update_time":1704153550000,"done_time":-1}}}}. Partitions are written in + * ascending id order for deterministic output. + * + *

    The payload schema version lives in the {@code TieringStateEntry} envelope, not here. The + * serde validates the JSON structure and field types; value ranges are enforced by the {@link + * PartitionState} constructor. Unknown fields are tolerated. + */ +public class PartitionMarkDoneStateJsonSerde + implements JsonSerializer, + JsonDeserializer { + + public static final PartitionMarkDoneStateJsonSerde INSTANCE = + new PartitionMarkDoneStateJsonSerde(); + + private static final String PARTITIONS_KEY = "partitions"; + private static final String UPDATE_TIME_KEY = "update_time"; + private static final String DONE_TIME_KEY = "done_time"; + + @Override + public void serialize(PartitionMarkDoneState state, JsonGenerator generator) + throws IOException { + generator.writeStartObject(); + generator.writeObjectFieldStart(PARTITIONS_KEY); + for (Map.Entry entry : + new TreeMap<>(state.getPartitionStates()).entrySet()) { + generator.writeObjectFieldStart(String.valueOf(entry.getKey())); + generator.writeNumberField(UPDATE_TIME_KEY, entry.getValue().getUpdateTime()); + generator.writeNumberField(DONE_TIME_KEY, entry.getValue().getDoneTime()); + generator.writeEndObject(); + } + generator.writeEndObject(); + generator.writeEndObject(); + } + + @Override + public PartitionMarkDoneState deserialize(JsonNode node) { + if (node == null || !node.isObject()) { + throw new IllegalArgumentException( + "Corrupt partition mark-done state: expected a JSON object but got " + node); + } + + Map partitionStates = new HashMap<>(); + JsonNode partitionsNode = node.get(PARTITIONS_KEY); + if (partitionsNode != null && !partitionsNode.isNull()) { + if (!partitionsNode.isObject()) { + throw new IllegalArgumentException( + "Corrupt partition mark-done state: '" + + PARTITIONS_KEY + + "' must be a JSON object but got " + + partitionsNode); + } + Iterator> fields = partitionsNode.fields(); + while (fields.hasNext()) { + Map.Entry field = fields.next(); + long partitionId; + try { + partitionId = Long.parseLong(field.getKey()); + } catch (NumberFormatException e) { + throw new IllegalArgumentException( + "Corrupt partition mark-done state: invalid partitionId key '" + + field.getKey() + + "' in " + + PARTITIONS_KEY); + } + if (partitionId <= 0) { + throw new IllegalArgumentException( + "Corrupt partition mark-done state: partitionId must be positive but got " + + partitionId); + } + partitionStates.put( + partitionId, deserializePartitionState(partitionId, field.getValue())); + } + } + + return new PartitionMarkDoneState(partitionStates); + } + + private static PartitionState deserializePartitionState(long partitionId, JsonNode node) { + JsonNode updateTimeNode = node == null ? null : node.get(UPDATE_TIME_KEY); + JsonNode doneTimeNode = node == null ? null : node.get(DONE_TIME_KEY); + if (node == null + || !node.isObject() + || updateTimeNode == null + || !updateTimeNode.canConvertToLong() + || doneTimeNode == null + || !doneTimeNode.canConvertToLong()) { + throw new IllegalArgumentException( + "Corrupt partition mark-done state: partition " + + partitionId + + " must be an object with numeric '" + + UPDATE_TIME_KEY + + "' and '" + + DONE_TIME_KEY + + "' but got " + + node); + } + try { + return new PartitionState(updateTimeNode.asLong(), doneTimeNode.asLong()); + } catch (IllegalArgumentException e) { + throw new IllegalArgumentException( + "Corrupt partition mark-done state: partition " + + partitionId + + ": " + + e.getMessage()); + } + } +} 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..32c1e0004f 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 @@ -18,19 +18,25 @@ package org.apache.fluss.utils.json; +import org.apache.fluss.lake.committer.TieringStateEntry; import org.apache.fluss.metadata.TableBucket; +import java.util.Collections; +import java.util.List; 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 the keyed tiering-state entries. * *

    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 the table-level {@link TieringStateEntry}s (e.g. the partition mark-done + * state), passed through without parsing the payloads. + * *

    The offsets map contains entries for each bucket that has a valid offset. Missing buckets are * not included in the map. * @@ -47,15 +53,31 @@ public class TableBucketOffsets { */ private final Map offsets; + /** The keyed tiering-state entries, passed through by the serde without parsing payloads. */ + private final List tieringStates; + /** - * Creates a new {@link TableBucketOffsets} instance. + * Creates a new {@link TableBucketOffsets} instance without tiering states. * * @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, Collections.emptyList()); + } + + /** + * Creates a new {@link TableBucketOffsets} instance with tiering-state entries. + * + * @param tableId the table ID that all buckets belong to + * @param offsets the mapping from {@link TableBucket} to their offsets + * @param tieringStates the keyed tiering-state entries (passed through) + */ + public TableBucketOffsets( + long tableId, Map offsets, List tieringStates) { this.tableId = tableId; this.offsets = offsets; + this.tieringStates = tieringStates; } /** @@ -76,6 +98,11 @@ public Map getOffsets() { return offsets; } + /** Returns the keyed tiering-state entries; empty when absent. */ + public List getTieringStates() { + return tieringStates; + } + /** * Serialize to a JSON byte array. * @@ -103,16 +130,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) + && Objects.equals(tieringStates, that.tieringStates); } @Override public int hashCode() { - return Objects.hash(tableId, offsets); + return Objects.hash(tableId, offsets, tieringStates); } @Override public String toString() { - return "TableBucketOffsets{" + "tableId=" + tableId + ", offsets=" + offsets + '}'; + return "TableBucketOffsets{" + + "tableId=" + + tableId + + ", offsets=" + + offsets + + ", tieringStates=" + + tieringStates + + '}'; } } 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..d1d7d436c7 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 @@ -18,12 +18,15 @@ package org.apache.fluss.utils.json; +import org.apache.fluss.lake.committer.TieringStateEntry; import org.apache.fluss.metadata.TableBucket; 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.nio.charset.StandardCharsets; import java.util.ArrayList; +import java.util.Collections; import java.util.Comparator; import java.util.HashMap; import java.util.List; @@ -58,6 +61,8 @@ *

  • "bucket_offsets": array of offsets for non-partitioned table buckets (optional) *
  • "partition_offsets": array of partition offset objects for partitioned table buckets * (optional) + *
  • "tiering_states": array of keyed tiering-state entries, each {@code {"key", "version", + * "payload"}} with the payload embedded as a JSON object but not parsed (optional) * */ public class TableBucketOffsetsJsonSerde @@ -70,8 +75,13 @@ 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_STATES_KEY = "tiering_states"; + private static final String STATE_ENTRY_KEY = "key"; + private static final String STATE_ENTRY_VERSION = "version"; + private static final String STATE_ENTRY_PAYLOAD = "payload"; - 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 +104,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 +159,38 @@ public void serialize(TableBucketOffsets tableBucketOffsets, JsonGenerator gener } } + // embed the tiering-state entries (optional); each payload must be a JSON object, its + // content is not parsed. + List tieringStates = tableBucketOffsets.getTieringStates(); + if (!tieringStates.isEmpty()) { + generator.writeArrayFieldStart(TIERING_STATES_KEY); + for (TieringStateEntry entry : tieringStates) { + JsonNode payloadNode; + try { + payloadNode = JsonSerdeUtils.readTree(entry.getPayload()); + } catch (IOException e) { + throw new IllegalArgumentException( + "tiering-state payload of key '" + + entry.getStateKey() + + "' is not valid JSON", + e); + } + if (payloadNode == null || !payloadNode.isObject()) { + throw new IllegalArgumentException( + "tiering-state payload of key '" + + entry.getStateKey() + + "' must be a JSON object"); + } + generator.writeStartObject(); + generator.writeStringField(STATE_ENTRY_KEY, entry.getStateKey()); + generator.writeNumberField(STATE_ENTRY_VERSION, entry.getStateVersion()); + generator.writeFieldName(STATE_ENTRY_PAYLOAD); + generator.writeTree(payloadNode); + generator.writeEndObject(); + } + generator.writeEndArray(); + } + generator.writeEndObject(); } @@ -165,7 +207,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 +253,29 @@ public TableBucketOffsets deserialize(JsonNode node) { } } - return new TableBucketOffsets(tableId, offsets); + // read the tiering-state entries (optional); payloads are kept as raw JSON bytes. + List tieringStates = Collections.emptyList(); + JsonNode tieringStatesNode = node.get(TIERING_STATES_KEY); + if (tieringStatesNode != null && !tieringStatesNode.isNull()) { + tieringStates = new ArrayList<>(); + for (JsonNode entryNode : tieringStatesNode) { + JsonNode keyNode = entryNode.get(STATE_ENTRY_KEY); + JsonNode versionNode = entryNode.get(STATE_ENTRY_VERSION); + JsonNode payloadNode = entryNode.get(STATE_ENTRY_PAYLOAD); + if (keyNode == null || versionNode == null || payloadNode == null) { + throw new IllegalArgumentException( + "Corrupt tiering-state entry: missing key/version/payload in " + + entryNode); + } + tieringStates.add( + new TieringStateEntry( + keyNode.asText(), + versionNode.asInt(), + payloadNode.toString().getBytes(StandardCharsets.UTF_8))); + } + } + + return new TableBucketOffsets(tableId, offsets, tieringStates); } private void serializeBucketLogEndOffset( diff --git a/fluss-common/src/test/java/org/apache/fluss/lake/committer/PartitionMarkDoneStateTest.java b/fluss-common/src/test/java/org/apache/fluss/lake/committer/PartitionMarkDoneStateTest.java new file mode 100644 index 0000000000..545deb2d03 --- /dev/null +++ b/fluss-common/src/test/java/org/apache/fluss/lake/committer/PartitionMarkDoneStateTest.java @@ -0,0 +1,149 @@ +/* + * 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.lake.committer.PartitionMarkDoneState.PartitionState; + +import org.junit.jupiter.api.Test; + +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.Map; + +import static org.apache.fluss.lake.committer.PartitionMarkDoneState.NOT_DONE; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Test for {@link PartitionMarkDoneState} and its {@link TieringStateEntry} conversion. */ +class PartitionMarkDoneStateTest { + + @Test + void testStateEntryRoundTrip() { + Map partitionStates = new HashMap<>(); + partitionStates.put(5L, new PartitionState(1704153550000L, NOT_DONE)); + partitionStates.put(7L, new PartitionState(1704153560000L, 1704153570000L)); + PartitionMarkDoneState state = new PartitionMarkDoneState(partitionStates); + + TieringStateEntry entry = state.toStateEntry(); + assertThat(entry.getStateKey()).isEqualTo(PartitionMarkDoneState.STATE_KEY); + assertThat(entry.getStateVersion()).isEqualTo(PartitionMarkDoneState.CURRENT_VERSION); + assertThat(PartitionMarkDoneState.fromStateEntry(entry)).isEqualTo(state); + + // empty state round-trips too (null map normalizes to empty). + PartitionMarkDoneState empty = new PartitionMarkDoneState(null); + assertThat(PartitionMarkDoneState.fromStateEntry(empty.toStateEntry())).isEqualTo(empty); + } + + @Test + void testPartitionStateValidation() { + assertThat(new PartitionState(1000L, NOT_DONE).isDone()).isFalse(); + assertThat(new PartitionState(1000L, 2000L).isDone()).isTrue(); + assertThatThrownBy(() -> new PartitionState(-5L, NOT_DONE)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("updateTime"); + // doneTime 0 and negatives other than the NOT_DONE sentinel are invalid. + assertThatThrownBy(() -> new PartitionState(1000L, 0L)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("doneTime"); + assertThatThrownBy(() -> new PartitionState(1000L, -2L)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("doneTime"); + } + + @Test + void testFromStateEntryRejectsWrongKey() { + TieringStateEntry entry = + new TieringStateEntry("other-key", 1, "{}".getBytes(StandardCharsets.UTF_8)); + assertThatThrownBy(() -> PartitionMarkDoneState.fromStateEntry(entry)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Expected state key"); + } + + @Test + void testFromStateEntryRejectsNewerVersion() { + // a newer build's payload cannot be interpreted; the entry must be passed through. + TieringStateEntry entry = + new TieringStateEntry( + PartitionMarkDoneState.STATE_KEY, + PartitionMarkDoneState.CURRENT_VERSION + 1, + "{}".getBytes(StandardCharsets.UTF_8)); + assertThatThrownBy(() -> PartitionMarkDoneState.fromStateEntry(entry)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("pass the entry through unchanged"); + } + + @Test + void testFromStateEntryRejectsCorruptPayload() { + assertThatThrownBy(() -> PartitionMarkDoneState.fromStateEntry(markDoneEntry("[1,2]"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("expected a JSON object"); + assertThatThrownBy( + () -> + PartitionMarkDoneState.fromStateEntry( + markDoneEntry("{\"partitions\":{\"abc\":{}}}"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("invalid partitionId"); + // record missing done_time + assertThatThrownBy( + () -> + PartitionMarkDoneState.fromStateEntry( + markDoneEntry( + "{\"partitions\":{\"5\":{\"update_time\":1}}}"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("done_time"); + // invalid done_time value, rejected by the PartitionState constructor with context + assertThatThrownBy( + () -> + PartitionMarkDoneState.fromStateEntry( + markDoneEntry( + "{\"partitions\":{\"5\":{\"update_time\":1,\"done_time\":0}}}"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("partition 5"); + } + + @Test + void testPayloadTolerantToUnknownFields() { + // unknown fields (added by a compatible newer build) are ignored at both levels. + PartitionMarkDoneState state = + PartitionMarkDoneState.fromStateEntry( + markDoneEntry( + "{\"partitions\":{\"5\":{\"update_time\":1000,\"done_time\":-1," + + "\"max_timestamp\":9}},\"future_field\":123}")); + assertThat(state.getPartitionStates()) + .containsOnlyKeys(5L) + .containsEntry(5L, new PartitionState(1000L, NOT_DONE)); + } + + @Test + void testEntryEnvelopeValidation() { + byte[] payload = "{}".getBytes(StandardCharsets.UTF_8); + assertThatThrownBy(() -> new TieringStateEntry("", 1, payload)) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> new TieringStateEntry("key", 0, payload)) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy(() -> new TieringStateEntry("key", 1, null)) + .isInstanceOf(NullPointerException.class); + } + + private static TieringStateEntry markDoneEntry(String payloadJson) { + return new TieringStateEntry( + PartitionMarkDoneState.STATE_KEY, + PartitionMarkDoneState.CURRENT_VERSION, + payloadJson.getBytes(StandardCharsets.UTF_8)); + } +} 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..23acb53248 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,20 @@ package org.apache.fluss.utils.json; +import org.apache.fluss.lake.committer.PartitionMarkDoneState; +import org.apache.fluss.lake.committer.PartitionMarkDoneState.PartitionState; +import org.apache.fluss.lake.committer.TieringStateEntry; 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 +82,32 @@ protected TableBucketOffsets[] createObjects() { TableBucketOffsets tableBucketOffsets5 = new TableBucketOffsets(tableId, bucketLogEndOffset); + // Test case 6: Partition table with a tiering-state entry + 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); + TableBucketOffsets tableBucketOffsets6 = + new TableBucketOffsets( + tableId, + bucketLogEndOffset, + Collections.singletonList( + new PartitionMarkDoneState( + Collections.singletonMap( + 5L, + new PartitionState( + 1704153550000L, + PartitionMarkDoneState.NOT_DONE))) + .toStateEntry())); + return new TableBucketOffsets[] { tableBucketOffsets1, tableBucketOffsets2, tableBucketOffsets3, tableBucketOffsets4, tableBucketOffsets5, + tableBucketOffsets6, }; } @@ -102,7 +130,67 @@ 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 a tiering-state entry + "{\"version\":1,\"table_id\":8,\"partition_offsets\":[{\"partition_id\":4,\"bucket_offsets\":[420]},{\"partition_id\":5,\"bucket_offsets\":[100,230]}],\"tiering_states\":[{\"key\":\"fluss.partition-mark-done\",\"version\":1,\"payload\":{\"partitions\":{\"5\":{\"update_time\":1704153550000,\"done_time\":-1}}}}]}" }; } + + /** + * 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 states, exact JSON and round-trip) is covered by {@link + * #createObjects()} / {@link #expectedJsons()}. + */ + @Test + void testSerializeRejectsBadTieringStatePayload() { + Map offsets = new HashMap<>(); + offsets.put(new TableBucket(1L, 1L, 0), 100L); + + assertThatThrownBy( + () -> + new TableBucketOffsets( + 1L, + offsets, + Collections.singletonList( + new TieringStateEntry( + "some-key", + 1, + "{not-json" + .getBytes( + StandardCharsets + .UTF_8)))) + .toJsonBytes()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("not valid JSON"); + + assertThatThrownBy( + () -> + new TableBucketOffsets( + 1L, + offsets, + Collections.singletonList( + new TieringStateEntry( + "some-key", + 1, + "[1,2,3]" + .getBytes( + StandardCharsets + .UTF_8)))) + .toJsonBytes()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("must be a JSON object"); + } + + /** A tiering-state entry missing key/version/payload fails as a controlled parse error. */ + @Test + void testDeserializeRejectsIncompleteTieringStateEntry() { + String json = "{\"version\":1,\"table_id\":1,\"tiering_states\":[{\"key\":\"some-key\"}]}"; + assertThatThrownBy( + () -> + TableBucketOffsets.fromJsonBytes( + json.getBytes(StandardCharsets.UTF_8))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("missing key/version/payload"); + } } 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 e9f18b3d67..b8a43931b4 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 @@ -70,7 +70,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), @@ -97,7 +98,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 keyed tiering-state entries (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 2e6104b4cf..bb12b70180 100644 --- a/fluss-rpc/src/main/proto/FlussApi.proto +++ b/fluss-rpc/src/main/proto/FlussApi.proto @@ -477,6 +477,8 @@ message GetLakeSnapshotResponse { required int64 table_id = 1; required int64 snapshotId = 2; repeated PbLakeSnapshotForBucket bucket_snapshots = 3; + // The keyed tiering-state entries (e.g. partition mark-done state), passed through unparsed. + repeated PbTieringStateEntry tiering_states = 4; } message GetFileSystemSecurityTokenRequest { @@ -1250,6 +1252,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; @@ -1257,6 +1261,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 { @@ -1289,6 +1296,16 @@ message PbTableOffsets { required int64 table_id = 1; required PbTablePath table_path = 2; repeated PbBucketOffset bucket_offsets = 3; + // The keyed tiering-state entries (e.g. partition mark-done state), passed through unparsed. + repeated PbTieringStateEntry tiering_states = 4; +} + +// A keyed, versioned tiering-state entry carried by the lake offsets file. The payload is an +// opaque JSON object owned by the state key (see TieringStateEntry); servers never parse it. +message PbTieringStateEntry { + required string state_key = 1; + required int32 state_version = 2; + required bytes payload = 3; } 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 c5b90f5904..60a0aa5cf8 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 @@ -2300,6 +2300,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 9498c45ecd..31d62e7577 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 @@ -516,6 +516,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 e96be16c17..58e6a554ff 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 @@ -31,6 +31,7 @@ import org.apache.fluss.fs.FsPath; import org.apache.fluss.fs.token.ObtainedSecurityToken; import org.apache.fluss.lake.committer.LakeCommitResult; +import org.apache.fluss.lake.committer.TieringStateEntry; import org.apache.fluss.metadata.AggFunction; import org.apache.fluss.metadata.AggFunctionType; import org.apache.fluss.metadata.AggFunctions; @@ -159,6 +160,7 @@ import org.apache.fluss.rpc.messages.PbTablePath; import org.apache.fluss.rpc.messages.PbTableStatsReqForBucket; import org.apache.fluss.rpc.messages.PbTableStatsRespForBucket; +import org.apache.fluss.rpc.messages.PbTieringStateEntry; import org.apache.fluss.rpc.messages.PbValue; import org.apache.fluss.rpc.messages.PbValueList; import org.apache.fluss.rpc.messages.PrefixLookupRequest; @@ -1781,7 +1783,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) @@ -1802,6 +1805,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( @@ -1809,7 +1816,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(); @@ -1828,7 +1836,17 @@ public static TableBucketOffsets toTableBucketOffsets(PbTableOffsets pbTableOffs pbBucketOffset.getBucketId()); bucketOffsets.put(tableBucket, pbBucketOffset.getLogEndOffset()); } - return new TableBucketOffsets(tableId, bucketOffsets); + + // pass through the tiering-state entries unparsed + List tieringStates = new ArrayList<>(); + for (PbTieringStateEntry pbEntry : pbTableOffsets.getTieringStatesList()) { + tieringStates.add( + new TieringStateEntry( + pbEntry.getStateKey(), + pbEntry.getStateVersion(), + pbEntry.getPayload())); + } + return new TableBucketOffsets(tableId, bucketOffsets, tieringStates); } /** @@ -1966,6 +1984,15 @@ public static GetLakeSnapshotResponse makeGetLakeSnapshotResponse( } } + // pass through the tiering-state entries unparsed + for (TieringStateEntry entry : lakeTableSnapshot.getTieringStates()) { + getLakeTableSnapshotResponse + .addTieringState() + .setStateKey(entry.getStateKey()) + .setStateVersion(entry.getStateVersion()) + .setPayload(entry.getPayload()); + } + 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..7f38ff97ad 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.getTieringStates()); } } 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..8d0feeb2d3 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 @@ -21,6 +21,7 @@ import org.apache.fluss.fs.FSDataOutputStream; import org.apache.fluss.fs.FileSystem; import org.apache.fluss.fs.FsPath; +import org.apache.fluss.lake.committer.TieringStateEntry; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TablePath; import org.apache.fluss.server.zk.ZooKeeperClient; @@ -33,6 +34,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Optional; @@ -63,12 +65,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 states. This path is not expected to run + // on a state-bearing table; if it does (e.g. an old committer), drop the states with a + // warning instead of failing the commit. + if (!previousSnapshot.getTieringStates().isEmpty()) { + LOG.warn( + "Dropping tiering states for table {} on a legacy (v1) lake commit; " + + "the v1 format cannot store them.", + 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 +182,28 @@ 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); + + // merge tiering states by key (upsert): entries not carried by the request are inherited + // from the previous snapshot, so a writer only sends the keys it wants to update and never + // touches states owned by others (including keys it does not understand). + Map mergedStates = new LinkedHashMap<>(); + for (TieringStateEntry entry : previousSnapshot.getTieringStates()) { + mergedStates.put(entry.getStateKey(), entry); + } + for (TieringStateEntry entry : newTableBucketOffsets.getTieringStates()) { + mergedStates.put(entry.getStateKey(), entry); + } + + return new TableBucketOffsets( + newTableBucketOffsets.getTableId(), + bucketLogEndOffset, + new ArrayList<>(mergedStates.values())); } 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..68c55b490b 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 @@ -18,8 +18,11 @@ package org.apache.fluss.server.zk.data.lake; +import org.apache.fluss.lake.committer.TieringStateEntry; import org.apache.fluss.metadata.TableBucket; +import java.util.Collections; +import java.util.List; 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; + // the keyed tiering-state entries; empty when absent. + private final List tieringStates; + public LakeTableSnapshot(long snapshotId, Map bucketLogEndOffset) { + this(snapshotId, bucketLogEndOffset, Collections.emptyList()); + } + + public LakeTableSnapshot( + long snapshotId, + Map bucketLogEndOffset, + List tieringStates) { this.snapshotId = snapshotId; this.bucketLogEndOffset = bucketLogEndOffset; + this.tieringStates = tieringStates; } public long getSnapshotId() { @@ -52,6 +66,11 @@ public Map getBucketLogEndOffset() { return bucketLogEndOffset; } + /** Returns the keyed tiering-state entries; empty when absent. */ + public List getTieringStates() { + return tieringStates; + } + @Override public boolean equals(Object o) { if (this == o) { @@ -62,12 +81,13 @@ public boolean equals(Object o) { } LakeTableSnapshot that = (LakeTableSnapshot) o; return snapshotId == that.snapshotId - && Objects.equals(bucketLogEndOffset, that.bucketLogEndOffset); + && Objects.equals(bucketLogEndOffset, that.bucketLogEndOffset) + && Objects.equals(tieringStates, that.tieringStates); } @Override public int hashCode() { - return Objects.hash(snapshotId, bucketLogEndOffset); + return Objects.hash(snapshotId, bucketLogEndOffset, tieringStates); } @Override @@ -77,6 +97,8 @@ public String toString() { + snapshotId + ", bucketLogEndOffset=" + bucketLogEndOffset + + ", tieringStates=" + + tieringStates + '}'; } } 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..95c44cd8ac 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 states are a v2-only feature; the legacy v1 format cannot carry them (fail fast + // rather than silently drop them). + if (!lakeTableSnapshot.getTieringStates().isEmpty()) { + throw new IllegalStateException( + "The legacy (v1) lake table format does not support tiering states."); + } 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..8745604d83 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,26 @@ import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; +import org.apache.fluss.lake.committer.PartitionMarkDoneState; +import org.apache.fluss.lake.committer.PartitionMarkDoneState.PartitionState; +import org.apache.fluss.lake.committer.TieringStateEntry; 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.PbTieringStateEntry; +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 +50,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 +149,150 @@ 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; + PartitionMarkDoneState state1 = stateOf(1000L, PartitionMarkDoneState.NOT_DONE); + 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 (partition 1 marked done). It appends another entry with the same id; reads + // resolve to the latest. + PartitionMarkDoneState state2 = stateOf(1000L, 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(parseState(latest)).isEqualTo(state2); + assertThat(readState(coordinatorGateway, byIdRequest(snapshotId))).isEqualTo(state2); + } + + private static PartitionMarkDoneState stateOf(long updateTime, long doneTime) { + return new PartitionMarkDoneState( + Collections.singletonMap(1L, new PartitionState(updateTime, doneTime))); + } + + private static String prepareOffsetsFile( + CoordinatorGateway gateway, + long tableId, + Map offsets, + PartitionMarkDoneState 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()); + TieringStateEntry entry = state.toStateEntry(); + pbTableOffsets + .addTieringState() + .setStateKey(entry.getStateKey()) + .setStateVersion(entry.getStateVersion()) + .setPayload(entry.getPayload()); + for (Map.Entry offsetEntry : offsets.entrySet()) { + PbBucketOffset pbBucketOffset = pbTableOffsets.addBucketOffset(); + pbBucketOffset.setBucketId(offsetEntry.getKey().getBucket()); + pbBucketOffset.setLogEndOffset(offsetEntry.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 PartitionMarkDoneState readState( + CoordinatorGateway gateway, GetLakeSnapshotRequest request) throws Exception { + return parseState(gateway.getLakeSnapshot(request).get()); + } + + private static PartitionMarkDoneState parseState(GetLakeSnapshotResponse response) { + for (PbTieringStateEntry pbEntry : response.getTieringStatesList()) { + if (PartitionMarkDoneState.STATE_KEY.equals(pbEntry.getStateKey())) { + return PartitionMarkDoneState.fromStateEntry( + new TieringStateEntry( + pbEntry.getStateKey(), + pbEntry.getStateVersion(), + pbEntry.getPayload())); + } + } + throw new AssertionError("partition mark-done state not found in response"); + } + + 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..e697dc2978 --- /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 states when absent (PREPARE read + * yields empty, GET fill leaves the field unset). The present-state passthrough is covered + * end-to-end by {@code CommitLakeTableSnapshotITCase}. + */ +class ServerRpcMessageUtilsTieringStateTest { + + @Test + void testAbsentTieringStates() { + // PREPARE read without tiering states -> empty. + PbTableOffsets pbTableOffsets = new PbTableOffsets(); + pbTableOffsets.setTableId(2L); + pbTableOffsets.setTablePath().setDatabaseName("db").setTableName("t"); + pbTableOffsets.addBucketOffset().setBucketId(0).setLogEndOffset(100L); + assertThat(ServerRpcMessageUtils.toTableBucketOffsets(pbTableOffsets).getTieringStates()) + .isEmpty(); + + // GET fill without tiering states -> response omits them. + Map bucketOffsets = new HashMap<>(); + bucketOffsets.put(new TableBucket(2L, 0), 100L); + GetLakeSnapshotResponse response = + ServerRpcMessageUtils.makeGetLakeSnapshotResponse( + 2L, new LakeTableSnapshot(9L, bucketOffsets)); + assertThat(response.getTieringStatesCount()).isZero(); + } +} 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..c8d11cc7e8 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 @@ -17,6 +17,7 @@ package org.apache.fluss.server.zk.data; +import org.apache.fluss.lake.committer.TieringStateEntry; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.server.zk.data.lake.LakeTableSnapshot; import org.apache.fluss.server.zk.data.lake.LakeTableSnapshotLegacyJsonSerde; @@ -26,10 +27,12 @@ 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.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Test for {@link LakeTableSnapshotLegacyJsonSerde}. */ class LakeTableSnapshotLegacyJsonSerdeTest extends JsonSerdeTestBase { @@ -104,4 +107,25 @@ void testBackwardCompatibility() { LakeTableSnapshotLegacyJsonSerde.INSTANCE); assertThat(snapshot3.getSnapshotId()).isEqualTo(3); } + + @Test + void testSerializeRejectsTieringStates() { + // Tiering states are a version-2 only feature; the legacy v1 format must reject them + // rather than silently drop them. + Map offsets = new HashMap<>(); + offsets.put(new TableBucket(1L, 0), 100L); + LakeTableSnapshot withState = + new LakeTableSnapshot( + 1L, + offsets, + Collections.singletonList( + new TieringStateEntry( + "some-key", 1, "{}".getBytes(StandardCharsets.UTF_8)))); + assertThatThrownBy( + () -> + JsonSerdeUtils.writeValueAsBytes( + withState, LakeTableSnapshotLegacyJsonSerde.INSTANCE)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("does not support tiering states"); + } } 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..52fc9c40fc 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,9 @@ 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.PartitionMarkDoneState; +import org.apache.fluss.lake.committer.PartitionMarkDoneState.PartitionState; +import org.apache.fluss.lake.committer.TieringStateEntry; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TableDescriptor; import org.apache.fluss.metadata.TablePath; @@ -42,7 +45,9 @@ import org.junit.jupiter.api.extension.RegisterExtension; import org.junit.jupiter.api.io.TempDir; +import java.nio.charset.StandardCharsets; import java.nio.file.Path; +import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -253,6 +258,147 @@ void testRegisterLakeTableSnapshotWithRetention(@TempDir Path tempDir) throws Ex .containsExactly(4L, 5L, 6L); } + /** + * Verifies the tiering states merge by key (upsert): a request entry replaces the previous + * entry with the same key, entries of other keys (even unrecognized ones) are inherited, and + * bucket offsets are still merged. + */ + @Test + void testMergeUpsertsTieringStateByKey(@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 mark-done state and a foreign 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 prevStates = new HashMap<>(); + prevStates.put(1L, new PartitionState(1000L, PartitionMarkDoneState.NOT_DONE)); + prevStates.put(2L, new PartitionState(2000L, PartitionMarkDoneState.NOT_DONE)); + prevStates.put(3L, new PartitionState(3000L, PartitionMarkDoneState.NOT_DONE)); + TieringStateEntry foreignEntry = + new TieringStateEntry( + "future-key", 7, "{\"x\":1}".getBytes(StandardCharsets.UTF_8)); + FsPath prevPath = + lakeTableHelper.storeLakeTableOffsetsFile( + tablePath, + new TableBucketOffsets( + tableId, + prevOffsets, + Arrays.asList( + new PartitionMarkDoneState(prevStates).toStateEntry(), + foreignEntry))); + lakeTableHelper.registerLakeTableSnapshotV2( + tableId, new LakeTable.LakeSnapshotMetadata(1L, prevPath, prevPath)); + LakeTable previousLakeTable = zookeeperClient.getLakeTable(tableId).get(); + + // --- New offsets: advance partitions 1 & 3, carry only the mark-done state (partition 2 + // now done). --- + Map newOffsets = new HashMap<>(); + newOffsets.put(new TableBucket(tableId, 1L, 0), 150L); + newOffsets.put(new TableBucket(tableId, 3L, 0), 350L); + Map newStates = new HashMap<>(); + newStates.put(1L, new PartitionState(1500L, PartitionMarkDoneState.NOT_DONE)); + newStates.put(2L, new PartitionState(2000L, 2500L)); + newStates.put(3L, new PartitionState(3500L, PartitionMarkDoneState.NOT_DONE)); + PartitionMarkDoneState newState = new PartitionMarkDoneState(newStates); + TableBucketOffsets merged = + lakeTableHelper.mergeTableBucketOffsets( + previousLakeTable, + new TableBucketOffsets( + tableId, + newOffsets, + Collections.singletonList(newState.toStateEntry()))); + + // mark-done entry replaced by the new value; the foreign entry survives untouched + assertThat(merged.getTieringStates()) + .containsExactlyInAnyOrder(newState.toStateEntry(), foreignEntry); + // 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 upsert semantics for absent states: a request without tiering states leaves the + * previous entries untouched (a writer that sends nothing touches nothing), while bucket + * offsets are still merged. + */ + @Test + void testMergePreservesTieringStatesWhenAbsent(@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); + TieringStateEntry prevEntry = + new PartitionMarkDoneState( + Collections.singletonMap( + 1L, + new PartitionState(1000L, PartitionMarkDoneState.NOT_DONE))) + .toStateEntry(); + FsPath prevPath = + lakeTableHelper.storeLakeTableOffsetsFile( + tablePath, + new TableBucketOffsets( + tableId, prevOffsets, Collections.singletonList(prevEntry))); + 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 states -> previous entries preserved, offsets merged. + TableBucketOffsets merged = + lakeTableHelper.mergeTableBucketOffsets( + previousLakeTable, new TableBucketOffsets(tableId, newOffsets)); + + assertThat(merged.getTieringStates()).containsExactly(prevEntry); + assertThat(merged.getOffsets()).containsEntry(new TableBucket(tableId, 1L, 0), 150L); + } + + /** + * A legacy (v1) commit on a state-bearing table drops the states (with a warning), not fails. + */ + @Test + void testRegisterV1DropsExistingTieringStates(@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 a tiering state + Map offsets = new HashMap<>(); + offsets.put(new TableBucket(tableId, 0), 100L); + TieringStateEntry entry = + new PartitionMarkDoneState( + Collections.singletonMap( + 1L, + new PartitionState(1000L, PartitionMarkDoneState.NOT_DONE))) + .toStateEntry(); + FsPath path = + lakeTableHelper.storeLakeTableOffsetsFile( + tablePath, + new TableBucketOffsets(tableId, offsets, Collections.singletonList(entry))); + lakeTableHelper.registerLakeTableSnapshotV2( + tableId, new LakeTable.LakeSnapshotMetadata(1L, path, path)); + + // a v1 commit drops the states 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.getTieringStates()).isEmpty(); + 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 {