[lake/paimon] Introduce scan-based lake table lookuper - #4124
Conversation
b412192 to
0f3bdb9
Compare
| partitionKeyExtractor::partition); | ||
| ReadBuilder readBuilder = | ||
| table.newReadBuilder() | ||
| .withFilter(predicates) |
There was a problem hiding this comment.
I think withFilter defeats the withBucket below it as Paimon derives the bucket itself when the
predicate covers the bucket-key, and if it disagrees with context.bucketId() we return null instead
of the row.
Wrote a few keys into one bucket and looked them up there: SCAN misses some, SST finds all.
Am I reading this right?
| try { | ||
| FileStoreTable table = table(); | ||
| // Paimon tables contain mutable lazy store state; isolate it per lookup. | ||
| return scanLookup(table.copy(table.schema()), key, context); |
There was a problem hiding this comment.
This re-plans the snapshot on every key, and HistoricalLakeLookupManager calls lookup() per key.
Noticeably slower than SST. Worth hoisting the setup out?
There was a problem hiding this comment.
Thanks for pointing this out.
I don't think an additional plan cache would help here. The catalog and table are initialized only once, and Paimon's CachingCatalog already caches snapshot and manifest metadata; table.copy() preserves those caches. The remaining plan is key-specific because each lookup has a different primary-key predicate, so it cannot be reused safely and caching it could also pin an old snapshot. The per-key planning is the expected trade-off of SCAN mode for avoiding local SST downloads.
Future historical-partition writes should not depend on the SST-based lookuper, so making the lookup implementation pluggable is appropriate. That said, the SST mode and its local cache remain important and deserve further investment. Potential optimizations include remote SST lookup files and global indexes.
To share our initial production benchmark results for scan-based lookuper:
- With the default SST-based lookuper under a highly sparse lookup workload, building local SST files requires reading an entire 128 MB Parquet file. On our relatively slow HDFS cluster, this became the main bottleneck, with P50 lookup latency reaching approximately 5 seconds. This assumes the concurrent lookup issue has already been addressed (however this haven't. and it also will result in the tablet potential OOM).
- After reducing the Parquet row-group size to approximately 2 MB while keeping the file size at 128 MB, scan-based lookup can push down the primary-key predicate and read only the relevant row group. This reduced P50 lookup latency to approximately 2 seconds.
In this workload, the reduction in remote I/O outweighs the additional per-key planning overhead.
There was a problem hiding this comment.
This re-plans the snapshot on every key, and HistoricalLakeLookupManager calls lookup() per key.
Nice catch, this is necessary to make it parallelize. let me follow up in another PR.
And one possible way to use the multi filter that could cover multi lookup key by using the OR-filer scan instead of using concurrency in the invoking side.
|
|
||
| /** Lookup strategy for historical partitions stored in lake storage. */ | ||
| public static final ConfigOption<LookupMode> TABLE_DATALAKE_HISTORICAL_PARTITION_LOOKUP_MODE = | ||
| key("table.datalake.historical-partition.lookup-mode") |
There was a problem hiding this comment.
Mind adding this to options.md/ddl.md/lookups.md too? historical-partition.enabled is in all three.
| lookupContext(schema, "20240101", 1, SCHEMA_ID))) | ||
| .isNull(); | ||
| assertThat(lookuper.lookup(compactedKey(schema, 1, "20240101"), context)).isNull(); | ||
| if (lookupMode == LookupMode.SST) { |
There was a problem hiding this comment.
Curious why this one is SST-only, what does SCAN return for a compacted-encoded key?
There was a problem hiding this comment.
for the KV format v1 with compacted format, this encoding style is illegal for the paimon side. anyway, the V2 is the compatible way, let me remove this assert logic.
| } | ||
|
|
||
| /** Mode used to look up historical data in lake storage. */ | ||
| enum LookupMode { |
There was a problem hiding this comment.
Should this live in org.apache.fluss.metadata, next to DataLakeFormat and KvFormat?
Yes, and I think we should prioritize this. |
0f3bdb9 to
20aff07
Compare
2 participants
Purpose
this is the sub-task for #3631
The SST-based Paimon lake table lookuper introduced in #3632 caches Paimon SST files on TabletServer local disks. This approach can also benefit from future Fluss-level optimizations, such as a global index.
However, for highly sparse lookup workloads, the cache hit ratio can be very low, resulting in frequent file downloads and additional I/O overhead. Scan-based lookup can reduce this overhead by using predicate pushdown to read only the relevant Parquet row groups instead of entire files. This is particularly beneficial when row groups are small relative to the overall Parquet file size.
Brief change log
Tests
API and Format
Documentation