diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java index 150b84e5679..93733b32bb8 100644 --- a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java +++ b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java @@ -124,6 +124,7 @@ import static org.apache.fluss.testutils.DataTestUtils.row; import static org.apache.fluss.testutils.InternalRowAssert.assertThatRow; import static org.apache.fluss.testutils.common.CommonTestUtils.waitUntil; +import static org.apache.fluss.utils.PartitionUtils.HISTORICAL_PARTITION_VALUE; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -1102,22 +1103,33 @@ void testListPartitionInfos() throws Exception { .column("id", DataTypes.STRING()) .column("name", DataTypes.STRING()) .column("pt", DataTypes.STRING()) + .primaryKey("id", "pt") .build()) .comment("test table") .distributedBy(3, "id") .partitionedBy("pt") .property(ConfigOptions.TABLE_AUTO_PARTITION_ENABLED, true) + .property(ConfigOptions.TABLE_AUTO_PARTITION_KEY, "pt") .property( ConfigOptions.TABLE_AUTO_PARTITION_TIME_UNIT, AutoPartitionTimeUnit.YEAR) + .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true) + .property(ConfigOptions.TABLE_DATALAKE_FORMAT, PAIMON) + .property(ConfigOptions.TABLE_DATALAKE_HISTORICAL_PARTITION_ENABLED, true) .build(); TablePath partitionedTablePath = TablePath.of(dbName, "test_partitioned_table"); admin.createTable(partitionedTablePath, partitionedTable, true).get(); Map partitionIdByNames = - FLUSS_CLUSTER_EXTENSION.waitUntilPartitionAllReady(partitionedTablePath); + FLUSS_CLUSTER_EXTENSION.waitUntilPartitionAllReady( + partitionedTablePath, + ConfigOptions.TABLE_AUTO_PARTITION_NUM_PRECREATE.defaultValue() + 1); + assertThat(partitionIdByNames).containsKey(HISTORICAL_PARTITION_VALUE); List partitionInfos = admin.listPartitionInfos(partitionedTablePath).get(); - assertThat(partitionInfos).hasSize(partitionIdByNames.size()); + assertThat(partitionInfos) + .hasSize(partitionIdByNames.size() - 1) + .extracting(PartitionInfo::getPartitionName) + .doesNotContain(HISTORICAL_PARTITION_VALUE); for (PartitionInfo partitionInfo : partitionInfos) { assertThat(partitionIdByNames.get(partitionInfo.getPartitionName())) .isEqualTo(partitionInfo.getPartitionId()); 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 23c50097ba6..455096713a7 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 @@ -428,6 +428,24 @@ public class ConfigOptions { "The maximum number of threads used for historical partition operations, such as lake lookups and writes. " + "Threads are started lazily and released after the keep-alive timeout when idle."); + public static final ConfigOption + SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO = + key("server.historical-partition.lookup-cache.max-disk-ratio") + .doubleType() + .defaultValue(0.10) + .withDescription( + "The maximum fraction of the total capacity of the volume containing the first available data directory allocated to historical partition lookup caches on a TabletServer. " + + "Up to ten table lookupers are cached, and each receives one tenth of this capacity. Historical lookup cache files are stored under that data directory; additional data volumes are not used. " + + "The valid range is (0.0, 1.0]."); + + public static final ConfigOption + SERVER_HISTORICAL_PARTITION_LOOKUPER_CACHE_EXPIRE_AFTER_ACCESS = + key("server.historical-partition.lookuper-cache.expire-after-access") + .durationType() + .defaultValue(Duration.ofHours(3)) + .withDescription( + "The duration after which an idle historical partition table lookuper is removed from the cache."); + public static final ConfigOption SERVER_DATA_DISK_WRITE_LIMIT_RATIO = key("server.data-disk.write-limit-ratio") .doubleType() diff --git a/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java b/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java index 593b0d61741..5f1489baac3 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java @@ -222,6 +222,11 @@ protected static void validateServerConfigs(Configuration conf) { validMinValue(conf, ConfigOptions.SERVER_IO_POOL_SIZE, 1); validMinValue(conf, ConfigOptions.BACKGROUND_THREADS, 1); validMinDuration(conf, ConfigOptions.LOG_RETENTION_CHECK_INTERVAL, 1); + validMinDuration( + conf, + ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUPER_CACHE_EXPIRE_AFTER_ACCESS, + 1); + validateHistoricalLookupCacheRatio(conf); if (conf.get(ConfigOptions.LOG_SEGMENT_FILE_SIZE).getBytes() > Integer.MAX_VALUE) { throw new IllegalConfigurationException( @@ -231,6 +236,16 @@ protected static void validateServerConfigs(Configuration conf) { } } + private static void validateHistoricalLookupCacheRatio(Configuration conf) { + double historicalLookupCacheMaxRatio = + conf.get(ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO); + if (!(historicalLookupCacheMaxRatio > 0.0 && historicalLookupCacheMaxRatio <= 1.0)) { + throw new IllegalConfigurationException( + "Invalid configuration for %s, it must be within (0.0, 1.0].", + ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO.key()); + } + } + private static void validMinValue( Configuration conf, ConfigOption option, int minValue) { validMinValue(option, conf.get(option), minValue); 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 fa0aa1dccd8..446aa1184c4 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 @@ -23,6 +23,7 @@ import org.apache.fluss.lake.writer.LakeTieringFactory; import org.apache.fluss.metadata.TablePath; +import static org.apache.fluss.utils.Preconditions.checkArgument; import static org.apache.fluss.utils.Preconditions.checkNotNull; /** @@ -71,16 +72,28 @@ default LakeTableLookuper createLakeTableLookuper( final class LookuperContext { private final String ioTmpDir; private final TableConfig tableConfig; + private final long lookupCacheMaxDiskBytes; + private final Runnable diskWriteGuard; /** * Creates a lookuper context. * * @param ioTmpDir local directory for temporary files used by the lookuper * @param tableConfig configuration of the Fluss table + * @param lookupCacheMaxDiskBytes maximum local lookup cache size in bytes + * @param diskWriteGuard guard invoked before creating a local lookup cache file */ - public LookuperContext(String ioTmpDir, TableConfig tableConfig) { + public LookuperContext( + String ioTmpDir, + TableConfig tableConfig, + long lookupCacheMaxDiskBytes, + Runnable diskWriteGuard) { this.ioTmpDir = checkNotNull(ioTmpDir, "ioTmpDir must not be null."); this.tableConfig = checkNotNull(tableConfig, "tableConfig must not be null."); + checkArgument( + lookupCacheMaxDiskBytes > 0, "lookupCacheMaxDiskBytes must be greater than 0."); + this.lookupCacheMaxDiskBytes = lookupCacheMaxDiskBytes; + this.diskWriteGuard = checkNotNull(diskWriteGuard, "diskWriteGuard must not be null."); } /** Returns the local directory for temporary files used by the lookuper. */ @@ -92,5 +105,15 @@ public String ioTmpDir() { public TableConfig tableConfig() { return tableConfig; } + + /** Returns the maximum local lookup cache size in bytes. */ + public long lookupCacheMaxDiskBytes() { + return lookupCacheMaxDiskBytes; + } + + /** Returns the guard invoked before creating a local lookup cache file. */ + public Runnable diskWriteGuard() { + return diskWriteGuard; + } } } diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeTableLookuper.java b/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeTableLookuper.java index b4dbb046273..d5d22888157 100644 --- a/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeTableLookuper.java +++ b/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeTableLookuper.java @@ -33,6 +33,19 @@ @PublicEvolving public interface LakeTableLookuper extends AutoCloseable { + /** Records metrics for a lake table point lookup. */ + @FunctionalInterface + interface LookupMetricRecorder { + + /** + * Records a completed lake table point lookup. + * + * @param lookupTimeNanos time spent on the lake table point lookup, in nanoseconds + * @param lookupFileDownloaded whether the lookup downloaded a lookup file + */ + void recordLookup(long lookupTimeNanos, boolean lookupFileDownloaded); + } + /** * Looks up one key from the lake table. * @@ -52,6 +65,7 @@ final class LookupContext { private final int bucketId; private final short schemaId; private final RowType valueRowType; + private final LookupMetricRecorder lookupMetricRecorder; /** * Creates a lookup context. @@ -60,16 +74,20 @@ final class LookupContext { * @param bucketId target bucket id in the lake table * @param schemaId schema id to encode the returned Fluss value with * @param valueRowType row type to encode the returned Fluss value with + * @param lookupMetricRecorder recorder for lake table point lookup metrics */ public LookupContext( ResolvedPartitionSpec partitionSpec, int bucketId, short schemaId, - RowType valueRowType) { + RowType valueRowType, + LookupMetricRecorder lookupMetricRecorder) { this.partitionSpec = checkNotNull(partitionSpec, "partitionSpec must not be null."); this.bucketId = bucketId; this.schemaId = schemaId; this.valueRowType = checkNotNull(valueRowType, "valueRowType must not be null."); + this.lookupMetricRecorder = + checkNotNull(lookupMetricRecorder, "lookupMetricRecorder must not be null."); } /** Returns the resolved Fluss partition spec for the lookup. */ @@ -91,5 +109,10 @@ public short schemaId() { public RowType valueRowType() { return valueRowType; } + + /** Returns the recorder for lake table point lookup metrics. */ + public LookupMetricRecorder lookupMetricRecorder() { + return lookupMetricRecorder; + } } } diff --git a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java index 7a9c50a86d9..98f93ba8ad1 100644 --- a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java +++ b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java @@ -92,6 +92,7 @@ public class MetricNames { "delayedFetchFromFollowerExpiresPerSecond"; public static final String DELAYED_FETCH_FROM_CLIENT_EXPIRES_RATE = "delayedFetchFromClientExpiresPerSecond"; + public static final String HISTORICAL_INFLIGHT_REQUESTS = "inflightRequests"; public static final String SERVER_LOGICAL_STORAGE_LOG_SIZE = "logSize"; public static final String SERVER_LOGICAL_STORAGE_KV_SIZE = "kvSize"; @@ -102,6 +103,12 @@ public class MetricNames { public static final String DISK_USAGE_RATIO = "diskUsageRatio"; public static final String DISK_WRITE_LOCKED = "diskWriteLocked"; + // for historical lookup cache + public static final String HISTORICAL_LOOKUP_CACHE_DISK_SIZE = "lookupCacheDiskSize"; + public static final String HISTORICAL_LOOKUP_CACHE_TABLE_COUNT = "lookupCacheTableCount"; + public static final String HISTORICAL_LOOKUP_CACHE_CAPACITY_EVICTIONS = + "lookupCacheCapacityEvictions"; + // -------------------------------------------------------------------------------------------- // metrics for user // -------------------------------------------------------------------------------------------- @@ -130,6 +137,8 @@ public class MetricNames { public static final String TOTAL_LOOKUP_REQUESTS_RATE = "totalLookupRequestsPerSecond"; public static final String FAILED_LOOKUP_REQUESTS_RATE = "failedLookupRequestsPerSecond"; + public static final String LAKE_LOOKUPS_RATE = "lakeLookupsPerSecond"; + public static final String LAKE_LOOKUP_TIME_MS = "lakeLookupTimeMs"; public static final String TOTAL_PUT_KV_REQUESTS_RATE = "totalPutKvRequestsPerSecond"; public static final String FAILED_PUT_KV_REQUESTS_RATE = "failedPutKvRequestsPerSecond"; public static final String TOTAL_LIMIT_SCAN_REQUESTS_RATE = "totalLimitScanRequestsPerSecond"; diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/FlussPaths.java b/fluss-common/src/main/java/org/apache/fluss/utils/FlussPaths.java index 1c75663ba39..fef00c6c558 100644 --- a/fluss-common/src/main/java/org/apache/fluss/utils/FlussPaths.java +++ b/fluss-common/src/main/java/org/apache/fluss/utils/FlussPaths.java @@ -37,7 +37,8 @@ import static org.apache.fluss.utils.Preconditions.checkState; /** - * Central place for defining all the paths of kv and log local/remote files/directories. + * Central place for defining all the paths of kv, log, and historical lookup local/remote + * files/directories. * *

All the local file/directories returns the {@link File java.io.File} interface. * @@ -52,6 +53,9 @@ public class FlussPaths { /** Prefix of a local kv tablet directory to store kv files for a specific kv tablet. */ public static final String KV_TABLET_DIR_PREFIX = "kv-"; + /** The directory name for historical lookup cache files under a local data directory. */ + public static final String HISTORICAL_LOOKUP_CACHE_DIR_NAME = ".historical-lookup-cache"; + /** Prefix for a partition id to distinguish between table id and partition id. */ public static final String PARTITION_DIR_PREFIX = "p"; @@ -147,6 +151,37 @@ public static File kvTabletDir( return tabletParentDir.resolve(KV_TABLET_DIR_PREFIX + tableBucket.getBucket()).toFile(); } + /** + * Returns the historical lookup cache root under the local data directory. + * + * @param dataDir the local data directory + */ + public static File historicalLookupRootDir(File dataDir) { + return new File(dataDir, HISTORICAL_LOOKUP_CACHE_DIR_NAME); + } + + /** + * Returns the local directory path for storing historical lookup files for a table. + * + *

The path contract: + * + * 

+     * {$lookupRoot}/{databaseName}/{tableName}-{tableId}
+     *
+ * + * @param lookupRoot the historical lookup root directory + * @param tablePath the table path + * @param tableId the table ID + */ + public static File historicalLookupTableDir( + File lookupRoot, TablePath tablePath, long tableId) { + return Paths.get( + lookupRoot.getAbsolutePath(), + tablePath.getDatabaseName(), + tablePath.getTableName() + "-" + tableId) + .toFile(); + } + private static Path tabletParentDir( File dataDir, PhysicalTablePath tablePath, TableBucket tableBucket) { String dbName = tablePath.getDatabaseName(); diff --git a/fluss-common/src/test/java/org/apache/fluss/config/FlussConfigUtilsTest.java b/fluss-common/src/test/java/org/apache/fluss/config/FlussConfigUtilsTest.java index 2b9e815ee7d..61a00c01d2e 100644 --- a/fluss-common/src/test/java/org/apache/fluss/config/FlussConfigUtilsTest.java +++ b/fluss-common/src/test/java/org/apache/fluss/config/FlussConfigUtilsTest.java @@ -215,6 +215,34 @@ void testValidateLogRetentionCheckInterval() { validateCoordinatorConfigs(conf); } + @Test + void testValidateHistoricalLookupCacheConfigs() { + Configuration conf = new Configuration(); + conf.set(ConfigOptions.REMOTE_DATA_DIR, "s3://bucket/path"); + conf.set(ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO, 0.0); + + assertThatThrownBy(() -> validateCoordinatorConfigs(conf)) + .isInstanceOf(IllegalConfigurationException.class) + .hasMessageContaining( + ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO.key()) + .hasMessageContaining("within (0.0, 1.0]"); + + conf.set(ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO, 1.01); + assertThatThrownBy(() -> validateCoordinatorConfigs(conf)) + .isInstanceOf(IllegalConfigurationException.class) + .hasMessageContaining("within (0.0, 1.0]"); + + conf.set(ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO, 0.1); + conf.set( + ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUPER_CACHE_EXPIRE_AFTER_ACCESS, + Duration.ZERO); + assertThatThrownBy(() -> validateCoordinatorConfigs(conf)) + .isInstanceOf(IllegalConfigurationException.class) + .hasMessageContaining( + ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUPER_CACHE_EXPIRE_AFTER_ACCESS + .key()); + } + @Test void testValidateClientConfigs() { // valid defaults should pass diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/SetClusterConfigsProcedure.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/SetClusterConfigsProcedure.java index 7246706af8b..b474a37ae38 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/SetClusterConfigsProcedure.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/SetClusterConfigsProcedure.java @@ -46,6 +46,7 @@ * -- Set a configuration * CALL sys.set_cluster_configs('kv.rocksdb.shared-rate-limiter.bytes-per-sec', '200MB'); * CALL sys.set_cluster_configs('datalake.format', 'paimon'); + * CALL sys.set_cluster_configs('server.historical-partition.lookup-cache.max-disk-ratio', '0.12'); * * -- Set multiple configurations at one time * CALL sys.set_cluster_configs('kv.rocksdb.shared-rate-limiter.bytes-per-sec', '200MB','datalake.format', 'paimon'); diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java index 872932f3ca9..f1c71f0c26c 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/procedure/FlinkProcedureITCase.java @@ -404,13 +404,16 @@ void testSetClusterConfigs() throws Exception { try (CloseableIterator resultIterator = tEnv.executeSql( String.format( - "Call %s.sys.set_cluster_configs('%s', '300MB', '%s', 'paimon')", + "Call %s.sys.set_cluster_configs('%s', '300MB', '%s', 'paimon', '%s', '0.12')", CATALOG_NAME, ConfigOptions.KV_SHARED_RATE_LIMITER_BYTES_PER_SEC.key(), - ConfigOptions.DATALAKE_FORMAT.key())) + ConfigOptions.DATALAKE_FORMAT.key(), + ConfigOptions + .SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO + .key())) .collect()) { List results = CollectionUtil.iteratorToList(resultIterator); - assertThat(results).hasSize(2); + assertThat(results).hasSize(3); assertThat(results.get(0).getField(0)) .asString() .contains("Successfully set to '300MB'") @@ -420,6 +423,13 @@ void testSetClusterConfigs() throws Exception { .asString() .contains("Successfully set to 'paimon'") .contains(ConfigOptions.DATALAKE_FORMAT.key()); + + assertThat(results.get(2).getField(0)) + .asString() + .contains("Successfully set to '0.12'") + .contains( + ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO + .key()); } // Verify the config was actually set @@ -435,13 +445,30 @@ void testSetClusterConfigs() throws Exception { assertThat(results.get(0).getField(1)).isEqualTo("300MB"); } + try (CloseableIterator resultIterator = + tEnv.executeSql( + String.format( + "Call %s.sys.get_cluster_configs('%s')", + CATALOG_NAME, + ConfigOptions + .SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO + .key())) + .collect()) { + List results = CollectionUtil.iteratorToList(resultIterator); + assertThat(results).hasSize(1); + assertThat(results.get(0).getField(1)).isEqualTo("0.12"); + } + // reset cluster configs. tEnv.executeSql( String.format( - "Call %s.sys.reset_cluster_configs('%s', '%s')", + "Call %s.sys.reset_cluster_configs('%s', '%s', '%s')", CATALOG_NAME, ConfigOptions.KV_SHARED_RATE_LIMITER_BYTES_PER_SEC.key(), - ConfigOptions.DATALAKE_FORMAT.key())) + ConfigOptions.DATALAKE_FORMAT.key(), + ConfigOptions + .SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO + .key())) .await(); } 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 9b7e97eb4ff..80e66398985 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 @@ -57,6 +57,11 @@ public LakeSource createLakeSource(TablePath tablePath) { @Override public LakeTableLookuper createLakeTableLookuper(TablePath tablePath, LookuperContext context) { return new PaimonLakeTableLookuper( - paimonConfig, tablePath, context.ioTmpDir(), context.tableConfig()); + paimonConfig, + tablePath, + context.ioTmpDir(), + context.tableConfig(), + context.lookupCacheMaxDiskBytes(), + context.diskWriteGuard()); } } 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 d15a182ed18..18afe2d982e 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 @@ -18,7 +18,9 @@ package org.apache.fluss.lake.paimon.lookup; 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.exception.KvStorageException; import org.apache.fluss.lake.lakestorage.LakeTableLookuper; import org.apache.fluss.lake.paimon.utils.PaimonPartitionBucket; @@ -32,16 +34,19 @@ 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; 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.BufferFileReader; +import org.apache.paimon.disk.BufferFileWriter; +import org.apache.paimon.disk.FileIOChannel; import org.apache.paimon.disk.IOManager; import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.memory.MemorySegment; -import org.apache.paimon.options.MemorySize; import org.apache.paimon.options.Options; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.query.LocalTableQuery; @@ -54,6 +59,7 @@ import javax.annotation.Nullable; import java.io.IOException; +import java.io.UncheckedIOException; import java.util.ArrayList; import java.util.Collections; import java.util.HashSet; @@ -65,6 +71,7 @@ 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.checkArgument; import static org.apache.fluss.utils.Preconditions.checkNotNull; /** @@ -86,17 +93,12 @@ */ public class PaimonLakeTableLookuper implements LakeTableLookuper { - // Each TabletServer caches at most ten table lookupers, bounding their retained lookup cache - // capacity to 20GB. See HistoricalLakeLookupManager for the follow-up to make this configurable - // and use a global Paimon IOManager limit. - private static final String LOOKUP_CACHE_MAX_DISK_SIZE = "2gb"; - private static final MemorySize LOOKUP_CACHE_MAX_DISK_MEMORY_SIZE = - MemorySize.parse(LOOKUP_CACHE_MAX_DISK_SIZE); - private final Configuration paimonConfig; private final TablePath tablePath; private final String ioTmpDir; private final TableConfig tableConfig; + private final long lookupCacheMaxDiskBytes; + private final Runnable diskWriteGuard; private final Set initializedBuckets; @@ -106,6 +108,7 @@ public class PaimonLakeTableLookuper implements LakeTableLookuper { private @Nullable LocalTableQuery localTableQuery; private @Nullable RowPartitionKeyExtractor partitionKeyExtractor; private int primaryKeyFieldCount; + private long lookupFileDownloadCount; // Both encoders are initialized only for a kv-format-v2 table whose bucket key differs from // its physical primary key. They remain null when the incoming Fluss key already uses Paimon's @@ -118,15 +121,22 @@ public class PaimonLakeTableLookuper implements LakeTableLookuper { private @Nullable InternalRow.FieldGetter[] cachedValueFieldGetters; private boolean closed; + /** Creates a lookuper with the specified local lookup cache limit. */ public PaimonLakeTableLookuper( Configuration paimonConfig, TablePath tablePath, String ioTmpDir, - TableConfig tableConfig) { + TableConfig tableConfig, + long lookupCacheMaxDiskBytes, + Runnable diskWriteGuard) { 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.tableConfig = checkNotNull(tableConfig, "tableConfig must not be null."); + checkArgument( + lookupCacheMaxDiskBytes > 0, "lookupCacheMaxDiskBytes must be greater than 0."); + this.lookupCacheMaxDiskBytes = lookupCacheMaxDiskBytes; + this.diskWriteGuard = checkNotNull(diskWriteGuard, "diskWriteGuard must not be null."); this.initializedBuckets = new HashSet<>(); } @@ -143,9 +153,28 @@ public PaimonLakeTableLookuper( org.apache.paimon.data.BinaryRow keyRow = toPaimonLookupKey(key); initializeFilesIfNeeded(partition, context.bucketId()); - org.apache.paimon.data.InternalRow paimonRow = - lookupWithFileRefresh( - partition, context.bucketId(), keyRow, context.valueRowType()); + long downloadCountBeforeLookup = lookupFileDownloadCount; + long lookupStartNanos = System.nanoTime(); + org.apache.paimon.data.InternalRow paimonRow; + try { + paimonRow = + lookupWithFileRefresh( + partition, context.bucketId(), keyRow, context.valueRowType()); + } catch (Exception e) { + DiskWriteLockedException diskWriteLockedException = + ExceptionUtils.findThrowable(e, DiskWriteLockedException.class).orElse(null); + if (diskWriteLockedException != null) { + throw diskWriteLockedException; + } + throw e; + } finally { + context.lookupMetricRecorder() + .recordLookup( + System.nanoTime() - lookupStartNanos, + // An increase means this lookup downloaded at least one lookup file + // through the tracking IO manager. + lookupFileDownloadCount > downloadCountBeforeLookup); + } if (paimonRow == null) { return null; } @@ -239,18 +268,12 @@ private void ensureInitialized(RowType valueRowType) throws Exception { private FileStoreTable withLookupCacheOptions(FileStoreTable table) { String key = CoreOptions.LOOKUP_CACHE_MAX_DISK_SIZE.key(); - String configuredMaxDiskSize = table.options().get(key); - if (configuredMaxDiskSize != null - && MemorySize.parse(configuredMaxDiskSize) - .compareTo(LOOKUP_CACHE_MAX_DISK_MEMORY_SIZE) - <= 0) { - return table; - } - return table.copy(Collections.singletonMap(key, LOOKUP_CACHE_MAX_DISK_SIZE)); + String maxDiskSize = new MemorySize(lookupCacheMaxDiskBytes).toString(); + return table.copy(Collections.singletonMap(key, maxDiskSize)); } - private static IOManager createIOManager(String ioTmpDir) { - return IOManager.create(ioTmpDir); + private IOManager createIOManager(String ioTmpDir) { + return new TrackingIOManager(IOManager.create(ioTmpDir)); } private static int[] businessFieldProjection(FileStoreTable fileStoreTable) { @@ -304,7 +327,6 @@ private void initializeFilesIfNeeded(org.apache.paimon.data.BinaryRow partition, return; } - LinkedHashMap beforeFilesByName = new LinkedHashMap<>(); LinkedHashMap dataFilesByName = new LinkedHashMap<>(); InnerTableScan tableScan = @@ -317,7 +339,6 @@ private void initializeFilesIfNeeded(org.apache.paimon.data.BinaryRow partition, continue; } DataSplit dataSplit = (DataSplit) split; - addFilesByName(beforeFilesByName, dataSplit.beforeFiles()); addFilesByName(dataFilesByName, dataSplit.dataFiles()); } @@ -326,11 +347,13 @@ private void initializeFilesIfNeeded(org.apache.paimon.data.BinaryRow partition, // been dropped. This PR does not support writes to expired partitions, so no new rows are // expected and initializing the file set once is sufficient. Compaction-related missing // files are handled by the IOException refresh path below. + // This partition-bucket has no registered lookup levels yet, so there are no old files to + // remove when building its lookup state from the active data files. localTableQuery() .refreshFiles( partition, bucketId, - new ArrayList<>(beforeFilesByName.values()), + Collections.emptyList(), new ArrayList<>(dataFilesByName.values())); initializedBuckets.add(partitionBucket); } @@ -436,4 +459,59 @@ private LocalTableQuery localTableQuery() { private RowPartitionKeyExtractor partitionKeyExtractor() { return checkNotNull(partitionKeyExtractor, "partitionKeyExtractor must be initialized."); } + + /** Tracks creation of Paimon lookup files while delegating all local I/O operations. */ + private final class TrackingIOManager implements IOManager { + + private final IOManager delegate; + + private TrackingIOManager(IOManager delegate) { + this.delegate = delegate; + } + + @Override + public FileIOChannel.ID createChannel() { + return delegate.createChannel(); + } + + @Override + public FileIOChannel.ID createChannel(String prefix) { + try { + diskWriteGuard.run(); + } catch (DiskWriteLockedException e) { + // IOManager does not allow createChannel to declare IOException. Preserve the + // I/O boundary here and unwrap the retriable Fluss exception in lookup(). + throw new UncheckedIOException(new IOException(e)); + } + lookupFileDownloadCount++; + return delegate.createChannel(prefix); + } + + @Override + public String[] tempDirs() { + return delegate.tempDirs(); + } + + @Override + public FileIOChannel.Enumerator createChannelEnumerator() { + return delegate.createChannelEnumerator(); + } + + @Override + public BufferFileWriter createBufferFileWriter(FileIOChannel.ID channelID) + throws IOException { + return delegate.createBufferFileWriter(channelID); + } + + @Override + public BufferFileReader createBufferFileReader(FileIOChannel.ID channelID) + throws IOException { + return delegate.createBufferFileReader(channelID); + } + + @Override + public void close() throws Exception { + delegate.close(); + } + } } diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/HistoricalPartitionLookupITCase.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/HistoricalPartitionLookupITCase.java index e4ab4130db8..091c72bc21b 100644 --- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/HistoricalPartitionLookupITCase.java +++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/HistoricalPartitionLookupITCase.java @@ -34,6 +34,7 @@ import org.apache.fluss.metadata.TablePath; import org.apache.fluss.row.InternalRow; import org.apache.fluss.server.testutils.FlussClusterExtension; +import org.apache.fluss.server.zk.data.PartitionRegistration; import org.apache.fluss.types.DataTypes; import org.apache.flink.core.execution.JobClient; @@ -46,6 +47,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.Optional; import java.util.concurrent.CompletableFuture; import static org.apache.fluss.testutils.DataTestUtils.row; @@ -98,13 +100,12 @@ void testLookupExpiredPartitionFromPaimon(boolean defaultBucketKey) throws Excep false) .get(); // The ALTER RPC must not complete until the required system partition is persisted. - assertThat( - FLUSS_CLUSTER_EXTENSION - .getZooKeeperClient() - .getPartition(tablePath, HISTORICAL_PARTITION_VALUE)) - .isPresent(); - waitUntilPartitionCreated(tablePath, HISTORICAL_PARTITION_VALUE); - long historicalPartitionId = getPartitionId(tablePath, HISTORICAL_PARTITION_VALUE); + Optional historicalPartition = + FLUSS_CLUSTER_EXTENSION + .getZooKeeperClient() + .getPartition(tablePath, HISTORICAL_PARTITION_VALUE); + assertThat(historicalPartition).isPresent(); + long historicalPartitionId = historicalPartition.get().getPartitionId(); FLUSS_CLUSTER_EXTENSION.waitUntilTablePartitionReady(tableId, historicalPartitionId); // Keep the initial retention wide enough so this old partition can be created and written @@ -347,12 +348,4 @@ private static void waitUntilPartitionDropped(TablePath tablePath, String partit assertThat(admin.listPartitionInfos(tablePath).get()) .noneMatch(p -> partitionName.equals(p.getPartitionName()))); } - - private static void waitUntilPartitionCreated(TablePath tablePath, String partitionName) { - retry( - Duration.ofMinutes(1), - () -> - assertThat(admin.listPartitionInfos(tablePath).get()) - .anyMatch(p -> partitionName.equals(p.getPartitionName()))); - } } 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 5faaacb5f25..1bf0ec3a16f 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 @@ -19,7 +19,9 @@ 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.LakeTableLookuper; import org.apache.fluss.lake.lakestorage.TestingLakeCatalogContext; import org.apache.fluss.lake.paimon.PaimonLakeCatalog; @@ -62,6 +64,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; import static org.apache.fluss.config.ConfigOptions.KV_FORMAT_VERSION_2; import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon; @@ -77,6 +80,10 @@ class PaimonLakeTableLookuperTest { private static final String DB = "lookup_db"; private static final short SCHEMA_ID = 1; private static final short EVOLVED_SCHEMA_ID = 2; + private static final long LOOKUP_CACHE_MAX_DISK_BYTES = MemorySize.parse("8gb").getBytes(); + private static final LakeTableLookuper.LookupMetricRecorder NO_OP_LOOKUP_METRIC_RECORDER = + (lookupTimeNanos, lookupFileDownloaded) -> {}; + private static final Runnable NO_OP_DISK_WRITE_GUARD = () -> {}; @TempDir private File tempWarehouseDir; @@ -120,9 +127,18 @@ void testLookupPartitionedPrimaryKeyTable() throws Exception { paimonConfig, tablePath, tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED))) { + tableConfig(KvFormat.COMPACTED), + LOOKUP_CACHE_MAX_DISK_BYTES, + NO_OP_DISK_WRITE_GUARD)) { + List lookupFileDownloads = new ArrayList<>(); LakeTableLookuper.LookupContext context = - lookupContext(schema, "20240101", 0, SCHEMA_ID); + lookupContext( + schema, + "20240101", + 0, + SCHEMA_ID, + (lookupTimeNanos, lookupFileDownloaded) -> + lookupFileDownloads.add(lookupFileDownloaded)); byte[] value = lookuper.lookup(paimonKey(schema, 1, "20240101"), context); BinaryValue decodedValue = decodeValue(value, SCHEMA_ID, schema); @@ -136,6 +152,61 @@ void testLookupPartitionedPrimaryKeyTable() throws Exception { lookupContext(schema, "20240101", 1, SCHEMA_ID))) .isNull(); 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); + } + } + + @Test + void testDiskWriteLockBlocksOnlyLookupFileDownloads() throws Exception { + TablePath tablePath = TablePath.of(DB, "disk_write_lock"); + Schema schema = pkSchema(); + FileStoreTable table = createPaimonTable(tablePath, partitionedPkDescriptor(schema)); + writeAndCommitData( + table, + Collections.singletonMap( + 0, + Arrays.asList( + paimonRow(1, "20240101", "Alice"), + paimonRow(2, "20240102", "Bob")))); + AtomicBoolean diskWriteLocked = new AtomicBoolean(); + Runnable diskWriteGuard = + () -> { + if (diskWriteLocked.get()) { + throw new DiskWriteLockedException("Data disk is write-locked."); + } + }; + + try (LakeTableLookuper lookuper = + new PaimonLakeTableLookuper( + paimonConfig, + tablePath, + tempWarehouseDir.getAbsolutePath(), + tableConfig(KvFormat.COMPACTED), + LOOKUP_CACHE_MAX_DISK_BYTES, + diskWriteGuard)) { + LakeTableLookuper.LookupContext cachedPartition = + lookupContext(schema, "20240101", 0, SCHEMA_ID); + LakeTableLookuper.LookupContext uncachedPartition = + lookupContext(schema, "20240102", 0, SCHEMA_ID); + + assertThat(lookuper.lookup(paimonKey(schema, 1, "20240101"), cachedPartition)) + .isNotNull(); + diskWriteLocked.set(true); + + // Cache hits remain available, while a lookup that needs a new local file is rejected. + assertThat(lookuper.lookup(paimonKey(schema, 1, "20240101"), cachedPartition)) + .isNotNull(); + assertThatThrownBy( + () -> + lookuper.lookup( + paimonKey(schema, 2, "20240102"), uncachedPartition)) + .isInstanceOf(DiskWriteLockedException.class); + + diskWriteLocked.set(false); + assertThat(lookuper.lookup(paimonKey(schema, 2, "20240102"), uncachedPartition)) + .isNotNull(); } } @@ -169,7 +240,9 @@ void testLookupPartitionsWithSameHashCode() throws Exception { paimonConfig, tablePath, tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED))) { + tableConfig(KvFormat.COMPACTED), + LOOKUP_CACHE_MAX_DISK_BYTES, + NO_OP_DISK_WRITE_GUARD)) { BinaryValue firstValue = decodeValue( lookuper.lookup( @@ -209,7 +282,9 @@ void testLookupWithIndexedKvFormat() throws Exception { paimonConfig, tablePath, tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.INDEXED))) { + tableConfig(KvFormat.INDEXED), + LOOKUP_CACHE_MAX_DISK_BYTES, + NO_OP_DISK_WRITE_GUARD)) { LakeTableLookuper.LookupContext context = lookupContext(schema, "20240101", 0, SCHEMA_ID); @@ -253,7 +328,9 @@ void testLookupKvFormatV2WithNonDefaultBucketKey() throws Exception { paimonConfig, tablePath, tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED, KV_FORMAT_VERSION_2))) { + tableConfig(KvFormat.COMPACTED, KV_FORMAT_VERSION_2), + LOOKUP_CACHE_MAX_DISK_BYTES, + NO_OP_DISK_WRITE_GUARD)) { LakeTableLookuper.LookupContext context = lookupContext(schema, "20240101", 0, SCHEMA_ID); byte[] compactedKey = @@ -308,7 +385,9 @@ void testRetriesInitializationAfterLookupKeyConverterFailure() throws Exception paimonConfig, tablePath, tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED, KV_FORMAT_VERSION_2))) { + tableConfig(KvFormat.COMPACTED, KV_FORMAT_VERSION_2), + LOOKUP_CACHE_MAX_DISK_BYTES, + 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. assertThatThrownBy( @@ -350,7 +429,9 @@ void testRefreshFilesAfterCompactionAndSnapshotExpiration() throws Exception { paimonConfig, tablePath, tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED))) { + tableConfig(KvFormat.COMPACTED), + LOOKUP_CACHE_MAX_DISK_BYTES, + NO_OP_DISK_WRITE_GUARD)) { LakeTableLookuper.LookupContext context = lookupContext(schema, "20240101", 0, SCHEMA_ID); assertThat(lookuper.lookup(paimonKey(schema, 5, "20240101"), context)).isNotNull(); @@ -416,14 +497,17 @@ void testLookupWithNonStringPartitionKey() throws Exception { paimonConfig, tablePath, tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED))) { + tableConfig(KvFormat.COMPACTED), + LOOKUP_CACHE_MAX_DISK_BYTES, + NO_OP_DISK_WRITE_GUARD)) { LakeTableLookuper.LookupContext context = new LakeTableLookuper.LookupContext( ResolvedPartitionSpec.fromPartitionName( Collections.singletonList("pt"), "7"), 0, SCHEMA_ID, - schema.getRowType()); + schema.getRowType(), + NO_OP_LOOKUP_METRIC_RECORDER); BinaryValue decodedValue = decodeValue( @@ -451,14 +535,17 @@ void testRejectAppendOnlyTable() throws Exception { paimonConfig, tablePath, tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED))) { + tableConfig(KvFormat.COMPACTED), + LOOKUP_CACHE_MAX_DISK_BYTES, + NO_OP_DISK_WRITE_GUARD)) { LakeTableLookuper.LookupContext context = new LakeTableLookuper.LookupContext( new ResolvedPartitionSpec( Collections.emptyList(), Collections.emptyList()), 0, SCHEMA_ID, - schema.getRowType()); + schema.getRowType(), + NO_OP_LOOKUP_METRIC_RECORDER); assertThatThrownBy(() -> lookuper.lookup(new byte[0], context)) .isInstanceOf(UnsupportedOperationException.class) @@ -507,7 +594,9 @@ void testLookupAfterSchemaEvolutionPadsNewColumnsWithNull() throws Exception { paimonConfig, tablePath, tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED))) { + tableConfig(KvFormat.COMPACTED), + LOOKUP_CACHE_MAX_DISK_BYTES, + NO_OP_DISK_WRITE_GUARD)) { BinaryValue oldSchemaValue = decodeValue( lookuper.lookup( @@ -599,7 +688,23 @@ private static LakeTableLookuper.LookupContext lookupContext( Collections.singletonList("dt"), partitionName), bucket, schemaId, - schema.getRowType()); + schema.getRowType(), + NO_OP_LOOKUP_METRIC_RECORDER); + } + + private static LakeTableLookuper.LookupContext lookupContext( + Schema schema, + String partitionName, + int bucket, + short schemaId, + LakeTableLookuper.LookupMetricRecorder lookupMetricRecorder) { + return new LakeTableLookuper.LookupContext( + ResolvedPartitionSpec.fromPartitionName( + Collections.singletonList("dt"), partitionName), + bucket, + schemaId, + schema.getRowType(), + lookupMetricRecorder); } private static List dataFiles( diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java index 3d22c35238e..25328fe44e6 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java @@ -26,6 +26,7 @@ import org.apache.fluss.rpc.messages.AuthenticateRequest; import org.apache.fluss.rpc.messages.AuthenticateResponse; import org.apache.fluss.rpc.messages.FetchLogRequest; +import org.apache.fluss.rpc.messages.LookupRequest; import org.apache.fluss.rpc.protocol.ApiError; import org.apache.fluss.rpc.protocol.ApiKeys; import org.apache.fluss.rpc.protocol.ApiManager; @@ -55,6 +56,7 @@ import static org.apache.fluss.rpc.protocol.MessageCodec.encodeErrorResponse; import static org.apache.fluss.rpc.protocol.MessageCodec.encodeServerFailure; import static org.apache.fluss.rpc.protocol.MessageCodec.encodeSuccessResponse; +import static org.apache.fluss.rpc.util.CommonRpcMessageUtils.hasHistoricalLookup; /** Implementation of the channel handler to process inbound requests for RPC server. */ public final class NettyServerHandler extends ChannelInboundHandlerAdapter { @@ -298,13 +300,16 @@ private void updateRequestMetrics(FlussRequest request, long requestEndTimeMs) { private Optional getMetrics(FlussRequest request) { boolean isFromFollower = false; + boolean isHistorical = false; ApiMessage requestMessage = request.getMessage(); if (request.getApiKey() == ApiKeys.FETCH_LOG.id) { // for fetch, we need to identify it's from client or follower FetchLogRequest fetchLogRequest = (FetchLogRequest) requestMessage; isFromFollower = fetchLogRequest.getFollowerServerId() >= 0; + } else if (request.getApiKey() == ApiKeys.LOOKUP.id) { + isHistorical = hasHistoricalLookup((LookupRequest) requestMessage); } - return requestsMetrics.getMetrics(request.getApiKey(), isFromFollower); + return requestsMetrics.getMetrics(request.getApiKey(), isFromFollower, isHistorical); } @VisibleForTesting diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestsMetrics.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestsMetrics.java index 592eaf30d91..3fff71f70a6 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestsMetrics.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestsMetrics.java @@ -51,12 +51,13 @@ private RequestsMetrics(MetricGroup serverMetricsGroup, Collection apiK for (ApiKeys apiKey : apiKeys) { // we create a metrics group for each type of request, with the request type // as variable + addMetrics(serverMetricsGroup, toRequestName(apiKey, false, false)); if (apiKey == ApiKeys.FETCH_LOG) { - // if it's fetch, we need two metrics group, one for client, one for follower - addMetrics(serverMetricsGroup, toRequestName(apiKey, true)); - addMetrics(serverMetricsGroup, toRequestName(apiKey, false)); - } else { - addMetrics(serverMetricsGroup, toRequestName(apiKey, false)); + // For fetch, register separate metric groups for clients and followers. + addMetrics(serverMetricsGroup, toRequestName(apiKey, true, false)); + } + if (apiKey == ApiKeys.LOOKUP) { + addMetrics(serverMetricsGroup, toRequestName(apiKey, false, true)); } } this.requestMetricGroup = serverMetricsGroup.addGroup("request"); @@ -96,14 +97,15 @@ private void addMetrics(MetricGroup parentMetricGroup, String requestName) { requestName, new Metrics(parentMetricGroup.addGroup("request", requestName))); } - private static String toRequestName(ApiKeys apiKeys, boolean isFromFollower) { + private static String toRequestName( + ApiKeys apiKeys, boolean isFromFollower, boolean isHistorical) { switch (apiKeys) { case PRODUCE_LOG: return "produceLog"; case PUT_KV: return "putKv"; case LOOKUP: - return "lookup"; + return isHistorical ? "historicalLookup" : "lookup"; case PREFIX_LOOKUP: return "prefixLookup"; case FETCH_LOG: @@ -115,8 +117,9 @@ private static String toRequestName(ApiKeys apiKeys, boolean isFromFollower) { } } - public Optional getMetrics(short apiKey, boolean isFromFollower) { - String requestName = toRequestName(ApiKeys.forId(apiKey), isFromFollower); + public Optional getMetrics( + short apiKey, boolean isFromFollower, boolean isHistorical) { + String requestName = toRequestName(ApiKeys.forId(apiKey), isFromFollower, isHistorical); return Optional.ofNullable(metricsByRequest.get(requestName)); } diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/util/CommonRpcMessageUtils.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/util/CommonRpcMessageUtils.java index 2cf783a1184..a5aa22cb3b8 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/util/CommonRpcMessageUtils.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/util/CommonRpcMessageUtils.java @@ -26,6 +26,7 @@ import org.apache.fluss.remote.RemoteLogFetchInfo; import org.apache.fluss.remote.RemoteLogSegment; import org.apache.fluss.rpc.entity.FetchLogResultForBucket; +import org.apache.fluss.rpc.messages.LookupRequest; import org.apache.fluss.rpc.messages.PbAclFilter; import org.apache.fluss.rpc.messages.PbAclInfo; import org.apache.fluss.rpc.messages.PbFetchLogRespForBucket; @@ -59,6 +60,17 @@ */ public class CommonRpcMessageUtils { + /** + * Returns whether the lookup request is for historical partition lookup. + * + *

Normal and historical lookup buckets cannot be mixed in the same request, so the first + * bucket determines the request type. + */ + public static boolean hasHistoricalLookup(LookupRequest lookupRequest) { + return lookupRequest.getBucketsReqsCount() > 0 + && lookupRequest.getBucketsReqAt(0).hasOriginalPartitionName(); + } + public static List toPbAclInfos(Collection aclBindings) { return aclBindings.stream() .map(CommonRpcMessageUtils::toPbAclInfo) diff --git a/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java b/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java index 37ae794f538..98d721e152e 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java @@ -56,6 +56,8 @@ import static org.apache.fluss.config.ConfigOptions.REMOTE_DATA_DIRS_WEIGHTS; import static org.apache.fluss.config.ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO; import static org.apache.fluss.config.ConfigOptions.SERVER_DATA_DISK_WRITE_RECOVER_RATIO; +import static org.apache.fluss.config.ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUPER_CACHE_EXPIRE_AFTER_ACCESS; +import static org.apache.fluss.config.ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO; import static org.apache.fluss.config.ConfigOptions.SERVER_SASL_CREDENTIALS; import static org.apache.fluss.config.ConfigOptions.SERVER_SASL_PLAIN_JAAS_CONFIG; import static org.apache.fluss.utils.concurrent.LockUtils.inReadLock; @@ -82,6 +84,8 @@ class DynamicServerConfig { KV_SNAPSHOT_INTERVAL.key(), SERVER_DATA_DISK_WRITE_RECOVER_RATIO.key(), SERVER_DATA_DISK_WRITE_LIMIT_RATIO.key(), + SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO.key(), + SERVER_HISTORICAL_PARTITION_LOOKUPER_CACHE_EXPIRE_AFTER_ACCESS.key(), // Config options for remote.data.dirs REMOTE_DATA_DIRS.key(), REMOTE_DATA_DIRS_STRATEGY.key(), diff --git a/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java b/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java index 4ae6b09a331..a534a35516d 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java @@ -127,6 +127,7 @@ import static org.apache.fluss.server.utils.ServerRpcMessageUtils.toPbConfigEntries; import static org.apache.fluss.server.utils.ServerRpcMessageUtils.toPbDatabaseSummary; import static org.apache.fluss.server.utils.ServerRpcMessageUtils.toTablePath; +import static org.apache.fluss.utils.PartitionUtils.HISTORICAL_PARTITION_VALUE; import static org.apache.fluss.utils.Preconditions.checkState; /** @@ -486,6 +487,8 @@ public CompletableFuture listPartitionInfos( } else { partitionRegistrations = metadataManager.listPartitions(tablePath); } + // TODO: Return the actual lake partitions instead of the internal historical partition. + partitionRegistrations.remove(HISTORICAL_PARTITION_VALUE); TableInfo tableInfo = metadataManager.getTable(tablePath); List partitionKeys = tableInfo.getPartitionKeys(); return CompletableFuture.completedFuture( diff --git a/fluss-server/src/main/java/org/apache/fluss/server/TabletManagerBase.java b/fluss-server/src/main/java/org/apache/fluss/server/TabletManagerBase.java index 6ab15dc8397..389c85918a1 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/TabletManagerBase.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/TabletManagerBase.java @@ -56,8 +56,10 @@ import java.util.function.Consumer; import java.util.stream.Collectors; +import static org.apache.fluss.utils.FlussPaths.HISTORICAL_LOOKUP_CACHE_DIR_NAME; import static org.apache.fluss.utils.FlussPaths.KV_TABLET_DIR_PREFIX; import static org.apache.fluss.utils.FlussPaths.LOG_TABLET_DIR_PREFIX; +import static org.apache.fluss.utils.FlussPaths.REMOTE_LOG_INDEX_LOCAL_CACHE; import static org.apache.fluss.utils.FlussPaths.isPartitionDir; /** @@ -115,6 +117,10 @@ protected List listTabletsToLoad(File dataDir) { // Get all database directory. File[] dbDirs = FileUtils.listDirectories(dataDir); for (File dbDir : dbDirs) { + if (dbDir.getName().equals(HISTORICAL_LOOKUP_CACHE_DIR_NAME) + || dbDir.getName().equals(REMOTE_LOG_INDEX_LOCAL_CACHE)) { + continue; + } // Get all table path directory. File[] tableDirs = FileUtils.listDirectories(dbDir); for (File tableDir : tableDirs) { diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java index 0f610eb42a3..a2f9b4fad97 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java @@ -306,6 +306,7 @@ protected void initCoordinatorStandby() throws Exception { dynamicConfigManager.register(remoteDirDynamicLoader); dynamicConfigManager.register(replicaCapacityController); // Register stateless validators for coordinator-side upfront validation + dynamicConfigManager.register(new HistoricalLookupCacheConfigValidator()); dynamicConfigManager.register(new DiskWriteLimitConfigValidator()); rpcServer.getServerReconfigurables().forEach(dynamicConfigManager::register); dynamicConfigManager.startup(); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/HistoricalLookupCacheConfigValidator.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/HistoricalLookupCacheConfigValidator.java new file mode 100644 index 00000000000..c9d763f410a --- /dev/null +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/HistoricalLookupCacheConfigValidator.java @@ -0,0 +1,59 @@ +/* + * 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.server.coordinator; + +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.config.cluster.ServerReconfigurable; +import org.apache.fluss.exception.ConfigException; + +import java.time.Duration; + +/** Validates dynamic historical lookup cache settings. */ +final class HistoricalLookupCacheConfigValidator implements ServerReconfigurable { + + @Override + public void validate(Configuration newConfig) throws ConfigException { + double newMaxRatio = + newConfig.get( + ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO); + if (!(newMaxRatio > 0.0 && newMaxRatio <= 1.0)) { + throw new ConfigException( + String.format( + "Invalid configuration for %s, it must be within (0.0, 1.0].", + ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO + .key())); + } + + Duration newExpiration = + newConfig.get( + ConfigOptions + .SERVER_HISTORICAL_PARTITION_LOOKUPER_CACHE_EXPIRE_AFTER_ACCESS); + if (newExpiration.toMillis() < 1) { + throw new ConfigException( + String.format( + "Invalid configuration for %s, it must be greater than or equal to 1 ms.", + ConfigOptions + .SERVER_HISTORICAL_PARTITION_LOOKUPER_CACHE_EXPIRE_AFTER_ACCESS + .key())); + } + } + + @Override + public void reconfigure(Configuration newConfig) {} +} diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TableMetricGroup.java b/fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TableMetricGroup.java index 68387b94195..303ed580514 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TableMetricGroup.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TableMetricGroup.java @@ -21,11 +21,14 @@ import org.apache.fluss.metadata.TablePath; import org.apache.fluss.metrics.CharacterFilter; import org.apache.fluss.metrics.Counter; +import org.apache.fluss.metrics.DescriptiveStatisticsHistogram; +import org.apache.fluss.metrics.Histogram; import org.apache.fluss.metrics.MeterView; import org.apache.fluss.metrics.MetricNames; import org.apache.fluss.metrics.NoOpCounter; import org.apache.fluss.metrics.ThreadSafeSimpleCounter; import org.apache.fluss.metrics.groups.AbstractMetricGroup; +import org.apache.fluss.metrics.groups.MetricGroup; import org.apache.fluss.metrics.registry.MetricRegistry; import org.apache.fluss.server.kv.rocksdb.RocksDBStatistics; @@ -33,6 +36,7 @@ import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; import java.util.stream.Stream; import static org.apache.fluss.metrics.utils.MetricGroupUtils.makeScope; @@ -104,6 +108,16 @@ protected String getGroupName(CharacterFilter filter) { return "table"; } + /** Closes this table metric group and its directly created tablet metric groups. */ + @Override + public void close() { + if (kvMetrics != null) { + kvMetrics.close(); + } + logMetrics.close(); + super.close(); + } + public void incLogMessageIn(long n) { logMetrics.messagesIn.inc(n); serverMetrics.messageIn().inc(n); @@ -189,6 +203,36 @@ public Counter failedLookupRequests() { } } + /** Returns the counter for historical lookup requests received by this table. */ + public Counter totalHistoricalLookupRequests() { + if (kvMetrics == null) { + return NoOpCounter.INSTANCE; + } else { + return kvMetrics.totalHistoricalLookupRequests; + } + } + + /** Returns the counter for failed historical lookup requests for this table. */ + public Counter failedHistoricalLookupRequests() { + if (kvMetrics == null) { + return NoOpCounter.INSTANCE; + } else { + return kvMetrics.failedHistoricalLookupRequests; + } + } + + /** + * Records a historical lake table point lookup. + * + * @param lookupTimeNanos time spent on the lake table point lookup, in nanoseconds + * @param lookupFileDownloaded whether the lookup downloaded a lookup file + */ + public void recordHistoricalLakeLookup(long lookupTimeNanos, boolean lookupFileDownloaded) { + if (kvMetrics != null) { + kvMetrics.recordHistoricalLakeLookup(lookupTimeNanos, lookupFileDownloaded); + } + } + public Counter totalPutKvRequests() { if (kvMetrics == null) { return NoOpCounter.INSTANCE; @@ -528,8 +572,14 @@ protected String getGroupName(CharacterFilter filter) { private static class KvMetricGroup extends TabletMetricGroup { + private static final String LOOKUP_FILE_DOWNLOADED = "lookup_file_downloaded"; + private final Counter totalLookupRequests; private final Counter failedLookupRequests; + private final Counter totalHistoricalLookupRequests; + private final Counter failedHistoricalLookupRequests; + private final LookupFileDownloadedMetricGroup downloadedHistoricalLookupMetrics; + private final LookupFileDownloadedMetricGroup nonDownloadedHistoricalLookupMetrics; private final Counter totalPutKvRequests; private final Counter failedPutKvRequests; private final Counter totalLimitScanRequests; @@ -545,6 +595,22 @@ public KvMetricGroup(TableMetricGroup tableMetricGroup) { meter(MetricNames.TOTAL_LOOKUP_REQUESTS_RATE, new MeterView(totalLookupRequests)); failedLookupRequests = new ThreadSafeSimpleCounter(); meter(MetricNames.FAILED_LOOKUP_REQUESTS_RATE, new MeterView(failedLookupRequests)); + // for historical lookup request + MetricGroup historicalLookupMetrics = addGroup("historical"); + totalHistoricalLookupRequests = new ThreadSafeSimpleCounter(); + historicalLookupMetrics.meter( + MetricNames.TOTAL_LOOKUP_REQUESTS_RATE, + new MeterView(totalHistoricalLookupRequests)); + failedHistoricalLookupRequests = new ThreadSafeSimpleCounter(); + historicalLookupMetrics.meter( + MetricNames.FAILED_LOOKUP_REQUESTS_RATE, + new MeterView(failedHistoricalLookupRequests)); + // Separate groups expose the same metric names with different downloaded-file labels + // without adding the label key to the logical metric scope. + downloadedHistoricalLookupMetrics = + new LookupFileDownloadedMetricGroup(registry, this, true); + nonDownloadedHistoricalLookupMetrics = + new LookupFileDownloadedMetricGroup(registry, this, false); // for put kv request totalPutKvRequests = new ThreadSafeSimpleCounter(); meter(MetricNames.TOTAL_PUT_KV_REQUESTS_RATE, new MeterView(totalPutKvRequests)); @@ -571,12 +637,65 @@ public KvMetricGroup(TableMetricGroup tableMetricGroup) { new MeterView(failedPrefixLookupRequests)); } + @Override + public void close() { + downloadedHistoricalLookupMetrics.close(); + nonDownloadedHistoricalLookupMetrics.close(); + super.close(); + } + + private void recordHistoricalLakeLookup( + long lookupTimeNanos, boolean lookupFileDownloaded) { + LookupFileDownloadedMetricGroup metricGroup = + lookupFileDownloaded + ? downloadedHistoricalLookupMetrics + : nonDownloadedHistoricalLookupMetrics; + metricGroup.recordLookup(lookupTimeNanos); + } + @Override protected String getGroupName(CharacterFilter filter) { return super.getGroupName(filter); } } + private static final class LookupFileDownloadedMetricGroup extends AbstractMetricGroup { + + private static final int WINDOW_SIZE = 64; + + private final boolean lookupFileDownloaded; + private final Counter lakeLookups; + private final Histogram lakeLookupTimeMs; + + private LookupFileDownloadedMetricGroup( + MetricRegistry registry, KvMetricGroup parent, boolean lookupFileDownloaded) { + super(registry, makeScope(parent, "historical"), parent); + this.lookupFileDownloaded = lookupFileDownloaded; + lakeLookups = new ThreadSafeSimpleCounter(); + meter(MetricNames.LAKE_LOOKUPS_RATE, new MeterView(lakeLookups)); + lakeLookupTimeMs = + histogram( + MetricNames.LAKE_LOOKUP_TIME_MS, + new DescriptiveStatisticsHistogram(WINDOW_SIZE)); + } + + private void recordLookup(long lookupTimeNanos) { + lakeLookups.inc(); + lakeLookupTimeMs.update(TimeUnit.NANOSECONDS.toMillis(lookupTimeNanos)); + } + + @Override + protected void putVariables(Map variables) { + variables.put( + KvMetricGroup.LOOKUP_FILE_DOWNLOADED, String.valueOf(lookupFileDownloaded)); + } + + @Override + protected String getGroupName(CharacterFilter filter) { + return "historical"; + } + } + private enum TabletType { LOG("log"), KV("kv"), diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalLakeLookupManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalLakeLookupManager.java index fe1b8028d95..977b608402b 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalLakeLookupManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalLakeLookupManager.java @@ -35,12 +35,16 @@ import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TableInfo; import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.metrics.Counter; +import org.apache.fluss.metrics.ThreadSafeSimpleCounter; import org.apache.fluss.plugin.PluginManager; import org.apache.fluss.rpc.entity.LookupResultForBucket; import org.apache.fluss.rpc.protocol.ApiError; import org.apache.fluss.server.entity.LookupDataForBucket; +import org.apache.fluss.server.storage.LocalDiskManager; import org.apache.fluss.utils.ExecutorUtils; import org.apache.fluss.utils.FileUtils; +import org.apache.fluss.utils.FlussPaths; import org.apache.fluss.utils.IOUtils; import org.apache.fluss.utils.concurrent.ExecutorThreadFactory; import org.apache.fluss.utils.concurrent.Scheduler; @@ -49,16 +53,21 @@ import com.github.benmanes.caffeine.cache.Caffeine; import com.github.benmanes.caffeine.cache.RemovalCause; import com.github.benmanes.caffeine.cache.Ticker; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import javax.annotation.Nullable; import java.io.File; import java.io.IOException; +import java.io.UncheckedIOException; import java.nio.file.Files; +import java.nio.file.Path; import java.time.Duration; import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; @@ -67,10 +76,12 @@ import java.util.concurrent.Semaphore; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.stream.Stream; import static org.apache.fluss.server.utils.LakeStorageUtils.extractLakeProperties; import static org.apache.fluss.utils.Preconditions.checkArgument; import static org.apache.fluss.utils.Preconditions.checkNotNull; +import static org.apache.fluss.utils.Preconditions.checkState; /** * Handles server-side point lookup for historical partitions stored in lake storage. @@ -82,53 +93,80 @@ *

Creating a lake table lookuper may initialize catalog, table, and query state and allocate * local lookup files, so lookupers are cached and reused. The cache is keyed by table ID rather * than table path to prevent a deleted and recreated table from reusing the old table's lookuper. A - * cached lookuper is replaced when its schema ID no longer matches the requested table schema. + * cached lookuper is replaced when its schema ID or lake configuration version no longer matches + * the current request. Active lookups can finish on the old lookuper, which is closed after its + * last lookup releases it. + * + *

Up to ten table lookupers are cached. Each lookuper receives one tenth of the server-level + * disk budget, and Caffeine evicts lookupers when the table limit is exceeded. + * + *

Historical lookup cache I/O participates in TabletServer disk write protection. Existing cache + * hits remain available when the data disk is write-locked, while lookups that need to download new + * cache files are rejected until the disk recovers. * *

A lookuper is closed when replaced, explicitly invalidated by a replica lifecycle event, - * evicted after the cache reaches ten tables, the manager shuts down, or after three hours without - * access. Caffeine expiration is scheduled on the shared TabletServer scheduler, allowing idle + * evicted when the table limit is exceeded, the manager shuts down, or after the configured idle + * expiration. Caffeine expiration is scheduled on the shared TabletServer scheduler, allowing idle * resources to be released even if no subsequent lookup accesses the cache. */ class HistoricalLakeLookupManager implements AutoCloseable { - private static final String PAIMON_LOOKUP_DIR_NAME = "paimon-lookup"; + private static final Logger LOG = LoggerFactory.getLogger(HistoricalLakeLookupManager.class); + private static final String LOOKUPER_CACHE_EXPIRATION_TASK_NAME = "historical-lookuper-cache-expiration"; - private static final Duration LOOKUPER_CACHE_EXPIRATION = Duration.ofHours(3); + private static final String LOOKUP_CACHE_DISK_SIZE_TASK_NAME = + "historical-lookup-cache-disk-size"; + private static final Duration LOOKUP_CACHE_DISK_SIZE_CHECK_INTERVAL = Duration.ofMinutes(3); private static final Duration HISTORICAL_PARTITION_THREAD_KEEP_ALIVE = Duration.ofMinutes(10); private static final Duration HISTORICAL_PARTITION_EXECUTOR_SHUTDOWN_TIMEOUT = Duration.ofSeconds(10); - private static final int MAX_CACHED_LOOKUPERS = 10; private static final String HISTORICAL_PARTITION_THREAD_NAME_PREFIX = "historical-partition-io"; + // TODO: Share one Paimon IOManager disk budget across all table lookupers and evict cached + // entries by data file instead of reserving fixed per-table capacity. See + // https://github.com/apache/fluss/issues/3955. + private static final int MAX_CACHED_TABLES = 10; - // TODO: MAX_CACHED_LOOKUPERS and the 2GB per-table limit configured by - // PaimonLakeTableLookuper through Paimon's "lookup.cache-max-disk-size" option are hard-coded. - // Make them configurable, and prefer a Paimon IOManager-level global disk limit shared by all - // table lookupers because fixed per-table limits can underutilize cache for hot tables while - // reserving too much for cold tables. - - private final Configuration conf; + private volatile Configuration conf; + private volatile long lakeConfigVersion; private final @Nullable PluginManager pluginManager; - private final int serverId; + private final Counter capacityEvictions; + private final int maxQueuedHistoricalRequests; private final Semaphore lookupPermits; // Accepted lookup futures tracked so close() can cancel tasks left after executor shutdown. private final Set> pendingLookups; private final Cache lakeTableLookupers; private final ExecutorService historicalPartitionExecutor; - private @Nullable String paimonLookupTempDir; + private final File historicalLookupCacheRootDir; + private final long dataDirVolumeBytes; + // TODO: Introduce a minimum lookup cache disk ratio (default 0.01). When disk usage is high, + // evict cached entries down to the minimum ratio instead of clearing the entire cache; allow + // the cache to grow back to the maximum ratio after disk usage recovers. + private final Runnable diskWriteGuard; + + private volatile long lookupCacheMaxDiskBytesPerTable; + private volatile long lookupCacheDiskSize; + private volatile boolean started; + + /** Creates a historical lake lookup manager. */ HistoricalLakeLookupManager( Configuration conf, @Nullable PluginManager pluginManager, - int serverId, + LocalDiskManager localDiskManager, + File dataDir, + long dataDirVolumeBytes, Scheduler scheduler) { this( conf, pluginManager, null, - serverId, + dataDir, + dataDirVolumeBytes, Ticker.systemTicker(), - createCacheScheduler(scheduler)); + createCacheScheduler(scheduler), + checkNotNull(localDiskManager, "localDiskManager must not be null.") + ::ensureWritable); } @VisibleForTesting @@ -136,13 +174,26 @@ class HistoricalLakeLookupManager implements AutoCloseable { Configuration conf, @Nullable PluginManager pluginManager, @Nullable ExecutorService historicalPartitionExecutor, - int serverId, + File dataDir, + long dataDirVolumeBytes, Ticker ticker, - com.github.benmanes.caffeine.cache.Scheduler cacheScheduler) { + com.github.benmanes.caffeine.cache.Scheduler cacheScheduler, + Runnable diskWriteGuard) { this.conf = checkNotNull(conf, "conf must not be null."); this.pluginManager = pluginManager; - this.serverId = serverId; - int maxQueuedHistoricalRequests = + this.historicalLookupCacheRootDir = + FlussPaths.historicalLookupRootDir( + checkNotNull(dataDir, "dataDir must not be null.")); + checkArgument(dataDirVolumeBytes > 0, "dataDirVolumeBytes must be greater than 0."); + this.dataDirVolumeBytes = dataDirVolumeBytes; + this.diskWriteGuard = checkNotNull(diskWriteGuard, "diskWriteGuard must not be null."); + this.lookupCacheMaxDiskBytesPerTable = + cacheBytesPerTable( + conf.get( + ConfigOptions + .SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO)); + this.capacityEvictions = new ThreadSafeSimpleCounter(); + this.maxQueuedHistoricalRequests = conf.get(ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS); checkArgument( maxQueuedHistoricalRequests > 0, @@ -160,19 +211,15 @@ class HistoricalLakeLookupManager implements AutoCloseable { : historicalPartitionExecutor; this.lakeTableLookupers = Caffeine.newBuilder() - .expireAfterAccess(LOOKUPER_CACHE_EXPIRATION) - .maximumSize(MAX_CACHED_LOOKUPERS) + .maximumSize(MAX_CACHED_TABLES) + .expireAfterAccess( + conf.get( + ConfigOptions + .SERVER_HISTORICAL_PARTITION_LOOKUPER_CACHE_EXPIRE_AFTER_ACCESS)) .ticker(checkNotNull(ticker, "ticker must not be null.")) .scheduler(checkNotNull(cacheScheduler, "cacheScheduler must not be null.")) .executor(Runnable::run) - .removalListener( - (Long ignored, - CachedLakeTableLookuper cachedLookuper, - RemovalCause ignoredCause) -> { - if (cachedLookuper != null) { - cachedLookuper.invalidate(); - } - }) + .removalListener(this::onLookuperRemoved) .build(); this.lookupPermits = new Semaphore(maxQueuedHistoricalRequests); this.pendingLookups = ConcurrentHashMap.newKeySet(); @@ -190,8 +237,48 @@ private static com.github.benmanes.caffeine.cache.Scheduler createCacheScheduler timeUnit.toMillis(delay)); } + /** + * Attempts to clean lookup cache files left by a previous TabletServer process. + * + *

The cache root under this server's first data directory is removed and recreated before + * lookups are accepted. + */ + synchronized void startup(Scheduler scheduler) { + checkNotNull(scheduler, "scheduler must not be null."); + if (started) { + return; + } + try { + FileUtils.deleteDirectory(historicalLookupCacheRootDir); + } catch (IOException e) { + LOG.warn( + "Failed to clean historical lookup cache directory {}.", + historicalLookupCacheRootDir, + e); + } + try { + Files.createDirectories(historicalLookupCacheRootDir.toPath()); + } catch (IOException e) { + throw new FlussRuntimeException( + "Failed to create historical lookup cache directory: " + + historicalLookupCacheRootDir, + e); + } + scheduler.schedule( + LOOKUP_CACHE_DISK_SIZE_TASK_NAME, + this::updateLookupCacheDiskSize, + 0L, + LOOKUP_CACHE_DISK_SIZE_CHECK_INTERVAL.toMillis()); + started = true; + } + + /** Looks up a batch of keys from one historical lake partition. */ CompletableFuture lookup( - LookupDataForBucket lookupData, TableInfo tableInfo, SchemaInfo schemaInfo) { + LookupDataForBucket lookupData, + TableInfo tableInfo, + SchemaInfo schemaInfo, + LakeTableLookuper.LookupMetricRecorder lookupMetricRecorder) { + checkState(started, "Historical lake lookup manager has not been started."); TableBucket tableBucket = lookupData.tableBucket(); if (!lookupPermits.tryAcquire()) { return CompletableFuture.completedFuture( @@ -208,7 +295,14 @@ CompletableFuture lookup( CompletableFuture future; try { - future = submitLookup(lookupData, tableInfo, schemaInfo); + future = + submitLookup( + lookupData, + tableInfo, + schemaInfo, + checkNotNull( + lookupMetricRecorder, + "lookupMetricRecorder must not be null.")); } catch (RuntimeException e) { lookupPermits.release(); throw e; @@ -233,10 +327,15 @@ public void close() { } private CompletableFuture submitLookup( - LookupDataForBucket lookupData, TableInfo tableInfo, SchemaInfo schemaInfo) { + LookupDataForBucket lookupData, + TableInfo tableInfo, + SchemaInfo schemaInfo, + LakeTableLookuper.LookupMetricRecorder lookupMetricRecorder) { CompletableFuture future = CompletableFuture.supplyAsync( - () -> lookupInternal(lookupData, tableInfo, schemaInfo), + () -> + lookupInternal( + lookupData, tableInfo, schemaInfo, lookupMetricRecorder), historicalPartitionExecutor); pendingLookups.add(future); return future; @@ -255,16 +354,89 @@ private ExecutorService createHistoricalPartitionExecutor(int maxThreadPoolSize) return executor; } + /** Invalidates the cached lake lookuper for the given table. */ void invalidateTableLookuper(long tableId) { lakeTableLookupers.invalidate(tableId); } + /** Returns the number of table lookupers currently cached. */ + int cachedTableCount() { + return lakeTableLookupers.asMap().size(); + } + + /** Returns the counter for table lookuper evictions caused by the cached table limit. */ + Counter capacityEvictions() { + return capacityEvictions; + } + + /** Returns the number of accepted historical lookup requests that have not completed. */ + int numInflightRequests() { + return maxQueuedHistoricalRequests - lookupPermits.availablePermits(); + } + + /** Applies dynamic historical lookup configuration changes. */ + void reconfigure(Configuration newConf) { + checkNotNull(newConf, "newConf must not be null."); + boolean lakeConfigChanged; + boolean cacheLimitChanged; + boolean expirationChanged; + Duration newExpiration = + newConf.get( + ConfigOptions + .SERVER_HISTORICAL_PARTITION_LOOKUPER_CACHE_EXPIRE_AFTER_ACCESS); + synchronized (this) { + long newMaxBytesPerTable = + cacheBytesPerTable( + newConf.get( + ConfigOptions + .SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO)); + cacheLimitChanged = newMaxBytesPerTable != lookupCacheMaxDiskBytesPerTable; + lookupCacheMaxDiskBytesPerTable = newMaxBytesPerTable; + + lakeConfigChanged = hasLakeConfigChanged(conf, newConf); + expirationChanged = + !newExpiration.equals( + conf.get( + ConfigOptions + .SERVER_HISTORICAL_PARTITION_LOOKUPER_CACHE_EXPIRE_AFTER_ACCESS)); + // Publish the configuration before its version. A lookup that observes the new version + // must also observe the matching configuration snapshot. + conf = newConf; + if (lakeConfigChanged) { + lakeConfigVersion++; + } + } + if (expirationChanged) { + lakeTableLookupers + .policy() + .expireAfterAccess() + .get() + .setExpiresAfter(newExpiration.toMillis(), TimeUnit.MILLISECONDS); + } + if (lakeConfigChanged || cacheLimitChanged) { + // Do not invalidate while holding this monitor: lookuper creation holds a cache key + // lock before preparing the lookup directory under the same monitor. Invalidation + // closes inactive lookupers immediately and active lookupers after their last lookup + // releases them. After a cache limit change, the next lookup creates a Paimon lookuper + // with the updated per-table limit. + lakeTableLookupers.invalidateAll(); + lakeTableLookupers.cleanUp(); + } + } + private LookupResultForBucket lookupInternal( - LookupDataForBucket lookupData, TableInfo tableInfo, SchemaInfo schemaInfo) { + LookupDataForBucket lookupData, + TableInfo tableInfo, + SchemaInfo schemaInfo, + LakeTableLookuper.LookupMetricRecorder lookupMetricRecorder) { TableBucket tableBucket = lookupData.tableBucket(); CachedLakeTableLookuper cachedLookuper = null; try { - LookupContext context = createLookupContext(lookupData, tableInfo, schemaInfo); + LookupContext context = + createLookupContext(lookupData, tableInfo, schemaInfo, lookupMetricRecorder); + long currentLakeConfigVersion = lakeConfigVersion; + Configuration currentConf = conf; + long cacheSizeBytes = lookupCacheMaxDiskBytesPerTable; cachedLookuper = lakeTableLookupers .asMap() @@ -272,19 +444,37 @@ private LookupResultForBucket lookupInternal( context.tableId, (ignored, currentLookuper) -> { CachedLakeTableLookuper selectedLookuper = currentLookuper; - // Create the lookuper lazily, and recreate it after schema - // evolution so it reloads lake table/query state and - // encodes values with the requested schema. + // Create the lookuper lazily, and recreate it after schema, + // lake configuration, or server cache size changes so it + // reloads lake table/query state and uses the current + // settings. if (selectedLookuper == null - || selectedLookuper.schemaId != context.schemaId) { - LakeTableLookuper newLookuper = + || selectedLookuper.schemaId != context.schemaId + || selectedLookuper.lakeConfigVersion + != currentLakeConfigVersion + || selectedLookuper.cacheSizeBytes + != cacheSizeBytes) { + File tableLookupDir = + FlussPaths.historicalLookupTableDir( + historicalLookupCacheRootDir, + context.tablePath, + context.tableId); + LakeTableLookuper lookuper = createLakeTableLookuper( context.tablePath, - getOrPreparePaimonLookupTempDir(), - tableInfo.getTableConfig()); + tableLookupDir.getAbsolutePath(), + tableInfo.getTableConfig(), + cacheSizeBytes, + currentConf); selectedLookuper = new CachedLakeTableLookuper( - context.schemaId, newLookuper); + context.tableId, + context.tablePath, + context.schemaId, + currentLakeConfigVersion, + cacheSizeBytes, + tableLookupDir, + lookuper); } // Pin the lookuper before leaving the atomic cache update. // Eviction or invalidation can then defer closing it until @@ -311,8 +501,27 @@ private LookupResultForBucket lookupInternal( } } + private void onLookuperRemoved( + Long ignored, @Nullable CachedLakeTableLookuper cachedLookuper, RemovalCause cause) { + if (cachedLookuper == null) { + return; + } + if (cause == RemovalCause.SIZE) { + capacityEvictions.inc(); + LOG.info( + "Evicted historical lookup cache for table {} (table ID {}) because the cache retains at most {} tables.", + cachedLookuper.tablePath, + cachedLookuper.tableId, + MAX_CACHED_TABLES); + } + cachedLookuper.invalidate(); + } + private LookupContext createLookupContext( - LookupDataForBucket lookupData, TableInfo tableInfo, SchemaInfo schemaInfo) { + LookupDataForBucket lookupData, + TableInfo tableInfo, + SchemaInfo schemaInfo, + LakeTableLookuper.LookupMetricRecorder lookupMetricRecorder) { TableBucket tableBucket = lookupData.tableBucket(); String originalPartitionName = lookupData.originalPartitionName(); if (originalPartitionName == null) { @@ -339,14 +548,19 @@ private LookupContext createLookupContext( originalPartitionSpec, tableBucket.getBucket(), (short) schemaInfo.getSchemaId(), - schemaInfo.getSchema().getRowType()); + schemaInfo.getSchema().getRowType(), + lookupMetricRecorder); return new LookupContext( tableInfo.getTableId(), schemaInfo.getSchemaId(), tablePath, lookupContext); } LakeTableLookuper createLakeTableLookuper( - TablePath tablePath, String ioTmpDir, TableConfig tableConfig) { - DataLakeFormat dataLakeFormat = conf.get(ConfigOptions.DATALAKE_FORMAT); + TablePath tablePath, + String ioTmpDir, + TableConfig tableConfig, + long cacheSizeBytes, + Configuration clusterConf) { + DataLakeFormat dataLakeFormat = clusterConf.get(ConfigOptions.DATALAKE_FORMAT); if (dataLakeFormat == null) { throw new LakeStorageNotConfiguredException( "Historical lookup requires cluster lake storage to be configured."); @@ -358,7 +572,7 @@ LakeTableLookuper createLakeTableLookuper( dataLakeFormat)); } - Map lakeProperties = extractLakeProperties(conf); + Map lakeProperties = extractLakeProperties(clusterConf); if (lakeProperties == null) { throw new LakeStorageNotConfiguredException( "Historical lookup requires cluster lake storage properties to be configured."); @@ -369,37 +583,80 @@ LakeTableLookuper createLakeTableLookuper( LakeStorage lakeStorage = lakeStoragePlugin.createLakeStorage(Configuration.fromMap(lakeProperties)); return lakeStorage.createLakeTableLookuper( - tablePath, new LakeStorage.LookuperContext(ioTmpDir, tableConfig)); + tablePath, + new LakeStorage.LookuperContext( + ioTmpDir, tableConfig, cacheSizeBytes, diskWriteGuard)); + } + + private static boolean hasLakeConfigChanged(Configuration currentConf, Configuration newConf) { + return currentConf.get(ConfigOptions.DATALAKE_FORMAT) + != newConf.get(ConfigOptions.DATALAKE_FORMAT) + || !Objects.equals( + extractLakeProperties(currentConf), extractLakeProperties(newConf)); + } + + private long cacheBytesPerTable(double ratio) { + checkArgument(ratio > 0.0 && ratio <= 1.0, "ratio must be within (0.0, 1.0]."); + long totalCacheBytes = + Math.min(dataDirVolumeBytes, (long) Math.ceil(dataDirVolumeBytes * ratio)); + return Math.max(1L, totalCacheBytes / MAX_CACHED_TABLES); } - private synchronized String getOrPreparePaimonLookupTempDir() { - if (paimonLookupTempDir == null) { - paimonLookupTempDir = preparePaimonLookupTempDir(conf, serverId); + /** Returns the most recently sampled historical lookup cache footprint, in bytes. */ + long lookupCacheDiskSize() { + return lookupCacheDiskSize; + } + + private void updateLookupCacheDiskSize() { + if (!historicalLookupCacheRootDir.exists()) { + lookupCacheDiskSize = 0L; + return; + } + try (Stream paths = Files.walk(historicalLookupCacheRootDir.toPath())) { + lookupCacheDiskSize = + paths.filter(Files::isRegularFile) + .mapToLong(HistoricalLakeLookupManager::fileSize) + .sum(); + } catch (IOException | UncheckedIOException e) { + LOG.warn( + "Failed to calculate historical lookup cache usage under {}. Keeping the last sampled value of {} bytes.", + historicalLookupCacheRootDir, + lookupCacheDiskSize, + e); } - return paimonLookupTempDir; } - private static String preparePaimonLookupTempDir(Configuration conf, int serverId) { - File paimonLookupTempDir = - new File( - new File(conf.get(ConfigOptions.SERVER_IO_TMP_DIR), PAIMON_LOOKUP_DIR_NAME), - String.valueOf(serverId)); + private static long fileSize(Path path) { try { - // A crashed server cannot close the Paimon IOManager, so lookup cache files may be - // left behind. Clean only this server's directory before creating the first table - // lookuper; cleaning in each table lookuper would delete files used by other tables. - FileUtils.deleteDirectory(paimonLookupTempDir); - Files.createDirectories(paimonLookupTempDir.toPath()); - return paimonLookupTempDir.getAbsolutePath(); + return Files.size(path); } catch (IOException e) { - throw new FlussRuntimeException( - "Failed to prepare Paimon lookup temporary directory: " + paimonLookupTempDir, - e); + throw new UncheckedIOException(e); } } private static void closeLookuper(CachedLakeTableLookuper cachedLookuper) { - IOUtils.closeQuietly(cachedLookuper.lookuper, "historical lake table lookuper"); + closeLookuper(cachedLookuper.lookuper, cachedLookuper.tableLookupDir); + } + + private static void closeLookuper(LakeTableLookuper lookuper, File tableLookupDir) { + try { + IOUtils.closeQuietly(lookuper, "historical lake table lookuper"); + } finally { + deleteTableLookupDirIfEmpty(tableLookupDir); + } + } + + private static void deleteTableLookupDirIfEmpty(File tableLookupDir) { + if (FileUtils.isDirectoryEmpty(tableLookupDir)) { + try { + Files.deleteIfExists(tableLookupDir.toPath()); + } catch (IOException e) { + LOG.debug( + "Failed to delete empty historical lookup directory {}.", + tableLookupDir, + e); + } + } } private static final class LookupContext { @@ -421,14 +678,31 @@ private LookupContext( } private static final class CachedLakeTableLookuper { + private final long tableId; + private final TablePath tablePath; private final int schemaId; + private final long lakeConfigVersion; + private final long cacheSizeBytes; + private final File tableLookupDir; private final LakeTableLookuper lookuper; private int activeLookups; private boolean invalidated; private boolean closed; - private CachedLakeTableLookuper(int schemaId, LakeTableLookuper lookuper) { + private CachedLakeTableLookuper( + long tableId, + TablePath tablePath, + int schemaId, + long lakeConfigVersion, + long cacheSizeBytes, + File tableLookupDir, + LakeTableLookuper lookuper) { + this.tableId = tableId; + this.tablePath = tablePath; this.schemaId = schemaId; + this.lakeConfigVersion = lakeConfigVersion; + this.cacheSizeBytes = cacheSizeBytes; + this.tableLookupDir = tableLookupDir; this.lookuper = lookuper; } @@ -439,22 +713,32 @@ private synchronized void acquire() { activeLookups++; } - private synchronized void release() { - if (activeLookups <= 0) { - throw new IllegalStateException("Lake table lookuper is not acquired."); + private void release() { + synchronized (this) { + if (activeLookups <= 0) { + throw new IllegalStateException("Lake table lookuper is not acquired."); + } + activeLookups--; } - activeLookups--; closeIfUnused(); } - private synchronized void invalidate() { - invalidated = true; + private void invalidate() { + synchronized (this) { + invalidated = true; + } closeIfUnused(); } private void closeIfUnused() { - if (invalidated && activeLookups == 0 && !closed) { - closed = true; + boolean shouldClose; + synchronized (this) { + shouldClose = invalidated && activeLookups == 0 && !closed; + if (shouldClose) { + closed = true; + } + } + if (shouldClose) { closeLookuper(this); } } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java index 5b0bebfc3e2..75b0f2ed1c3 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java @@ -24,6 +24,7 @@ import org.apache.fluss.config.cluster.ServerReconfigurable; import org.apache.fluss.exception.ConfigException; import org.apache.fluss.exception.FencedLeaderEpochException; +import org.apache.fluss.exception.HistoricalPartitionThrottledException; import org.apache.fluss.exception.InvalidColumnProjectionException; import org.apache.fluss.exception.InvalidCoordinatorException; import org.apache.fluss.exception.InvalidPartitionException; @@ -359,13 +360,24 @@ public ReplicaManager( this.ioExecutor = ioExecutor; this.minInSyncReplicas = conf.get(ConfigOptions.LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER); this.scannerManager = checkNotNull(scannerManager, "scannerManager"); + // Historical lookup cache capacity currently uses only the first data volume. + File dataDir = localDiskManager.dataDirs().get(0); + long dataDirVolumeBytes = Files.getFileStore(dataDir.toPath()).getTotalSpace(); this.historicalLakeLookupManager = - new HistoricalLakeLookupManager(conf, pluginManager, serverId, scheduler); + new HistoricalLakeLookupManager( + conf, + pluginManager, + localDiskManager, + dataDir, + dataDirVolumeBytes, + scheduler); registerMetrics(); } public void startup() { + historicalLakeLookupManager.startup(scheduler); + // start up ISR expiration thread. // A follower can log behind leader for up tp configOptions#LOG_REPLICA_MAX_LAG_TIME x 1.5 // before it is removed from ISR. @@ -415,6 +427,7 @@ public void validate(Configuration newConfig) throws ConfigException { @Override public void reconfigure(Configuration newConfig) { + historicalLakeLookupManager.reconfigure(newConfig); int newMinInSyncReplicas = newConfig.get(ConfigOptions.LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER); if (newMinInSyncReplicas == minInSyncReplicas) { @@ -431,6 +444,21 @@ public void reconfigure(Configuration newConfig) { } private void registerMetrics() { + // for historical lookup metrics + MetricGroup historicalMetrics = serverMetricGroup.addGroup("historical"); + historicalMetrics.gauge( + MetricNames.HISTORICAL_INFLIGHT_REQUESTS, + historicalLakeLookupManager::numInflightRequests); + historicalMetrics.gauge( + MetricNames.HISTORICAL_LOOKUP_CACHE_DISK_SIZE, + historicalLakeLookupManager::lookupCacheDiskSize); + historicalMetrics.gauge( + MetricNames.HISTORICAL_LOOKUP_CACHE_TABLE_COUNT, + historicalLakeLookupManager::cachedTableCount); + historicalMetrics.counter( + MetricNames.HISTORICAL_LOOKUP_CACHE_CAPACITY_EVICTIONS, + historicalLakeLookupManager.capacityEvictions()); + serverMetricGroup.gauge( MetricNames.REPLICA_LEADER_COUNT, () -> onlineReplicas().filter(Replica::isLeader).count()); @@ -821,9 +849,11 @@ public void historicalLookups( Collections.synchronizedList(new ArrayList<>(lookupData.size())); AtomicInteger remainingLookups = new AtomicInteger(lookupData.size()); for (LookupDataForBucket data : lookupData) { + Replica replica; CompletableFuture lookupFuture; try { - Replica replica = getReplicaOrException(data.tableBucket()); + replica = getReplicaOrException(data.tableBucket()); + replica.tableMetrics().totalHistoricalLookupRequests().inc(); if (!replica.isKvTable()) { throw new NonPrimaryKeyTableException( "Historical lookup is only supported for primary key tables, but " @@ -837,28 +867,38 @@ public void historicalLookups( SchemaInfo latestSchemaInfo = replica.getSchemaGetter().getLatestSchemaInfo(); lookupFuture = historicalLakeLookupManager.lookup( - data, replica.getTableInfo(), latestSchemaInfo); + data, + replica.getTableInfo(), + latestSchemaInfo, + replica.tableMetrics()::recordHistoricalLakeLookup); } catch (Exception e) { - lookupFuture = - CompletableFuture.completedFuture( - new LookupResultForBucket( - data.tableBucket(), - null, - data.originalPartitionName(), - ApiError.fromThrowable(e))); + result.add( + new LookupResultForBucket( + data.tableBucket(), + null, + data.originalPartitionName(), + ApiError.fromThrowable(e))); + if (remainingLookups.decrementAndGet() == 0) { + responseCallback.accept(result); + } + continue; } lookupFuture.whenComplete( (bucketResult, error) -> { - if (error == null) { - result.add(bucketResult); - } else { - result.add( - new LookupResultForBucket( - data.tableBucket(), - null, - data.originalPartitionName(), - ApiError.fromThrowable(error))); + LookupResultForBucket completedResult = + error == null + ? bucketResult + : new LookupResultForBucket( + data.tableBucket(), + null, + data.originalPartitionName(), + ApiError.fromThrowable(error)); + if (completedResult.failed() + && isUnexpectedHistoricalLookupException( + completedResult.getError().exception())) { + replica.tableMetrics().failedHistoricalLookupRequests().inc(); } + result.add(completedResult); if (remainingLookups.decrementAndGet() == 0) { responseCallback.accept(result); } @@ -1818,6 +1858,13 @@ private boolean isUnexpectedException(Exception e) { || e instanceof StorageBackpressureException); } + private boolean isUnexpectedHistoricalLookupException(Exception e) { + return isUnexpectedException(e) + && !(e instanceof HistoricalPartitionThrottledException + || e instanceof InvalidPartitionException + || e instanceof NonPrimaryKeyTableException); + } + /** * Start the high watermark check point thread to periodically flush the high watermark value * for all buckets to the high watermark checkpoint file. diff --git a/fluss-server/src/main/java/org/apache/fluss/server/storage/LocalDiskManager.java b/fluss-server/src/main/java/org/apache/fluss/server/storage/LocalDiskManager.java index b12e897ab62..e34283e4548 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/storage/LocalDiskManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/storage/LocalDiskManager.java @@ -512,9 +512,9 @@ public double getDiskWriteRecoverRatio() { /** * Throws {@link DiskWriteLockedException} when the local data disk usage has crossed the - * configured write-limit ratio. Only client-driven writes ({@code appendLog} / {@code putKv}) - * should call this; follower replication paths must bypass this check to preserve replica - * consistency. + * configured write-limit ratio. Client-driven writes ({@code appendLog} / {@code putKv}) and + * lookup cache file download paths should call this; follower replication paths must bypass + * this check to preserve replica consistency. */ public void ensureWritable() { if (diskWriteLocked) { diff --git a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java index 9452842ab67..f7a004de351 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java @@ -62,7 +62,6 @@ import org.apache.fluss.rpc.messages.NotifyLeaderAndIsrResponse; import org.apache.fluss.rpc.messages.NotifyRemoteLogOffsetsRequest; import org.apache.fluss.rpc.messages.NotifyRemoteLogOffsetsResponse; -import org.apache.fluss.rpc.messages.PbLookupReqForBucket; import org.apache.fluss.rpc.messages.PbScanReqForBucket; import org.apache.fluss.rpc.messages.PrefixLookupRequest; import org.apache.fluss.rpc.messages.PrefixLookupResponse; @@ -123,6 +122,7 @@ import java.util.function.Consumer; import java.util.stream.Collectors; +import static org.apache.fluss.rpc.util.CommonRpcMessageUtils.hasHistoricalLookup; import static org.apache.fluss.security.acl.OperationType.DESCRIBE; import static org.apache.fluss.security.acl.OperationType.READ; import static org.apache.fluss.security.acl.OperationType.WRITE; @@ -368,18 +368,6 @@ public CompletableFuture prefixLookup(PrefixLookupRequest return response; } - private boolean hasHistoricalLookup(LookupRequest request) { - for (PbLookupReqForBucket lookupReqForBucket : request.getBucketsReqsList()) { - if (lookupReqForBucket.hasOriginalPartitionName()) { - // An original partition name is only set for historical lookups, so route the - // whole request to the historical path. Conversion rejects requests that mix - // normal and historical lookup batches. - return true; - } - } - return false; - } - @Override public CompletableFuture limitScan(LimitScanRequest request) { authorizeTable(READ, request.getTableId()); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java index 60212736d6c..89ae6f7138e 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java @@ -1123,6 +1123,10 @@ public static Map> toLookupData(LookupRequest lookupRe long tableId = lookupRequest.getTableId(); Map> lookupEntryData = new HashMap<>(); for (PbLookupReqForBucket lookupReqForBucket : lookupRequest.getBucketsReqsList()) { + if (lookupReqForBucket.hasOriginalPartitionName()) { + throw new IllegalArgumentException( + "Normal and historical lookups cannot be mixed in the same request."); + } TableBucket tb = new TableBucket( tableId, diff --git a/fluss-server/src/test/java/org/apache/fluss/server/TabletManagerBaseTest.java b/fluss-server/src/test/java/org/apache/fluss/server/TabletManagerBaseTest.java index 498c9b95f15..01d43c5562f 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/TabletManagerBaseTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/TabletManagerBaseTest.java @@ -19,9 +19,12 @@ import org.apache.fluss.config.Configuration; import org.apache.fluss.exception.FlussRuntimeException; +import org.apache.fluss.utils.FlussPaths; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import java.io.File; import java.util.Arrays; import java.util.Collections; import java.util.List; @@ -39,6 +42,21 @@ /** Test for {@link TabletManagerBase}. */ final class TabletManagerBaseTest { + @TempDir private File tempDir; + + @Test + void testIgnoresHistoricalLookupCacheDirectoryWhenLoadingTablets() { + File fakeTabletDir = + new File( + new File(FlussPaths.historicalLookupRootDir(tempDir), "database"), + "kv-table-1"); + assertThat(fakeTabletDir.mkdirs()).isTrue(); + + TestingTabletManager tabletManager = new TestingTabletManager(tempDir); + + assertThat(tabletManager.tabletsToLoad(tempDir)).isEmpty(); + } + @Test void testCloseTabletsConcurrentlyWaitsForAllTasksAndShutsDownPoolOnFailure() throws Exception { TestingTabletManager tabletManager = new TestingTabletManager(2); @@ -92,6 +110,14 @@ private TestingTabletManager(int closingThreads) { super(TabletType.KV, Collections.emptyList(), new Configuration(), closingThreads); } + private TestingTabletManager(File dataDir) { + super(TabletType.KV, Collections.singletonList(dataDir), new Configuration(), 1); + } + + private List tabletsToLoad(File dataDir) { + return listTabletsToLoad(dataDir); + } + private CompletableFuture closeTablets( List tablets, Consumer closeAction) { return closeTabletsConcurrently(tablets, "testing-tablet-closing", closeAction); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/HistoricalLakeLookupManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/HistoricalLakeLookupManagerTest.java index ecf3a5e4e2e..afa53e5516b 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/replica/HistoricalLakeLookupManagerTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/HistoricalLakeLookupManagerTest.java @@ -19,9 +19,11 @@ 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.HistoricalPartitionThrottledException; import org.apache.fluss.lake.lakestorage.LakeTableLookuper; +import org.apache.fluss.metadata.DataLakeFormat; import org.apache.fluss.metadata.KvFormat; import org.apache.fluss.metadata.Schema; import org.apache.fluss.metadata.SchemaInfo; @@ -33,6 +35,7 @@ import org.apache.fluss.rpc.protocol.Errors; import org.apache.fluss.server.entity.LookupDataForBucket; import org.apache.fluss.types.DataTypes; +import org.apache.fluss.utils.FlussPaths; import com.github.benmanes.caffeine.cache.Scheduler; import com.github.benmanes.caffeine.cache.Ticker; @@ -42,6 +45,7 @@ import org.junit.jupiter.params.provider.ValueSource; import java.io.File; +import java.io.RandomAccessFile; import java.time.Duration; import java.util.ArrayList; import java.util.Collections; @@ -52,6 +56,7 @@ import java.util.concurrent.FutureTask; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; @@ -64,8 +69,13 @@ /** Tests for {@link HistoricalLakeLookupManager}. */ class HistoricalLakeLookupManagerTest { - private static final int SERVER_ID = 1; + private static final long DATA_DIR_VOLUME_BYTES = MemorySize.parse("800gb").getBytes(); private static final TableBucket HISTORICAL_BUCKET = new TableBucket(PARTITION_TABLE_ID, 1L, 0); + private static final LakeTableLookuper.LookupMetricRecorder NO_OP_LOOKUP_METRIC_RECORDER = + (lookupTimeNanos, lookupFileDownloaded) -> {}; + private static final Runnable NO_OP_DISK_WRITE_GUARD = () -> {}; + private static final org.apache.fluss.utils.concurrent.Scheduler NO_OP_SCHEDULER = + new NoOpScheduler(); @TempDir private File ioTmpDir; @@ -73,21 +83,25 @@ class HistoricalLakeLookupManagerTest { void testHistoricalLookupThrottledWhenPermitsExhausted() throws Exception { ManualExecutor executor = new ManualExecutor(); HistoricalLakeLookupManager manager = createManager(1, executor); + assertThat(manager.numInflightRequests()).isZero(); CompletableFuture first = manager.lookup( lookupData(HISTORICAL_BUCKET), PARTITION_TABLE_INFO, - PARTITION_TABLE_INFO.getSchemaInfo()); + PARTITION_TABLE_INFO.getSchemaInfo(), + NO_OP_LOOKUP_METRIC_RECORDER); assertThat(first).isNotDone(); assertThat(executor.numQueuedTasks()).isEqualTo(1); + assertThat(manager.numInflightRequests()).isOne(); TableBucket secondBucket = new TableBucket(PARTITION_TABLE_ID, 2L, 0); LookupResultForBucket second = manager.lookup( lookupData(secondBucket), PARTITION_TABLE_INFO, - PARTITION_TABLE_INFO.getSchemaInfo()) + PARTITION_TABLE_INFO.getSchemaInfo(), + NO_OP_LOOKUP_METRIC_RECORDER) .get(1, TimeUnit.SECONDS); assertThat(second.failed()).isTrue(); @@ -95,6 +109,7 @@ void testHistoricalLookupThrottledWhenPermitsExhausted() throws Exception { assertThat(second.getError().exception()) .isInstanceOf(HistoricalPartitionThrottledException.class); assertThat(executor.numQueuedTasks()).isEqualTo(1); + assertThat(manager.numInflightRequests()).isOne(); } @Test @@ -106,18 +121,21 @@ void testHistoricalLookupReleasesPermitOnFailure() throws Exception { manager.lookup( lookupData(HISTORICAL_BUCKET), PARTITION_TABLE_INFO, - PARTITION_TABLE_INFO.getSchemaInfo()); + PARTITION_TABLE_INFO.getSchemaInfo(), + NO_OP_LOOKUP_METRIC_RECORDER); executor.runNext(); LookupResultForBucket firstResult = first.get(1, TimeUnit.SECONDS); assertThat(firstResult.failed()).isTrue(); assertThat(firstResult.getError().error()) .isNotEqualTo(Errors.HISTORICAL_PARTITION_THROTTLED); + assertThat(manager.numInflightRequests()).isZero(); CompletableFuture second = manager.lookup( lookupData(HISTORICAL_BUCKET), PARTITION_TABLE_INFO, - PARTITION_TABLE_INFO.getSchemaInfo()); + PARTITION_TABLE_INFO.getSchemaInfo(), + NO_OP_LOOKUP_METRIC_RECORDER); assertThat(second).isNotDone(); assertThat(executor.numQueuedTasks()).isEqualTo(1); } @@ -131,17 +149,20 @@ void testHistoricalLookupMaxQueuedRequestsUsesExplicitConfig() throws Exception manager.lookup( lookupData(new TableBucket(PARTITION_TABLE_ID, 1L, 0)), PARTITION_TABLE_INFO, - PARTITION_TABLE_INFO.getSchemaInfo()); + PARTITION_TABLE_INFO.getSchemaInfo(), + NO_OP_LOOKUP_METRIC_RECORDER); CompletableFuture second = manager.lookup( lookupData(new TableBucket(PARTITION_TABLE_ID, 2L, 0)), PARTITION_TABLE_INFO, - PARTITION_TABLE_INFO.getSchemaInfo()); + PARTITION_TABLE_INFO.getSchemaInfo(), + NO_OP_LOOKUP_METRIC_RECORDER); LookupResultForBucket third = manager.lookup( lookupData(new TableBucket(PARTITION_TABLE_ID, 3L, 0)), PARTITION_TABLE_INFO, - PARTITION_TABLE_INFO.getSchemaInfo()) + PARTITION_TABLE_INFO.getSchemaInfo(), + NO_OP_LOOKUP_METRIC_RECORDER) .get(1, TimeUnit.SECONDS); assertThat(first).isNotDone(); @@ -161,9 +182,11 @@ void testRejectNonPositiveHistoricalLookupMaxQueuedRequests() { conf, null, executor, - SERVER_ID, + ioTmpDir, + DATA_DIR_VOLUME_BYTES, Ticker.systemTicker(), - Scheduler.disabledScheduler())) + Scheduler.disabledScheduler(), + NO_OP_DISK_WRITE_GUARD)) .isInstanceOf(IllegalArgumentException.class) .hasMessageContaining( ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS.key()); @@ -181,30 +204,38 @@ void testRejectNonPositiveHistoricalPartitionThreadPoolMaxSize(int maxThreadPool conf, null, null, - SERVER_ID, + ioTmpDir, + DATA_DIR_VOLUME_BYTES, Ticker.systemTicker(), - Scheduler.disabledScheduler())) + Scheduler.disabledScheduler(), + NO_OP_DISK_WRITE_GUARD)) .isInstanceOf(IllegalArgumentException.class) .hasMessageContaining( ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE.key()); } @Test - void testLazilyCleansPaimonLookupTempDirectory() throws Exception { - File serverLookupDir = - new File(new File(ioTmpDir, "paimon-lookup"), String.valueOf(SERVER_ID)); + void testCleansAndCreatesLookupCacheDirectoryOnStartup() throws Exception { + File serverLookupDir = FlussPaths.historicalLookupRootDir(ioTmpDir); assertThat(serverLookupDir.mkdirs()).isTrue(); File staleLookupFile = new File(serverLookupDir, "stale-lookup-file"); assertThat(staleLookupFile.createNewFile()).isTrue(); ManualExecutor executor = new ManualExecutor(); - TestingHistoricalLakeLookupManager manager = createTestingManager(executor); + TestingHistoricalLakeLookupManager manager = + new TestingHistoricalLakeLookupManager(conf(1), executor); assertThat(staleLookupFile).exists(); - lookupAndRun(manager, executor, PARTITION_TABLE_INFO); + manager.startup(NO_OP_SCHEDULER); assertThat(staleLookupFile).doesNotExist(); assertThat(serverLookupDir).isDirectory(); - assertThat(manager.createdIoTmpDirs).containsExactly(serverLookupDir.getAbsolutePath()); + lookupAndRun(manager, executor, PARTITION_TABLE_INFO); + assertThat(manager.createdIoTmpDirs.get(0)).startsWith(serverLookupDir.getAbsolutePath()); + + File liveLookupFile = new File(serverLookupDir, "live-lookup-file"); + assertThat(liveLookupFile.createNewFile()).isTrue(); + manager.startup(NO_OP_SCHEDULER); + assertThat(liveLookupFile).exists(); } @Test @@ -283,7 +314,36 @@ void testInvalidatesLookuperOnSchemaAndLifecycleChanges() throws Exception { } @Test - void testExpiresIdleLookuperWithoutAnotherLookup() throws Exception { + void testDoesNotReplaceLookuperForUnrelatedTableConfigChange() throws Exception { + ManualExecutor executor = new ManualExecutor(); + TestingHistoricalLakeLookupManager manager = createTestingManager(executor); + + lookupAndRun(manager, executor, PARTITION_TABLE_INFO); + TestingLakeTableLookuper initialLookuper = manager.createdLookupers.get(0); + + TableDescriptor changedDescriptor = + TableDescriptor.builder(PARTITION_TABLE_INFO.toTableDescriptor()) + .property( + ConfigOptions.TABLE_TIERED_LOG_LOCAL_SEGMENTS, + ConfigOptions.TABLE_TIERED_LOG_LOCAL_SEGMENTS.defaultValue() + 1) + .build(); + TableInfo changedTableInfo = + TableInfo.of( + PARTITION_TABLE_INFO.getTablePath(), + PARTITION_TABLE_INFO.getTableId(), + PARTITION_TABLE_INFO.getSchemaId(), + changedDescriptor, + PARTITION_TABLE_INFO.getRemoteDataDir(), + PARTITION_TABLE_INFO.getCreatedTime(), + PARTITION_TABLE_INFO.getModifiedTime()); + lookupAndRun(manager, executor, changedTableInfo); + + assertThat(manager.createdLookupers).hasSize(1); + assertThat(initialLookuper.closed).isFalse(); + } + + @Test + void testDynamicallyUpdatesExpirationAndExpiresIdleLookuper() throws Exception { ManualExecutor executor = new ManualExecutor(); AtomicLong tickerNanos = new AtomicLong(); AtomicReference> expirationTask = new AtomicReference<>(); @@ -300,12 +360,17 @@ void testExpiresIdleLookuperWithoutAnotherLookup() throws Exception { }; TestingHistoricalLakeLookupManager manager = new TestingHistoricalLakeLookupManager( - conf(1), executor, tickerNanos::get, cacheScheduler); + confWithExpiration(Duration.ofHours(1)), + executor, + tickerNanos::get, + cacheScheduler); + manager.startup(NO_OP_SCHEDULER); lookupAndRun(manager, executor, PARTITION_TABLE_INFO); TestingLakeTableLookuper expiredLookuper = manager.createdLookupers.get(0); - tickerNanos.addAndGet(Duration.ofHours(4).toNanos()); + manager.reconfigure(confWithExpiration(Duration.ofMinutes(30))); + tickerNanos.addAndGet(Duration.ofMinutes(31).toNanos()); assertThat(expirationTask.get()).isNotNull(); expirationTask.get().run(); @@ -315,9 +380,19 @@ void testExpiresIdleLookuperWithoutAnotherLookup() throws Exception { } @Test - void testLimitsCachedLookupersToTen() throws Exception { + void testEvictsLookuperWhenCachedTableLimitIsExceeded() throws Exception { ManualExecutor executor = new ManualExecutor(); - TestingHistoricalLakeLookupManager manager = createTestingManager(executor); + Configuration conf = conf(1); + conf.set(ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO, 0.20); + TestingHistoricalLakeLookupManager manager = + new TestingHistoricalLakeLookupManager( + conf, + executor, + Ticker.systemTicker(), + Scheduler.disabledScheduler(), + 100, + 0); + manager.startup(NO_OP_SCHEDULER); for (int i = 0; i < 11; i++) { lookupAndRun( @@ -328,21 +403,57 @@ void testLimitsCachedLookupersToTen() throws Exception { assertThat(manager.createdLookupers).hasSize(11); assertThat(manager.createdLookupers).filteredOn(lookuper -> lookuper.closed).hasSize(1); + assertThat(manager.createdCacheSizes).containsOnly(2L); + assertThat(manager.cachedTableCount()).isEqualTo(10); + assertThat(manager.capacityEvictions().getCount()).isEqualTo(1); + } + + @Test + void testReconfiguresLakePropertiesAndInvalidatesLookuper() throws Exception { + Configuration initialConf = conf(1); + initialConf.set(ConfigOptions.DATALAKE_FORMAT, DataLakeFormat.PAIMON); + initialConf.setString("datalake.paimon.warehouse", "old-warehouse"); + ManualExecutor executor = new ManualExecutor(); + TestingHistoricalLakeLookupManager manager = + new TestingHistoricalLakeLookupManager(initialConf, executor); + manager.startup(NO_OP_SCHEDULER); + + lookupAndRun(manager, executor, PARTITION_TABLE_INFO); + TestingLakeTableLookuper initialLookuper = manager.createdLookupers.get(0); + + Configuration newConf = new Configuration(initialConf); + newConf.setString("datalake.paimon.warehouse", "new-warehouse"); + manager.reconfigure(newConf); + + assertThat(initialLookuper.closed).isTrue(); + assertThat(manager.cachedTableCount()).isZero(); + lookupAndRun(manager, executor, PARTITION_TABLE_INFO); + assertThat(manager.createdLookupers).hasSize(2); + assertThat(manager.createdClusterConfigs.get(1).toMap()) + .containsEntry("datalake.paimon.warehouse", "new-warehouse"); } private HistoricalLakeLookupManager createManager( int maxQueuedHistoricalRequests, ManualExecutor executor) { - return new HistoricalLakeLookupManager( - conf(maxQueuedHistoricalRequests), - null, - executor, - SERVER_ID, - Ticker.systemTicker(), - Scheduler.disabledScheduler()); + HistoricalLakeLookupManager manager = + new HistoricalLakeLookupManager( + conf(maxQueuedHistoricalRequests), + null, + executor, + ioTmpDir, + DATA_DIR_VOLUME_BYTES, + Ticker.systemTicker(), + Scheduler.disabledScheduler(), + NO_OP_DISK_WRITE_GUARD); + manager.startup(NO_OP_SCHEDULER); + return manager; } private TestingHistoricalLakeLookupManager createTestingManager(ManualExecutor executor) { - return new TestingHistoricalLakeLookupManager(conf(1), executor); + TestingHistoricalLakeLookupManager manager = + new TestingHistoricalLakeLookupManager(conf(1), executor); + manager.startup(NO_OP_SCHEDULER); + return manager; } private Configuration conf(int maxQueuedHistoricalRequests) { @@ -350,7 +461,15 @@ private Configuration conf(int maxQueuedHistoricalRequests) { conf.set( ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS, maxQueuedHistoricalRequests); - conf.set(ConfigOptions.SERVER_IO_TMP_DIR, ioTmpDir.getAbsolutePath()); + conf.set(ConfigOptions.DATA_DIR, ioTmpDir.getAbsolutePath()); + return conf; + } + + private Configuration confWithExpiration(Duration expiration) { + Configuration conf = conf(1); + conf.set( + ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUPER_CACHE_EXPIRE_AFTER_ACCESS, + expiration); return conf; } @@ -359,6 +478,15 @@ private static LookupDataForBucket lookupData(TableBucket tableBucket) { tableBucket, Collections.singletonList(new byte[] {1}), "2024"); } + private static CompletableFuture lookup( + HistoricalLakeLookupManager manager, TableInfo tableInfo) { + return manager.lookup( + lookupData(new TableBucket(tableInfo.getTableId(), 1L, 0)), + tableInfo, + tableInfo.getSchemaInfo(), + NO_OP_LOOKUP_METRIC_RECORDER); + } + private static TableInfo tableInfo(long tableId, int schemaId) { return TableInfo.of( PARTITION_TABLE_INFO.getTablePath(), @@ -382,13 +510,32 @@ private static void lookupAndRun( TableInfo tableInfo, SchemaInfo schemaInfo) throws Exception { + LookupResultForBucket result = lookupResultAndRun(manager, executor, tableInfo, schemaInfo); + assertThat(result.failed()).isFalse(); + assertThat(result.originalPartitionName()).isEqualTo("2024"); + } + + private static LookupResultForBucket lookupResultAndRun( + HistoricalLakeLookupManager manager, ManualExecutor executor, TableInfo tableInfo) + throws Exception { + return lookupResultAndRun(manager, executor, tableInfo, tableInfo.getSchemaInfo()); + } + + private static LookupResultForBucket lookupResultAndRun( + HistoricalLakeLookupManager manager, + ManualExecutor executor, + TableInfo tableInfo, + SchemaInfo schemaInfo) + throws Exception { TableBucket tableBucket = new TableBucket(tableInfo.getTableId(), 1L, 0); CompletableFuture future = - manager.lookup(lookupData(tableBucket), tableInfo, schemaInfo); + manager.lookup( + lookupData(tableBucket), + tableInfo, + schemaInfo, + NO_OP_LOOKUP_METRIC_RECORDER); executor.runNext(); - LookupResultForBucket result = future.get(1, TimeUnit.SECONDS); - assertThat(result.failed()).isFalse(); - assertThat(result.originalPartitionName()).isEqualTo("2024"); + return future.get(1, TimeUnit.SECONDS); } private static final class TestingHistoricalLakeLookupManager @@ -396,15 +543,21 @@ private static final class TestingHistoricalLakeLookupManager private final List createdLookupers = new ArrayList<>(); private final List createdIoTmpDirs = new ArrayList<>(); private final List createdTableConfigs = new ArrayList<>(); + private final List createdCacheSizes = new ArrayList<>(); + private final List createdClusterConfigs = new ArrayList<>(); + private final long lookupCacheFileBytes; private TestingHistoricalLakeLookupManager(Configuration conf, ManualExecutor executor) { super( conf, null, executor, - SERVER_ID, + new File(conf.get(ConfigOptions.DATA_DIR)), + DATA_DIR_VOLUME_BYTES, Ticker.systemTicker(), - Scheduler.disabledScheduler()); + Scheduler.disabledScheduler(), + NO_OP_DISK_WRITE_GUARD); + this.lookupCacheFileBytes = 0L; } private TestingHistoricalLakeLookupManager( @@ -412,36 +565,110 @@ private TestingHistoricalLakeLookupManager( ManualExecutor executor, Ticker ticker, Scheduler cacheScheduler) { - super(conf, null, executor, SERVER_ID, ticker, cacheScheduler); + super( + conf, + null, + executor, + new File(conf.get(ConfigOptions.DATA_DIR)), + DATA_DIR_VOLUME_BYTES, + ticker, + cacheScheduler, + NO_OP_DISK_WRITE_GUARD); + this.lookupCacheFileBytes = 0L; + } + + private TestingHistoricalLakeLookupManager( + Configuration conf, + ManualExecutor executor, + Ticker ticker, + Scheduler cacheScheduler, + long dataDirVolumeBytes, + long lookupCacheFileBytes) { + super( + conf, + null, + executor, + new File(conf.get(ConfigOptions.DATA_DIR)), + dataDirVolumeBytes, + ticker, + cacheScheduler, + NO_OP_DISK_WRITE_GUARD); + this.lookupCacheFileBytes = lookupCacheFileBytes; } @Override LakeTableLookuper createLakeTableLookuper( - TablePath tablePath, String ioTmpDir, TableConfig tableConfig) { - TestingLakeTableLookuper lookuper = new TestingLakeTableLookuper(); + TablePath tablePath, + String ioTmpDir, + TableConfig tableConfig, + long cacheSizeBytes, + Configuration clusterConf) { + TestingLakeTableLookuper lookuper = + new TestingLakeTableLookuper(new File(ioTmpDir), lookupCacheFileBytes); createdLookupers.add(lookuper); createdIoTmpDirs.add(ioTmpDir); createdTableConfigs.add(tableConfig); + createdCacheSizes.add(cacheSizeBytes); + createdClusterConfigs.add(clusterConf); return lookuper; } } private static final class TestingLakeTableLookuper implements LakeTableLookuper { + private final File cacheFile; + private final long cacheFileBytes; private boolean closed; + private boolean cacheFileDownloaded; private final List lookupContexts = new ArrayList<>(); + private TestingLakeTableLookuper(File lookupDir, long cacheFileBytes) { + this.cacheFile = new File(lookupDir, "cache-file"); + this.cacheFileBytes = cacheFileBytes; + } + @Override - public byte[] lookup(byte[] key, LookupContext context) { + public byte[] lookup(byte[] key, LookupContext context) throws Exception { if (closed) { throw new IllegalStateException("Lookuper is already closed."); } lookupContexts.add(context); + boolean downloaded = false; + if (!cacheFileDownloaded && cacheFileBytes > 0) { + java.nio.file.Files.createDirectories(cacheFile.getParentFile().toPath()); + try (RandomAccessFile file = new RandomAccessFile(cacheFile, "rw")) { + file.setLength(cacheFileBytes); + } + cacheFileDownloaded = true; + downloaded = true; + } + context.lookupMetricRecorder().recordLookup(1L, downloaded); return key; } @Override - public void close() { + public void close() throws Exception { closed = true; + java.nio.file.Files.deleteIfExists(cacheFile.toPath()); + } + } + + private static final class NoOpScheduler + implements org.apache.fluss.utils.concurrent.Scheduler { + + @Override + public void startup() { + // no-op + } + + @Override + public void shutdown() { + // no-op + } + + @Override + public ScheduledFuture schedule( + String name, Runnable task, long delayMs, long periodMs) { + return null; } } diff --git a/fluss-test-coverage/pom.xml b/fluss-test-coverage/pom.xml index b742c2d8e1e..485dbf0ed51 100644 --- a/fluss-test-coverage/pom.xml +++ b/fluss-test-coverage/pom.xml @@ -500,6 +500,9 @@ org.apache.fluss.flink.DummyClass120 org.apache.fluss.lake.batch.ArrowRecordBatch org.apache.fluss.lake.committer.CommittedLakeSnapshot + + org.apache.fluss.lake.paimon.lookup.PaimonLakeTableLookuper.TrackingIOManager + org.apache.fluss.lake.paimon.utils.FlussDataTypeToPaimonDataType org.apache.paimon.arrow.converter.Arrow2PaimonVectorConverter* diff --git a/website/docs/maintenance/observability/monitor-metrics.md b/website/docs/maintenance/observability/monitor-metrics.md index 18a4edccb2a..4bb313e61bc 100644 --- a/website/docs/maintenance/observability/monitor-metrics.md +++ b/website/docs/maintenance/observability/monitor-metrics.md @@ -463,7 +463,7 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM - tabletserver + tabletserver - messagesInPerSecond The number of messages written per second to this server. @@ -589,6 +589,27 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM The number of kv pre-write buffer truncate due to the error happened when writing cdc to log per second. Meter + + historical + inflightRequests + The number of accepted historical requests that have not completed. + Gauge + + + lookupCacheDiskSize + The current historical lookup cache footprint on local disk, in bytes. + Gauge + + + lookupCacheTableCount + The number of table lookupers currently retained in the historical lookup cache. + Gauge + + + lookupCacheCapacityEvictions + The cumulative number of cached table lookupers evicted because the cache retains at most ten tables. + Counter + logicalStorage logSize @@ -679,6 +700,7 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM request_produceLog request_putKv request_lookup + request_historicalLookup request_prefixLookup request_metadata request_fetchLogClient @@ -767,7 +789,7 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM - tabletserver + tabletserver table messagesInPerSecond The number of messages written per second to this table. @@ -867,6 +889,27 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM remoteLogDeleteErrorPerSecond The number of failed delete remote log requests to delete remote log after log ttl per second. Meter + + + table_historical + totalLookupRequestsPerSecond + The number of historical lookup requests to this table per second. + Meter + + + failedLookupRequestsPerSecond + The number of historical lookup requests that failed unexpectedly for this table per second. + Meter + + + lakeLookupsPerSecond + The number of historical lake point lookups performed for this table per second, labeled with lookup_file_downloaded. + Meter + + + lakeLookupTimeMs + The time spent on a historical lake point lookup, in milliseconds, labeled with lookup_file_downloaded. + Histogram table_bucket_log @@ -941,6 +984,10 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM +For lakeLookupsPerSecond and lakeLookupTimeMs, +lookup_file_downloaded="true" means that the lookup downloaded at least one local +lookup file; false means that it did not download a local lookup file. + ### RocksDB RocksDB metrics provide insights into the performance and health of the underlying RocksDB storage engine used by Fluss. These metrics are categorized into table-level metrics (aggregated from all buckets of a table) and server-level metrics (aggregated from all tables in a server).