Skip to content

[lake/paimon] Explictly snapshotId setting to improve scan-based lookup performance #4480

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zuston wants to merge 2 commits into
base: main
Choose a base branch
from
+112 −10
Open

[lake/paimon] Explictly snapshotId setting to improve scan-based lookup performance #4480

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
cbed23c
[lake/paimon] Explictly snapshotId setting to improve scan-based look…
Sep 23, 2026
a50e5d0
comment
Sep 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@

package org.apache.fluss.config;

import org.apache.fluss.annotation.Internal;
import org.apache.fluss.annotation.PublicEvolving;
import org.apache.fluss.compression.ArrowCompressionInfo;
import org.apache.fluss.metadata.ChangelogImage;
Expand Down Expand Up @@ -136,6 +137,7 @@ public boolean isHistoricalPartitionEnabled() {
}

/** Gets the lookup mode for historical partitions of the table. */
@Internal
public LakeLookupMode getHistoricalLookupMode() {
return config.get(ConfigOptions.TABLE_DATALAKE_HISTORICAL_PARTITION_LOOKUP_MODE);
}
Expand Down
23 changes: 22 additions & 1 deletion fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeTableLookuper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.fluss.lake.lakestorage;

import org.apache.fluss.annotation.PublicEvolving;
import org.apache.fluss.annotation.VisibleForTesting;
import org.apache.fluss.metadata.ResolvedPartitionSpec;
import org.apache.fluss.types.RowType;

Expand Down Expand Up @@ -78,6 +79,18 @@ final class LookupContext {
private final short schemaId;
private final RowType valueRowType;
private final LookupMetricRecorder lookupMetricRecorder;
private final @Nullable Long lakeSnapshotId;

/** Creates a lookup context when the lake snapshot is unknown. */
@VisibleForTesting

@luoyuxia luoyuxia Sep 26, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we remove this test-only overload? The tests can call the six-argument constructor directly and pass null when the lake snapshot is unknown.

public LookupContext(
ResolvedPartitionSpec partitionSpec,
@Nullable Integer bucketId,
short schemaId,
RowType valueRowType,
LookupMetricRecorder lookupMetricRecorder) {
this(partitionSpec, bucketId, schemaId, valueRowType, lookupMetricRecorder, null);
}

/**
* Creates a lookup context.
Expand All @@ -88,19 +101,22 @@ final class LookupContext {
* @param schemaId schema id to encode the returned Fluss value with
* @param valueRowType row type to encode the returned Fluss value with
* @param lookupMetricRecorder recorder for lake table point lookup metrics
* @param lakeSnapshotId known lake snapshot ID, or null if unknown
*/
public LookupContext(
ResolvedPartitionSpec partitionSpec,
@Nullable Integer bucketId,
short schemaId,
RowType valueRowType,
LookupMetricRecorder lookupMetricRecorder) {
LookupMetricRecorder lookupMetricRecorder,
@Nullable Long lakeSnapshotId) {
this.partitionSpec = checkNotNull(partitionSpec, "partitionSpec must not be null.");
this.bucketId = bucketId;
this.schemaId = schemaId;
this.valueRowType = checkNotNull(valueRowType, "valueRowType must not be null.");
this.lookupMetricRecorder =
checkNotNull(lookupMetricRecorder, "lookupMetricRecorder must not be null.");
this.lakeSnapshotId = lakeSnapshotId;
}

/** Returns the resolved Fluss partition spec for the lookup. */
Expand Down Expand Up @@ -130,5 +146,10 @@ public RowType valueRowType() {
public LookupMetricRecorder lookupMetricRecorder() {
return lookupMetricRecorder;
}

/** Returns the known lake snapshot ID, or null if unknown. */
public @Nullable Long lakeSnapshotId() {
return lakeSnapshotId;
}
}
}
19 changes: 16 additions & 3 deletions ...aimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonScanBasedTableLookuper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import org.apache.fluss.utils.ExceptionUtils;
import org.apache.fluss.utils.IOUtils;

import org.apache.paimon.CoreOptions;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.CatalogFactory;
Expand Down Expand Up @@ -61,7 +62,8 @@
import static org.apache.fluss.utils.concurrent.LockUtils.inWriteLock;

/**
* Looks up a primary key by scanning the latest Paimon snapshot with a limit of one row.
* Looks up a primary key by scanning the requested Paimon snapshot, or the latest snapshot when no
* snapshot ID is provided, with a limit of one row.
*
* <p>Each scan is restricted to the requested partition, bucket, and complete primary key. It does
* not create local lookup files. Lookups use independent readers and encoders and may run in
Expand Down Expand Up @@ -119,7 +121,7 @@ public PaimonScanBasedTableLookuper(

@Override
public void requestRefresh() {
// Each lookup already plans a fresh scan of the latest snapshot.
// Each lookup plans a fresh scan, so there are no cached data files to refresh.
}

@Override
Expand Down Expand Up @@ -161,8 +163,19 @@ private FileStoreTable table() throws Exception {

private @Nullable byte[] scanLookup(FileStoreTable table, byte[] key, LookupContext context)
throws Exception {
FileStoreTable scanTable = table;
Long lakeSnapshotId = context.lakeSnapshotId();
if (lakeSnapshotId != null) {
// Paimon propagates the table's snapshot and manifest caches to this copy.
Comment on lines +167 to +169
scanTable =
scanTable.copy(
Collections.singletonMap(
CoreOptions.SCAN_SNAPSHOT_ID.key(),
String.valueOf(lakeSnapshotId)));
}
ReadBuilder readBuilder =
table.newReadBuilder()
scanTable
.newReadBuilder()
.withFilter(createKeyPredicates(table, key, context))
.withPartitionFilter(createPartitionPredicate(table, context))
.withReadType(
Expand Down
75 changes: 70 additions & 5 deletions ...paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuperTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -178,6 +178,67 @@ void testLookupPartitionedPrimaryKeyTable(LakeLookupMode lookupMode) throws Exce
}
}

@Test
void testScanLookupUsesRequestedSnapshot() throws Exception {
TablePath tablePath = TablePath.of(DB, "scan_snapshot");
Schema schema = pkSchema();
FileStoreTable table = createPaimonTable(tablePath, partitionedPkDescriptor(schema));
long firstSnapshotId =
writeAndCommitData(
table,
Collections.singletonMap(
0, Collections.singletonList(paimonRow(1, "20240101", "Alice"))));
ResolvedPartitionSpec partitionSpec =
ResolvedPartitionSpec.fromPartitionName(
Collections.singletonList("dt"), "20240101");
LakeTableLookuper.LookupContext firstContext =
new LakeTableLookuper.LookupContext(
partitionSpec,
0,
SCHEMA_ID,
schema.getRowType(),
NO_OP_LOOKUP_METRIC_RECORDER,
firstSnapshotId);
byte[] key = paimonKey(schema, 1, "20240101");

try (LakeTableLookuper lookuper =
createLookuper(LakeLookupMode.SCAN, tablePath, KvFormat.COMPACTED)) {
assertRow(
decodeValue(lookuper.lookup(key, firstContext), SCHEMA_ID, schema).row,
1,
"20240101",
"Alice");

long secondSnapshotId =
writeAndCommitData(
table,
Collections.singletonMap(
0,
Collections.singletonList(
paimonRow(1, "20240101", "Updated Alice"))));
// A committed newer snapshot does not change the snapshot pinned by this context.
assertRow(
decodeValue(lookuper.lookup(key, firstContext), SCHEMA_ID, schema).row,
1,
"20240101",
"Alice");

LakeTableLookuper.LookupContext secondContext =
new LakeTableLookuper.LookupContext(
partitionSpec,
0,
SCHEMA_ID,
schema.getRowType(),
NO_OP_LOOKUP_METRIC_RECORDER,
secondSnapshotId);
assertRow(
decodeValue(lookuper.lookup(key, secondContext), SCHEMA_ID, schema).row,
1,
"20240101",
"Updated Alice");
}
}

@ParameterizedTest(name = "lookupMode={0}")
@EnumSource(LakeLookupMode.class)
void testLookupKeysInComputedBuckets(LakeLookupMode lookupMode) throws Exception {
Expand Down Expand Up @@ -836,9 +897,11 @@ void testLookupWithNonStringPartitionKey(LakeLookupMode lookupMode) throws Excep
.distributedBy(2, "id")
.build();
FileStoreTable table = createPaimonTable(tablePath, tableDescriptor);
writeAndCommitData(
table,
Collections.singletonMap(0, Collections.singletonList(paimonRow(1, 7, "Alice"))));
long snapshotId =
writeAndCommitData(
table,
Collections.singletonMap(
0, Collections.singletonList(paimonRow(1, 7, "Alice"))));

try (LakeTableLookuper lookuper =
createLookuper(lookupMode, tablePath, KvFormat.COMPACTED)) {
Expand All @@ -849,7 +912,8 @@ void testLookupWithNonStringPartitionKey(LakeLookupMode lookupMode) throws Excep
0,
SCHEMA_ID,
schema.getRowType(),
NO_OP_LOOKUP_METRIC_RECORDER);
NO_OP_LOOKUP_METRIC_RECORDER,
snapshotId);

BinaryValue decodedValue =
decodeValue(
Expand Down Expand Up @@ -882,7 +946,8 @@ void testRejectAppendOnlyTableAndLookupAfterClose(LakeLookupMode lookupMode) thr
0,
SCHEMA_ID,
schema.getRowType(),
NO_OP_LOOKUP_METRIC_RECORDER);
NO_OP_LOOKUP_METRIC_RECORDER,
null);

assertThatThrownBy(() -> lookuper.lookup(new byte[0], context))
.isInstanceOf(UnsupportedOperationException.class)
Expand Down
3 changes: 2 additions & 1 deletion ...src/main/java/org/apache/fluss/server/replica/historical/HistoricalLakeLookupManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -373,7 +373,8 @@ private LookupContext createLookupContext(
lakeBucketId,
(short) schemaInfo.getSchemaId(),
schemaInfo.getSchema().getRowType(),
lookupMetricRecorder);
lookupMetricRecorder,
requiredLakeSnapshotIds.get(tableInfo.getTableId()));

@zuston zuston Sep 24, 2026 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

all the modification origins from the fluss, so this won't happen

@luoyuxia luoyuxia Sep 26, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There is a stale-snapshot case here even when the context and acquireLookuper() observe the same snapshot ID:

  1. The historical bucket records requiredLakeSnapshotIds[tableId] = S1, and its leader remains unchanged.
  2. A regular partition P writes a new key and tiers it to snapshot S2. If the historical bucket has no progress in that commit, its lookup manager can still retain S1.
  3. P expires, and the first historical lookup for that key misses the local historical state.
  4. Both context creation and acquireLookuper() see S1. SCAN explicitly reads S1 and misses the key committed in S2.

This can happen even while S1 is still retained, without concurrent writes. Making snapshot capture and acquire atomic would still select S1.

SST avoids this particular case because its file cache is keyed by the original partition and bucket. On the first lookup of P, getOrInitializeFiles() scans the latest snapshot to initialize that entry.

Could we address this case so that SCAN does not remain pinned to a snapshot that predates the queried partition’s tiered data?

@zuston zuston Sep 26, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nice catch, this is uncovered (described into the following first case) in the current PR. I think we should distinguish two cases:

Data written before the regular partition expires.

The data is written through the regular bucket and committed in S2. The historical bucket may have no progress in that commit, so its lookup manager can remain on S1. When the regular partition expires, its existing rows are not copied into the historical bucket’s local KV. The first historical lookup therefore falls back to S1 and can miss the data. This is the gap you pointed out, and making snapshot capture and acquire atomic would not fix it.

For this case, I’ll make partition expiration trigger a refresh of the historical lookup manager’s cached snapshot to the latest version. This relies on the guarantee established by #3820 that the partition’s data has been fully tiered to the lake before expiration. I’ll include this fix in the current PR.

Updates or deletes written after the partition has expired.

Updates and deletes to expired partitions go through the historical bucket. Local values and deletion markers take priority over lake data. After tiering, the snapshot is updated before local state can be cleaned up. so for this case, there is no correctness risk for the cached expired snapshotId (that may be delayed updated with a potential async updating interval)

return new LookupContext(
tableInfo.getTableId(), schemaInfo.getSchemaId(), tablePath, lookupContext);
}
Expand Down
Loading