From ff5ba99341ce96c337608eee00fc637923d70637 Mon Sep 17 00:00:00 2001 From: zhangjunfan Date: Thu, 27 Aug 2026 16:35:17 +0800 Subject: [PATCH 1/7] [lake/paimon] Introduce scan-based lake table lookuper --- .../apache/fluss/config/ConfigOptions.java | 13 + .../org/apache/fluss/config/TableConfig.java | 6 + .../fluss/lake/lakestorage/LakeStorage.java | 16 ++ .../fluss/lake/paimon/PaimonLakeStorage.java | 4 + .../lookup/PaimonScanBasedTableLookuper.java | 231 ++++++++++++++++++ 5 files changed, 270 insertions(+) create mode 100644 fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonScanBasedTableLookuper.java diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java index 19e71dfee21..42acb521f42 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java @@ -20,6 +20,7 @@ import org.apache.fluss.annotation.Internal; import org.apache.fluss.annotation.PublicEvolving; import org.apache.fluss.compression.ArrowCompressionType; +import org.apache.fluss.lake.lakestorage.LakeStorage.LookupMode; import org.apache.fluss.metadata.ChangelogImage; import org.apache.fluss.metadata.DataLakeFormat; import org.apache.fluss.metadata.DeleteBehavior; @@ -1933,6 +1934,18 @@ public class ConfigOptions { + "to look up historical partition data so that their clients load the " + "updated table configuration."); + /** Lookup strategy for historical partitions stored in lake storage. */ + public static final ConfigOption TABLE_DATALAKE_HISTORICAL_PARTITION_LOOKUP_MODE = + key("table.datalake.historical-partition.lookup-mode") + .enumType(LookupMode.class) + .defaultValue(LookupMode.SST) + .withDescription( + "The lookup mode for historical partitions stored in Paimon. " + + "SST uses local lookup files cached from lake storage. " + + "SCAN scans the requested partition and bucket with primary-key filters " + + "and a limit of one row, without creating local lookup files. " + + "This option can only be set when creating the table and cannot be altered."); + public static final ConfigOption TABLE_DATALAKE_FORMAT = key("table.datalake.format") .enumType(DataLakeFormat.class) diff --git a/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java b/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java index ea046e3b11d..c86edfe9a0a 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java @@ -19,6 +19,7 @@ import org.apache.fluss.annotation.PublicEvolving; import org.apache.fluss.compression.ArrowCompressionInfo; +import org.apache.fluss.lake.lakestorage.LakeStorage.LookupMode; import org.apache.fluss.metadata.ChangelogImage; import org.apache.fluss.metadata.DataLakeFormat; import org.apache.fluss.metadata.DeleteBehavior; @@ -134,6 +135,11 @@ public boolean isHistoricalPartitionEnabled() { return config.get(ConfigOptions.TABLE_DATALAKE_HISTORICAL_PARTITION_ENABLED); } + /** Gets the lookup mode for historical partitions of the table. */ + public LookupMode getHistoricalLookupMode() { + return config.get(ConfigOptions.TABLE_DATALAKE_HISTORICAL_PARTITION_LOOKUP_MODE); + } + /** * Return the data lake format of the table. It'll be the datalake format configured in Fluss * whiling creating the table. Return empty if no datalake format configured while creating. diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeStorage.java b/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeStorage.java index 446aa1184c4..f93d7ef0268 100644 --- a/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeStorage.java +++ b/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeStorage.java @@ -68,12 +68,22 @@ default LakeTableLookuper createLakeTableLookuper( "Point lookup is not supported for this lake storage."); } + /** Mode used to look up historical data in lake storage. */ + enum LookupMode { + /** Use local lookup files cached from lake storage. */ + SST, + + /** Scan lake storage with primary-key filters and return at most one row. */ + SCAN + } + /** Runtime context for creating a lake table lookuper. */ final class LookuperContext { private final String ioTmpDir; private final TableConfig tableConfig; private final long lookupCacheMaxDiskBytes; private final Runnable diskWriteGuard; + private final LookupMode lookupMode; /** * Creates a lookuper context. @@ -94,6 +104,7 @@ public LookuperContext( lookupCacheMaxDiskBytes > 0, "lookupCacheMaxDiskBytes must be greater than 0."); this.lookupCacheMaxDiskBytes = lookupCacheMaxDiskBytes; this.diskWriteGuard = checkNotNull(diskWriteGuard, "diskWriteGuard must not be null."); + this.lookupMode = tableConfig.getHistoricalLookupMode(); } /** Returns the local directory for temporary files used by the lookuper. */ @@ -115,5 +126,10 @@ public long lookupCacheMaxDiskBytes() { public Runnable diskWriteGuard() { return diskWriteGuard; } + + /** Returns the mode used to look up historical data. */ + public LookupMode lookupMode() { + return lookupMode; + } } } diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeStorage.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeStorage.java index 80e66398985..dabb7c38434 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeStorage.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeStorage.java @@ -21,6 +21,7 @@ import org.apache.fluss.lake.lakestorage.LakeStorage; import org.apache.fluss.lake.lakestorage.LakeTableLookuper; import org.apache.fluss.lake.paimon.lookup.PaimonLakeTableLookuper; +import org.apache.fluss.lake.paimon.lookup.PaimonScanBasedTableLookuper; import org.apache.fluss.lake.paimon.source.PaimonLakeSource; import org.apache.fluss.lake.paimon.source.PaimonSplit; import org.apache.fluss.lake.paimon.tiering.PaimonCommittable; @@ -56,6 +57,9 @@ public LakeSource createLakeSource(TablePath tablePath) { @Override public LakeTableLookuper createLakeTableLookuper(TablePath tablePath, LookuperContext context) { + if (context.lookupMode() == LookupMode.SCAN) { + return new PaimonScanBasedTableLookuper(paimonConfig, tablePath, context.tableConfig()); + } return new PaimonLakeTableLookuper( paimonConfig, tablePath, diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonScanBasedTableLookuper.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonScanBasedTableLookuper.java new file mode 100644 index 00000000000..9d12e978ad3 --- /dev/null +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonScanBasedTableLookuper.java @@ -0,0 +1,231 @@ +/* + * 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. + */ + +package org.apache.fluss.lake.paimon.lookup; + +import org.apache.fluss.config.Configuration; +import org.apache.fluss.config.TableConfig; +import org.apache.fluss.exception.KvStorageException; +import org.apache.fluss.lake.lakestorage.LakeTableLookuper; +import org.apache.fluss.lake.paimon.source.FlussRowAsPaimonRow; +import org.apache.fluss.lake.paimon.utils.PaimonRowAsFlussRow; +import org.apache.fluss.metadata.DataLakeFormat; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.row.InternalRow; +import org.apache.fluss.row.decode.KeyDecoder; +import org.apache.fluss.row.encode.RowEncoder; +import org.apache.fluss.row.encode.ValueEncoder; +import org.apache.fluss.utils.IOUtils; + +import org.apache.paimon.catalog.Catalog; +import org.apache.paimon.catalog.CatalogContext; +import org.apache.paimon.catalog.CatalogFactory; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.options.Options; +import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.reader.RecordReader; +import org.apache.paimon.reader.RecordReader.RecordIterator; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.sink.RowPartitionKeyExtractor; +import org.apache.paimon.table.source.ReadBuilder; +import org.apache.paimon.types.RowType; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; + +import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon; +import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimonPartition; +import static org.apache.fluss.utils.Preconditions.checkNotNull; +import static org.apache.fluss.utils.Preconditions.checkState; +import static org.apache.fluss.utils.concurrent.LockUtils.inReadLock; +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. + * + *

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 + * parallel. The catalog and table are initialized once, and close waits for active lookups to + * finish. + */ +public class PaimonScanBasedTableLookuper implements LakeTableLookuper { + + private final Configuration paimonConfig; + private final TablePath tablePath; + private final TableConfig tableConfig; + private final ReadWriteLock lifecycleLock = new ReentrantReadWriteLock(); + private final Object initializationLock = new Object(); + + private @Nullable Catalog catalog; + private @Nullable FileStoreTable fileStoreTable; + private boolean closed; + + /** Creates a scan-based lookuper for the specified Paimon table. */ + public PaimonScanBasedTableLookuper( + Configuration paimonConfig, TablePath tablePath, TableConfig tableConfig) { + this.paimonConfig = checkNotNull(paimonConfig, "paimonConfig must not be null."); + this.tablePath = checkNotNull(tablePath, "tablePath must not be null."); + this.tableConfig = checkNotNull(tableConfig, "tableConfig must not be null."); + } + + @Override + public @Nullable byte[] lookup(byte[] key, LookupContext context) throws Exception { + checkNotNull(key, "key must not be null."); + checkNotNull(context, "context must not be null."); + return inReadLock( + lifecycleLock, + () -> { + checkState(!closed, "Paimon scan-based lookuper has been closed."); + long lookupStartNanos = System.nanoTime(); + try { + FileStoreTable table = table(); + // Paimon tables contain mutable lazy store state; isolate it per lookup. + return scanLookup(table.copy(table.schema()), key, context); + } catch (IOException | UncheckedIOException e) { + // The next RPC retry plans a fresh scan after compaction or expiration. + throw new KvStorageException( + "Failed to scan historical data from Paimon for " + tablePath + ".", + e); + } finally { + context.lookupMetricRecorder() + .recordLookup(System.nanoTime() - lookupStartNanos, false); + } + }); + } + + @Override + public void close() { + inWriteLock( + lifecycleLock, + () -> { + if (!closed) { + closed = true; + IOUtils.closeQuietly(catalog, "Paimon catalog"); + } + }); + } + + private FileStoreTable table() throws Exception { + synchronized (initializationLock) { + if (fileStoreTable == null) { + Catalog newCatalog = + CatalogFactory.createCatalog( + CatalogContext.create(Options.fromMap(paimonConfig.toMap()))); + try { + FileStoreTable table = + (FileStoreTable) newCatalog.getTable(toPaimon(tablePath)); + if (table.primaryKeys().isEmpty()) { + throw new UnsupportedOperationException( + "Point lookup is only supported for primary-key Paimon tables."); + } + catalog = newCatalog; + fileStoreTable = table; + } finally { + if (fileStoreTable == null) { + IOUtils.closeQuietly(newCatalog, "Paimon catalog"); + } + } + } + return fileStoreTable; + } + } + + private @Nullable byte[] scanLookup(FileStoreTable table, byte[] key, LookupContext context) + throws Exception { + RowType rowType = table.rowType(); + List primaryKeys = table.schema().trimmedPrimaryKeys(); + KeyDecoder keyDecoder = + KeyDecoder.ofPrimaryKeyDecoder( + context.valueRowType(), + primaryKeys, + tableConfig.getKvFormatVersion().orElse(1).shortValue(), + DataLakeFormat.PAIMON, + table.schema().bucketKeys().equals(primaryKeys)); + FlussRowAsPaimonRow keyRow = + new FlussRowAsPaimonRow(keyDecoder.decodeKey(key), rowType.project(primaryKeys)); + PredicateBuilder predicateBuilder = new PredicateBuilder(rowType); + List predicates = new ArrayList<>(primaryKeys.size()); + for (int i = 0; i < primaryKeys.size(); i++) { + int fieldIndex = predicateBuilder.indexOf(primaryKeys.get(i)); + Object value = + org.apache.paimon.data.InternalRow.createFieldGetter( + rowType.getTypeAt(fieldIndex), i) + .getFieldOrNull(keyRow); + predicates.add(predicateBuilder.equal(fieldIndex, value)); + } + + RowPartitionKeyExtractor partitionKeyExtractor = + new RowPartitionKeyExtractor(table.schema()); + BinaryRow partition = + toPaimonPartition( + context.partitionSpec(), + context.valueRowType(), + rowType, + partitionKeyExtractor::partition); + ReadBuilder readBuilder = + table.newReadBuilder() + .withFilter(predicates) + .withPartitionFilter( + PartitionPredicate.fromMultiple( + rowType.project(table.partitionKeys()), + Collections.singletonList(partition))) + .withBucket(context.bucketId()) + .withReadType(rowType.project(context.valueRowType().getFieldNames())) + .withLimit(1); + // Pushdown alone may only prune files. Filter each row before applying the limit. + try (RecordReader reader = + readBuilder.newRead().executeFilter().createReader(readBuilder.newScan().plan())) { + RecordIterator batch; + while ((batch = reader.readBatch()) != null) { + try { + org.apache.paimon.data.InternalRow row = batch.next(); + if (row != null) { + // Encode while the batch still owns the row's backing storage. + return encodeValue(row, context); + } + } finally { + batch.releaseBatch(); + } + } + } + return null; + } + + private byte[] encodeValue(org.apache.paimon.data.InternalRow row, LookupContext context) + throws Exception { + PaimonRowAsFlussRow flussRow = new PaimonRowAsFlussRow(row); + InternalRow.FieldGetter[] fieldGetters = + InternalRow.createFieldGetters(context.valueRowType()); + try (RowEncoder encoder = + RowEncoder.create(tableConfig.getKvFormat(), context.valueRowType())) { + encoder.startNewRow(); + for (int i = 0; i < fieldGetters.length; i++) { + encoder.encodeField(i, fieldGetters[i].getFieldOrNull(flussRow)); + } + return ValueEncoder.encodeValue(context.schemaId(), encoder.finishRow()); + } + } +} From 5adc62932ad34631188b0e54746a3696f0436e8b Mon Sep 17 00:00:00 2001 From: zhangjunfan Date: Sun, 30 Aug 2026 21:38:03 +0800 Subject: [PATCH 2/7] extract the same abstraction func --- .../lookup/PaimonLakeTableLookuper.java | 31 ++++++------------- .../lookup/PaimonScanBasedTableLookuper.java | 26 ++++------------ .../lake/paimon/utils/PaimonConversions.java | 21 +++++++++++++ 3 files changed, 37 insertions(+), 41 deletions(-) diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java index 43497c23781..497cbcfeeb9 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java @@ -24,13 +24,9 @@ import org.apache.fluss.exception.KvStorageException; import org.apache.fluss.lake.lakestorage.LakeTableLookuper; import org.apache.fluss.lake.paimon.utils.PaimonPartitionBucket; -import org.apache.fluss.lake.paimon.utils.PaimonRowAsFlussRow; import org.apache.fluss.metadata.TablePath; -import org.apache.fluss.row.BinaryRow; import org.apache.fluss.row.InternalRow; import org.apache.fluss.row.decode.CompactedKeyDecoder; -import org.apache.fluss.row.encode.RowEncoder; -import org.apache.fluss.row.encode.ValueEncoder; import org.apache.fluss.row.encode.paimon.PaimonKeyEncoder; import org.apache.fluss.types.RowType; import org.apache.fluss.utils.ExceptionUtils; @@ -68,6 +64,7 @@ import static org.apache.fluss.config.ConfigOptions.KV_FORMAT_VERSION_2; import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.LEGACY_SYSTEM_COLUMNS; +import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toFlussValue; import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon; import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimonPartition; import static org.apache.fluss.utils.Preconditions.checkArgument; @@ -330,7 +327,15 @@ private org.apache.paimon.data.BinaryRow getKey(byte[] key, LookupContext contex if (paimonRow == null) { return null; } - return encodeValue(paimonRow, context.schemaId(), context.valueRowType()); + try { + return toFlussValue( + paimonRow, + context.schemaId(), + context.valueRowType(), + tableConfig.getKvFormat()); + } catch (Exception e) { + throw new RuntimeException("Failed to encode Paimon lookup row as Fluss value.", e); + } } private @Nullable org.apache.paimon.data.InternalRow lookupPaimon( @@ -418,22 +423,6 @@ private List scanDataFiles( return Collections.unmodifiableList(new ArrayList<>(dataFilesByName.values())); } - private byte[] encodeValue( - org.apache.paimon.data.InternalRow paimonRow, short schemaId, RowType valueRowType) { - PaimonRowAsFlussRow flussRow = new PaimonRowAsFlussRow(paimonRow); - InternalRow.FieldGetter[] fieldGetters = InternalRow.createFieldGetters(valueRowType); - try (RowEncoder rowEncoder = RowEncoder.create(tableConfig.getKvFormat(), valueRowType)) { - rowEncoder.startNewRow(); - for (int i = 0; i < fieldGetters.length; i++) { - rowEncoder.encodeField(i, fieldGetters[i].getFieldOrNull(flussRow)); - } - BinaryRow row = rowEncoder.finishRow(); - return ValueEncoder.encodeValue(schemaId, row); - } catch (Exception e) { - throw new RuntimeException("Failed to encode Paimon lookup row as Fluss value.", e); - } - } - /** Tracks creation of Paimon lookup files while delegating all local I/O operations. */ private final class TrackingIOManager implements IOManager { diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonScanBasedTableLookuper.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonScanBasedTableLookuper.java index 9d12e978ad3..db347e49986 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonScanBasedTableLookuper.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonScanBasedTableLookuper.java @@ -22,13 +22,9 @@ import org.apache.fluss.exception.KvStorageException; import org.apache.fluss.lake.lakestorage.LakeTableLookuper; import org.apache.fluss.lake.paimon.source.FlussRowAsPaimonRow; -import org.apache.fluss.lake.paimon.utils.PaimonRowAsFlussRow; import org.apache.fluss.metadata.DataLakeFormat; import org.apache.fluss.metadata.TablePath; -import org.apache.fluss.row.InternalRow; import org.apache.fluss.row.decode.KeyDecoder; -import org.apache.fluss.row.encode.RowEncoder; -import org.apache.fluss.row.encode.ValueEncoder; import org.apache.fluss.utils.IOUtils; import org.apache.paimon.catalog.Catalog; @@ -56,6 +52,7 @@ import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; +import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toFlussValue; import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon; import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimonPartition; import static org.apache.fluss.utils.Preconditions.checkNotNull; @@ -204,7 +201,11 @@ private FileStoreTable table() throws Exception { org.apache.paimon.data.InternalRow row = batch.next(); if (row != null) { // Encode while the batch still owns the row's backing storage. - return encodeValue(row, context); + return toFlussValue( + row, + context.schemaId(), + context.valueRowType(), + tableConfig.getKvFormat()); } } finally { batch.releaseBatch(); @@ -213,19 +214,4 @@ private FileStoreTable table() throws Exception { } return null; } - - private byte[] encodeValue(org.apache.paimon.data.InternalRow row, LookupContext context) - throws Exception { - PaimonRowAsFlussRow flussRow = new PaimonRowAsFlussRow(row); - InternalRow.FieldGetter[] fieldGetters = - InternalRow.createFieldGetters(context.valueRowType()); - try (RowEncoder encoder = - RowEncoder.create(tableConfig.getKvFormat(), context.valueRowType())) { - encoder.startNewRow(); - for (int i = 0; i < fieldGetters.length; i++) { - encoder.encodeField(i, fieldGetters[i].getFieldOrNull(flussRow)); - } - return ValueEncoder.encodeValue(context.schemaId(), encoder.finishRow()); - } - } } diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java index c0a1a3a4c90..d5f416f20c6 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java @@ -21,6 +21,7 @@ import org.apache.fluss.exception.InvalidConfigException; import org.apache.fluss.exception.InvalidTableException; import org.apache.fluss.lake.paimon.source.FlussRowAsPaimonRow; +import org.apache.fluss.metadata.KvFormat; import org.apache.fluss.metadata.ResolvedPartitionSpec; import org.apache.fluss.metadata.TableChange; import org.apache.fluss.metadata.TableDescriptor; @@ -28,6 +29,8 @@ import org.apache.fluss.record.ChangeType; import org.apache.fluss.row.GenericRow; import org.apache.fluss.row.InternalRow; +import org.apache.fluss.row.encode.RowEncoder; +import org.apache.fluss.row.encode.ValueEncoder; import org.apache.fluss.types.DataTypeRoot; import org.apache.fluss.utils.PartitionUtils; @@ -141,6 +144,24 @@ public static org.apache.fluss.types.RowType toFlussRowType(RowType paimonRowTyp return builder.build(); } + /** Encodes a Paimon row as a Fluss KV value. */ + public static byte[] toFlussValue( + org.apache.paimon.data.InternalRow paimonRow, + short schemaId, + org.apache.fluss.types.RowType valueRowType, + KvFormat kvFormat) + throws Exception { + PaimonRowAsFlussRow flussRow = new PaimonRowAsFlussRow(paimonRow); + InternalRow.FieldGetter[] fieldGetters = InternalRow.createFieldGetters(valueRowType); + try (RowEncoder rowEncoder = RowEncoder.create(kvFormat, valueRowType)) { + rowEncoder.startNewRow(); + for (int i = 0; i < fieldGetters.length; i++) { + rowEncoder.encodeField(i, fieldGetters[i].getFieldOrNull(flussRow)); + } + return ValueEncoder.encodeValue(schemaId, rowEncoder.finishRow()); + } + } + /** * Renders a Paimon partition row into Fluss partition value strings, in partition-key order. */ From a357fee668a08c6454fc482996cb31e69c23348f Mon Sep 17 00:00:00 2001 From: zhangjunfan Date: Sun, 30 Aug 2026 23:10:42 +0800 Subject: [PATCH 3/7] add test --- .../lookup/PaimonLakeTableLookuperTest.java | 408 ++++++++++-------- 1 file changed, 228 insertions(+), 180 deletions(-) diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuperTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuperTest.java index f493f62317c..bb7c0018f80 100644 --- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuperTest.java +++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuperTest.java @@ -22,9 +22,12 @@ import org.apache.fluss.config.MemorySize; import org.apache.fluss.config.TableConfig; import org.apache.fluss.exception.DiskWriteLockedException; +import org.apache.fluss.lake.lakestorage.LakeStorage.LookupMode; +import org.apache.fluss.lake.lakestorage.LakeStorage.LookuperContext; import org.apache.fluss.lake.lakestorage.LakeTableLookuper; import org.apache.fluss.lake.lakestorage.TestingLakeCatalogContext; import org.apache.fluss.lake.paimon.PaimonLakeCatalog; +import org.apache.fluss.lake.paimon.PaimonLakeStorage; import org.apache.fluss.metadata.KvFormat; import org.apache.fluss.metadata.ResolvedPartitionSpec; import org.apache.fluss.metadata.Schema; @@ -59,6 +62,8 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; import java.io.File; import java.util.ArrayList; @@ -80,7 +85,7 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; -/** Tests for {@link PaimonLakeTableLookuper}. */ +/** Tests for the Paimon {@link LakeTableLookuper} implementations. */ class PaimonLakeTableLookuperTest { private static final String DB = "lookup_db"; @@ -117,8 +122,9 @@ void tearDown() throws Exception { } } - @Test - void testLookupPartitionedPrimaryKeyTable() throws Exception { + @ParameterizedTest(name = "lookupMode={0}") + @EnumSource(LookupMode.class) + void testLookupPartitionedPrimaryKeyTable(LookupMode lookupMode) throws Exception { TablePath tablePath = TablePath.of(DB, "partitioned_pk"); Schema schema = pkSchema(); TableDescriptor tableDescriptor = partitionedPkDescriptor(schema); @@ -129,13 +135,7 @@ void testLookupPartitionedPrimaryKeyTable() throws Exception { 0, Collections.singletonList(paimonRow(1, "20240101", "Alice")))); try (LakeTableLookuper lookuper = - new PaimonLakeTableLookuper( - paimonConfig, - tablePath, - tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED), - LOOKUP_CACHE_MAX_DISK_BYTES, - NO_OP_DISK_WRITE_GUARD)) { + createLookuper(lookupMode, tablePath, KvFormat.COMPACTED)) { List lookupFileDownloads = new ArrayList<>(); LakeTableLookuper.LookupContext context = lookupContext( @@ -157,10 +157,16 @@ void testLookupPartitionedPrimaryKeyTable() throws Exception { paimonKey(schema, 1, "20240101"), lookupContext(schema, "20240101", 1, SCHEMA_ID))) .isNull(); - assertThat(lookuper.lookup(compactedKey(schema, 1, "20240101"), context)).isNull(); + if (lookupMode == LookupMode.SST) { + assertThat(lookuper.lookup(compactedKey(schema, 1, "20240101"), context)).isNull(); + } - // The first lookup creates the local lookup file, while subsequent lookups reuse it. - assertThat(lookupFileDownloads).containsExactly(true, false, false); + // SST creates a local lookup file on the first lookup; SCAN never creates one. + assertThat(lookupFileDownloads) + .containsExactlyElementsOf( + lookupMode == LookupMode.SST + ? Arrays.asList(true, false, false) + : Arrays.asList(false, false)); } } @@ -202,7 +208,7 @@ void testConcurrentFirstLookupsForDifferentPartitions() throws Exception { paimonConfig, tablePath, tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED), + tableConfig(KvFormat.COMPACTED, 1, LookupMode.SST), LOOKUP_CACHE_MAX_DISK_BYTES, diskWriteGuard)) { Future firstLookup = @@ -231,8 +237,9 @@ void testConcurrentFirstLookupsForDifferentPartitions() throws Exception { } } - @Test - void testDiskWriteLockBlocksOnlyLookupFileDownloads() throws Exception { + @ParameterizedTest(name = "lookupMode={0}") + @EnumSource(LookupMode.class) + void testDiskWriteLockBlocksOnlyLookupFileDownloads(LookupMode lookupMode) throws Exception { TablePath tablePath = TablePath.of(DB, "disk_write_lock"); Schema schema = pkSchema(); FileStoreTable table = createPaimonTable(tablePath, partitionedPkDescriptor(schema)); @@ -252,13 +259,7 @@ void testDiskWriteLockBlocksOnlyLookupFileDownloads() throws Exception { }; try (LakeTableLookuper lookuper = - new PaimonLakeTableLookuper( - paimonConfig, - tablePath, - tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED), - LOOKUP_CACHE_MAX_DISK_BYTES, - diskWriteGuard)) { + createLookuper(lookupMode, tablePath, KvFormat.COMPACTED, 1, diskWriteGuard)) { LakeTableLookuper.LookupContext cachedPartition = lookupContext(schema, "20240101", 0, SCHEMA_ID); LakeTableLookuper.LookupContext uncachedPartition = @@ -268,14 +269,21 @@ void testDiskWriteLockBlocksOnlyLookupFileDownloads() throws Exception { .isNotNull(); diskWriteLocked.set(true); - // Cache hits remain available, while a lookup that needs a new local file is rejected. + // SST cache hits remain available, while a lookup that needs a new local file is + // rejected. SCAN performs no local writes and is unaffected by the guard. assertThat(lookuper.lookup(paimonKey(schema, 1, "20240101"), cachedPartition)) .isNotNull(); - assertThatThrownBy( - () -> - lookuper.lookup( - paimonKey(schema, 2, "20240102"), uncachedPartition)) - .isInstanceOf(DiskWriteLockedException.class); + if (lookupMode == LookupMode.SST) { + assertThatThrownBy( + () -> + lookuper.lookup( + paimonKey(schema, 2, "20240102"), + uncachedPartition)) + .isInstanceOf(DiskWriteLockedException.class); + } else { + assertThat(lookuper.lookup(paimonKey(schema, 2, "20240102"), uncachedPartition)) + .isNotNull(); + } diskWriteLocked.set(false); assertThat(lookuper.lookup(paimonKey(schema, 2, "20240102"), uncachedPartition)) @@ -283,8 +291,9 @@ void testDiskWriteLockBlocksOnlyLookupFileDownloads() throws Exception { } } - @Test - void testLookupPartitionsWithSameHashCode() throws Exception { + @ParameterizedTest(name = "lookupMode={0}") + @EnumSource(LookupMode.class) + void testLookupPartitionsWithSameHashCode(LookupMode lookupMode) throws Exception { // These distinct partition values produce the same BinaryRow hash code, reproducing the // mutable-key collision that previously made one partition reuse another partition's files. String firstPartition = "b8"; @@ -309,13 +318,7 @@ void testLookupPartitionsWithSameHashCode() throws Exception { 0, Collections.singletonList(paimonRow(2, secondPartition, "Bob")))); try (LakeTableLookuper lookuper = - new PaimonLakeTableLookuper( - paimonConfig, - tablePath, - tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED), - LOOKUP_CACHE_MAX_DISK_BYTES, - NO_OP_DISK_WRITE_GUARD)) { + createLookuper(lookupMode, tablePath, KvFormat.COMPACTED)) { BinaryValue firstValue = decodeValue( lookuper.lookup( @@ -336,8 +339,9 @@ void testLookupPartitionsWithSameHashCode() throws Exception { } } - @Test - void testLookupWithIndexedKvFormat() throws Exception { + @ParameterizedTest(name = "lookupMode={0}") + @EnumSource(LookupMode.class) + void testLookupWithIndexedKvFormat(LookupMode lookupMode) throws Exception { TablePath tablePath = TablePath.of(DB, "indexed_kv_format"); Schema schema = pkSchema(); TableDescriptor tableDescriptor = @@ -350,14 +354,7 @@ void testLookupWithIndexedKvFormat() throws Exception { Collections.singletonMap( 0, Collections.singletonList(paimonRow(1, "20240101", "Alice")))); - try (LakeTableLookuper lookuper = - new PaimonLakeTableLookuper( - paimonConfig, - tablePath, - tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.INDEXED), - LOOKUP_CACHE_MAX_DISK_BYTES, - NO_OP_DISK_WRITE_GUARD)) { + try (LakeTableLookuper lookuper = createLookuper(lookupMode, tablePath, KvFormat.INDEXED)) { LakeTableLookuper.LookupContext context = lookupContext(schema, "20240101", 0, SCHEMA_ID); @@ -373,8 +370,9 @@ void testLookupWithIndexedKvFormat() throws Exception { } } - @Test - void testLookupKvFormatV2WithNonDefaultBucketKey() throws Exception { + @ParameterizedTest(name = "lookupMode={0}") + @EnumSource(LookupMode.class) + void testLookupKvFormatV2WithNonDefaultBucketKey(LookupMode lookupMode) throws Exception { TablePath tablePath = TablePath.of(DB, "non_default_bucket_key"); Schema schema = Schema.newBuilder() @@ -397,12 +395,11 @@ void testLookupKvFormatV2WithNonDefaultBucketKey() throws Exception { 0, Collections.singletonList(paimonRow(1, "sub-1", "20240101", "Alice")))); try (LakeTableLookuper lookuper = - new PaimonLakeTableLookuper( - paimonConfig, + createLookuper( + lookupMode, tablePath, - tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED, KV_FORMAT_VERSION_2), - LOOKUP_CACHE_MAX_DISK_BYTES, + KvFormat.COMPACTED, + KV_FORMAT_VERSION_2, NO_OP_DISK_WRITE_GUARD)) { LakeTableLookuper.LookupContext context = lookupContext(schema, "20240101", 0, SCHEMA_ID); @@ -418,8 +415,10 @@ void testLookupKvFormatV2WithNonDefaultBucketKey() throws Exception { } } - @Test - void testRetriesInitializationAfterLookupKeyConverterFailure() throws Exception { + @ParameterizedTest(name = "lookupMode={0}") + @EnumSource(LookupMode.class) + void testRetriesInitializationAfterLookupKeyConverterFailure(LookupMode lookupMode) + throws Exception { TablePath tablePath = TablePath.of(DB, "retry_initialization"); Schema schema = Schema.newBuilder() @@ -454,12 +453,11 @@ void testRetriesInitializationAfterLookupKeyConverterFailure() throws Exception .build(); try (LakeTableLookuper lookuper = - new PaimonLakeTableLookuper( - paimonConfig, + createLookuper( + lookupMode, tablePath, - tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED, KV_FORMAT_VERSION_2), - LOOKUP_CACHE_MAX_DISK_BYTES, + KvFormat.COMPACTED, + KV_FORMAT_VERSION_2, NO_OP_DISK_WRITE_GUARD)) { // Inject a late initialization failure: the Paimon table requires sub_id in its // lookup key, but the first lookup's value row type deliberately omits that field. @@ -471,8 +469,8 @@ void testRetriesInitializationAfterLookupKeyConverterFailure() throws Exception .isInstanceOf(IllegalArgumentException.class) .hasMessageContaining("sub_id"); - // The failed attempt must not leave localTableQuery as a false initialization marker. - // A second lookup on the same object should rebuild all resources and succeed. + // A failed attempt must not poison lazy initialization. A second lookup on the same + // object should initialize with the valid schema and succeed. byte[] value = lookuper.lookup(compactedKey, lookupContext(schema, "20240101", 0, SCHEMA_ID)); assertThat(value).isNotNull(); @@ -481,110 +479,104 @@ void testRetriesInitializationAfterLookupKeyConverterFailure() throws Exception } } - @Test - void testRefreshFilesAfterCompactionAndSnapshotExpiration() throws Exception { - TablePath tablePath = TablePath.of(DB, "compacted_pk"); + @ParameterizedTest(name = "lookupMode={0}") + @EnumSource(LookupMode.class) + void testLookupAfterCompactionAndSnapshotExpiration(LookupMode lookupMode) throws Exception { + TablePath tablePath = TablePath.of(DB, "compacted_lookup"); Schema schema = pkSchema(); - FileStoreTable table = createPaimonTable(tablePath, partitionedPkDescriptor(schema)); - for (int id = 1; id <= 5; id++) { - writeAndCommitData( - table, - Collections.singletonMap( - 0, Collections.singletonList(paimonRow(id, "20240101", "name-" + id)))); - } - writeAndCommitData( - table, - Collections.singletonMap( - 0, Collections.singletonList(paimonRow(6, "20240102", "name-6")))); - + FileStoreTable table = createCompactionTable(tablePath, schema); BinaryRow partition = BinaryRow.singleColumn(BinaryString.fromString("20240101")); - List filesBeforeCompaction = dataFiles(table, partition, 0); - assertThat(filesBeforeCompaction).hasSize(5); try (LakeTableLookuper lookuper = - new PaimonLakeTableLookuper( - paimonConfig, - tablePath, - tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED), - LOOKUP_CACHE_MAX_DISK_BYTES, - NO_OP_DISK_WRITE_GUARD)) { + createLookuper(lookupMode, tablePath, KvFormat.COMPACTED)) { assertThat( lookuper.lookup( paimonKey(schema, 5, "20240101"), lookupContext(schema, "20240101", 0, SCHEMA_ID))) .isNotNull(); - assertThat( - lookuper.lookup( - paimonKey(schema, 6, "20240102"), - lookupContext(schema, "20240102", 0, SCHEMA_ID))) - .isNotNull(); - new CompactHelper(table, new File(tempWarehouseDir, "compact")) - .compactBucket(partition, 0) - .commit(); - assertThat(dataFiles(table, partition, 0)).hasSize(1); - - DataFilePathFactory pathFactory = - table.store().pathFactory().createDataFilePathFactory(partition, 0); - List filesBeforeCompactionPaths = new ArrayList<>(); - for (DataFileMeta file : filesBeforeCompaction) { - Path path = pathFactory.toPath(file); - assertThat(table.store().snapshotManager().fileIO().exists(path)).isTrue(); - filesBeforeCompactionPaths.add(path); - } + compactAndExpire(table, partition); - Options expireOptions = new Options(); - expireOptions.set(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN, 1); - expireOptions.set(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX, 1); - // Compaction only marks the replaced files as deleted. Snapshot expiration performs - // the physical cleanup that makes the stale lookuper fail to open an old data file. - try (TableCommitImpl commit = table.copy(expireOptions.toMap()).newCommit("")) { - commit.expireSnapshots(); - } - for (Path path : filesBeforeCompactionPaths) { - assertThat(table.store().snapshotManager().fileIO().exists(path)).isFalse(); - } - - List refreshedLookupDownloads = new ArrayList<>(); - List cachedLookupDownloads = new ArrayList<>(); - LakeTableLookuper.LookupContext refreshedContext = - lookupContext( - schema, - "20240101", - 0, - SCHEMA_ID, - (lookupTimeNanos, lookupFileDownloaded) -> - refreshedLookupDownloads.add(lookupFileDownloaded)); - LakeTableLookuper.LookupContext cachedContext = - lookupContext( - schema, - "20240102", - 0, - SCHEMA_ID, - (lookupTimeNanos, lookupFileDownloaded) -> - cachedLookupDownloads.add(lookupFileDownloaded)); - BinaryValue refreshedValue = + BinaryValue value = decodeValue( - lookuper.lookup(paimonKey(schema, 1, "20240101"), refreshedContext), - SCHEMA_ID, - schema); - assertRow(refreshedValue.row, 1, "20240101", "name-1"); - - BinaryValue cachedValue = - decodeValue( - lookuper.lookup(paimonKey(schema, 6, "20240102"), cachedContext), + lookuper.lookup( + paimonKey(schema, 1, "20240101"), + lookupContext(schema, "20240101", 0, SCHEMA_ID)), SCHEMA_ID, schema); - assertRow(cachedValue.row, 6, "20240102", "name-6"); + assertRow(value.row, 1, "20240101", "name-1"); + } + } - assertThat(refreshedLookupDownloads).containsExactly(true); - assertThat(cachedLookupDownloads).containsExactly(false); + @ParameterizedTest(name = "lookupMode={0}") + @EnumSource(LookupMode.class) + void testLookupsRunConcurrently(LookupMode lookupMode) throws Exception { + TablePath tablePath = TablePath.of(DB, "concurrent_lookups"); + Schema schema = pkSchema(); + FileStoreTable table = createPaimonTable(tablePath, partitionedPkDescriptor(schema)); + writeAndCommitData( + table, + Collections.singletonMap( + 0, + Arrays.asList( + paimonRow(1, "20240101", "Alice"), + paimonRow(2, "20240101", "Bob")))); + + CountDownLatch firstLookupAtRecorder = new CountDownLatch(1); + CountDownLatch releaseFirstLookup = new CountDownLatch(1); + CountDownLatch secondLookupAtRecorder = new CountDownLatch(1); + LakeTableLookuper.LookupContext firstContext = + lookupContext( + schema, + "20240101", + 0, + SCHEMA_ID, + (lookupTimeNanos, lookupFileDownloaded) -> { + firstLookupAtRecorder.countDown(); + try { + releaseFirstLookup.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + }); + LakeTableLookuper.LookupContext secondContext = + lookupContext( + schema, + "20240101", + 0, + SCHEMA_ID, + (lookupTimeNanos, lookupFileDownloaded) -> + secondLookupAtRecorder.countDown()); + ExecutorService executor = Executors.newFixedThreadPool(2); + try (LakeTableLookuper lookuper = + createLookuper(lookupMode, tablePath, KvFormat.COMPACTED)) { + Future firstLookup = + executor.submit( + () -> lookuper.lookup(paimonKey(schema, 1, "20240101"), firstContext)); + assertThat(firstLookupAtRecorder.await(30, TimeUnit.SECONDS)).isTrue(); + + Future secondLookup = + executor.submit( + () -> lookuper.lookup(paimonKey(schema, 2, "20240101"), secondContext)); + assertThat(secondLookupAtRecorder.await(30, TimeUnit.SECONDS)).isTrue(); + + releaseFirstLookup.countDown(); + BinaryValue firstValue = + decodeValue(firstLookup.get(30, TimeUnit.SECONDS), SCHEMA_ID, schema); + BinaryValue secondValue = + decodeValue(secondLookup.get(30, TimeUnit.SECONDS), SCHEMA_ID, schema); + assertRow(firstValue.row, 1, "20240101", "Alice"); + assertRow(secondValue.row, 2, "20240101", "Bob"); + } finally { + releaseFirstLookup.countDown(); + ExecutorUtils.gracefulShutdown(30, TimeUnit.SECONDS, executor); } } - @Test - void testLookupWithNonStringPartitionKey() throws Exception { + @ParameterizedTest(name = "lookupMode={0}") + @EnumSource(LookupMode.class) + void testLookupWithNonStringPartitionKey(LookupMode lookupMode) throws Exception { TablePath tablePath = TablePath.of(DB, "int_partition_pk"); Schema schema = Schema.newBuilder() @@ -605,13 +597,7 @@ void testLookupWithNonStringPartitionKey() throws Exception { Collections.singletonMap(0, Collections.singletonList(paimonRow(1, 7, "Alice")))); try (LakeTableLookuper lookuper = - new PaimonLakeTableLookuper( - paimonConfig, - tablePath, - tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED), - LOOKUP_CACHE_MAX_DISK_BYTES, - NO_OP_DISK_WRITE_GUARD)) { + createLookuper(lookupMode, tablePath, KvFormat.COMPACTED)) { LakeTableLookuper.LookupContext context = new LakeTableLookuper.LookupContext( ResolvedPartitionSpec.fromPartitionName( @@ -631,8 +617,9 @@ void testLookupWithNonStringPartitionKey() throws Exception { } } - @Test - void testRejectAppendOnlyTable() throws Exception { + @ParameterizedTest(name = "lookupMode={0}") + @EnumSource(LookupMode.class) + void testRejectAppendOnlyTableAndLookupAfterClose(LookupMode lookupMode) throws Exception { TablePath tablePath = TablePath.of(DB, "append_only"); Schema schema = Schema.newBuilder() @@ -643,13 +630,7 @@ void testRejectAppendOnlyTable() throws Exception { createPaimonTable(tablePath, tableDescriptor); try (LakeTableLookuper lookuper = - new PaimonLakeTableLookuper( - paimonConfig, - tablePath, - tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED), - LOOKUP_CACHE_MAX_DISK_BYTES, - NO_OP_DISK_WRITE_GUARD)) { + createLookuper(lookupMode, tablePath, KvFormat.COMPACTED)) { LakeTableLookuper.LookupContext context = new LakeTableLookuper.LookupContext( new ResolvedPartitionSpec( @@ -662,11 +643,18 @@ void testRejectAppendOnlyTable() throws Exception { assertThatThrownBy(() -> lookuper.lookup(new byte[0], context)) .isInstanceOf(UnsupportedOperationException.class) .hasMessageContaining("primary-key Paimon tables"); + + lookuper.close(); + assertThatThrownBy(() -> lookuper.lookup(new byte[0], context)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("closed"); } } - @Test - void testLookupAfterSchemaEvolutionPadsNewColumnsWithNull() throws Exception { + @ParameterizedTest(name = "lookupMode={0}") + @EnumSource(LookupMode.class) + void testLookupAfterSchemaEvolutionPadsNewColumnsWithNull(LookupMode lookupMode) + throws Exception { TablePath tablePath = TablePath.of(DB, "schema_evolution_pk"); Schema oldSchema = pkSchema(); TableDescriptor oldDescriptor = partitionedPkDescriptor(oldSchema); @@ -702,13 +690,7 @@ void testLookupAfterSchemaEvolutionPadsNewColumnsWithNull() throws Exception { Collections.singletonList(paimonRow(2, "20240101", "Bob", "new-value")))); try (LakeTableLookuper lookuper = - new PaimonLakeTableLookuper( - paimonConfig, - tablePath, - tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED), - LOOKUP_CACHE_MAX_DISK_BYTES, - NO_OP_DISK_WRITE_GUARD)) { + createLookuper(lookupMode, tablePath, KvFormat.COMPACTED)) { BinaryValue oldSchemaValue = decodeValue( lookuper.lookup( @@ -744,6 +726,74 @@ void testLookupAfterSchemaEvolutionPadsNewColumnsWithNull() throws Exception { } } + private LakeTableLookuper createLookuper( + LookupMode lookupMode, TablePath tablePath, KvFormat kvFormat) { + return createLookuper(lookupMode, tablePath, kvFormat, 1, NO_OP_DISK_WRITE_GUARD); + } + + private LakeTableLookuper createLookuper( + LookupMode lookupMode, + TablePath tablePath, + KvFormat kvFormat, + int kvFormatVersion, + Runnable diskWriteGuard) { + TableConfig tableConfig = tableConfig(kvFormat, kvFormatVersion, lookupMode); + LookuperContext context = + new LookuperContext( + tempWarehouseDir.getAbsolutePath(), + tableConfig, + LOOKUP_CACHE_MAX_DISK_BYTES, + diskWriteGuard); + return new PaimonLakeStorage(paimonConfig).createLakeTableLookuper(tablePath, context); + } + + private FileStoreTable createCompactionTable(TablePath tablePath, Schema schema) + throws Exception { + FileStoreTable table = createPaimonTable(tablePath, partitionedPkDescriptor(schema)); + for (int id = 1; id <= 5; id++) { + writeAndCommitData( + table, + Collections.singletonMap( + 0, Collections.singletonList(paimonRow(id, "20240101", "name-" + id)))); + } + writeAndCommitData( + table, + Collections.singletonMap( + 0, Collections.singletonList(paimonRow(6, "20240102", "name-6")))); + return table; + } + + private void compactAndExpire(FileStoreTable table, BinaryRow partition) throws Exception { + List filesBeforeCompaction = dataFiles(table, partition, 0); + assertThat(filesBeforeCompaction).hasSize(5); + + new CompactHelper(table, new File(tempWarehouseDir, "compact")) + .compactBucket(partition, 0) + .commit(); + assertThat(dataFiles(table, partition, 0)).hasSize(1); + + DataFilePathFactory pathFactory = + table.store().pathFactory().createDataFilePathFactory(partition, 0); + List filesBeforeCompactionPaths = new ArrayList<>(); + for (DataFileMeta file : filesBeforeCompaction) { + Path path = pathFactory.toPath(file); + assertThat(table.store().snapshotManager().fileIO().exists(path)).isTrue(); + filesBeforeCompactionPaths.add(path); + } + + Options expireOptions = new Options(); + expireOptions.set(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN, 1); + expireOptions.set(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX, 1); + // Compaction only marks the replaced files as deleted. Snapshot expiration performs the + // physical cleanup that makes an SST lookuper refresh its stale file set. + try (TableCommitImpl commit = table.copy(expireOptions.toMap()).newCommit("")) { + commit.expireSnapshots(); + } + for (Path path : filesBeforeCompactionPaths) { + assertThat(table.store().snapshotManager().fileIO().exists(path)).isFalse(); + } + } + private FileStoreTable createPaimonTable(TablePath tablePath, TableDescriptor tableDescriptor) throws Exception { lakeCatalog.createTable( @@ -765,14 +815,12 @@ private void refreshPaimonCatalog() throws Exception { CatalogContext.create(Options.fromMap(paimonConfig.toMap()))); } - private static TableConfig tableConfig(KvFormat kvFormat) { - return tableConfig(kvFormat, 1); - } - - private static TableConfig tableConfig(KvFormat kvFormat, int kvFormatVersion) { + private static TableConfig tableConfig( + KvFormat kvFormat, int kvFormatVersion, LookupMode lookupMode) { Configuration config = new Configuration(); config.set(ConfigOptions.TABLE_KV_FORMAT, kvFormat); config.set(ConfigOptions.TABLE_KV_FORMAT_VERSION, kvFormatVersion); + config.set(ConfigOptions.TABLE_DATALAKE_HISTORICAL_PARTITION_LOOKUP_MODE, lookupMode); return new TableConfig(config); } From ec5243fe83dce2ff4e60659a5419ca888c9f690b Mon Sep 17 00:00:00 2001 From: zhangjunfan Date: Tue, 1 Sep 2026 14:06:43 +0800 Subject: [PATCH 4/7] docs: document historical partition lookup modes --- website/docs/engine-flink/ddl.md | 1 + website/docs/engine-flink/lookups.md | 7 ++++++- website/docs/engine-flink/options.md | 1 + 3 files changed, 8 insertions(+), 1 deletion(-) diff --git a/website/docs/engine-flink/ddl.md b/website/docs/engine-flink/ddl.md index 37ca0090a9a..730f9e828c4 100644 --- a/website/docs/engine-flink/ddl.md +++ b/website/docs/engine-flink/ddl.md @@ -312,6 +312,7 @@ ALTER TABLE my_table SET ('table.log.tiered.local-segments' = '5'); **Limits** - If lakehouse storage (`table.datalake.enabled`) is already enabled for a table, options with lakehouse format prefixes (e.g., `paimon.*`) cannot be modified again. - After changing `table.datalake.historical-partition.enabled`, restart existing lookup jobs that need to look up historical partition data so that their clients load the updated table configuration. +- `table.datalake.historical-partition.lookup-mode` can only be configured when creating the table and cannot be modified with `ALTER TABLE SET` or `ALTER TABLE RESET`. ### RESET properties diff --git a/website/docs/engine-flink/lookups.md b/website/docs/engine-flink/lookups.md index 57b2196c627..e00d2a4bf08 100644 --- a/website/docs/engine-flink/lookups.md +++ b/website/docs/engine-flink/lookups.md @@ -248,7 +248,8 @@ PARTITIONED BY (`dt`) WITH ( 'bucket.key' = 'c_custkey', 'table.auto-partition.enabled' = 'true', - 'table.auto-partition.time-unit' = 'year' + 'table.auto-partition.time-unit' = 'year', + 'table.datalake.historical-partition.lookup-mode' = 'SCAN' ); ``` @@ -284,6 +285,10 @@ ALTER TABLE customer_partitioned_with_bucket_key SET ( ); ``` +The lookup mode can be configured only when the table is created and cannot be altered later. +`SST`, the default, creates and caches local lookup files. `SCAN` applies primary-key filters while +scanning Paimon and does not create local lookup files. + This option is disabled by default and currently supports only Paimon primary-key tables with auto partitioning enabled and exactly one partition key. When enabled, the Coordinator creates and retains the `__historical__` system partition used to route lookups to Paimon. Disabling the option diff --git a/website/docs/engine-flink/options.md b/website/docs/engine-flink/options.md index 44189d9e1a9..870ed47525f 100644 --- a/website/docs/engine-flink/options.md +++ b/website/docs/engine-flink/options.md @@ -89,6 +89,7 @@ See more details about [ALTER TABLE ... SET](engine-flink/ddl.md#set-properties) | table.datalake.enabled | Boolean | false | Whether enable lakehouse storage for the table. Disabled by default. When this option is set to ture and the datalake tiering service is up, the table will be tiered and compacted into datalake format stored on lakehouse storage. | | table.datalake.format | Enum | (None) | The data lake format of the table specifies the tiered Lakehouse storage format. Currently, supported formats are `paimon`, `iceberg`, and `lance`. In the future, more kinds of data lake format will be supported, such as DeltaLake or Hudi. Once the `table.datalake.format` property is configured, Fluss adopts the key encoding and bucketing strategy used by the corresponding data lake format. This ensures consistency in key encoding and bucketing, enabling seamless **Union Read** functionality across Fluss and Lakehouse. The `table.datalake.format` can be pre-defined before enabling `table.datalake.enabled`. This allows the data lake feature to be dynamically enabled on the table without requiring table recreation. If `table.datalake.format` is not explicitly set during table creation, the table will default to the format specified by the `datalake.format` configuration in the Fluss cluster. | | table.datalake.historical-partition.enabled | Boolean | false | Whether to enable historical partition lookup for the table. When enabled, the coordinator creates and retains a system partition for routing lookups of expired partitions to lake storage. Currently, this option only supports auto-partitioned Paimon primary key tables with a single partition key. After changing this option, restart existing lookup jobs that need to look up historical partition data so that their clients load the updated table configuration. | +| table.datalake.historical-partition.lookup-mode | Enum | SST | The mode used to look up historical partitions in Paimon. `SST` creates and caches local lookup files. `SCAN` scans the requested partition and bucket with primary-key filters without creating local lookup files. This option can only be set when creating the table and cannot be altered. | | table.datalake.freshness | Duration | 3min | It defines the maximum amount of time that the datalake table's content should lag behind updates to the Fluss table. Based on this target freshness, the Fluss service automatically moves data from the Fluss table and updates to the datalake table, so that the data in the datalake table is kept up to date within this target. If the data does not need to be as fresh, you can specify a longer target freshness time to reduce costs. | | table.datalake.auto-compaction | Boolean | false | If true, compaction will be triggered automatically when tiering service writes to the datalake. It is disabled by default. | | table.datalake.auto-expire-snapshot | Boolean | false | If true, snapshot expiration will be triggered automatically when tiering service commits to the datalake. It is disabled by default. | From 09432ce005012cb82298e42787f1e12e31fbefcc Mon Sep 17 00:00:00 2001 From: zhangjunfan Date: Tue, 1 Sep 2026 14:12:35 +0800 Subject: [PATCH 5/7] refactor(common): move lake lookup mode to metadata --- .../apache/fluss/config/ConfigOptions.java | 23 +++---- .../org/apache/fluss/config/TableConfig.java | 4 +- .../fluss/lake/lakestorage/LakeStorage.java | 14 +---- .../apache/fluss/metadata/LakeLookupMode.java | 27 ++++++++ .../fluss/lake/paimon/PaimonLakeStorage.java | 3 +- .../lookup/PaimonLakeTableLookuperTest.java | 62 ++++++++++--------- 6 files changed, 78 insertions(+), 55 deletions(-) create mode 100644 fluss-common/src/main/java/org/apache/fluss/metadata/LakeLookupMode.java diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java index 42acb521f42..9714987093f 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java @@ -20,11 +20,11 @@ import org.apache.fluss.annotation.Internal; import org.apache.fluss.annotation.PublicEvolving; import org.apache.fluss.compression.ArrowCompressionType; -import org.apache.fluss.lake.lakestorage.LakeStorage.LookupMode; import org.apache.fluss.metadata.ChangelogImage; import org.apache.fluss.metadata.DataLakeFormat; import org.apache.fluss.metadata.DeleteBehavior; import org.apache.fluss.metadata.KvFormat; +import org.apache.fluss.metadata.LakeLookupMode; import org.apache.fluss.metadata.LogFormat; import org.apache.fluss.metadata.MergeEngineType; import org.apache.fluss.rpc.protocol.FetchLogReadPreference; @@ -1935,16 +1935,17 @@ public class ConfigOptions { + "updated table configuration."); /** Lookup strategy for historical partitions stored in lake storage. */ - public static final ConfigOption TABLE_DATALAKE_HISTORICAL_PARTITION_LOOKUP_MODE = - key("table.datalake.historical-partition.lookup-mode") - .enumType(LookupMode.class) - .defaultValue(LookupMode.SST) - .withDescription( - "The lookup mode for historical partitions stored in Paimon. " - + "SST uses local lookup files cached from lake storage. " - + "SCAN scans the requested partition and bucket with primary-key filters " - + "and a limit of one row, without creating local lookup files. " - + "This option can only be set when creating the table and cannot be altered."); + public static final ConfigOption + TABLE_DATALAKE_HISTORICAL_PARTITION_LOOKUP_MODE = + key("table.datalake.historical-partition.lookup-mode") + .enumType(LakeLookupMode.class) + .defaultValue(LakeLookupMode.SST) + .withDescription( + "The lookup mode for historical partitions stored in Paimon. " + + "SST uses local lookup files cached from lake storage. " + + "SCAN scans the requested partition and bucket with primary-key filters " + + "and a limit of one row, without creating local lookup files. " + + "This option can only be set when creating the table and cannot be altered."); public static final ConfigOption TABLE_DATALAKE_FORMAT = key("table.datalake.format") diff --git a/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java b/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java index c86edfe9a0a..49f7ec2fa86 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java @@ -19,11 +19,11 @@ import org.apache.fluss.annotation.PublicEvolving; import org.apache.fluss.compression.ArrowCompressionInfo; -import org.apache.fluss.lake.lakestorage.LakeStorage.LookupMode; import org.apache.fluss.metadata.ChangelogImage; import org.apache.fluss.metadata.DataLakeFormat; import org.apache.fluss.metadata.DeleteBehavior; import org.apache.fluss.metadata.KvFormat; +import org.apache.fluss.metadata.LakeLookupMode; import org.apache.fluss.metadata.LogFormat; import org.apache.fluss.metadata.MergeEngineType; import org.apache.fluss.utils.AutoPartitionStrategy; @@ -136,7 +136,7 @@ public boolean isHistoricalPartitionEnabled() { } /** Gets the lookup mode for historical partitions of the table. */ - public LookupMode getHistoricalLookupMode() { + public LakeLookupMode getHistoricalLookupMode() { return config.get(ConfigOptions.TABLE_DATALAKE_HISTORICAL_PARTITION_LOOKUP_MODE); } diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeStorage.java b/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeStorage.java index f93d7ef0268..478e8d4a3ec 100644 --- a/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeStorage.java +++ b/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeStorage.java @@ -21,6 +21,7 @@ import org.apache.fluss.config.TableConfig; import org.apache.fluss.lake.source.LakeSource; import org.apache.fluss.lake.writer.LakeTieringFactory; +import org.apache.fluss.metadata.LakeLookupMode; import org.apache.fluss.metadata.TablePath; import static org.apache.fluss.utils.Preconditions.checkArgument; @@ -68,22 +69,13 @@ default LakeTableLookuper createLakeTableLookuper( "Point lookup is not supported for this lake storage."); } - /** Mode used to look up historical data in lake storage. */ - enum LookupMode { - /** Use local lookup files cached from lake storage. */ - SST, - - /** Scan lake storage with primary-key filters and return at most one row. */ - SCAN - } - /** Runtime context for creating a lake table lookuper. */ final class LookuperContext { private final String ioTmpDir; private final TableConfig tableConfig; private final long lookupCacheMaxDiskBytes; private final Runnable diskWriteGuard; - private final LookupMode lookupMode; + private final LakeLookupMode lookupMode; /** * Creates a lookuper context. @@ -128,7 +120,7 @@ public Runnable diskWriteGuard() { } /** Returns the mode used to look up historical data. */ - public LookupMode lookupMode() { + public LakeLookupMode lookupMode() { return lookupMode; } } diff --git a/fluss-common/src/main/java/org/apache/fluss/metadata/LakeLookupMode.java b/fluss-common/src/main/java/org/apache/fluss/metadata/LakeLookupMode.java new file mode 100644 index 00000000000..ca825dc51c5 --- /dev/null +++ b/fluss-common/src/main/java/org/apache/fluss/metadata/LakeLookupMode.java @@ -0,0 +1,27 @@ +/* + * 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. + */ + +package org.apache.fluss.metadata; + +/** Mode used to look up historical partitions in lake storage. */ +public enum LakeLookupMode { + /** Use local lookup files cached from lake storage. */ + SST, + + /** Scan lake storage with primary-key filters. */ + SCAN +} diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeStorage.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeStorage.java index dabb7c38434..cace6129a87 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeStorage.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeStorage.java @@ -29,6 +29,7 @@ import org.apache.fluss.lake.paimon.tiering.PaimonWriteResult; import org.apache.fluss.lake.source.LakeSource; import org.apache.fluss.lake.writer.LakeTieringFactory; +import org.apache.fluss.metadata.LakeLookupMode; import org.apache.fluss.metadata.TablePath; /** Paimon implementation of {@link LakeStorage}. */ @@ -57,7 +58,7 @@ public LakeSource createLakeSource(TablePath tablePath) { @Override public LakeTableLookuper createLakeTableLookuper(TablePath tablePath, LookuperContext context) { - if (context.lookupMode() == LookupMode.SCAN) { + if (context.lookupMode() == LakeLookupMode.SCAN) { return new PaimonScanBasedTableLookuper(paimonConfig, tablePath, context.tableConfig()); } return new PaimonLakeTableLookuper( diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuperTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuperTest.java index bb7c0018f80..51ce3f029bc 100644 --- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuperTest.java +++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuperTest.java @@ -22,13 +22,13 @@ import org.apache.fluss.config.MemorySize; import org.apache.fluss.config.TableConfig; import org.apache.fluss.exception.DiskWriteLockedException; -import org.apache.fluss.lake.lakestorage.LakeStorage.LookupMode; import org.apache.fluss.lake.lakestorage.LakeStorage.LookuperContext; import org.apache.fluss.lake.lakestorage.LakeTableLookuper; import org.apache.fluss.lake.lakestorage.TestingLakeCatalogContext; import org.apache.fluss.lake.paimon.PaimonLakeCatalog; import org.apache.fluss.lake.paimon.PaimonLakeStorage; import org.apache.fluss.metadata.KvFormat; +import org.apache.fluss.metadata.LakeLookupMode; import org.apache.fluss.metadata.ResolvedPartitionSpec; import org.apache.fluss.metadata.Schema; import org.apache.fluss.metadata.TableChange; @@ -123,8 +123,8 @@ void tearDown() throws Exception { } @ParameterizedTest(name = "lookupMode={0}") - @EnumSource(LookupMode.class) - void testLookupPartitionedPrimaryKeyTable(LookupMode lookupMode) throws Exception { + @EnumSource(LakeLookupMode.class) + void testLookupPartitionedPrimaryKeyTable(LakeLookupMode lookupMode) throws Exception { TablePath tablePath = TablePath.of(DB, "partitioned_pk"); Schema schema = pkSchema(); TableDescriptor tableDescriptor = partitionedPkDescriptor(schema); @@ -157,14 +157,14 @@ void testLookupPartitionedPrimaryKeyTable(LookupMode lookupMode) throws Exceptio paimonKey(schema, 1, "20240101"), lookupContext(schema, "20240101", 1, SCHEMA_ID))) .isNull(); - if (lookupMode == LookupMode.SST) { + if (lookupMode == LakeLookupMode.SST) { assertThat(lookuper.lookup(compactedKey(schema, 1, "20240101"), context)).isNull(); } // SST creates a local lookup file on the first lookup; SCAN never creates one. assertThat(lookupFileDownloads) .containsExactlyElementsOf( - lookupMode == LookupMode.SST + lookupMode == LakeLookupMode.SST ? Arrays.asList(true, false, false) : Arrays.asList(false, false)); } @@ -208,7 +208,7 @@ void testConcurrentFirstLookupsForDifferentPartitions() throws Exception { paimonConfig, tablePath, tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED, 1, LookupMode.SST), + tableConfig(KvFormat.COMPACTED, 1, LakeLookupMode.SST), LOOKUP_CACHE_MAX_DISK_BYTES, diskWriteGuard)) { Future firstLookup = @@ -238,8 +238,9 @@ void testConcurrentFirstLookupsForDifferentPartitions() throws Exception { } @ParameterizedTest(name = "lookupMode={0}") - @EnumSource(LookupMode.class) - void testDiskWriteLockBlocksOnlyLookupFileDownloads(LookupMode lookupMode) throws Exception { + @EnumSource(LakeLookupMode.class) + void testDiskWriteLockBlocksOnlyLookupFileDownloads(LakeLookupMode lookupMode) + throws Exception { TablePath tablePath = TablePath.of(DB, "disk_write_lock"); Schema schema = pkSchema(); FileStoreTable table = createPaimonTable(tablePath, partitionedPkDescriptor(schema)); @@ -273,7 +274,7 @@ void testDiskWriteLockBlocksOnlyLookupFileDownloads(LookupMode lookupMode) throw // rejected. SCAN performs no local writes and is unaffected by the guard. assertThat(lookuper.lookup(paimonKey(schema, 1, "20240101"), cachedPartition)) .isNotNull(); - if (lookupMode == LookupMode.SST) { + if (lookupMode == LakeLookupMode.SST) { assertThatThrownBy( () -> lookuper.lookup( @@ -292,8 +293,8 @@ void testDiskWriteLockBlocksOnlyLookupFileDownloads(LookupMode lookupMode) throw } @ParameterizedTest(name = "lookupMode={0}") - @EnumSource(LookupMode.class) - void testLookupPartitionsWithSameHashCode(LookupMode lookupMode) throws Exception { + @EnumSource(LakeLookupMode.class) + void testLookupPartitionsWithSameHashCode(LakeLookupMode lookupMode) throws Exception { // These distinct partition values produce the same BinaryRow hash code, reproducing the // mutable-key collision that previously made one partition reuse another partition's files. String firstPartition = "b8"; @@ -340,8 +341,8 @@ void testLookupPartitionsWithSameHashCode(LookupMode lookupMode) throws Exceptio } @ParameterizedTest(name = "lookupMode={0}") - @EnumSource(LookupMode.class) - void testLookupWithIndexedKvFormat(LookupMode lookupMode) throws Exception { + @EnumSource(LakeLookupMode.class) + void testLookupWithIndexedKvFormat(LakeLookupMode lookupMode) throws Exception { TablePath tablePath = TablePath.of(DB, "indexed_kv_format"); Schema schema = pkSchema(); TableDescriptor tableDescriptor = @@ -371,8 +372,8 @@ void testLookupWithIndexedKvFormat(LookupMode lookupMode) throws Exception { } @ParameterizedTest(name = "lookupMode={0}") - @EnumSource(LookupMode.class) - void testLookupKvFormatV2WithNonDefaultBucketKey(LookupMode lookupMode) throws Exception { + @EnumSource(LakeLookupMode.class) + void testLookupKvFormatV2WithNonDefaultBucketKey(LakeLookupMode lookupMode) throws Exception { TablePath tablePath = TablePath.of(DB, "non_default_bucket_key"); Schema schema = Schema.newBuilder() @@ -416,8 +417,8 @@ void testLookupKvFormatV2WithNonDefaultBucketKey(LookupMode lookupMode) throws E } @ParameterizedTest(name = "lookupMode={0}") - @EnumSource(LookupMode.class) - void testRetriesInitializationAfterLookupKeyConverterFailure(LookupMode lookupMode) + @EnumSource(LakeLookupMode.class) + void testRetriesInitializationAfterLookupKeyConverterFailure(LakeLookupMode lookupMode) throws Exception { TablePath tablePath = TablePath.of(DB, "retry_initialization"); Schema schema = @@ -480,8 +481,9 @@ void testRetriesInitializationAfterLookupKeyConverterFailure(LookupMode lookupMo } @ParameterizedTest(name = "lookupMode={0}") - @EnumSource(LookupMode.class) - void testLookupAfterCompactionAndSnapshotExpiration(LookupMode lookupMode) throws Exception { + @EnumSource(LakeLookupMode.class) + void testLookupAfterCompactionAndSnapshotExpiration(LakeLookupMode lookupMode) + throws Exception { TablePath tablePath = TablePath.of(DB, "compacted_lookup"); Schema schema = pkSchema(); FileStoreTable table = createCompactionTable(tablePath, schema); @@ -509,8 +511,8 @@ void testLookupAfterCompactionAndSnapshotExpiration(LookupMode lookupMode) throw } @ParameterizedTest(name = "lookupMode={0}") - @EnumSource(LookupMode.class) - void testLookupsRunConcurrently(LookupMode lookupMode) throws Exception { + @EnumSource(LakeLookupMode.class) + void testLookupsRunConcurrently(LakeLookupMode lookupMode) throws Exception { TablePath tablePath = TablePath.of(DB, "concurrent_lookups"); Schema schema = pkSchema(); FileStoreTable table = createPaimonTable(tablePath, partitionedPkDescriptor(schema)); @@ -575,8 +577,8 @@ void testLookupsRunConcurrently(LookupMode lookupMode) throws Exception { } @ParameterizedTest(name = "lookupMode={0}") - @EnumSource(LookupMode.class) - void testLookupWithNonStringPartitionKey(LookupMode lookupMode) throws Exception { + @EnumSource(LakeLookupMode.class) + void testLookupWithNonStringPartitionKey(LakeLookupMode lookupMode) throws Exception { TablePath tablePath = TablePath.of(DB, "int_partition_pk"); Schema schema = Schema.newBuilder() @@ -618,8 +620,8 @@ void testLookupWithNonStringPartitionKey(LookupMode lookupMode) throws Exception } @ParameterizedTest(name = "lookupMode={0}") - @EnumSource(LookupMode.class) - void testRejectAppendOnlyTableAndLookupAfterClose(LookupMode lookupMode) throws Exception { + @EnumSource(LakeLookupMode.class) + void testRejectAppendOnlyTableAndLookupAfterClose(LakeLookupMode lookupMode) throws Exception { TablePath tablePath = TablePath.of(DB, "append_only"); Schema schema = Schema.newBuilder() @@ -652,8 +654,8 @@ void testRejectAppendOnlyTableAndLookupAfterClose(LookupMode lookupMode) throws } @ParameterizedTest(name = "lookupMode={0}") - @EnumSource(LookupMode.class) - void testLookupAfterSchemaEvolutionPadsNewColumnsWithNull(LookupMode lookupMode) + @EnumSource(LakeLookupMode.class) + void testLookupAfterSchemaEvolutionPadsNewColumnsWithNull(LakeLookupMode lookupMode) throws Exception { TablePath tablePath = TablePath.of(DB, "schema_evolution_pk"); Schema oldSchema = pkSchema(); @@ -727,12 +729,12 @@ void testLookupAfterSchemaEvolutionPadsNewColumnsWithNull(LookupMode lookupMode) } private LakeTableLookuper createLookuper( - LookupMode lookupMode, TablePath tablePath, KvFormat kvFormat) { + LakeLookupMode lookupMode, TablePath tablePath, KvFormat kvFormat) { return createLookuper(lookupMode, tablePath, kvFormat, 1, NO_OP_DISK_WRITE_GUARD); } private LakeTableLookuper createLookuper( - LookupMode lookupMode, + LakeLookupMode lookupMode, TablePath tablePath, KvFormat kvFormat, int kvFormatVersion, @@ -816,7 +818,7 @@ private void refreshPaimonCatalog() throws Exception { } private static TableConfig tableConfig( - KvFormat kvFormat, int kvFormatVersion, LookupMode lookupMode) { + KvFormat kvFormat, int kvFormatVersion, LakeLookupMode lookupMode) { Configuration config = new Configuration(); config.set(ConfigOptions.TABLE_KV_FORMAT, kvFormat); config.set(ConfigOptions.TABLE_KV_FORMAT_VERSION, kvFormatVersion); From a946789ee7e88815cda180c1c4a08813db42aca1 Mon Sep 17 00:00:00 2001 From: zhangjunfan Date: Tue, 1 Sep 2026 14:16:30 +0800 Subject: [PATCH 6/7] test(paimon): simplify lake table lookup assertions --- .../paimon/lookup/PaimonLakeTableLookuperTest.java | 11 +---------- 1 file changed, 1 insertion(+), 10 deletions(-) diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuperTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuperTest.java index 51ce3f029bc..9061090ba01 100644 --- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuperTest.java +++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuperTest.java @@ -157,15 +157,11 @@ void testLookupPartitionedPrimaryKeyTable(LakeLookupMode lookupMode) throws Exce paimonKey(schema, 1, "20240101"), lookupContext(schema, "20240101", 1, SCHEMA_ID))) .isNull(); - if (lookupMode == LakeLookupMode.SST) { - assertThat(lookuper.lookup(compactedKey(schema, 1, "20240101"), context)).isNull(); - } - // SST creates a local lookup file on the first lookup; SCAN never creates one. assertThat(lookupFileDownloads) .containsExactlyElementsOf( lookupMode == LakeLookupMode.SST - ? Arrays.asList(true, false, false) + ? Arrays.asList(true, false) : Arrays.asList(false, false)); } } @@ -911,11 +907,6 @@ private static byte[] paimonKey(Schema schema, List keys, Object... fiel return new PaimonKeyEncoder(schema.getRowType(), keys).encodeKey(row(fields)); } - private static byte[] compactedKey(Schema schema, int id, String dt) { - return CompactedKeyEncoder.createKeyEncoder(schema.getRowType(), Arrays.asList("id", "dt")) - .encodeKey(row(id, dt, "")); - } - private static BinaryValue decodeValue(byte[] value, short schemaId, Schema schema) { return decodeValue(value, schemaId, schema, KvFormat.COMPACTED); } From 20aff0720a5e6faaf0e43c6af930c451d8068726 Mon Sep 17 00:00:00 2001 From: zhangjunfan Date: Tue, 1 Sep 2026 14:21:59 +0800 Subject: [PATCH 7/7] test(paimon): cover lookups in computed buckets --- .../lookup/PaimonLakeTableLookuperTest.java | 49 +++++++++++++++++++ 1 file changed, 49 insertions(+) diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuperTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuperTest.java index 9061090ba01..dc3c4b0bf64 100644 --- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuperTest.java +++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuperTest.java @@ -17,6 +17,7 @@ package org.apache.fluss.lake.paimon.lookup; +import org.apache.fluss.bucketing.PaimonBucketingFunction; import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; import org.apache.fluss.config.MemorySize; @@ -166,6 +167,54 @@ void testLookupPartitionedPrimaryKeyTable(LakeLookupMode lookupMode) throws Exce } } + @ParameterizedTest(name = "lookupMode={0}") + @EnumSource(LakeLookupMode.class) + void testLookupKeysInComputedBuckets(LakeLookupMode lookupMode) throws Exception { + TablePath tablePath = TablePath.of(DB, "computed_buckets"); + Schema schema = pkSchema(); + FileStoreTable table = createPaimonTable(tablePath, partitionedPkDescriptor(schema)); + PaimonKeyEncoder bucketKeyEncoder = + new PaimonKeyEncoder(schema.getRowType(), Collections.singletonList("id")); + PaimonBucketingFunction bucketingFunction = new PaimonBucketingFunction(); + int firstBucket = + bucketingFunction.bucketing( + bucketKeyEncoder.encodeKey(row(1, "20240101", "Alice")), 2); + int secondBucket = + bucketingFunction.bucketing( + bucketKeyEncoder.encodeKey(row(3, "20240101", "Bob")), 2); + assertThat(firstBucket).isNotEqualTo(secondBucket); + + writeAndCommitData( + table, + Collections.singletonMap( + firstBucket, Collections.singletonList(paimonRow(1, "20240101", "Alice")))); + writeAndCommitData( + table, + Collections.singletonMap( + secondBucket, Collections.singletonList(paimonRow(3, "20240101", "Bob")))); + + try (LakeTableLookuper lookuper = + createLookuper(lookupMode, tablePath, KvFormat.COMPACTED)) { + BinaryValue firstValue = + decodeValue( + lookuper.lookup( + paimonKey(schema, 1, "20240101"), + lookupContext(schema, "20240101", firstBucket, SCHEMA_ID)), + SCHEMA_ID, + schema); + BinaryValue secondValue = + decodeValue( + lookuper.lookup( + paimonKey(schema, 3, "20240101"), + lookupContext(schema, "20240101", secondBucket, SCHEMA_ID)), + SCHEMA_ID, + schema); + + assertRow(firstValue.row, 1, "20240101", "Alice"); + assertRow(secondValue.row, 3, "20240101", "Bob"); + } + } + @Test void testConcurrentFirstLookupsForDifferentPartitions() throws Exception { TablePath tablePath = TablePath.of(DB, "concurrent_first_lookups");