You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ab...@apache.org on 2026/10/18 21:54:54 UTC
(fluss) branch main updated: [python] Add prefix lookup example and documentation (#4632)
This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new d3e50c036 [python] Add prefix lookup example and documentation (#4632)
d3e50c036 is described below
commit d3e50c0362b17d1c3e81aae89a12361ddb7d64b5
Author: Jandy <35...@users.noreply.github.com>
AuthorDate: Sun Oct 18 22:54:50 2026 +0100
[python] Add prefix lookup example and documentation (#4632)
* [python] Add prefix lookup example and documentation
* [python] Align prefix lookup example with Rust and C++
* [docs] Align Python prefix lookup docs with Rust and C++
* [docs] List Python prefix lookup example in development guide
* [docs] Simplify Python prefix lookup validation wording
* [python] Align prefix lookup tests with C++ coverage
---
fluss-rust/bindings/python/DEVELOPMENT.md | 2 +
.../bindings/python/example/prefix_lookup.py | 140 +++++++++++++++++++++
fluss-rust/bindings/python/test/test_examples.py | 3 +
fluss-rust/bindings/python/test/test_kv_table.py | 73 ++++++++++-
website/docs/apis/python/example/prefix-lookup.md | 98 +++++++++++++++
5 files changed, 313 insertions(+), 3 deletions(-)
diff --git a/fluss-rust/bindings/python/DEVELOPMENT.md b/fluss-rust/bindings/python/DEVELOPMENT.md
index c756b796b..8ceae5225 100644
--- a/fluss-rust/bindings/python/DEVELOPMENT.md
+++ b/fluss-rust/bindings/python/DEVELOPMENT.md
@@ -71,6 +71,7 @@ cluster at `127.0.0.1:9123`; override with `FLUSS_BOOTSTRAP_SERVERS`.
```bash
uv run python example/log_table.py
uv run python example/pk_table.py
+uv run python example/prefix_lookup.py
uv run python example/complex_types.py
uv run python example/partitioned_table.py
uv run python example/partitioned_kv_table.py
@@ -130,6 +131,7 @@ bindings/python/
├── example/ # Standalone, CI-checked examples
│ ├── log_table.py
│ ├── pk_table.py
+│ ├── prefix_lookup.py
│ ├── complex_types.py
│ ├── partitioned_table.py
│ └── partitioned_kv_table.py
diff --git a/fluss-rust/bindings/python/example/prefix_lookup.py b/fluss-rust/bindings/python/example/prefix_lookup.py
new file mode 100644
index 000000000..099158dde
--- /dev/null
+++ b/fluss-rust/bindings/python/example/prefix_lookup.py
@@ -0,0 +1,140 @@
+# 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.
+
+"""Prefix lookup example: fetch all events for a user session without scanning.
+
+Run standalone against a local cluster:
+
+ python example/prefix_lookup.py
+
+Or point it at a specific cluster:
+
+ FLUSS_BOOTSTRAP_SERVERS=host:port python example/prefix_lookup.py
+"""
+
+import asyncio
+import os
+from typing import Optional
+
+import pyarrow as pa
+
+import fluss
+
+DEFAULT_BOOTSTRAP_SERVERS = "127.0.0.1:9123"
+
+
+async def main(bootstrap_servers: Optional[str] = None):
+ bootstrap_servers = bootstrap_servers or os.environ.get(
+ "FLUSS_BOOTSTRAP_SERVERS", DEFAULT_BOOTSTRAP_SERVERS
+ )
+ config = fluss.Config({"bootstrap.servers": bootstrap_servers})
+ conn = await fluss.FlussConnection.create(config)
+ try:
+ await _run(conn)
+ finally:
+ await conn.close()
+ print("\nConnection closed")
+
+
+async def _run(conn):
+ schema = fluss.Schema(
+ pa.schema(
+ [
+ pa.field("user_id", pa.int32()),
+ pa.field("session_id", pa.string()),
+ pa.field("event_seq", pa.int64()),
+ pa.field("event_data", pa.string()),
+ ]
+ ),
+ primary_keys=["user_id", "session_id", "event_seq"],
+ )
+ # The bucket key must be a strict prefix of the primary key.
+ descriptor = fluss.TableDescriptor(
+ schema, bucket_count=3, bucket_keys=["user_id", "session_id"]
+ )
+ admin = conn.get_admin()
+ table_path = fluss.TablePath("fluss", "example_prefix_lookup")
+ await admin.drop_table(table_path, ignore_if_not_exists=True)
+ await admin.create_table(table_path, descriptor, ignore_if_exists=True)
+ print(f"Created session events table: {table_path}")
+
+ try:
+ table = await conn.get_table(table_path)
+ events = [
+ {
+ "user_id": 1,
+ "session_id": "sess-a",
+ "event_seq": 1,
+ "event_data": "open",
+ },
+ {
+ "user_id": 1,
+ "session_id": "sess-a",
+ "event_seq": 2,
+ "event_data": "click",
+ },
+ {
+ "user_id": 1,
+ "session_id": "sess-a",
+ "event_seq": 3,
+ "event_data": "close",
+ },
+ {
+ "user_id": 1,
+ "session_id": "sess-b",
+ "event_seq": 1,
+ "event_data": "open",
+ },
+ {
+ "user_id": 2,
+ "session_id": "sess-a",
+ "event_seq": 1,
+ "event_data": "open",
+ },
+ ]
+ writer = table.new_upsert().create_writer()
+ for event in events:
+ writer.upsert(event)
+ # Wait for acknowledgment before reading the events back.
+ await writer.flush()
+
+ lookuper = (
+ table.new_lookup().lookup_by(["user_id", "session_id"]).create_lookuper()
+ )
+ result = await lookuper.lookup({"user_id": 1, "session_id": "sess-a"})
+ assert result == events[:3]
+ print(f"Events for user_id=1, session_id=sess-a: {result}")
+
+ # List and tuple keys follow the column order supplied to lookup_by().
+ result = await lookuper.lookup([1, "sess-b"])
+ assert result == [events[3]]
+ print(f"Events for user_id=1, session_id=sess-b: {result}")
+
+ result = await lookuper.lookup((2, "sess-a"))
+ assert result == [events[4]]
+ print(f"Events for user_id=2, session_id=sess-a: {result}")
+
+ result = await lookuper.lookup({"user_id": 2, "session_id": "sess-missing"})
+ assert result == [], "Expected no events for user_id=2, session_id=sess-missing"
+ print("Events for user_id=2, session_id=sess-missing: [] (no matches)")
+ finally:
+ await admin.drop_table(table_path, ignore_if_not_exists=True)
+ print(f"\nDropped session events table: {table_path}")
+
+
+if __name__ == "__main__":
+ asyncio.run(main())
diff --git a/fluss-rust/bindings/python/test/test_examples.py b/fluss-rust/bindings/python/test/test_examples.py
index b2a1d6bca..d10796126 100644
--- a/fluss-rust/bindings/python/test/test_examples.py
+++ b/fluss-rust/bindings/python/test/test_examples.py
@@ -59,6 +59,9 @@ EXAMPLES = _discover_examples()
def test_examples_discovered():
assert EXAMPLES, f"No runnable examples found in {EXAMPLES_DIR}"
+ assert "prefix_lookup" in {example.id for example in EXAMPLES}, (
+ "The prefix lookup example must be included in the CI example suite"
+ )
@pytest.mark.parametrize("example_main", EXAMPLES)
diff --git a/fluss-rust/bindings/python/test/test_kv_table.py b/fluss-rust/bindings/python/test/test_kv_table.py
index e11a874e0..fe09f9fdb 100644
--- a/fluss-rust/bindings/python/test/test_kv_table.py
+++ b/fluss-rust/bindings/python/test/test_kv_table.py
@@ -200,7 +200,7 @@ async def test_batched_concurrent_prefix_lookups(connection, admin):
for id_, rows in zip(ids, results):
expected = [(seq, f"{id_}-{seq}") for seq in range(2)] if id_ < 64 else []
- assert sorted((row["seq"], row["name"]) for row in rows) == expected
+ assert [(row["seq"], row["name"]) for row in rows] == expected
await admin.drop_table(table_path, ignore_if_not_exists=False)
@@ -279,7 +279,7 @@ async def test_composite_primary_keys(connection, admin):
# Prefix (US, 1) should match 2 rows (event_id 1 and 2)
rows = await prefix_lookuper.lookup({"region": "US", "user_id": 1})
assert len(rows) == 2
- event_ids = sorted(row["event_id"] for row in rows)
+ event_ids = [row["event_id"] for row in rows]
assert event_ids == [1, 2]
# Also validate list/tuple prefix input
@@ -824,6 +824,67 @@ async def test_all_supported_datatypes(connection, admin):
await admin.drop_table(table_path, ignore_if_not_exists=False)
+async def test_prefix_lookup_partitioned(connection, admin):
+ """Test ordered prefix results and key isolation across populated partitions."""
+ table_path = fluss.TablePath("fluss", "py_test_prefix_lookup_partitioned")
+ await admin.drop_table(table_path, ignore_if_not_exists=True)
+
+ schema = fluss.Schema(
+ pa.schema(
+ [
+ pa.field("region", pa.string()),
+ pa.field("a", pa.int32()),
+ pa.field("b", pa.string()),
+ pa.field("c", pa.int64()),
+ pa.field("d", pa.string()),
+ ]
+ ),
+ primary_keys=["region", "a", "b", "c"],
+ )
+ descriptor = fluss.TableDescriptor(
+ schema,
+ partition_keys=["region"],
+ bucket_count=3,
+ bucket_keys=["a", "b"],
+ )
+ await admin.create_table(table_path, descriptor, ignore_if_exists=False)
+
+ try:
+ for region in ["US", "EU"]:
+ await admin.create_partition(
+ table_path, {"region": region}, ignore_if_exists=False
+ )
+
+ table = await connection.get_table(table_path)
+ writer = table.new_upsert().create_writer()
+ test_data = [
+ ("US", 1, "aaaaaaaaa", 1, "us-1"),
+ ("US", 1, "aaaaaaaaa", 2, "us-2"),
+ ("US", 2, "aaaaaaaaa", 3, "us-3"),
+ ("EU", 1, "aaaaaaaaa", 4, "eu-1"),
+ ("EU", 1, "bbbbbbbbb", 5, "eu-2"),
+ ]
+ rows = [
+ {"region": region, "a": a, "b": b, "c": c, "d": d}
+ for region, a, b, c, d in test_data
+ ]
+ for row in rows:
+ writer.upsert(row)
+ await writer.flush()
+
+ lookuper = table.new_lookup().lookup_by(["region", "a", "b"]).create_lookuper()
+ result = await lookuper.lookup({"region": "US", "a": 1, "b": "aaaaaaaaa"})
+ assert result == rows[:2]
+
+ result = await lookuper.lookup({"region": "EU", "a": 1, "b": "bbbbbbbbb"})
+ assert result == [rows[4]]
+
+ result = await lookuper.lookup({"region": "APAC", "a": 1, "b": "aaaaaaaaa"})
+ assert result == []
+ finally:
+ await admin.drop_table(table_path, ignore_if_not_exists=False)
+
+
async def test_prefix_lookup_validation_errors(connection, admin):
"""Test that prefix lookup raises errors for invalid column configurations."""
table_path = fluss.TablePath("fluss", "py_test_prefix_lookup_validation")
@@ -847,8 +908,14 @@ async def test_prefix_lookup_validation_errors(connection, admin):
table = await connection.get_table(table_path)
# lookup_by with columns equal to full PK should error
- with pytest.raises(fluss.FlussError, match="prefix lookup"):
+ with pytest.raises(fluss.FlussError, match="prefix lookup") as exc_info:
table.new_lookup().lookup_by(["a", "b", "c"]).create_lookuper()
+ assert exc_info.value.error_code == fluss.ErrorCode.CLIENT_ERROR
+
+ # Lookup columns must list the bucket keys in order.
+ with pytest.raises(fluss.FlussError, match="bucket keys") as exc_info:
+ table.new_lookup().lookup_by(["b", "a"]).create_lookuper()
+ assert exc_info.value.error_code == fluss.ErrorCode.CLIENT_ERROR
# lookup_by with wrong column names should error
with pytest.raises(fluss.FlussError, match="bucket keys"):
diff --git a/website/docs/apis/python/example/prefix-lookup.md b/website/docs/apis/python/example/prefix-lookup.md
new file mode 100644
index 000000000..6f6a4a08b
--- /dev/null
+++ b/website/docs/apis/python/example/prefix-lookup.md
@@ -0,0 +1,98 @@
+---
+sidebar_position: 7
+---
+# Prefix Lookup
+
+Prefix lookup returns all rows whose primary key starts with a given prefix. It's enabled by choosing a **bucket key that is a strict prefix of the primary key** — rows sharing the same bucket-key prefix land in the same bucket, so one bucket lookup returns them all.
+
+## Table Requirements
+
+- The table must have a primary key.
+- The bucket key must be a strict prefix of the primary key (on partitioned tables, of the *non-partition* portion of the primary key).
+- The bucket key cannot equal the full primary key — that's a normal primary-key lookup, use [`Lookuper`](./primary-key-tables.md#upsert-delete-lookup) instead.
+- The `lookup_by` columns passed to the client must equal `partition_keys ++ bucket_key` (in that order, if partitioned).
+
+`create_lookuper()` validates the prefix rules and raises `fluss.FlussError` on mismatch, with a message describing the violation.
+
+## Non-Partitioned Table
+
+Pick a schema where the bucket key is a prefix of the primary key:
+
+```python
+import pyarrow as pa
+import fluss
+
+schema = fluss.Schema(
+ pa.schema([
+ pa.field("user_id", pa.int32()),
+ pa.field("session_id", pa.string()),
+ pa.field("event_seq", pa.int64()),
+ pa.field("event_data", pa.string()),
+ ]),
+ primary_keys=["user_id", "session_id", "event_seq"],
+)
+# Bucket key (user_id, session_id) is a prefix of the primary key.
+table_descriptor = fluss.TableDescriptor(
+ schema, bucket_count=3, bucket_keys=["user_id", "session_id"]
+)
+```
+
+Create the lookuper with `lookup_by(columns)` naming the prefix columns, then call `lookup(prefix_row)`:
+
+```python
+prefix_lookuper = (
+ table.new_lookup()
+ .lookup_by(["user_id", "session_id"])
+ .create_lookuper()
+)
+
+prefix = {"user_id": 1, "session_id": "sess-a"}
+
+result = await prefix_lookuper.lookup(prefix)
+for row in result:
+ print(f"seq={row['event_seq']}, data={row['event_data']}")
+```
+
+Unlike primary-key lookup (which returns a single row dictionary or `None`), prefix lookup returns zero or more rows as a list of full-row dictionaries.
+
+## Partitioned Table
+
+On a partitioned table, the partition columns are stripped from the primary key before the bucket-prefix rule is evaluated. The lookup key, though, must still carry the partition values so the client can route the request to the right partition — so the `lookup_by` columns are `partition_keys ++ bucket_key`.
+
+```python
+schema = fluss.Schema(
+ pa.schema([
+ pa.field("region", pa.string()),
+ pa.field("user_id", pa.int32()),
+ pa.field("session_id", pa.string()),
+ pa.field("event_seq", pa.int64()),
+ pa.field("event_data", pa.string()),
+ ]),
+ primary_keys=["region", "user_id", "session_id", "event_seq"],
+)
+# Bucket key (user_id, session_id) is a prefix of the pk minus partition cols.
+table_descriptor = fluss.TableDescriptor(
+ schema,
+ partition_keys=["region"],
+ bucket_count=3,
+ bucket_keys=["user_id", "session_id"],
+)
+```
+
+```python
+prefix_lookuper = (
+ table.new_lookup()
+ .lookup_by(["region", "user_id", "session_id"])
+ .create_lookuper()
+)
+
+prefix = {
+ "region": "US", # region (partition column)
+ "user_id": 1,
+ "session_id": "sess-a",
+}
+
+result = await prefix_lookuper.lookup(prefix)
+for row in result:
+ print(f"seq={row['event_seq']}, data={row['event_data']}")
+```