You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "luoyuxia (via GitHub)" <gi...@apache.org> on 2026/09/02 07:32:50 UTC
[PR] [server] Incrementally clean historical KV state [fluss]
luoyuxia opened a new pull request, #4199:
URL: https://github.com/apache/fluss/pull/4199
<!--
Generated-by: Codex (GPT-5) following https://github.com/apache/fluss/blob/main/AGENTS.md
-->
### Purpose
Linked issue: #4159.
Depends on #4178. This is a stacked PR; until #4178 merges, review commit `a6bd60971` only.
Historical writes retain local KV values and tombstones until their WAL offsets are covered by the lake snapshot. This change makes entries below the confirmed lake progress eligible for incremental reclamation, without waiting for the lake and local log end offsets to become equal or rebuilding the entire RocksDB instance.
Cleanup is best effort: the cleanup offset controls eligibility, while physical deletion occurs during normal RocksDB compactions. The implementation does not schedule an additional full-range compaction when the offset advances.
### Brief change log
- Tag historical values and tombstones with the WAL offset that produced them on leader writes, pre-write-buffer flushes, and WAL recovery.
- Maintain a monotonic, exclusive historical cleanup offset from confirmed lake progress, publishing the required lake snapshot before exposing a newer offset and restoring it safely during leader recovery.
- Generalize the row-TTL compaction-filter factory so historical cleanup can use the stored WAL-offset tag and remove entries whose tag is lower than the cleanup offset.
- Decode tagged local lookup results while keeping lake values in their existing plain layout. The resolved-probe context introduced by #4178 preserves lookup correctness if compaction runs between probe and apply.
- Add tests for offset tagging, value and tombstone cleanup boundaries, repeated progress notifications, recovery, and lookup behavior.
### Tests
- Targeted `fluss-server` unit tests: `KvValueLayoutTest`, `RowTtlCompactionFilterTest`, `HistoricalKvCompactionFilterTest`, `KvTabletTest`, and `HistoricalPartitionManagerTest` (56 tests passed).
- `./mvnw -pl fluss-server spotless:check`
- Checkstyle and RAT validation passed for the affected modules.
- `git diff --check`
- `./mvnw verify` was attempted. It stopped in `fluss-server` after 1,303 tests because the local disk was 95% full: five `ReplicaFetcherThreadTest` cases hit the 85% RocksDB disk guard, and two unrelated `ReplicaTest` local-log metric assertions failed. The historical-KV and compaction-filter tests passed in that run.
### API and Format
No public API changes.
The internal RocksDB value format for historical KV state now carries an 8-byte WAL-offset tag; a historical tombstone consists only of that tag. Lake values remain in the plain value layout.
### Documentation
No user-facing documentation changes. The design and trade-offs are described in #4159 and this PR.
### Generative AI disclosure
- [x] Yes — Codex (GPT-5), reviewed by the author.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Incrementally clean historical KV state [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg commented on code in PR #4199:
URL: https://github.com/apache/fluss/pull/4199#discussion_r3987357357
##########
fluss-server/src/main/java/org/apache/fluss/server/kv/historical/LocalValueLookupResult.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.kv.historical;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.record.BinaryValue;
+import org.apache.fluss.row.encode.ValueDecoder;
+import org.apache.fluss.server.kv.KvStateLookupResult;
+import org.apache.fluss.utils.ByteArrayWrapper;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.NotThreadSafe;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+import static org.apache.fluss.utils.Preconditions.checkState;
+
+/** Local lookup results for previous values required by a historical write request. */
+@Internal
+@NotThreadSafe
+public final class LocalValueLookupResult {
+
+ /** Decodes tagged values returned by local historical KV state. */
+ private final ValueDecoder localValueDecoder;
+
+ /** Decodes plain values returned by lake storage. */
+ private final ValueDecoder lakeValueDecoder;
+
+ /** Previous values found locally, including tombstones represented by an empty optional. */
+ private final Map<ByteArrayWrapper, Optional<BinaryValue>> localValuesByKey = new HashMap<>();
+
+ /** True local misses in lake request order. */
+ private final Set<ByteArrayWrapper> keysMissingLocally = new LinkedHashSet<>();
+
+ /** Creates a local lookup result with decoders for the local and lake value layouts. */
+ public LocalValueLookupResult(ValueDecoder localValueDecoder, ValueDecoder lakeValueDecoder) {
+ this.localValueDecoder = localValueDecoder;
+ this.lakeValueDecoder = lakeValueDecoder;
+ }
+
+ /** Records the local result for one key whose previous value is required. */
+ public void add(byte[] primaryKey, KvStateLookupResult localResult) {
+ ByteArrayWrapper wrappedKey = new ByteArrayWrapper(primaryKey);
+ checkState(
+ !localValuesByKey.containsKey(wrappedKey)
+ && !keysMissingLocally.contains(wrappedKey),
+ "Historical write key has already been probed");
+
+ if (localResult.status() == KvStateLookupResult.Status.NOT_FOUND) {
+ keysMissingLocally.add(wrappedKey);
+ } else {
+ localValuesByKey.put(wrappedKey, decode(localResult.value(), localValueDecoder));
+ }
+ }
+
+ /** Returns a snapshot of true local misses in lake request order. */
+ public List<byte[]> keysMissingLocally() {
+ List<byte[]> primaryKeys = new ArrayList<>(keysMissingLocally.size());
+ for (ByteArrayWrapper keyMissingLocally : keysMissingLocally) {
+ primaryKeys.add(keyMissingLocally.getData());
+ }
+ return Collections.unmodifiableList(primaryKeys);
+ }
+
+ /**
+ * Combines lake results with the local lookup results and creates an in-memory lookup for
+ * apply.
+ *
+ * <p>Lake values must have the same order as {@link #keysMissingLocally()}.
+ */
+ public HistoricalValueLookup createValueLookup(List<byte[]> lakeValues) {
Review Comment:
keysMissingLocally() and createValueLookup() seem to have an implicit contract between them — the caller has to know to call them in the right order. Pushing that onto the caller in Replica doesn't feel like good practice.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Incrementally clean historical KV state [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #4199:
URL: https://github.com/apache/fluss/pull/4199#discussion_r3977568593
##########
fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java:
##########
@@ -1192,7 +1233,7 @@ public KvStateLookupResult lookupHistoricalLocal(String originalPartitionName, b
if (value == null) {
return KvStateLookupResult.notFound();
}
- return value.length == 0
+ return HistoricalKvTombstone.isTombstone(value)
? KvStateLookupResult.deleted()
: KvStateLookupResult.present(value);
Review Comment:
This change forces historical tablets to always use `KvValueLayout.TAGGED` and treats tombstones as 8-byte tagged values. If an existing deployment has historical RocksDB state written in the prior layout (plain values + empty tombstones), reads/merges/compaction decisions after upgrade can misinterpret stored bytes (e.g., treating old plain values as tagged). If rolling upgrades or in-place upgrades are supported, consider adding a compatibility strategy: detect legacy values at read time (and decode as PLAIN / treat empty value as tombstone), and ensure the compaction filter does not delete legacy entries prematurely; alternatively, gate this behind a persisted format version and require/trigger a one-time rebuild/migration for pre-existing historical KV state.
##########
fluss-common/src/main/java/org/apache/fluss/row/encode/ValueEncoder.java:
##########
@@ -20,31 +20,41 @@
import org.apache.fluss.record.BinaryValue;
import org.apache.fluss.row.BinaryRow;
+import javax.annotation.Nullable;
+
import java.util.function.ToLongFunction;
import static org.apache.fluss.utils.Preconditions.checkNotNull;
+import static org.apache.fluss.utils.Preconditions.checkState;
/** An encoder to encode {@link BinaryRow} with a schema id as value to be stored in kv store. */
public final class ValueEncoder {
private static final ValueEncoder PLAIN_ENCODER = new ValueEncoder(KvValueLayout.PLAIN, null);
private final KvValueLayout kvValueLayout;
- private final ToLongFunction<BinaryRow> valueTagProvider;
- private ValueEncoder(KvValueLayout kvValueLayout, ToLongFunction<BinaryRow> valueTagProvider) {
+ /**
+ * Generates tags for {@link #encodeValue(BinaryValue)}. {@code null} for plain values or when
+ * callers supply each tag to {@link #encodeValue(BinaryValue, long)}.
+ */
+ @Nullable private final ToLongFunction<BinaryRow> valueTagProvider;
+
+ private ValueEncoder(
+ KvValueLayout kvValueLayout, @Nullable ToLongFunction<BinaryRow> valueTagProvider) {
this.kvValueLayout = kvValueLayout;
this.valueTagProvider = valueTagProvider;
}
- /** Returns an encoder for a layout without an internal value tag. */
+ /**
+ * Returns an encoder for the given layout. Tagged values must supply their tag to {@link
+ * #encodeValue(BinaryValue, long)}.
+ */
public static ValueEncoder forLayout(KvValueLayout kvValueLayout) {
checkNotNull(kvValueLayout, "kvValueLayout must not be null.");
- if (kvValueLayout != KvValueLayout.PLAIN) {
- throw new IllegalArgumentException(
- "A value tag provider is required for this KV value layout.");
- }
- return PLAIN_ENCODER;
+ return kvValueLayout == KvValueLayout.PLAIN
+ ? PLAIN_ENCODER
+ : new ValueEncoder(kvValueLayout, null);
}
Review Comment:
`ValueEncoder.forLayout(TAGGED)` is now allowed but produces an encoder that will throw at runtime if `encodeValue(BinaryValue)` is called (because there is no tag provider). This is easy to misuse and shifts failures from compile-time/early validation to runtime. Consider separating the factories to make intent explicit (e.g., keep `forLayout(TAGGED)` rejecting unless a provider is given, and introduce a dedicated factory like `forTaggedLayoutWithExplicitTag(...)`), or at least strengthen the Javadoc/parameter naming so callers understand that the 1-arg `encodeValue` is invalid for tagged layouts created via `forLayout`.
##########
fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java:
##########
@@ -424,11 +437,16 @@ private static KvTablet create(
TableConfig tableConfig)
throws IOException {
checkNotNull(tableConfig, "tableConfig must not be null.");
+ boolean historicalPartition =
+ HISTORICAL_PARTITION_VALUE.equals(tablePath.getPartitionName());
Optional<Duration> rowTtl = tableConfig.getKvTTL();
- KvValueLayout kvValueLayout = KvValueLayout.fromTableConfig(tableConfig);
+ KvValueLayout kvValueLayout =
+ historicalPartition
+ ? KvValueLayout.TAGGED
+ : KvValueLayout.fromTableConfig(tableConfig);
Review Comment:
This change forces historical tablets to always use `KvValueLayout.TAGGED` and treats tombstones as 8-byte tagged values. If an existing deployment has historical RocksDB state written in the prior layout (plain values + empty tombstones), reads/merges/compaction decisions after upgrade can misinterpret stored bytes (e.g., treating old plain values as tagged). If rolling upgrades or in-place upgrades are supported, consider adding a compatibility strategy: detect legacy values at read time (and decode as PLAIN / treat empty value as tombstone), and ensure the compaction filter does not delete legacy entries prematurely; alternatively, gate this behind a persisted format version and require/trigger a one-time rebuild/migration for pre-existing historical KV state.
##########
fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java:
##########
@@ -1316,12 +1320,49 @@ public LogAppendInfo putHistoricalRecordsToLeader(
targetColumns,
mergeMode,
originalPartitionName,
- memoizedLakeLookup);
+ historicalValueLookup);
maybeIncrementLeaderHW(logTablet, clock.milliseconds());
return appendInfo;
});
}
+ /**
+ * Updates the historical cleanup offset if it is valid.
+ *
+ * <p>{@code beforeUpdate} runs after validation and before the new cleanup offset becomes
+ * visible to the RocksDB compaction filter.
+ */
+ public void tryUpdateHistoricalCleanupOffset(long newCleanupOffset, Runnable beforeUpdate) {
+ checkNotNull(beforeUpdate, "beforeUpdate must not be null.");
+ inWriteLock(
+ leaderIsrUpdateLock,
+ () -> {
+ if (!isLeader() || !historicalPartition) {
+ return;
+ }
+ KvTablet currentKvTablet = kvTablet;
+ if (currentKvTablet == null) {
+ return;
+ }
+ long currentCleanupOffset = currentKvTablet.getHistoricalCleanupOffset();
+ long localLogEndOffset = logTablet.localLogEndOffset();
+ if (newCleanupOffset < currentCleanupOffset
+ || newCleanupOffset > localLogEndOffset) {
+ LOG.warn(
+ "Ignore invalid historical cleanup offset {} for {} with "
+ + "current cleanup offset {} and local log end "
+ + "offset {}.",
+ newCleanupOffset,
+ tableBucket,
+ currentCleanupOffset,
+ localLogEndOffset);
+ return;
+ }
+ beforeUpdate.run();
+ currentKvTablet.advanceHistoricalCleanupOffset(newCleanupOffset);
+ });
+ }
Review Comment:
`beforeUpdate.run()` executes while holding `leaderIsrUpdateLock` (write lock). In this PR, `beforeUpdate` publishes the required lake snapshot; if that path can block (e.g., sync RPC, IO, or waiting on other locks), it risks stalling leader ISR updates and historical writes, and may introduce lock-ordering hazards. Consider making `beforeUpdate` strictly non-blocking (document/enforce), or refactor to a two-phase update (validate under lock → publish snapshot without the lock → re-check + apply offset) to avoid executing potentially expensive work while holding the write lock.
##########
fluss-server/src/test/java/org/apache/fluss/server/replica/historical/HistoricalPartitionManagerTest.java:
##########
@@ -776,6 +959,39 @@ void testHistoricalLookupThrottledWhenPermitsExhausted() throws Exception {
}
}
+ private HistoricalPartitionManager createHistoricalPartitionManager(
+ TestingHistoricalLakeLookupManager lakeLookupManager) {
+ return new HistoricalPartitionManager(
+ new HistoricalPartitionTaskExecutor(lookupConfiguration()), lakeLookupManager);
+ }
+
+ private static LogAppendInfo writeBatch(
+ HistoricalPartitionManager manager,
+ Replica replica,
+ String originalPartition,
+ KvRecordBatch records)
+ throws Exception {
+ return manager.processPut(
+ replica,
+ new PutKvDataForBucket(TABLE_BUCKET, records, originalPartition),
+ null,
+ MergeMode.DEFAULT,
+ 1);
+ }
+
+ private static byte[] key(int id, String region) {
+ return new CompactedKeyEncoder(KEY_TYPE).encodeKey(row(id, region));
+ }
Review Comment:
This helper allocates a new `CompactedKeyEncoder` for every key generation. In tests that generate many keys/batches, this adds avoidable churn. Consider caching a static `CompactedKeyEncoder` instance (or reusing one per test) to reduce allocations and keep test runtime more stable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Incrementally clean historical KV state [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #4199:
URL: https://github.com/apache/fluss/pull/4199#issuecomment-5628263068
@wuchong @platinumhamburg Could you please help review this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Incrementally clean historical KV state [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #4199:
URL: https://github.com/apache/fluss/pull/4199#issuecomment-5724261710
Let me merge it first since the risk is realtely low. If there’s any further feedback, we can address it in a follow-up PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Incrementally clean historical KV state [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia merged PR #4199:
URL: https://github.com/apache/fluss/pull/4199
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org