[client] Support prefix lookups with multiple keys by polyzos · Pull Request #4981 · apache/fluss

polyzos · 2026-11-01T18:05:13Z

A prefix lookup only accepts exactly the bucket key as its lookup columns. For a table with
primary key (sale_id, provider_id, id) and bucket key sale_id, a lookup by
(sale_id, provider_id) is rejected by the client. The Flink connector, used by lookup joins and
delta joins, falls back to looking up by sale_id alone: the server returns the rows of every
provider_id for that sale_id, and the client filters them.

This is to make it work with Iceberg, which allows only one bucket key column, but it affects every primary
key table. With Lance, Hudi, Paimon or no lake format, the workaround of widening the bucket key to
(sale_id, provider_id) spreads a sale's rows across buckets and rules out lookups by sale_id
alone.

Solution
Allow the lookup columns to be the bucket key followed by the primary key columns that directly
follow it. For example, with primary key (a, b, c, d) and bucket key a, prefix lookups by (a),
(a, b) and (a, b, c) are allowed; a lookup by all of (a, b, c, d) remains a primary key
lookup. Lookups are still routed by the bucket key, and the server returns only the rows matching
all lookup columns, except on legacy tables (kv format version 1) with Paimon key encoding, where
it returns the bucket key's rows and the client filters them. The Flink connector passes the join
keys through, so lookup joins and delta joins fetch only the rows they need.

platinumhamburg

This PR makes a lot of sense, and it’s an optimization I’ve been particularly interested in. Using a longer primary-key prefix should substantially reduce unnecessary server-side reads, improve lookup selectivity, and reduce cache pollution. Thanks for working on this—I’d be very happy to help move it toward merging.

platinumhamburg · 2026-11-02T02:31:09Z

+                        && lookupKeySet.contains(encodedPrimaryKeys[prefixLength])) {
+                    prefixLength++;
+                }
+                int[] prefixKeys = Arrays.copyOf(encodedPrimaryKeys, prefixLength);


Could we gate this extension on the table’s key encoding? For legacy Paimon KV v1 tables with PK (a BIGINT, b STRING, c BIGINT) and bucket key a, extending the lookup to (a,b) silently misses rows when b exceeds 7 bytes: Paimon’s variable-length offsets depend on the encoded field count, so the lookup bytes no longer match the stored key prefix.

I reproduced an empty result here, while the base normalizer returns the expected row. Could we preserve bucket-only lookup with client-side filtering for incompatible encodings and add a regression test?

platinumhamburg · 2026-11-02T02:31:11Z

+     * columns must be the bucket key, optionally followed by the next primary key columns, but not
+     * the whole primary key. For example, with the primary key [a, b, c, d] and the bucket key [a],
+     * the lookup columns can be [a], [a, b] or [a, b, c]. Lookups are routed by the bucket key and
+     * return only the rows matching all lookup columns.


Could we also update website/docs/engine-flink/lookups.md with an example showing a lookup prefix longer than the bucket key? This would help users discover the improvement and understand the contiguous-prefix requirement.

SteNicholas

@polyzos, thanks for the simpler solution. I left one comment for an additional legacy Paimon KV v1 regression test failed with an empty extended-prefix lookup result. PTAL.

SteNicholas · 2026-11-02T05:21:49Z

+                while (prefixLength < encodedPrimaryKeys.length - 1
+                        && lookupKeySet.contains(encodedPrimaryKeys[prefixLength])) {
+                    prefixLength++;
+                }
+                int[] prefixKeys = Arrays.copyOf(encodedPrimaryKeys, prefixLength);


Gate longer prefixes on key-encoding compatibility. For legacy Paimon KV v1 tables with primary key (a BIGINT, b STRING, c BIGINT) and bucket key a, this extension silently loses matches when b exceeds seven bytes. Paimon's variable-length offsets depend on the encoded field count, so encoding (a,b) does not produce a prefix of the stored (a,b,c) key. Previously, Flink looked up a and filtered b locally.

I independently reproduced the issue raised above with a client/server regression test: after writing (1, 'abcdefgh', 3, 'v'), lookup by a=1 returned the row, while lookup by (a,b)=(1,'abcdefgh') returned []. The encoder probe also showed the string offset changing from 32 in the stored key to 24 in the lookup prefix.

Please preserve bucket-only lookup with client-side filtering for incompatible encodings, guard explicit client prefix lookups as well, and add regression coverage for legacy KV v1 tables.

platinumhamburg · 2026-11-02T07:13:15Z


-    private void validatePrefixLookup(TableInfo tableInfo, List<String> lookupColumns) {
+    /** Validates the lookup columns and returns them without the partition fields. */
+    private List<String> validatePrefixLookup(TableInfo tableInfo, List<String> lookupColumns) {


Could we rename this to validateAndGetPrefixKeys? It now returns the lookup columns with partition fields removed, so the name would make that responsibility clearer to callers.

SteNicholas · 2026-11-02T07:39:35Z

+        Object[] expected = new Object[clientFilterKeyGetters.length];
+        for (int i = 0; i < expected.length; i++) {
+            expected[i] = clientFilterKeyGetters[i].getFieldOrNull(prefixKey);


Snapshot mutable filter values before retaining them asynchronously: These getters can return BinaryString values backed by the caller's reusable row buffer; putting them in Object[] does not copy their contents. On a legacy Paimon KV v1 table with primary key (a BIGINT, b STRING, c BIGINT) and bucket key a, reusing that buffer after lookup() returns but before its future completes changes the predicate and can return rows for a different key.

I reproduced this on the current head with a controlled pending response: request (1, 'abcdefgh') using an IndexedRowEncoder, reuse the encoder for (1, 'ijklmnop'), then complete the bucket response containing both rows. The first lookup incorrectly returns the 'ijklmnop' row. The same probe passes when the captured string is copied.

Could we copy mutable filter values (BinaryString and byte arrays) when capturing them, and add a regression test that reuses the input buffer while the response is pending?

fresh-borzoni

@polyzos Thank you for the PR, LGTM overall, left some comments, PTAL

fresh-borzoni · 2026-11-02T11:40:18Z

+                        "SELECT a, b, h.address FROM src JOIN %s FOR SYSTEM_TIME AS OF src.proc as h"
+                                + " ON src.b = h.name AND h.address = CONCAT('address', CAST(src.a AS STRING))",
+                        dim);
+        // the lookup keys sent to Fluss are (name, address), not just the bucket key


this test passes on main as well, checked locally.
lookup=[name=b, address=...] comes from Flink's lookup join, it's the join keys, not what is sent to Fluss
so it doesn't verify the change, on the other hand LookupNormalizerTest does

can we drop the explain check or at least fix the comment?

fresh-borzoni · 2026-11-02T11:41:11Z

+
+    @BeforeEach
+    void setup() throws Exception {
+        conn = ConnectionFactory.createConnection(FLUSS_CLUSTER_EXTENSION.getClientConfig());


nit: every lookup(...).get() here waits for client.lookup.batch-timeout (100ms) since nothing else fills the batch
locally it is ~12.7s -> ~2.5s per testPrefixLookup case with 1ms

Suggested change

conn = ConnectionFactory.createConnection(FLUSS_CLUSTER_EXTENSION.getClientConfig());

Configuration clientConf = FLUSS_CLUSTER_EXTENSION.getClientConfig();

clientConf.set(ConfigOptions.CLIENT_LOOKUP_BATCH_TIMEOUT, Duration.ofMillis(1));

conn = ConnectionFactory.createConnection(clientConf);

getClientConfig() returns a new Configuration on each call, so setting the timeout there is safe.

fresh-borzoni · 2026-11-02T11:41:38Z

+ON `o`.`o_custkey` = `s`.`cs_custkey` AND `o`.`o_orderstatus` = `s`.`cs_orderstatus`;
+```
+
+The join keys must follow the primary key order without gaps: a join on `cs_custkey` and `cs_date` looks up by `cs_custkey` only.


"must" reads like such a join is rejected, however it works, it just looks up by cs_custkey

Suggested change

The join keys must follow the primary key order without gaps: a join on `cs_custkey` and `cs_date` looks up by `cs_custkey` only.

Fluss looks up by the bucket key and the primary key columns that directly follow it among the join keys. The other join keys are filtered after the lookup, so a join on `cs_custkey` and `cs_date` looks up by `cs_custkey` only.

fresh-borzoni · 2026-11-02T11:42:02Z

- Use a primary key table as a dimension table, and the join condition must a prefix subset of the primary keys of the dimension table.
- The bucket key of Fluss dimension table need to set as the join key when creating Fluss table. 
+- Use a primary key table as a dimension table, whose bucket key is a prefix of its primary key.
+- The join keys must include the bucket key, and can also include the primary key columns that follow it, in order. See [Examples (Multiple Keys)](#examples-multiple-keys).


the order of conditions in ON doesn't matter (b = .. AND a = .. is looked up as (a, b)), so "in order" may confuse here

Suggested change

- The join keys must include the bucket key, and can also include the primary key columns that follow it, in order. See [Examples (Multiple Keys)](#examples-multiple-keys).

- The join keys must include the bucket key, and can also include the primary key columns that directly follow it. See [Examples (Multiple Keys)](#examples-multiple-keys).

fresh-borzoni · 2026-11-02T11:55:35Z


+        List<String> bucketKeys = tableInfo.getBucketKeys();
+        List<String> clientFilterKeys =
+                KeyEncoder.supportsMultiKeyPrefixLookup(tableInfo.getTableConfig())


I wonder whether this could be derived from the encoder instead of duplicating ofPrimaryKeyEncoder logic, something like:

KeyEncoder.ofPrimaryKeyEncoder( lookupRowType, bucketKeys, tableInfo.getTableConfig(), tableInfo.isDefaultBucketKey()) instanceof CompactedKeyEncoder

hope you get the idea, let me know if I need to elaborate

SteNicholas

@polyzos, I left two follow-up suggestions on deterministic regression coverage and documenting the legacy Paimon fallback for updates.

SteNicholas · 2026-11-02T12:56:26Z

+            CompletableFuture<LookupResult> result =
+                    lookuper.lookup(GenericRow.of(1L, BinaryString.fromBytes(b)));
+            // the caller reuses the row's buffer before the lookup completes
+            byte[] next = "klmnopqrst".getBytes(StandardCharsets.UTF_8);
+            System.arraycopy(next, 0, b, 0, b.length);


Control response completion in the buffer-reuse regression test. Calling lookup() and immediately overwriting b does not guarantee that the response is still pending. If the test thread is descheduled before the overwrite, the response and filtering can finish first, so this test can pass even if the defensive copy is removed. The comment's "before the lookup completes" assumption is therefore not enforced.

Could we use a controllable CompletableFuture or test gateway, mutate the backing buffer while the response is held, and only then complete the response? That would make the regression check deterministic. Please also consider covering a mutable byte[] filter value to exercise the other copy branch. The current string-copy implementation passed a separate controlled pending-response probe; this concern is about the regression coverage.

SteNicholas · 2026-11-02T12:56:26Z

+ON `o`.`o_custkey` = `s`.`cs_custkey` AND `o`.`o_orderstatus` = `s`.`cs_orderstatus`;
+```
+
+Fluss looks up by the bucket key and the primary key columns that directly follow it among the join keys. The other join keys are filtered after the lookup, so a join on `cs_custkey` and `cs_date` looks up by `cs_custkey` only.


Document the legacy Paimon filtering exception. For a Paimon KV v1 table with primary key (a, b, c) and bucket key a, PrefixKeyLookuper still sends only a to the server and filters b on the client when looking up (a, b). The longer logical lookup prefix is supported, but it does not narrow the server response on that compatibility path.

Could we add a short note here and qualify the PR description's claim that the server returns only rows matching all lookup columns? This would prevent users of legacy tables from expecting the same reduction in returned data as tables using compacted key encoding.

polyzos mentioned this pull request Nov 1, 2026

[lake/iceberg] Support composite bucket keys for Iceberg-tiered tables #4936

Open

2 tasks done

polyzos force-pushed the generalize-prefix-lookup branch from 0d0970a to b2d3db3 Compare November 1, 2026 18:23

[client] Support prefix lookups with multiple keys

5a2a492

polyzos force-pushed the generalize-prefix-lookup branch from b2d3db3 to 5a2a492 Compare November 1, 2026 18:26

polyzos changed the title ~~[client] Support prefix lookups longer than the bucket key~~ [client] Support prefix lookups with multiple keys Nov 1, 2026

polyzos requested review from SteNicholas, fresh-borzoni, luoyuxia, swuferhong and wuchong and removed request for SteNicholas, luoyuxia and swuferhong November 1, 2026 18:31

platinumhamburg requested changes Nov 2, 2026

View reviewed changes

SteNicholas reviewed Nov 2, 2026

View reviewed changes

[client] address comments

4081d94

platinumhamburg reviewed Nov 2, 2026

View reviewed changes

[client] Rename validatePrefixLookup to validateAndGetPrefixKeys

7bdbb37

SteNicholas reviewed Nov 2, 2026

View reviewed changes

[client] Copy client-side filter values before the lookup completes

7d28f51

fresh-borzoni reviewed Nov 2, 2026

View reviewed changes

[client] Derive client-side filtering from the key encoder

e40c66f

SteNicholas reviewed Nov 2, 2026

View reviewed changes

[docs] Note that legacy Paimon tables look up by the bucket key only

4155cf7

-        conn = ConnectionFactory.createConnection(FLUSS_CLUSTER_EXTENSION.getClientConfig());
+        Configuration clientConf = FLUSS_CLUSTER_EXTENSION.getClientConfig();
+        clientConf.set(ConfigOptions.CLIENT_LOOKUP_BATCH_TIMEOUT, Duration.ofMillis(1));
+        conn = ConnectionFactory.createConnection(clientConf);

	The join keys must follow the primary key order without gaps: a join on `cs_custkey` and `cs_date` looks up by `cs_custkey` only.
	Fluss looks up by the bucket key and the primary key columns that directly follow it among the join keys. The other join keys are filtered after the lookup, so a join on `cs_custkey` and `cs_date` looks up by `cs_custkey` only.

	- The join keys must include the bucket key, and can also include the primary key columns that follow it, in order. See [Examples (Multiple Keys)](#examples-multiple-keys).
	- The join keys must include the bucket key, and can also include the primary key columns that directly follow it. See [Examples (Multiple Keys)](#examples-multiple-keys).

Conversation

polyzos commented Nov 1, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

platinumhamburg left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SteNicholas left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SteNicholas Nov 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SteNicholas Nov 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

fresh-borzoni left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SteNicholas left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SteNicholas Nov 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SteNicholas Nov 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

polyzos commented Nov 1, 2026 •

edited

Loading

SteNicholas left a comment •

edited

Loading

SteNicholas Nov 2, 2026 •

edited

Loading

SteNicholas Nov 2, 2026 •

edited

Loading

SteNicholas left a comment •

edited

Loading

SteNicholas Nov 2, 2026 •

edited

Loading

SteNicholas Nov 2, 2026 •

edited

Loading