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 a0104c7eba..b3d43be146 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; @@ -1902,6 +1903,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 1cdc529566..38a2610201 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; @@ -112,6 +113,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 446aa1184c..f93d7ef026 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 80e6639898..dabb7c3843 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 0000000000..9d12e978ad --- /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()); + } + } +}