You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "platinumhamburg (via GitHub)" <gi...@apache.org> on 2026/08/21 11:48:40 UTC
[PR] [kv] Batch old value lookups in write path [fluss]
platinumhamburg opened a new pull request, #4065:
URL: https://github.com/apache/fluss/pull/4065
Load distinct RocksDB misses with a single multi-get while preserving pre-write precedence and intra-batch record order. Include multi-get latency in KV read metrics.
<!--
*Thank you very much for contributing to Fluss - we are happy that you want to help us improve Fluss. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [GitHub issue](https://github.com/apache/fluss/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no issue.
- Name the pull request in the format "[component] Title of the pull request", where *[component]* should be replaced by the name of the component being changed. Typically, this corresponds to the component label assigned to the issue (e.g., [kv], [log], [client], [flink], [rust], [python], [c++], [elixir]). Skip *[component]* if you are unsure about which is the best component.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes.
- Each pull request should address only one issue, not mix up code from multiple issues.
- **Generative AI disclosure:** Indicate whether generative AI tools were used in authoring this PR. If yes, specify the tool below.
- [ ] No generative AI tools used
- [ ] Yes (please specify the tool below)
**(The sections below can be removed for hotfixes or typos)**
-->
<!--
Generated-by: [Tool Name and Version] following [the guidelines](https://github.com/apache/fluss/blob/main/AGENTS.md)
-->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #4063
<!-- What is the purpose of the change -->
### Brief change log
<!-- Please describe the changes made in this pull request and explain how they address the issue -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
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] [kv] Batch old value lookups in write path [fluss]
Posted by "polyzos (via GitHub)" <gi...@apache.org>.
polyzos commented on code in PR #4065:
URL: https://github.com/apache/fluss/pull/4065#discussion_r4410002977
##########
fluss-server/src/main/java/org/apache/fluss/server/kv/KvStateAccessor.java:
##########
@@ -109,6 +114,35 @@ public BinaryValue lookup(
return localResult.isPresent() ? valueDecoder.decodeValue(localResult.value()) : null;
}
+ /** Returns whether a buffered value or delete already determines the local state. */
+ boolean containsBufferedKey(Key key) {
+ return preWriteBuffer.get(key) != null;
+ }
+
+ /**
+ * Loads distinct RocksDB misses for a write batch. Buffered values and deletes are excluded.
+ * Callers must hold the KV write lock through prefetch and application, and recheck the buffer
+ * before using a saved result so earlier mutations in the batch remain visible.
+ */
+ Map<Key, byte[]> prefetchRocksDBValues(Set<Key> keys) throws IOException {
+ List<Key> missingKeys = new ArrayList<>();
+ List<byte[]> encodedKeys = new ArrayList<>();
+ for (Key key : keys) {
+ if (preWriteBuffer.get(key) == null) {
+ missingKeys.add(key);
+ encodedKeys.add(key.get());
+ }
+ }
+ Map<Key, byte[]> values = new HashMap<>();
+ if (!missingKeys.isEmpty()) {
+ List<byte[]> results = rocksDBKv.multiGet(encodedKeys);
+ for (int i = 0; i < missingKeys.size(); i++) {
+ values.put(missingKeys.get(i), results.get(i));
Review Comment:
multiGetAsList returns null both for a missing key and for a key whose read failed (#4484). The write path used get() before, which throws on a read error. With this change, a corrupted block turns an update into an INSERT and silently drops a delete. 269fb33 removed the null recheck, but the PR description still says it is there. Maybe restore the recheck, or land #4488 first and rebase on it.??
##########
fluss-server/src/main/java/org/apache/fluss/server/kv/KvWriteProcessor.java:
##########
@@ -344,11 +378,30 @@ private void processKvRecords(
logOffset,
stateAccessor,
keyBytes,
- historicalValueLookup);
+ previousValueLookup);
}
}
}
+ private Map<KvPreWriteBuffer.Key, byte[]> prefetchPreviousValues(
+ KvRecordBatch kvRecords,
+ RowMerger currentMerger,
+ AutoIncrementUpdater autoIncrementUpdater,
+ KvStateAccessor stateAccessor)
+ throws Exception {
+ Set<KvPreWriteBuffer.Key> keys = new LinkedHashSet<>();
+ KvRecordBatch.ReadContext readContext =
Review Comment:
Nit: this is a second ReadContext, and a second full decode of the batch,
sitting next to TODO at line 323.
Maybe use the existing one?
##########
fluss-server/src/main/java/org/apache/fluss/server/kv/KvWriteProcessor.java:
##########
@@ -315,12 +324,37 @@ private void processKvRecords(
KvRecordBatch.ReadContext readContext =
KvRecordReadContext.createReadContext(kvFormat, schemaGetter);
+ boolean prefetchEnabled =
+ originalPartitionName == null
+ && kvRecords.getRecordCount() >= MIN_BATCH_SIZE_FOR_PREFETCH;
+ Function<byte[], BinaryValue> previousValueLookup =
+ historicalValueLookup == null ? null : historicalValueLookup::lookup;
for (KvRecord kvRecord : kvRecords.records(readContext)) {
byte[] keyBytes = BytesUtils.toArray(kvRecord.getKey());
KvPreWriteBuffer.Key key = stateAccessor.encodeKey(keyBytes, originalPartitionName);
BinaryRow row = kvRecord.getRow();
BinaryValue currentValue = row == null ? null : new BinaryValue(schemaIdOfNewData, row);
+ if (prefetchEnabled
+ && previousValueLookup == null
+ && (currentValue == null
+ ? !shouldIgnoreDeletion(currentMerger)
+ : !canSkipOldValueLookup(currentMerger, autoIncrementUpdater))
+ && !stateAccessor.containsBufferedKey(key)) {
+ // Start prefetch only on a required buffer miss. Full-row WAL batches and batches
+ // served entirely by the buffer retain their single-pass processing.
+ Map<KvPreWriteBuffer.Key, byte[]> savedValues =
+ prefetchPreviousValues(
+ kvRecords, currentMerger, autoIncrementUpdater, stateAccessor);
+ previousValueLookup =
+ primaryKey -> {
+ KvPreWriteBuffer.Key savedKey =
+ stateAccessor.encodeKey(primaryKey, null);
+ byte[] value = savedValues.get(savedKey);
+ return value == null ? null : valueDecoder.decodeValue(value);
Review Comment:
This treats "key not in the map" the same as "key not in RocksDB". It's safe now because the buffer can't shrink while we hold the write lock, but nothing enforces that. Maybe enforce it?? so a future change fails loudly instead of producing a fake insert?
##########
fluss-server/src/main/java/org/apache/fluss/server/kv/KvStateAccessor.java:
##########
@@ -109,6 +114,35 @@ public BinaryValue lookup(
return localResult.isPresent() ? valueDecoder.decodeValue(localResult.value()) : null;
}
+ /** Returns whether a buffered value or delete already determines the local state. */
+ boolean containsBufferedKey(Key key) {
+ return preWriteBuffer.get(key) != null;
+ }
+
+ /**
+ * Loads distinct RocksDB misses for a write batch. Buffered values and deletes are excluded.
+ * Callers must hold the KV write lock through prefetch and application, and recheck the buffer
+ * before using a saved result so earlier mutations in the batch remain visible.
+ */
+ Map<Key, byte[]> prefetchRocksDBValues(Set<Key> keys) throws IOException {
+ List<Key> missingKeys = new ArrayList<>();
+ List<byte[]> encodedKeys = new ArrayList<>();
+ for (Key key : keys) {
+ if (preWriteBuffer.get(key) == null) {
+ missingKeys.add(key);
+ encodedKeys.add(key.get());
+ }
+ }
+ Map<Key, byte[]> values = new HashMap<>();
Review Comment:
Every distinct old value in the batch is held on the heap at once, with no upper bound. A partial-update batch of many small keys against wide rows can pull much more data than the request size. Maybe consider chunking the MultiGet by key count or by bytes??
--
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] [kv] Batch old value lookups in write path [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg commented on code in PR #4065:
URL: https://github.com/apache/fluss/pull/4065#discussion_r4357179365
##########
fluss-server/src/main/java/org/apache/fluss/server/kv/KvWriteProcessor.java:
##########
@@ -315,12 +324,36 @@ private void processKvRecords(
KvRecordBatch.ReadContext readContext =
KvRecordReadContext.createReadContext(kvFormat, schemaGetter);
+ boolean prefetchEnabled =
+ originalPartitionName == null
+ && kvRecords.getRecordCount() >= MIN_BATCH_SIZE_FOR_PREFETCH;
+ Function<byte[], BinaryValue> previousValueLookup =
+ historicalValueLookup == null ? null : historicalValueLookup::lookup;
for (KvRecord kvRecord : kvRecords.records(readContext)) {
byte[] keyBytes = BytesUtils.toArray(kvRecord.getKey());
KvPreWriteBuffer.Key key = stateAccessor.encodeKey(keyBytes, originalPartitionName);
BinaryRow row = kvRecord.getRow();
BinaryValue currentValue = row == null ? null : new BinaryValue(schemaIdOfNewData, row);
+ if (prefetchEnabled
+ && previousValueLookup == null
+ && (currentValue == null
+ ? !shouldIgnoreDeletion(currentMerger)
+ : !canSkipOldValueLookup(currentMerger, autoIncrementUpdater))
+ && !stateAccessor.containsBufferedKey(key)) {
+ // Start prefetch only on a required buffer miss. Full-row WAL batches and batches
+ // served entirely by the buffer retain their single-pass processing.
+ Map<KvPreWriteBuffer.Key, byte[]> savedValues =
+ prefetchPreviousValues(
+ kvRecords, currentMerger, autoIncrementUpdater, stateAccessor);
+ previousValueLookup =
+ primaryKey -> {
+ KvPreWriteBuffer.Key savedKey = KvPreWriteBuffer.Key.of(primaryKey);
Review Comment:
Aligned saved-value lookup with stateAccessor.encodeKey(primaryKey, null). Population and lookup now use the same encoding path.
--
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] [kv] Batch old value lookups in write path [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg commented on code in PR #4065:
URL: https://github.com/apache/fluss/pull/4065#discussion_r4357182932
##########
fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java:
##########
@@ -2356,8 +2469,9 @@ void testRocksDBMetrics() throws Exception {
}
List<ByteArraySlice> readValues = kvTablet.multiGet(keysToRead);
assertThat(readValues).hasSize(100);
+ assertThat(kvTablet.getRocksDBKv().get("key".getBytes())).isEqualTo("value".getBytes());
Review Comment:
The test writes and reads those bytes in the same process with the same default charset. It does not assert a cross-platform byte encoding.
--
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] [kv] Batch old value lookups in write path [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg closed pull request #4065: [kv] Batch old value lookups in write path
URL: https://github.com/apache/fluss/pull/4065
--
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] [kv] Batch old value lookups in write path [fluss]
Posted by "platinumhamburg (via GitHub) via issues" <is...@fluss.apache.org>.
platinumhamburg commented on PR #4065:
URL: https://github.com/apache/fluss/pull/4065#issuecomment-6593360198
Thanks @polyzos and @wuchong for the thoughtful reviews and suggestions. The shared `KvConfig` change is now in place, and #5016 tracks the `KvTablet` construction cleanup separately.
Could you take another look when you have time? Thanks!
--
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] [kv] Batch old value lookups in write path [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg commented on code in PR #4065:
URL: https://github.com/apache/fluss/pull/4065#discussion_r4411107399
##########
fluss-server/src/main/java/org/apache/fluss/server/kv/KvWriteProcessor.java:
##########
@@ -315,12 +324,37 @@ private void processKvRecords(
KvRecordBatch.ReadContext readContext =
KvRecordReadContext.createReadContext(kvFormat, schemaGetter);
+ boolean prefetchEnabled =
+ originalPartitionName == null
+ && kvRecords.getRecordCount() >= MIN_BATCH_SIZE_FOR_PREFETCH;
+ Function<byte[], BinaryValue> previousValueLookup =
+ historicalValueLookup == null ? null : historicalValueLookup::lookup;
for (KvRecord kvRecord : kvRecords.records(readContext)) {
byte[] keyBytes = BytesUtils.toArray(kvRecord.getKey());
KvPreWriteBuffer.Key key = stateAccessor.encodeKey(keyBytes, originalPartitionName);
BinaryRow row = kvRecord.getRow();
BinaryValue currentValue = row == null ? null : new BinaryValue(schemaIdOfNewData, row);
+ if (prefetchEnabled
+ && previousValueLookup == null
+ && (currentValue == null
+ ? !shouldIgnoreDeletion(currentMerger)
+ : !canSkipOldValueLookup(currentMerger, autoIncrementUpdater))
+ && !stateAccessor.containsBufferedKey(key)) {
+ // Start prefetch only on a required buffer miss. Full-row WAL batches and batches
+ // served entirely by the buffer retain their single-pass processing.
+ Map<KvPreWriteBuffer.Key, byte[]> savedValues =
+ prefetchPreviousValues(
+ kvRecords, currentMerger, autoIncrementUpdater, stateAccessor);
+ previousValueLookup =
+ primaryKey -> {
+ KvPreWriteBuffer.Key savedKey =
+ stateAccessor.encodeKey(primaryKey, null);
+ byte[] value = savedValues.get(savedKey);
+ return value == null ? null : valueDecoder.decodeValue(value);
Review Comment:
The write lock is held throughout prefetch and batch application. Flush completion takes the same lock, and buffered deletes retain tombstones rather than removing entries. Therefore, a key excluded from prefetch because it was buffered remains available through the live buffer.
Under this invariant, a buffer miss requiring an old value must have a corresponding entry in the prefetched map. I would keep the current implementation; changes to buffer eviction during batch processing would need to revisit this assumption.
--
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] [kv] Batch old value lookups in write path [fluss]
Posted by "platinumhamburg (via GitHub) via issues" <is...@fluss.apache.org>.
platinumhamburg commented on PR #4065:
URL: https://github.com/apache/fluss/pull/4065#issuecomment-6622077407
Iām closing this PR after evaluating the performance trade-offs.
The Mac mini JMH benchmarks showed CPU savings in some workloads, but also reproducible regressions of approximately 10% for batches with highly repeated keys and 4% for missing-key lookups. Java allocation increased by roughly 8ā16% in workloads that triggered prefetching. These results make the current implementation difficult to justify as a default optimization.
Thanks everyone for the reviews and discussion.
--
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] [kv] Batch old value lookups in write path [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg commented on PR #4065:
URL: https://github.com/apache/fluss/pull/4065#issuecomment-6293299378
Benchmarked the complete `KvTablet.putAsLeader` call against main on a Mac mini M4, using five independent JVM forks per implementation per scenario. Mean latency per `KvTablet.putAsLeader` invocation decreased by 7.4%/12.5% for batches of 32/256 missing keys and 18.3%/25.8% for batches of 32/256 existing keys. Single-key misses still regressed by 7.7%, and batch workloads incurred higher Java allocation. These are hot-cache, single-tablet write-path latency results, including lock acquisition, old-value reads, merging, encoding, and local WAL append; they do not measure isolated RocksDB read latency or end-to-end RPC throughput.
--
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] [kv] Batch old value lookups in write path [fluss]
Posted by "platinumhamburg (via GitHub) via issues" <is...@fluss.apache.org>.
platinumhamburg commented on code in PR #4065:
URL: https://github.com/apache/fluss/pull/4065#discussion_r4450015293
##########
fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java:
##########
@@ -349,7 +353,8 @@ static KvTablet create(
@Nullable Runnable flushCompleteListener,
AutoIncrementManager autoIncrementManager,
Clock clock,
- TableConfig tableConfig)
+ TableConfig tableConfig,
+ BooleanSupplier batchOldValueLookupEnabled)
Review Comment:
Opened #5016 to track simplifying `KvTablet` construction separately: group related required inputs, consolidate the production creation path, and keep test shortcuts in test utilities or non-public `@VisibleForTesting` helpers. The issue also preserves the requirement that missing inputs must not silently fall back to Builder defaults.
--
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] [kv] Batch old value lookups in write path [fluss]
Posted by "platinumhamburg (via GitHub) via issues" <is...@fluss.apache.org>.
platinumhamburg commented on PR #4065:
URL: https://github.com/apache/fluss/pull/4065#issuecomment-6600556079
Thanks @polyzos, agreed. Addressed in [2691eb962f](https://github.com/apache/fluss/commit/2691eb962f53d3a8012a660959c57706f53a5fc9).
This branch now includes current main and the `11.8.1-fluss-5` upgrade from #5015. The batched-path counter assertions now expect zero extra Get calls.
The existing real-SST corruption test in `KvTabletOldValueTest` now passes with the native fix. It checks that the write throws, the WAL offset stays unchanged, and already-applied buffered mutations are rolled back.
I've also updated the PR description to explain the native read-error dependency and current test coverage, and moved `kvConfig.update(newConfig)` to a single call at the start of `reconfigure()`.
Validation on JDK 11: 148 tests passed, including Checkstyle, Spotless and RAT. CI has been triggered for the updated head.
--
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] [kv] Batch old value lookups in write path [fluss]
Posted by "wuchong (via GitHub) via issues" <is...@fluss.apache.org>.
wuchong commented on code in PR #4065:
URL: https://github.com/apache/fluss/pull/4065#discussion_r4444598123
##########
fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java:
##########
@@ -225,7 +226,8 @@ private KvTablet(
@Nullable RowTtlTimestampProvider rowTtlTimestampProvider,
Clock clock,
boolean rowTtlEnabled,
- boolean skipUnchangedUpsert) {
+ boolean skipUnchangedUpsert,
+ BooleanSupplier batchOldValueLookupEnabled) {
Review Comment:
Could we replace this per-option `BooleanSupplier` plumbing with a shared `KvDynamicConfig`?
`batchOldValueLookupEnabled` now travels from `KvManager` through several `KvTablet.create` overloads and the tablet constructor into `KvWriteProcessor`. This adds another parameter to already long signatures and requires separate handling for shared dynamic values versus a fixed configuration snapshot (`null` plus a fallback lambda). Repeating this for each dynamic option would keep expanding the construction chain.
Instead, `KvManager` could own one `KvConfig`, initialize it from `Configuration`, and update it in `reconfigure()`. Existing, newly created, and recovered tablets and their processors would share the same non-null instance. This flag only needs a `volatile boolean` behind a typed getter; reading it once per batch would preserve the current behavior. Standalone tests could create their own instance from test configuration, removing the nullable-supplier fallback.
The broader name `KvConfig` also leaves room for dynamically updatable read settings, so future read/write options can be added without another constructor parameter for each setting.
##########
fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java:
##########
@@ -349,7 +353,8 @@ static KvTablet create(
@Nullable Runnable flushCompleteListener,
AutoIncrementManager autoIncrementManager,
Clock clock,
- TableConfig tableConfig)
+ TableConfig tableConfig,
+ BooleanSupplier batchOldValueLookupEnabled)
Review Comment:
Could you open a follow-up issue to simplify `KvTablet` construction? This can be handled separately from this PR.
The constructor currently has 28 parameters, and there are five `create` overloads. That makes the required dependencies difficult to audit and makes it easy for a caller to choose an overload with unintended behavior.
For the follow-up, I would suggest:
- Group related required parameters into cohesive parameter objects or abstractions, while validating that all mandatory inputs are supplied. For example, `KvConfig` can carry the dynamic settings together instead of adding another parameter throughout the construction chain for each setting.
- Consolidate the 5 `create` overloads into one canonical production creation method, so callers cannot accidentally choose the wrong variant.
- Keep test shortcuts in test utility methods. If production-side helpers are necessary for testing, provide only a limited set of non-public methods annotated with `@VisibleForTesting`, rather than additional public creation overloads.
I would avoid replacing `create` with a Builder here: these inputs are required, and optional setters or defaults can allow an omitted input to silently select incorrect default behavior. Grouping the parameters should preserve their required nature.
--
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] [kv] Batch old value lookups in write path [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg commented on code in PR #4065:
URL: https://github.com/apache/fluss/pull/4065#discussion_r4411112391
##########
fluss-server/src/main/java/org/apache/fluss/server/kv/KvStateAccessor.java:
##########
@@ -109,6 +114,35 @@ public BinaryValue lookup(
return localResult.isPresent() ? valueDecoder.decodeValue(localResult.value()) : null;
}
+ /** Returns whether a buffered value or delete already determines the local state. */
+ boolean containsBufferedKey(Key key) {
+ return preWriteBuffer.get(key) != null;
+ }
+
+ /**
+ * Loads distinct RocksDB misses for a write batch. Buffered values and deletes are excluded.
+ * Callers must hold the KV write lock through prefetch and application, and recheck the buffer
+ * before using a saved result so earlier mutations in the batch remain visible.
+ */
+ Map<Key, byte[]> prefetchRocksDBValues(Set<Key> keys) throws IOException {
+ List<Key> missingKeys = new ArrayList<>();
+ List<byte[]> encodedKeys = new ArrayList<>();
+ for (Key key : keys) {
+ if (preWriteBuffer.get(key) == null) {
+ missingKeys.add(key);
+ encodedKeys.add(key.get());
+ }
+ }
+ Map<Key, byte[]> values = new HashMap<>();
Review Comment:
Agreed that batching retains the fetched old values until the batch finishes, adding temporary memory compared with individual Gets. Partial updates already accumulate complete new rows, but retaining all fetched old rows is additional overhead introduced here.
Added `kv.write.batch-old-value-lookup.enabled` in 49b304c41, enabled by default and dynamically configurable at cluster scope. Disabling it restores individual Gets for subsequent write batches on existing and newly created tablets.
This provides an operational fallback; it does not bound memory while batching is enabled. Splitting MultiGet calls while retaining all results would not solve that either. A bounded-window implementation would require a separate performance and complexity tradeoff, so I would keep it outside this 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] [kv] Batch old value lookups in write path [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg commented on code in PR #4065:
URL: https://github.com/apache/fluss/pull/4065#discussion_r4411109771
##########
fluss-server/src/main/java/org/apache/fluss/server/kv/KvWriteProcessor.java:
##########
@@ -344,11 +378,30 @@ private void processKvRecords(
logOffset,
stateAccessor,
keyBytes,
- historicalValueLookup);
+ previousValueLookup);
}
}
}
+ private Map<KvPreWriteBuffer.Key, byte[]> prefetchPreviousValues(
+ KvRecordBatch kvRecords,
+ RowMerger currentMerger,
+ AutoIncrementUpdater autoIncrementUpdater,
+ KvStateAccessor stateAccessor)
+ throws Exception {
+ Set<KvPreWriteBuffer.Key> keys = new LinkedHashSet<>();
+ KvRecordBatch.ReadContext readContext =
Review Comment:
The extra scan comes from starting prefetch lazily, only when a record actually requires an old value and misses the buffer. This preserves the single-pass path for full-row WAL writes and batches served entirely by the buffer.
Sharing the existing ReadContext could avoid a small amount of allocation, but would not eliminate the second scan or row decoding. I would leave this as a possible follow-up rather than change the processing flow in this 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] [kv] Batch old value lookups in write path [fluss]
Posted by "polyzos (via GitHub) via issues" <is...@fluss.apache.org>.
polyzos commented on PR #4065:
URL: https://github.com/apache/fluss/pull/4065#issuecomment-6598972123
Thanks @platinumhamburg š
One thing before we merge: #4488 was reverted in #4951, so on main `RocksDBKv.multiGet` again returns `null` for a key whose read failed. With this PR, the normal write path moves from `get()` (which throws on a read error) to `multiGet`, and a prefetched `null` is treated as a missing row. A corrupted block would then turn an update into an insert and silently drop a delete. Main is still on `11.8.1-fluss-3`, which doesn't have the fix.
So I'd suggest we merge this after #5014 (upgrade to `11.8.1-fluss-5`) lands, ideally with a regression test that corrupts a real SST and asserts `multiGet` throws. If you'd rather not wait, we could default `kv.write.batch-old-value-lookup.enabled` to `false` here and flip it to `true` in the upgrade PR.
Two small things:
- The PR description is a bit stale: it still says null MultiGet results are rechecked with `Get` and mentions a read-error propagation test. Could you update it to say the read-error safety depends on #5014, and that the tests are now in `KvTabletOldValueTest`?
- Nit: in `KvManager.reconfigure()`, `kvConfig.update(newConfig)` is called in both branches. A single call at the top of the method would be simpler.
Otherwise LGTM š
--
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] [kv] Batch old value lookups in write path [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #4065:
URL: https://github.com/apache/fluss/pull/4065#discussion_r4357021531
##########
fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java:
##########
@@ -2356,8 +2469,9 @@ void testRocksDBMetrics() throws Exception {
}
List<ByteArraySlice> readValues = kvTablet.multiGet(keysToRead);
assertThat(readValues).hasSize(100);
+ assertThat(kvTablet.getRocksDBKv().get("key".getBytes())).isEqualTo("value".getBytes());
Review Comment:
Using `String#getBytes()` relies on the platform default charset, which can make tests platform-dependent. Prefer specifying the charset explicitly (e.g., `getBytes(StandardCharsets.UTF_8)`) for both key and value byte arrays.
##########
fluss-server/src/main/java/org/apache/fluss/server/kv/KvWriteProcessor.java:
##########
@@ -315,12 +324,36 @@ private void processKvRecords(
KvRecordBatch.ReadContext readContext =
KvRecordReadContext.createReadContext(kvFormat, schemaGetter);
+ boolean prefetchEnabled =
+ originalPartitionName == null
+ && kvRecords.getRecordCount() >= MIN_BATCH_SIZE_FOR_PREFETCH;
+ Function<byte[], BinaryValue> previousValueLookup =
+ historicalValueLookup == null ? null : historicalValueLookup::lookup;
for (KvRecord kvRecord : kvRecords.records(readContext)) {
byte[] keyBytes = BytesUtils.toArray(kvRecord.getKey());
KvPreWriteBuffer.Key key = stateAccessor.encodeKey(keyBytes, originalPartitionName);
BinaryRow row = kvRecord.getRow();
BinaryValue currentValue = row == null ? null : new BinaryValue(schemaIdOfNewData, row);
+ if (prefetchEnabled
+ && previousValueLookup == null
+ && (currentValue == null
+ ? !shouldIgnoreDeletion(currentMerger)
+ : !canSkipOldValueLookup(currentMerger, autoIncrementUpdater))
+ && !stateAccessor.containsBufferedKey(key)) {
+ // Start prefetch only on a required buffer miss. Full-row WAL batches and batches
+ // served entirely by the buffer retain their single-pass processing.
+ Map<KvPreWriteBuffer.Key, byte[]> savedValues =
+ prefetchPreviousValues(
+ kvRecords, currentMerger, autoIncrementUpdater, stateAccessor);
+ previousValueLookup =
+ primaryKey -> {
+ KvPreWriteBuffer.Key savedKey = KvPreWriteBuffer.Key.of(primaryKey);
Review Comment:
`savedValues` is keyed by `KvPreWriteBuffer.Key` instances produced via `stateAccessor.encodeKey(...)`, but the lookup uses `KvPreWriteBuffer.Key.of(primaryKey)`. If `encodeKey` applies any encoding/namespace beyond simply wrapping `primaryKey`, this will silently miss prefetched entries and fall back to treating existing rows as absent. Use the same key encoding for lookup as for population (e.g., compute the key via `stateAccessor.encodeKey(primaryKey, null)` in the lambda) to guarantee consistency.
##########
fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java:
##########
@@ -905,6 +910,110 @@ void testPartialUpdateFirstInsertThenUpdate() throws Exception {
checkEqual(actualLogRecords, expectedLogs, rowType);
}
+ @Test
+ void testBatchOldValuesPreserveOrderedChangelogAndBufferedTombstones() throws Exception {
+ initLogTabletAndKvTablet(DATA1_SCHEMA_PK, new HashMap<>());
+ kvTablet.putAsLeader(
+ kvRecordBatchFactory.ofRecords(
+ kvRecordFactory.ofRecord("k1", new Object[] {1, "v1"}),
+ kvRecordFactory.ofRecord("k2", new Object[] {2, "v2"})),
+ null);
+ flushAndWait(kvTablet, Long.MAX_VALUE);
+ kvTablet.putAsLeader(
+ kvRecordBatchFactory.ofRecords(
+ kvRecordFactory.ofRecord("k2", null),
+ kvRecordFactory.ofRecord("k3", new Object[] {3, "v3"})),
+ null);
+ long offset = logTablet.localLogEndOffset();
+ Statistics statistics = kvTablet.getRocksDBKv().getStatistics();
+ long gets = statistics.getTickerCount(TickerType.NUMBER_KEYS_READ);
+ long calls = statistics.getTickerCount(TickerType.NUMBER_MULTIGET_CALLS);
+ long keys = statistics.getTickerCount(TickerType.NUMBER_MULTIGET_KEYS_READ);
+ kvTablet.putAsLeader(
+ kvRecordBatchFactory.ofRecords(
+ kvRecordFactory.ofRecord("k1", new Object[] {1, "v11"}),
+ kvRecordFactory.ofRecord("k1", null),
+ kvRecordFactory.ofRecord("k1", new Object[] {1, "v12"}),
+ kvRecordFactory.ofRecord("missing", null),
+ kvRecordFactory.ofRecord("missing", null),
+ kvRecordFactory.ofRecord("k2", new Object[] {2, "v22"}),
+ kvRecordFactory.ofRecord("k3", new Object[] {3, "v33"}),
+ kvRecordFactory.ofRecord("k3", null)),
+ null);
+ assertThat(statistics.getTickerCount(TickerType.NUMBER_MULTIGET_CALLS) - calls)
+ .isEqualTo(1);
+ assertThat(statistics.getTickerCount(TickerType.NUMBER_MULTIGET_KEYS_READ) - keys)
+ .isEqualTo(2);
+ assertThat(statistics.getTickerCount(TickerType.NUMBER_KEYS_READ) - gets).isZero();
+ checkEqual(
+ readLogRecords(offset),
+ Collections.singletonList(
+ logRecords(
+ offset,
+ Arrays.asList(
+ ChangeType.UPDATE_BEFORE,
+ ChangeType.UPDATE_AFTER,
+ ChangeType.DELETE,
+ ChangeType.INSERT,
+ ChangeType.INSERT,
+ ChangeType.UPDATE_BEFORE,
+ ChangeType.UPDATE_AFTER,
+ ChangeType.DELETE),
+ Arrays.asList(
+ new Object[] {1, "v1"},
+ new Object[] {1, "v11"},
+ new Object[] {1, "v11"},
+ new Object[] {1, "v12"},
+ new Object[] {2, "v22"},
+ new Object[] {3, "v3"},
+ new Object[] {3, "v33"},
+ new Object[] {3, "v33"}))));
+ }
+
+ @ParameterizedTest
+ @CsvSource({"7, 1, 0", "8, 0, 1"})
+ void testWalBatchLookupAfterStreamingUpserts(
+ int batchSize, int expectedGets, int expectedMultiGetCalls) throws Exception {
+ Map<String, String> config = new HashMap<>();
+ config.put("table.changelog.image", "WAL");
+ initLogTabletAndKvTablet(DATA1_SCHEMA_PK, config);
+ kvTablet.putAsLeader(
+ kvRecordBatchFactory.ofRecords(
+ kvRecordFactory.ofRecord("new0", new Object[] {0, "old"}),
+ kvRecordFactory.ofRecord("k2", new Object[] {2, "v2"})),
+ null);
+ flushAndWait(kvTablet, Long.MAX_VALUE);
+ List<KvRecord> records = new ArrayList<>();
+ List<Object[]> expectedRows = new ArrayList<>();
+ for (int i = 0; i < batchSize - 2; i++) {
+ records.add(kvRecordFactory.ofRecord("new" + i, new Object[] {i, "new"}));
+ expectedRows.add(new Object[] {i, "new"});
+ }
+ records.add(kvRecordFactory.ofRecord("new0", null));
+ records.add(kvRecordFactory.ofRecord("k2", null));
+ expectedRows.add(new Object[] {0, "new"});
+ expectedRows.add(new Object[] {2, "v2"});
+ List<ChangeType> expectedChanges =
+ new ArrayList<>(Collections.nCopies(batchSize - 2, ChangeType.UPDATE_AFTER));
+ expectedChanges.add(ChangeType.DELETE);
+ expectedChanges.add(ChangeType.DELETE);
+ long offset = logTablet.localLogEndOffset();
+ Statistics statistics = kvTablet.getRocksDBKv().getStatistics();
+ long gets = statistics.getTickerCount(TickerType.NUMBER_KEYS_READ);
+ long calls = statistics.getTickerCount(TickerType.NUMBER_MULTIGET_CALLS);
+ long keys = statistics.getTickerCount(TickerType.NUMBER_MULTIGET_KEYS_READ);
+ kvTablet.putAsLeader(kvRecordBatchFactory.ofRecords(records), null);
+ assertThat(statistics.getTickerCount(TickerType.NUMBER_KEYS_READ) - gets)
+ .isEqualTo(expectedGets);
+ assertThat(statistics.getTickerCount(TickerType.NUMBER_MULTIGET_CALLS) - calls)
+ .isEqualTo(expectedMultiGetCalls);
+ assertThat(statistics.getTickerCount(TickerType.NUMBER_MULTIGET_KEYS_READ) - keys)
+ .isEqualTo(expectedMultiGetCalls);
Review Comment:
`NUMBER_MULTIGET_KEYS_READ` measures keys read via MultiGet, not the number of MultiGet calls. Asserting it equals `expectedMultiGetCalls` is only correct when each MultiGet reads exactly one key, making the test brittle if future batching reads multiple distinct keys. Consider introducing a separate `expectedMultiGetKeysRead` parameter (or deriving it from the test inputs) and asserting against that instead.
--
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] [kv] Batch old value lookups in write path [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg commented on code in PR #4065:
URL: https://github.com/apache/fluss/pull/4065#discussion_r4357181110
##########
fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java:
##########
@@ -905,6 +910,110 @@ void testPartialUpdateFirstInsertThenUpdate() throws Exception {
checkEqual(actualLogRecords, expectedLogs, rowType);
}
+ @Test
+ void testBatchOldValuesPreserveOrderedChangelogAndBufferedTombstones() throws Exception {
+ initLogTabletAndKvTablet(DATA1_SCHEMA_PK, new HashMap<>());
+ kvTablet.putAsLeader(
+ kvRecordBatchFactory.ofRecords(
+ kvRecordFactory.ofRecord("k1", new Object[] {1, "v1"}),
+ kvRecordFactory.ofRecord("k2", new Object[] {2, "v2"})),
+ null);
+ flushAndWait(kvTablet, Long.MAX_VALUE);
+ kvTablet.putAsLeader(
+ kvRecordBatchFactory.ofRecords(
+ kvRecordFactory.ofRecord("k2", null),
+ kvRecordFactory.ofRecord("k3", new Object[] {3, "v3"})),
+ null);
+ long offset = logTablet.localLogEndOffset();
+ Statistics statistics = kvTablet.getRocksDBKv().getStatistics();
+ long gets = statistics.getTickerCount(TickerType.NUMBER_KEYS_READ);
+ long calls = statistics.getTickerCount(TickerType.NUMBER_MULTIGET_CALLS);
+ long keys = statistics.getTickerCount(TickerType.NUMBER_MULTIGET_KEYS_READ);
+ kvTablet.putAsLeader(
+ kvRecordBatchFactory.ofRecords(
+ kvRecordFactory.ofRecord("k1", new Object[] {1, "v11"}),
+ kvRecordFactory.ofRecord("k1", null),
+ kvRecordFactory.ofRecord("k1", new Object[] {1, "v12"}),
+ kvRecordFactory.ofRecord("missing", null),
+ kvRecordFactory.ofRecord("missing", null),
+ kvRecordFactory.ofRecord("k2", new Object[] {2, "v22"}),
+ kvRecordFactory.ofRecord("k3", new Object[] {3, "v33"}),
+ kvRecordFactory.ofRecord("k3", null)),
+ null);
+ assertThat(statistics.getTickerCount(TickerType.NUMBER_MULTIGET_CALLS) - calls)
+ .isEqualTo(1);
+ assertThat(statistics.getTickerCount(TickerType.NUMBER_MULTIGET_KEYS_READ) - keys)
+ .isEqualTo(2);
+ assertThat(statistics.getTickerCount(TickerType.NUMBER_KEYS_READ) - gets).isZero();
+ checkEqual(
+ readLogRecords(offset),
+ Collections.singletonList(
+ logRecords(
+ offset,
+ Arrays.asList(
+ ChangeType.UPDATE_BEFORE,
+ ChangeType.UPDATE_AFTER,
+ ChangeType.DELETE,
+ ChangeType.INSERT,
+ ChangeType.INSERT,
+ ChangeType.UPDATE_BEFORE,
+ ChangeType.UPDATE_AFTER,
+ ChangeType.DELETE),
+ Arrays.asList(
+ new Object[] {1, "v1"},
+ new Object[] {1, "v11"},
+ new Object[] {1, "v11"},
+ new Object[] {1, "v12"},
+ new Object[] {2, "v22"},
+ new Object[] {3, "v3"},
+ new Object[] {3, "v33"},
+ new Object[] {3, "v33"}))));
+ }
+
+ @ParameterizedTest
+ @CsvSource({"7, 1, 0", "8, 0, 1"})
+ void testWalBatchLookupAfterStreamingUpserts(
+ int batchSize, int expectedGets, int expectedMultiGetCalls) throws Exception {
+ Map<String, String> config = new HashMap<>();
+ config.put("table.changelog.image", "WAL");
+ initLogTabletAndKvTablet(DATA1_SCHEMA_PK, config);
+ kvTablet.putAsLeader(
+ kvRecordBatchFactory.ofRecords(
+ kvRecordFactory.ofRecord("new0", new Object[] {0, "old"}),
+ kvRecordFactory.ofRecord("k2", new Object[] {2, "v2"})),
+ null);
+ flushAndWait(kvTablet, Long.MAX_VALUE);
+ List<KvRecord> records = new ArrayList<>();
+ List<Object[]> expectedRows = new ArrayList<>();
+ for (int i = 0; i < batchSize - 2; i++) {
+ records.add(kvRecordFactory.ofRecord("new" + i, new Object[] {i, "new"}));
+ expectedRows.add(new Object[] {i, "new"});
+ }
+ records.add(kvRecordFactory.ofRecord("new0", null));
+ records.add(kvRecordFactory.ofRecord("k2", null));
+ expectedRows.add(new Object[] {0, "new"});
+ expectedRows.add(new Object[] {2, "v2"});
+ List<ChangeType> expectedChanges =
+ new ArrayList<>(Collections.nCopies(batchSize - 2, ChangeType.UPDATE_AFTER));
+ expectedChanges.add(ChangeType.DELETE);
+ expectedChanges.add(ChangeType.DELETE);
+ long offset = logTablet.localLogEndOffset();
+ Statistics statistics = kvTablet.getRocksDBKv().getStatistics();
+ long gets = statistics.getTickerCount(TickerType.NUMBER_KEYS_READ);
+ long calls = statistics.getTickerCount(TickerType.NUMBER_MULTIGET_CALLS);
+ long keys = statistics.getTickerCount(TickerType.NUMBER_MULTIGET_KEYS_READ);
+ kvTablet.putAsLeader(kvRecordBatchFactory.ofRecords(records), null);
+ assertThat(statistics.getTickerCount(TickerType.NUMBER_KEYS_READ) - gets)
+ .isEqualTo(expectedGets);
+ assertThat(statistics.getTickerCount(TickerType.NUMBER_MULTIGET_CALLS) - calls)
+ .isEqualTo(expectedMultiGetCalls);
+ assertThat(statistics.getTickerCount(TickerType.NUMBER_MULTIGET_KEYS_READ) - keys)
+ .isEqualTo(expectedMultiGetCalls);
Review Comment:
This case reads one RocksDB key, so both counts are one. testBatchOldValuesPreserveOrderedChangelogAndBufferedTombstones separately asserts one MultiGet call reading two keys.
--
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] [kv] Batch old value lookups in write path [fluss]
Posted by "polyzos (via GitHub)" <gi...@apache.org>.
polyzos commented on PR #4065:
URL: https://github.com/apache/fluss/pull/4065#issuecomment-6492004336
Thanks @platinumhamburg, the design looks good š
I left a few comments, so before merging my suggestions would be - if they make sense:
- Add a MultiGet latency metric and document it.
- Add a checkState for keys missing from the prefetched map, and use one requiresPreviousValue helper.
- Replace the new Mockito usage, and add tests for a mid-batch failure and for skip-unchanged.
- Rebase on main
--
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] [kv] Batch old value lookups in write path [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg commented on code in PR #4065:
URL: https://github.com/apache/fluss/pull/4065#discussion_r4411101467
##########
fluss-server/src/main/java/org/apache/fluss/server/kv/KvStateAccessor.java:
##########
@@ -109,6 +114,35 @@ public BinaryValue lookup(
return localResult.isPresent() ? valueDecoder.decodeValue(localResult.value()) : null;
}
+ /** Returns whether a buffered value or delete already determines the local state. */
+ boolean containsBufferedKey(Key key) {
+ return preWriteBuffer.get(key) != null;
+ }
+
+ /**
+ * Loads distinct RocksDB misses for a write batch. Buffered values and deletes are excluded.
+ * Callers must hold the KV write lock through prefetch and application, and recheck the buffer
+ * before using a saved result so earlier mutations in the batch remain visible.
+ */
+ Map<Key, byte[]> prefetchRocksDBValues(Set<Key> keys) throws IOException {
+ List<Key> missingKeys = new ArrayList<>();
+ List<byte[]> encodedKeys = new ArrayList<>();
+ for (Key key : keys) {
+ if (preWriteBuffer.get(key) == null) {
+ missingKeys.add(key);
+ encodedKeys.add(key.get());
+ }
+ }
+ Map<Key, byte[]> values = new HashMap<>();
+ if (!missingKeys.isEmpty()) {
+ List<byte[]> results = rocksDBKv.multiGet(encodedKeys);
+ for (int i = 0; i < missingKeys.size(); i++) {
+ values.put(missingKeys.get(i), results.get(i));
Review Comment:
Thanks for pointing this out. This is a known RocksJava issue tracked in [#12741](https://github.com/meta/rocksdb/issues/12741), with [#12766](https://github.com/meta/rocksdb/pull/12766) proposing to throw `RocksDBException` for corruption. The current Fluss wrapper already propagates that exception as `IOException`, so no interface change is needed here.
We plan to address error propagation separately in the Fluss RocksDB fork, including checking errors beyond corruption. I would prefer to keep that fix separate from this batching change rather than add point-read rechecks here.
--
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] [kv] Batch old value lookups in write path [fluss]
Posted by "platinumhamburg (via GitHub)" <gi...@apache.org>.
platinumhamburg commented on PR #4065:
URL: https://github.com/apache/fluss/pull/4065#issuecomment-6496609904
Thanks for the review, @polyzos. Holding the old values temporarily is a trade-off of batching: it substantially reduces repeated RocksDB/JNI calls. These values are held only while processing the write batch, so we expect the memory overhead to be manageable in typical workloads. I've also added a cluster-level switch to disable the MultiGet path dynamically when needed.
--
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] [kv] Batch old value lookups in write path [fluss]
Posted by "platinumhamburg (via GitHub) via issues" <is...@fluss.apache.org>.
platinumhamburg commented on code in PR #4065:
URL: https://github.com/apache/fluss/pull/4065#discussion_r4449955732
##########
fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java:
##########
@@ -225,7 +226,8 @@ private KvTablet(
@Nullable RowTtlTimestampProvider rowTtlTimestampProvider,
Clock clock,
boolean rowTtlEnabled,
- boolean skipUnchangedUpsert) {
+ boolean skipUnchangedUpsert,
+ BooleanSupplier batchOldValueLookupEnabled) {
Review Comment:
Replaced the per-option supplier with a shared `KvConfig` in ee42f8f83. `KvManager` initializes it from the server configuration and updates it in `reconfigure()`; existing, newly created, and recovered tablets pass the same instance to their write processors. The flag is a `volatile boolean` read once per batch. Standalone tablet creation builds its own `KvConfig` from the supplied configuration, removing the nullable-supplier fallback.
Validation on JDK 11: 147 tests passed across the existing KV manager, old-value lookup, tablet, merge, schema evolution, historical lookup, and dynamic configuration suites, including Checkstyle, Spotless, and license checks.
--
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] [kv] Batch old value lookups in write path [fluss]
Posted by "platinumhamburg (via GitHub) via issues" <is...@fluss.apache.org>.
platinumhamburg closed pull request #4065: [kv] Batch old value lookups in write path
URL: https://github.com/apache/fluss/pull/4065
--
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] [kv] Batch old value lookups in write path [fluss]
Posted by "polyzos (via GitHub)" <gi...@apache.org>.
polyzos commented on PR #4065:
URL: https://github.com/apache/fluss/pull/4065#issuecomment-6501194616
Thanks @platinumhamburg! I was about to merge this, but then I realized it would be better to merge #4488 first and rebase. Without it, multiGet reports a failed read as a missing key, so the batched path could turn an update into an insert or drop a delete. So I merged, and pushed a commit here ontop to rebase and do some minor changes - like completely removing mockito:
PTAL and let me know if you are ok with these - otherwise i can revert them. Once CI is green we can merge š
--
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