luoyuxia commented on code in PR #3630: URL: https://github.com/apache/fluss/pull/3630#discussion_r3601658200
########## fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java: ########## @@ -0,0 +1,359 @@ +/* + * 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.ConfigOptions; +import org.apache.fluss.config.Configuration; +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.ResolvedPartitionSpec; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.row.BinaryRow; +import org.apache.fluss.row.InternalRow; +import org.apache.fluss.row.encode.CompactedRowEncoder; +import org.apache.fluss.row.encode.ValueEncoder; +import org.apache.fluss.types.DataType; +import org.apache.fluss.types.RowType; +import org.apache.fluss.utils.IOUtils; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.catalog.Catalog; +import org.apache.paimon.catalog.CatalogContext; +import org.apache.paimon.catalog.CatalogFactory; +import org.apache.paimon.disk.IOManager; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.memory.MemorySegment; +import org.apache.paimon.options.Options; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.query.LocalTableQuery; +import org.apache.paimon.table.sink.RowPartitionKeyExtractor; +import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.table.source.InnerTableScan; +import org.apache.paimon.table.source.Split; +import org.apache.paimon.types.DataField; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Set; + +import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.SYSTEM_COLUMNS; +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; + +/** + * Paimon implementation of {@link LakeTableLookuper} for primary-key tables. + * + * <p>The catalog, table, local query, and I/O manager are initialized lazily on the first lookup. + * For each partition and bucket, the lookuper scans the latest Paimon snapshot once and registers + * its data files with {@link LocalTableQuery}. Paimon then creates local lookup files lazily as + * individual remote data files are queried. + * + * <p>A cached partition-bucket file set can become stale when Paimon compaction replaces its data + * files and snapshot expiration physically deletes the old files. Because {@code FileIO} + * implementations may represent a missing file with different {@link IOException} types, the first + * lookup I/O failure closes the cached query state, reopens the table from the latest snapshot, and + * retries once. + * + * <p>Lookup and close operations are synchronized because they share mutable Paimon query, local + * cache, and value-encoding state. + */ +public class PaimonLakeTableLookuper implements LakeTableLookuper { + + // Hard-code a conservative 1GB default for now to avoid unbounded lookup cache growth. + private static final String DEFAULT_LOOKUP_CACHE_MAX_DISK_SIZE = "1gb"; + + private final Configuration paimonConfig; + private final TablePath tablePath; + private final String ioTmpDir; + + private final Set<PaimonPartitionBucket> initializedBuckets; + + private @Nullable Catalog catalog; + private @Nullable FileStoreTable fileStoreTable; + private @Nullable IOManager ioManager; + private @Nullable LocalTableQuery localTableQuery; + private @Nullable RowPartitionKeyExtractor partitionKeyExtractor; + private int primaryKeyFieldCount; + private boolean hasCachedValueEncoder; + private short cachedValueSchemaId; + private @Nullable CompactedRowEncoder cachedValueRowEncoder; + private @Nullable InternalRow.FieldGetter[] cachedValueFieldGetters; + private boolean closed; + + public PaimonLakeTableLookuper(Configuration paimonConfig, TablePath tablePath) { + this(paimonConfig, tablePath, ConfigOptions.IO_TMP_DIR.defaultValue()); + } + + public PaimonLakeTableLookuper( + Configuration paimonConfig, TablePath tablePath, String ioTmpDir) { + this.paimonConfig = checkNotNull(paimonConfig, "paimonConfig must not be null."); + this.tablePath = checkNotNull(tablePath, "tablePath must not be null."); + this.ioTmpDir = checkNotNull(ioTmpDir, "ioTmpDir must not be null."); + this.initializedBuckets = new HashSet<>(); + } + + @Override + public synchronized @Nullable byte[] lookup(byte[] key, LookupContext context) Review Comment: Good point. For the initial implementation, I’d prefer to keep using the shared executor. Historical lookup is optional, and introducing a dedicated pool and another configuration option would add resources and configuration surface for clusters that may never use it. We expect the initial lookup traffic to be limited and can tune the existing I/O pool for deployments that need it. I suggest revisiting a dedicated, possibly lazily created executor if real workloads show contention -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
