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..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 @@ -24,6 +24,7 @@ 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; @@ -1933,6 +1934,19 @@ 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(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") .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..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 @@ -23,6 +23,7 @@ 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; @@ -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 LakeLookupMode 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..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; @@ -74,6 +75,7 @@ final class LookuperContext { private final TableConfig tableConfig; private final long lookupCacheMaxDiskBytes; private final Runnable diskWriteGuard; + private final LakeLookupMode lookupMode; /** * Creates a lookuper context. @@ -94,6 +96,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 +118,10 @@ public long lookupCacheMaxDiskBytes() { public Runnable diskWriteGuard() { return diskWriteGuard; } + + /** Returns the mode used to look up historical data. */ + 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 80e66398985..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 @@ -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; @@ -28,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}. */ @@ -56,6 +58,9 @@ public LakeSource createLakeSource(TablePath tablePath) { @Override public LakeTableLookuper createLakeTableLookuper(TablePath tablePath, LookuperContext context) { + if (context.lookupMode() == LakeLookupMode.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/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 new file mode 100644 index 00000000000..db347e49986 --- /dev/null +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonScanBasedTableLookuper.java @@ -0,0 +1,217 @@ +/* + * 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.metadata.DataLakeFormat; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.row.decode.KeyDecoder; +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.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; +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 toFlussValue( + row, + context.schemaId(), + context.valueRowType(), + tableConfig.getKvFormat()); + } + } finally { + batch.releaseBatch(); + } + } + } + return null; + } +} 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. */ 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..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,15 +17,19 @@ 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; import org.apache.fluss.config.TableConfig; import org.apache.fluss.exception.DiskWriteLockedException; +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; @@ -59,6 +63,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 +86,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 +123,9 @@ void tearDown() throws Exception { } } - @Test - void testLookupPartitionedPrimaryKeyTable() throws Exception { + @ParameterizedTest(name = "lookupMode={0}") + @EnumSource(LakeLookupMode.class) + void testLookupPartitionedPrimaryKeyTable(LakeLookupMode lookupMode) throws Exception { TablePath tablePath = TablePath.of(DB, "partitioned_pk"); Schema schema = pkSchema(); TableDescriptor tableDescriptor = partitionedPkDescriptor(schema); @@ -129,13 +136,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 +158,60 @@ void testLookupPartitionedPrimaryKeyTable() throws Exception { paimonKey(schema, 1, "20240101"), lookupContext(schema, "20240101", 1, SCHEMA_ID))) .isNull(); - 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) + : Arrays.asList(false, false)); + } + } + + @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); - // The first lookup creates the local lookup file, while subsequent lookups reuse it. - assertThat(lookupFileDownloads).containsExactly(true, false, false); + assertRow(firstValue.row, 1, "20240101", "Alice"); + assertRow(secondValue.row, 3, "20240101", "Bob"); } } @@ -202,7 +253,7 @@ void testConcurrentFirstLookupsForDifferentPartitions() throws Exception { paimonConfig, tablePath, tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED), + tableConfig(KvFormat.COMPACTED, 1, LakeLookupMode.SST), LOOKUP_CACHE_MAX_DISK_BYTES, diskWriteGuard)) { Future firstLookup = @@ -231,8 +282,10 @@ void testConcurrentFirstLookupsForDifferentPartitions() throws Exception { } } - @Test - void testDiskWriteLockBlocksOnlyLookupFileDownloads() throws Exception { + @ParameterizedTest(name = "lookupMode={0}") + @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)); @@ -252,13 +305,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 +315,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 == LakeLookupMode.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 +337,9 @@ void testDiskWriteLockBlocksOnlyLookupFileDownloads() throws Exception { } } - @Test - void testLookupPartitionsWithSameHashCode() throws Exception { + @ParameterizedTest(name = "lookupMode={0}") + @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"; @@ -309,13 +364,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 +385,9 @@ void testLookupPartitionsWithSameHashCode() throws Exception { } } - @Test - void testLookupWithIndexedKvFormat() throws Exception { + @ParameterizedTest(name = "lookupMode={0}") + @EnumSource(LakeLookupMode.class) + void testLookupWithIndexedKvFormat(LakeLookupMode lookupMode) throws Exception { TablePath tablePath = TablePath.of(DB, "indexed_kv_format"); Schema schema = pkSchema(); TableDescriptor tableDescriptor = @@ -350,14 +400,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 +416,9 @@ void testLookupWithIndexedKvFormat() throws Exception { } } - @Test - void testLookupKvFormatV2WithNonDefaultBucketKey() throws Exception { + @ParameterizedTest(name = "lookupMode={0}") + @EnumSource(LakeLookupMode.class) + void testLookupKvFormatV2WithNonDefaultBucketKey(LakeLookupMode lookupMode) throws Exception { TablePath tablePath = TablePath.of(DB, "non_default_bucket_key"); Schema schema = Schema.newBuilder() @@ -397,12 +441,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 +461,10 @@ void testLookupKvFormatV2WithNonDefaultBucketKey() throws Exception { } } - @Test - void testRetriesInitializationAfterLookupKeyConverterFailure() throws Exception { + @ParameterizedTest(name = "lookupMode={0}") + @EnumSource(LakeLookupMode.class) + void testRetriesInitializationAfterLookupKeyConverterFailure(LakeLookupMode lookupMode) + throws Exception { TablePath tablePath = TablePath.of(DB, "retry_initialization"); Schema schema = Schema.newBuilder() @@ -454,12 +499,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 +515,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 +525,105 @@ void testRetriesInitializationAfterLookupKeyConverterFailure() throws Exception } } - @Test - void testRefreshFilesAfterCompactionAndSnapshotExpiration() throws Exception { - TablePath tablePath = TablePath.of(DB, "compacted_pk"); + @ParameterizedTest(name = "lookupMode={0}") + @EnumSource(LakeLookupMode.class) + void testLookupAfterCompactionAndSnapshotExpiration(LakeLookupMode 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 = - decodeValue( - lookuper.lookup(paimonKey(schema, 1, "20240101"), refreshedContext), - SCHEMA_ID, - schema); - assertRow(refreshedValue.row, 1, "20240101", "name-1"); - - BinaryValue cachedValue = + BinaryValue value = 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(LakeLookupMode.class) + void testLookupsRunConcurrently(LakeLookupMode 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(LakeLookupMode.class) + void testLookupWithNonStringPartitionKey(LakeLookupMode lookupMode) throws Exception { TablePath tablePath = TablePath.of(DB, "int_partition_pk"); Schema schema = Schema.newBuilder() @@ -605,13 +644,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 +664,9 @@ void testLookupWithNonStringPartitionKey() throws Exception { } } - @Test - void testRejectAppendOnlyTable() throws Exception { + @ParameterizedTest(name = "lookupMode={0}") + @EnumSource(LakeLookupMode.class) + void testRejectAppendOnlyTableAndLookupAfterClose(LakeLookupMode lookupMode) throws Exception { TablePath tablePath = TablePath.of(DB, "append_only"); Schema schema = Schema.newBuilder() @@ -643,13 +677,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 +690,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(LakeLookupMode.class) + void testLookupAfterSchemaEvolutionPadsNewColumnsWithNull(LakeLookupMode lookupMode) + throws Exception { TablePath tablePath = TablePath.of(DB, "schema_evolution_pk"); Schema oldSchema = pkSchema(); TableDescriptor oldDescriptor = partitionedPkDescriptor(oldSchema); @@ -702,13 +737,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 +773,74 @@ void testLookupAfterSchemaEvolutionPadsNewColumnsWithNull() throws Exception { } } + private LakeTableLookuper createLookuper( + LakeLookupMode lookupMode, TablePath tablePath, KvFormat kvFormat) { + return createLookuper(lookupMode, tablePath, kvFormat, 1, NO_OP_DISK_WRITE_GUARD); + } + + private LakeTableLookuper createLookuper( + LakeLookupMode 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 +862,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, LakeLookupMode 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); } @@ -861,11 +956,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); } 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. |