This is an automated email from the ASF dual-hosted git repository. yiguolei pushed a commit to branch branch-4.2 in repository https://gitbox.apache.org/repos/asf/doris.git
commit f7cfe170aa8c99afe9788faf6d3ac558414f52cc Author: Gabriel <[email protected]> AuthorDate: Mon Sep 21 15:59:09 2026 +0800 [opt](lance) Cache table access resolution in FE (#68305) ### What problem does this PR solve? Repeated Lance metadata reads call `describeTable` before opening the Dataset, even when the catalog already shares a Lance metadata Session. This repeats filesystem discovery or REST requests and serializes callers on the namespace lock during query planning. Cache immutable table URIs and normalized access options for both filesystem and REST catalogs. Cache hits bypass the namespace lock, and concurrent misses for the same table share one load. Dataset opens and snapshot selection still run for every read. - Add `lance.table_access_cache_ttl_seconds` (default `60`; `0` disables caching), with at most 10,000 entries per catalog client generation. Reads do not extend the TTL. - Resolve responses containing vended storage options on every read, even when `expires_at_millis` is present: the BE cannot renew credentials during an arbitrarily long scan. Also bypass caching for credential-bearing or unclassified URIs, including userinfo, query parameters, and fragments. Plain filesystem and REST responses without vended options remain cacheable. - Explicit table/database refresh, catalog invalidation, and namespace removal retire the access cache, including in-flight loads. Refresh replay invalidates before cache-only object lookup, including when local database/table objects are absent. Routine database-object eviction preserves access entries. Table/database refresh conservatively clears all access entries because Doris refresh names may be mapped names; it does not rotate the native Session. - Keep index inspection/admission and current index-job locator validation on an uncached path so they verify the current target. ### Release note Reduce repeated Lance query-planning work by caching table access resolution for filesystem and REST catalogs, with credential-safe cache eligibility and explicit refresh invalidation. ### Validation - Added regression tests that first failed on the original implementation: two reads caused two `describeTable` calls for both filesystem and REST catalogs. - Added regressions reproduced five failures before the review fixes: credential reuse, signed URIs, routine database-object eviction, and refresh replay with missing table/database objects. - 66 FE tests passed (0 failures/errors/skips): access-cache expiry, credential handling, concurrent loading/invalidation, catalog lifecycle, property validation, filesystem/REST catalogs, and metadata-cache routing. - FE Checkstyle passed with zero violations. - No end-to-end latency benchmark was run. ### Check List (For Author) - Test - [x] Unit Test - Behavior changed: - [x] Yes. Table URI/access-option changes can remain cached until TTL expiry or explicit refresh; Dataset versions are not cached here. Set `lance.table_access_cache_ttl_seconds=0` to retain per-read resolution. - Does this need documentation? - [x] Yes. English and Chinese documentation: https://github.com/apache/doris-website/pull/4159 (draft pending this implementation). Covers the TTL property, credential exclusions, refresh behavior, and FE/BE cache separation. ### Check List (For Reviewer who merge this PR) - [ ] Confirm the release note - [ ] Confirm test cases - [ ] Confirm document - [ ] Add branch pick label --- .../org/apache/doris/catalog/RefreshManager.java | 14 ++ .../doris/datasource/ExternalMetaCacheMgr.java | 12 + .../doris/datasource/lance/LanceCatalogClient.java | 27 +- .../datasource/lance/LanceExternalCatalog.java | 16 +- .../datasource/lance/LanceNamespaceClient.java | 120 ++++++++- .../metastore/AbstractLanceProperties.java | 18 ++ .../lance/LanceCatalogLifecycleTest.java | 169 +++++++++++++ .../lance/LanceTableAccessCacheTest.java | 279 +++++++++++++++++++++ .../property/metastore/LancePropertiesTest.java | 16 ++ 9 files changed, 654 insertions(+), 17 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/RefreshManager.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/RefreshManager.java index 86b664eaf36..2fa1c974c9b 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/RefreshManager.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/RefreshManager.java @@ -30,6 +30,7 @@ import org.apache.doris.datasource.hive.HMSExternalCatalog; import org.apache.doris.datasource.hive.HMSExternalTable; import org.apache.doris.datasource.hive.HiveExternalMetaCache; import org.apache.doris.datasource.iceberg.IcebergExternalTable; +import org.apache.doris.datasource.lance.LanceExternalCatalog; import org.apache.doris.persist.OperationType; import com.google.common.base.Strings; @@ -90,6 +91,8 @@ public class RefreshManager { throw new DdlException("Only support refresh database in external catalog"); } DatabaseIf db = catalog.getDbOrDdlException(dbName); + // Local DB-object eviction also resets metadata; only an explicit refresh retires access. + invalidateLanceTableAccess(catalog); refreshDbInternal((ExternalDatabase) db); ExternalObjectLog log = ExternalObjectLog.createForRefreshDb(catalog.getId(), db.getFullName()); @@ -100,7 +103,9 @@ public class RefreshManager { ExternalCatalog catalog = (ExternalCatalog) Env.getCurrentEnv().getCatalogMgr().getCatalog(log.getCatalogId()); if (catalog == null) { LOG.warn("failed to find catalog when replaying refresh db: {}", log.debugForRefreshDb()); + return; } + invalidateLanceTableAccess(catalog); Optional<ExternalDatabase<? extends ExternalTable>> db; if (!Strings.isNullOrEmpty(log.getDbName())) { db = catalog.getDbForReplay(log.getDbName()); @@ -115,6 +120,14 @@ public class RefreshManager { } } + private void invalidateLanceTableAccess(CatalogIf catalog) { + // Access entries outlive the bounded DB/table object caches. Replay must invalidate by + // catalog identity before its cache-only object lookup can return early, including ID logs. + if (catalog instanceof LanceExternalCatalog) { + ((LanceExternalCatalog) catalog).invalidateTableAccessCache(); + } + } + private void refreshDbInternal(ExternalDatabase db) { db.resetMetaToUninitialized(); LOG.info("refresh database {} in catalog {}", db.getFullName(), db.getCatalog().getName()); @@ -160,6 +173,7 @@ public class RefreshManager { LOG.warn("failed to find catalog when replaying refresh table: {}", log.debugForRefreshTable()); return; } + invalidateLanceTableAccess(catalog); Optional<ExternalDatabase<? extends ExternalTable>> db; if (!Strings.isNullOrEmpty(log.getDbName())) { db = catalog.getDbForReplay(log.getDbName()); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalMetaCacheMgr.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalMetaCacheMgr.java index bb62d178f96..5edc4a63259 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalMetaCacheMgr.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalMetaCacheMgr.java @@ -24,6 +24,7 @@ import org.apache.doris.datasource.doris.DorisExternalMetaCache; import org.apache.doris.datasource.hive.HiveExternalMetaCache; import org.apache.doris.datasource.hudi.HudiExternalMetaCache; import org.apache.doris.datasource.iceberg.IcebergExternalMetaCache; +import org.apache.doris.datasource.lance.LanceExternalCatalog; import org.apache.doris.datasource.maxcompute.MaxComputeExternalMetaCache; import org.apache.doris.datasource.metacache.AbstractExternalMetaCache; import org.apache.doris.datasource.metacache.ExternalMetaCache; @@ -379,6 +380,7 @@ public class ExternalMetaCacheMgr { } public void invalidateCatalog(long catalogId) { + invalidateLanceTableAccess(catalogId); routeCatalogEngines(catalogId, cache -> safeInvalidate( cache, catalogId, "invalidateCatalog", () -> cache.invalidateCatalogEntries(catalogId))); @@ -489,6 +491,7 @@ public class ExternalMetaCacheMgr { } public void invalidateTable(long catalogId, String dbName, String tableName) { + invalidateLanceTableAccess(catalogId); routeCatalogEngines(catalogId, cache -> safeInvalidate( cache, catalogId, "invalidateTable", () -> cache.invalidateTable(catalogId, dbName, tableName))); @@ -500,6 +503,15 @@ public class ExternalMetaCacheMgr { () -> cache.invalidateTable(catalogId, dbName, tableName))); } + private void invalidateLanceTableAccess(long catalogId) { + CatalogIf<?> catalog = getCatalog(catalogId); + if (catalog instanceof LanceExternalCatalog) { + // Access keys use remote namespace names, whereas refresh can use mapped Doris names. + // Retire this small catalog-wide cache to also fence concurrent loads and name remaps. + ((LanceExternalCatalog) catalog).invalidateTableAccessCache(); + } + } + public void invalidatePartitions(long catalogId, String dbName, String tableName, List<String> partitions) { routeCatalogEngines(catalogId, cache -> safeInvalidate( diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceCatalogClient.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceCatalogClient.java index b89c90c09d9..f7cd271389e 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceCatalogClient.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceCatalogClient.java @@ -36,6 +36,7 @@ import org.apache.doris.datasource.lance.profile.LanceMetadataMetrics.Stage; import org.apache.doris.datasource.property.metastore.AbstractLanceProperties; import org.apache.doris.datasource.property.storage.StorageProperties; +import com.github.benmanes.caffeine.cache.Ticker; import org.apache.arrow.memory.BufferAllocator; import org.apache.arrow.memory.RootAllocator; import org.apache.arrow.vector.types.pojo.Schema; @@ -90,7 +91,8 @@ final class LanceCatalogClient implements AutoCloseable { session = Session.builder().metadataCacheSizeBytes(METADATA_CACHE_SIZE_BYTES) .indexCacheSizeBytes(INDEX_CACHE_SIZE_BYTES).build(); return new LanceCatalogClient(namespace, allocator, session, properties.getLanceCatalogType(), - properties.getRootDatabase(), parent, storageProperties, namespaceOptions, catalogSecrets); + properties.getRootDatabase(), parent, storageProperties, namespaceOptions, catalogSecrets, + properties.getTableAccessCacheTtlSeconds()); } catch (RuntimeException | Error e) { closeResource(namespace); closeResource(session); @@ -103,12 +105,22 @@ final class LanceCatalogClient implements AutoCloseable { String catalogType, String rootDatabase, List<String> parentNamespace, List<StorageProperties> storageProperties, Map<String, String> namespaceStorageOptions, List<String> catalogSecrets) { + this(namespace, allocator, session, catalogType, rootDatabase, parentNamespace, + storageProperties, namespaceStorageOptions, catalogSecrets, + AbstractLanceProperties.DEFAULT_TABLE_ACCESS_CACHE_TTL_SECONDS); + } + + LanceCatalogClient(LanceNamespace namespace, BufferAllocator allocator, Session session, + String catalogType, String rootDatabase, List<String> parentNamespace, + List<StorageProperties> storageProperties, Map<String, String> namespaceStorageOptions, + List<String> catalogSecrets, int tableAccessCacheTtlSeconds) { this.catalogSecrets = Collections.unmodifiableList(new ArrayList<>(catalogSecrets)); this.namespace = namespace; this.namespaceAllocator = allocator; this.session = session; this.namespaceClient = new LanceNamespaceClient( - namespace, catalogType, rootDatabase, parentNamespace, storageProperties); + namespace, catalogType, rootDatabase, parentNamespace, storageProperties, + tableAccessCacheTtlSeconds, Ticker.systemTicker()); this.namespaceStorageOptions = Collections.unmodifiableMap(new HashMap<>(namespaceStorageOptions)); } @@ -189,6 +201,10 @@ final class LanceCatalogClient implements AutoCloseable { } } + void invalidateTableAccessCache() { + namespaceClient.invalidateTableAccessCache(); + } + List<String> listDatabaseNames() { return namespaceClient.listDatabaseNames(); } @@ -234,7 +250,7 @@ final class LanceCatalogClient implements AutoCloseable { (dataset, access, metrics) -> LanceMetadataLoader.read(dataset, access, mode, metrics)); } - /** Pins one resource generation, fresh table access, and the Dataset version for the whole read. */ + /** Pins one resource generation, resolved table access, and the Dataset version for the whole read. */ private <T> T readTableSnapshot(String dbName, String tableName, Optional<TableSnapshot> tableSnapshot, SnapshotReader<T> reader) { LanceTableAccess tableAccess = null; @@ -302,7 +318,7 @@ final class LanceCatalogClient implements AutoCloseable { String resolveCurrentIndexJobLocator(String dbName, String tableName) { return LanceIndexDatasetLocator.normalize( - namespaceClient.resolveTableAccess(dbName, tableName).getDatasetUri()); + namespaceClient.resolveTableAccessUncached(dbName, tableName).getDatasetUri()); } public LanceIndexAdmissionSnapshot loadTableIndexAdmissionSnapshot(String dbName, String tableName) { @@ -313,7 +329,8 @@ final class LanceCatalogClient implements AutoCloseable { LanceTableAccess tableAccess = null; try { // The worker owns its Dataset and Session even if the caller releases its lease on timeout. - tableAccess = namespaceClient.resolveTableAccess(dbName, tableName); + // Index admission must verify the current target even while query access is cached. + tableAccess = namespaceClient.resolveTableAccessUncached(dbName, tableName); LanceTableAccess access = tableAccess; return LanceIndexInspectionExecutor.execute(() -> { // The caller can time out while JNI is running; the worker must own resource cleanup. diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceExternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceExternalCatalog.java index 254610c276e..fd9eeb7011d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceExternalCatalog.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceExternalCatalog.java @@ -277,6 +277,20 @@ public class LanceExternalCatalog extends ExternalCatalog { return client.acquire(); } + @Override + public void unregisterDatabase(String dbName) { + // Dropping a namespace is a semantic change, unlike routine local DB-object eviction. + invalidateTableAccessCache(); + super.unregisterDatabase(dbName); + } + + public synchronized void invalidateTableAccessCache() { + // Invalidation must not initialize JNI resources or wait for an in-flight namespace call. + if (client != null) { + client.invalidateTableAccessCache(); + } + } + @Override public void onRefreshCache(boolean invalidCache) { if (invalidCache) { @@ -287,7 +301,7 @@ public class LanceExternalCatalog extends ExternalCatalog { /** * REFRESH CATALOG invalidates the entire Session, including same-URI dataset replacements. - * REFRESH TABLE only invalidates Doris metadata and does not rotate this native cache. + * REFRESH TABLE invalidates Doris metadata and table access, but does not rotate the native Session. * New reads use a cold cache; in-flight reads retain their previous generation. */ @VisibleForTesting diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceNamespaceClient.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceNamespaceClient.java index dc05be88e6c..79f47210b8b 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceNamespaceClient.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceNamespaceClient.java @@ -23,6 +23,10 @@ import org.apache.doris.datasource.lance.storage.LanceStorageOptions; import org.apache.doris.datasource.property.metastore.AbstractLanceProperties; import org.apache.doris.datasource.property.storage.StorageProperties; +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; +import com.github.benmanes.caffeine.cache.Expiry; +import com.github.benmanes.caffeine.cache.Ticker; import org.apache.commons.lang3.StringUtils; import org.lance.namespace.LanceNamespace; import org.lance.namespace.errors.NamespaceNotFoundException; @@ -35,6 +39,8 @@ import org.lance.namespace.model.ListTablesRequest; import org.lance.namespace.model.ListTablesResponse; import org.lance.namespace.model.TableExistsRequest; +import java.net.URI; +import java.net.URISyntaxException; import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Collections; @@ -44,6 +50,7 @@ import java.util.List; import java.util.Map; import java.util.Queue; import java.util.Set; +import java.util.concurrent.TimeUnit; /** * Namespace requests and table access resolution. The catalog client owns the native resources @@ -60,14 +67,28 @@ final class LanceNamespaceClient { private final List<String> parentNamespace; private final List<StorageProperties> storageProperties; private final Object namespaceLock = new Object(); + private final long tableAccessTtlNanos; + private final Ticker ticker; + private volatile Cache<List<String>, CachedTableAccess> tableAccessCache; LanceNamespaceClient(LanceNamespace namespace, String catalogType, String rootDatabase, List<String> parentNamespace, List<StorageProperties> storageProperties) { + this(namespace, catalogType, rootDatabase, parentNamespace, storageProperties, + AbstractLanceProperties.DEFAULT_TABLE_ACCESS_CACHE_TTL_SECONDS, + Ticker.systemTicker()); + } + + LanceNamespaceClient(LanceNamespace namespace, String catalogType, String rootDatabase, + List<String> parentNamespace, List<StorageProperties> storageProperties, + long tableAccessTtlSeconds, Ticker ticker) { this.namespace = namespace; this.catalogType = catalogType; this.rootDatabase = rootDatabase; this.parentNamespace = Collections.unmodifiableList(new ArrayList<>(parentNamespace)); this.storageProperties = Collections.unmodifiableList(new ArrayList<>(storageProperties)); + this.tableAccessTtlNanos = TimeUnit.SECONDS.toNanos(tableAccessTtlSeconds); + this.ticker = ticker; + this.tableAccessCache = newTableAccessCache(); } List<String> listDatabaseNames() { @@ -178,14 +199,41 @@ final class LanceNamespaceClient { } LanceTableAccess resolveTableAccess(String dbName, String tableName) { - DescribeTableResponse table = describeTable(dbName, tableName); + List<String> tableId = tableAccessKey(dbName, tableName); + if (tableAccessTtlNanos == 0) { + return loadTableAccess(tableId).access; + } + // Cache hits avoid the catalog-wide namespace lock as well as filesystem or REST I/O. + return tableAccessCache.get(tableId, this::loadTableAccess).access; + } + + LanceTableAccess resolveTableAccessUncached(String dbName, String tableName) { + return loadTableAccess(tableAccessKey(dbName, tableName)).access; + } + + void invalidateTableAccessCache() { + // Swap generations: a describe already in flight may finish for its caller, but must + // never repopulate the cache used by reads admitted after an explicit refresh. + tableAccessCache = newTableAccessCache(); + } + + private List<String> tableAccessKey(String dbName, String tableName) { + try { + return Collections.unmodifiableList(buildTableId(dbName, tableName)); + } catch (DdlException e) { + throw new RuntimeException(e); + } + } + + private CachedTableAccess loadTableAccess(List<String> tableId) { + DescribeTableResponse table = describeTable(tableId); if (Boolean.TRUE.equals(table.getManagedVersioning())) { throw new UnsupportedOperationException( "Lance managed versioning is not supported by the current BE reader"); } String datasetUri = StringUtils.firstNonBlank(table.getTableUri(), table.getLocation()); if (datasetUri == null) { - throw new RuntimeException("Lance namespace returned no table URI for " + dbName + "." + tableName); + throw new RuntimeException("Lance namespace returned no table URI for " + tableId); } // One option map serves both readers: the FE opens the dataset through the Lance Java SDK @@ -193,19 +241,69 @@ final class LanceNamespaceClient { // dataset URL picks the option vocabulary, the same way Lance picks a provider from it. Map<String, String> storageOptions = LanceStorageOptions.fromDorisAndVendedStorageOptions(datasetUri, storageProperties, table.getStorageOptions()); - return new LanceTableAccess(datasetUri, storageOptions); + return new CachedTableAccess(new LanceTableAccess(datasetUri, storageOptions), + tableAccessTtlNanos(datasetUri, table.getStorageOptions())); } - private DescribeTableResponse describeTable(String dbName, String tableName) { + private long tableAccessTtlNanos(String datasetUri, Map<String, String> vendedOptions) { + // The BE cannot renew credentials during a scan. A fixed expiry margin cannot cover + // arbitrary query durations, so preserve per-read vending even with a reported deadline. + if (vendedOptions != null && !vendedOptions.isEmpty()) { + return 0; + } try { - List<String> tableId = buildTableId(dbName, tableName); - DescribeTableRequest request = new DescribeTableRequest().id(tableId).withTableUri(true) - .vendCredentials(LANCE_REST.equals(catalogType)); - synchronized (namespaceLock) { - return namespace.describeTable(request); + URI uri = new URI(datasetUri.trim()); + // Presigned/SAS credentials may live in the URI even when storage_options is empty. + // Also check registry-based authorities, for which URI.getRawUserInfo() returns null. + if (uri.isOpaque() || uri.getRawUserInfo() != null || uri.getRawQuery() != null + || uri.getRawFragment() != null + || (uri.getRawAuthority() != null && uri.getRawAuthority().contains("@"))) { + return 0; } - } catch (DdlException e) { - throw new RuntimeException(e); + return tableAccessTtlNanos; + } catch (URISyntaxException e) { + // Unclassified locators remain usable but must not be assumed credential-free. + return 0; + } + } + + private Cache<List<String>, CachedTableAccess> newTableAccessCache() { + return Caffeine.newBuilder().maximumSize(10_000).ticker(ticker) + .expireAfter(new Expiry<List<String>, CachedTableAccess>() { + @Override + public long expireAfterCreate(List<String> key, CachedTableAccess value, long currentTime) { + return value.ttlNanos; + } + + @Override + public long expireAfterUpdate(List<String> key, CachedTableAccess value, + long currentTime, long currentDuration) { + return value.ttlNanos; + } + + @Override + public long expireAfterRead(List<String> key, CachedTableAccess value, + long currentTime, long currentDuration) { + return currentDuration; + } + }).build(); + } + + private static final class CachedTableAccess { + private final LanceTableAccess access; + private final long ttlNanos; + + private CachedTableAccess(LanceTableAccess access, long ttlNanos) { + this.access = access; + this.ttlNanos = ttlNanos; + } + } + + private DescribeTableResponse describeTable(List<String> tableId) { + DescribeTableRequest request = new DescribeTableRequest().id(tableId).withTableUri(true) + .vendCredentials(LANCE_REST.equals(catalogType)); + synchronized (namespaceLock) { + return namespace.describeTable(request); } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/AbstractLanceProperties.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/AbstractLanceProperties.java index 9b6806c34e8..1550554a2d6 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/AbstractLanceProperties.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/AbstractLanceProperties.java @@ -34,9 +34,19 @@ public abstract class AbstractLanceProperties extends MetastoreProperties { public static final String NAMESPACE_DELIMITER = "lance.namespace.delimiter"; public static final String ROOT_DATABASE = "lance.namespace.root_database"; + public static final String TABLE_ACCESS_CACHE_TTL_SECONDS = "lance.table_access_cache_ttl_seconds"; + public static final int DEFAULT_TABLE_ACCESS_CACHE_TTL_SECONDS = 60; + public static final String DEFAULT_DELIMITER = "$"; public static final String DEFAULT_ROOT_DATABASE = "default"; + @ConnectorProperty( + names = {TABLE_ACCESS_CACHE_TTL_SECONDS}, + required = false, + description = "Maximum lifetime of cached table URIs and access options in seconds. " + + "Default: 60. Set to 0 to describe the table on every read.") + private int tableAccessCacheTtlSeconds = DEFAULT_TABLE_ACCESS_CACHE_TTL_SECONDS; + @ConnectorProperty( names = {NAMESPACE_PARENT}, required = false, @@ -101,7 +111,15 @@ public abstract class AbstractLanceProperties extends MetastoreProperties { return rootDatabase; } + public int getTableAccessCacheTtlSeconds() { + return tableAccessCacheTtlSeconds; + } + private void validateCommonProperties() { + if (tableAccessCacheTtlSeconds < 0) { + throw new IllegalArgumentException("Property '" + TABLE_ACCESS_CACHE_TTL_SECONDS + + "' must be non-negative"); + } if (namespaceDelimiter.isEmpty() || namespaceDelimiter.indexOf('\\') >= 0) { throw new IllegalArgumentException("Property '" + NAMESPACE_DELIMITER + "' cannot be empty or contain the escape character '\\'"); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceCatalogLifecycleTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceCatalogLifecycleTest.java index ac30266577c..effd827cc01 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceCatalogLifecycleTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceCatalogLifecycleTest.java @@ -18,7 +18,13 @@ package org.apache.doris.datasource.lance; import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.RefreshManager; +import org.apache.doris.datasource.CatalogMgr; import org.apache.doris.datasource.ExternalCatalog; +import org.apache.doris.datasource.ExternalMetaCacheMgr; +import org.apache.doris.datasource.ExternalObjectLog; +import org.apache.doris.datasource.lance.job.LanceIndexDatasetLocator; +import org.apache.doris.persist.EditLog; import org.apache.arrow.memory.BufferAllocator; import org.junit.jupiter.api.Assertions; @@ -26,12 +32,15 @@ import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.lance.Session; import org.lance.namespace.LanceNamespace; +import org.lance.namespace.model.DescribeTableResponse; +import org.mockito.MockedStatic; import org.mockito.Mockito; import java.lang.reflect.Field; import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.Optional; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -187,6 +196,166 @@ public class LanceCatalogLifecycleTest { } } + @Test + public void testMetadataRefreshInvalidatesAccessWithoutClosingSession() throws Exception { + Session session = Mockito.mock(Session.class); + LanceCatalogClient client = Mockito.spy(client(session)); + LanceExternalCatalog catalog = catalog(client); + Env env = Mockito.mock(Env.class); + CatalogMgr catalogs = Mockito.mock(CatalogMgr.class); + Mockito.when(env.getCatalogMgr()).thenReturn(catalogs); + long catalogId = catalog.getId(); + Mockito.doReturn(catalog).when(catalogs).getCatalog(catalogId); + ExternalMetaCacheMgr caches = new ExternalMetaCacheMgr(true); + try (MockedStatic<Env> currentEnv = Mockito.mockStatic(Env.class)) { + currentEnv.when(Env::getCurrentEnv).thenReturn(env); + caches.invalidateTable(catalog.getId(), "mapped_db", "mapped_table"); + caches.invalidateDb(catalog.getId(), "mapped_db"); + caches.invalidateCatalog(catalog.getId()); + Mockito.verify(client, Mockito.times(2)).invalidateTableAccessCache(); + Mockito.verify(session, Mockito.never()).close(); + Mockito.verify(catalog, Mockito.never()).createClient(); + } finally { + catalog.onClose(); + } + } + + @Test + public void testIndexJobLocatorBypassesQueryAccessCache() throws Exception { + LanceNamespace namespace = Mockito.mock(LanceNamespace.class); + Mockito.when(namespace.describeTable(Mockito.any())).thenReturn( + new DescribeTableResponse().tableUri("file:///warehouse/original.lance"), + new DescribeTableResponse().tableUri("file:///warehouse/replacement.lance")); + try (LanceCatalogClient client = new LanceCatalogClient(namespace, Mockito.mock(BufferAllocator.class), + Mockito.mock(Session.class), "filesystem", "default", Collections.emptyList(), + Collections.emptyList(), Collections.emptyMap(), Collections.emptyList())) { + Field field = LanceCatalogClient.class.getDeclaredField("namespaceClient"); + field.setAccessible(true); + LanceNamespaceClient namespaceClient = (LanceNamespaceClient) field.get(client); + namespaceClient.resolveTableAccess("default", "items"); + Assertions.assertEquals(LanceIndexDatasetLocator.normalize("file:///warehouse/replacement.lance"), + client.resolveCurrentIndexJobLocator("default", "items")); + Mockito.verify(namespace, Mockito.times(2)).describeTable(Mockito.any()); + } + } + + @Test + public void testRoutineDatabaseObjectCleanupPreservesHotAccess() throws Exception { + try (AccessFixture fixture = new AccessFixture()) { + LanceNamespaceClient access = fixture.access(); + access.resolveTableAccess("default", "items"); + new LanceExternalDatabase(fixture.catalog, 1, "cold_db", "cold_db").resetMetaToUninitialized(); + access.resolveTableAccess("default", "items"); + Mockito.verify(fixture.namespace).describeTable(Mockito.any()); + } + } + + @Test + public void testRefreshReplayInvalidatesAccessWithMissingObjects() throws Exception { + for (boolean missingDatabase : new boolean[] {false, true}) { + for (boolean legacyIds : new boolean[] {false, true}) { + try (AccessFixture fixture = new AccessFixture()) { + LanceNamespaceClient access = fixture.access(); + access.resolveTableAccess("default", "items"); + LanceExternalDatabase database = Mockito.mock(LanceExternalDatabase.class); + Mockito.doReturn(missingDatabase ? Optional.empty() : Optional.of(database)) + .when(fixture.catalog).getDbForReplay("mapped_db"); + Mockito.doReturn(missingDatabase ? Optional.empty() : Optional.of(database)) + .when(fixture.catalog).getDbForReplay(1L); + ExternalObjectLog log = ExternalObjectLog.createForRefreshTable( + fixture.catalog.getId(), "mapped_db", "mapped_table", 0); + if (legacyIds) { + log.setDbName(null); + log.setTableName(null); + log.setDbId(1L); + log.setTableId(2L); + } + // Access entries can survive eviction of the smaller database/table object caches. + new RefreshManager().replayRefreshTable(log); + access.resolveTableAccess("default", "items"); + Mockito.verify(fixture.namespace, Mockito.times(2)).describeTable(Mockito.any()); + Mockito.verify(fixture.catalog, Mockito.never()).createClient(); + } + } + } + } + + @Test + public void testDatabaseRefreshReplayWithoutDatabaseObject() throws Exception { + try (AccessFixture fixture = new AccessFixture()) { + LanceNamespaceClient access = fixture.access(); + access.resolveTableAccess("default", "items"); + Mockito.doReturn(Optional.empty()).when(fixture.catalog).getDbForReplay("mapped_db"); + new RefreshManager().replayRefreshDb( + ExternalObjectLog.createForRefreshDb(fixture.catalog.getId(), "mapped_db")); + access.resolveTableAccess("default", "items"); + Mockito.verify(fixture.namespace, Mockito.times(2)).describeTable(Mockito.any()); + } + } + + @Test + public void testExplicitDatabaseRefreshInvalidatesAccess() throws Exception { + try (AccessFixture fixture = new AccessFixture()) { + LanceNamespaceClient access = fixture.access(); + access.resolveTableAccess("default", "items"); + LanceExternalDatabase database = new LanceExternalDatabase(fixture.catalog, 1, "mapped_db", "default"); + Mockito.doReturn(database).when(fixture.catalog).getDbOrDdlException("mapped_db"); + new RefreshManager().handleRefreshDb("lifecycle", "mapped_db"); + access.resolveTableAccess("default", "items"); + Mockito.verify(fixture.namespace, Mockito.times(2)).describeTable(Mockito.any()); + } + } + + @Test + public void testNamespaceRemovalInvalidatesAccessWithoutDatabaseObjects() throws Exception { + try (AccessFixture fixture = new AccessFixture()) { + LanceNamespaceClient access = fixture.access(); + access.resolveTableAccess("default", "items"); + setField(ExternalCatalog.class, fixture.catalog, "initialized", false); + fixture.catalog.unregisterDatabase("mapped_db"); + access.resolveTableAccess("default", "items"); + Mockito.verify(fixture.namespace, Mockito.times(2)).describeTable(Mockito.any()); + Mockito.verify(fixture.catalog, Mockito.never()).createClient(); + } + } + + private static final class AccessFixture implements AutoCloseable { + private final LanceNamespace namespace = Mockito.mock(LanceNamespace.class); + private final LanceCatalogClient client = new LanceCatalogClient(namespace, + Mockito.mock(BufferAllocator.class), Mockito.mock(Session.class), "filesystem", "default", + Collections.emptyList(), Collections.emptyList(), Collections.emptyMap(), Collections.emptyList()); + private final LanceExternalCatalog catalog = catalog(client); + private final MockedStatic<Env> currentEnv; + + private AccessFixture() throws Exception { + Mockito.when(namespace.describeTable(Mockito.any())).thenReturn( + new DescribeTableResponse().tableUri("file:///warehouse/items.lance")); + ExternalMetaCacheMgr caches = Env.getCurrentEnv().getExtMetaCacheMgr(); + Env env = Mockito.mock(Env.class); + CatalogMgr catalogs = Mockito.mock(CatalogMgr.class); + Mockito.when(env.getCatalogMgr()).thenReturn(catalogs); + Mockito.when(env.getExtMetaCacheMgr()).thenReturn(caches); + Mockito.when(env.getEditLog()).thenReturn(Mockito.mock(EditLog.class)); + Mockito.doReturn(catalog).when(catalogs).getCatalog("lifecycle"); + long catalogId = catalog.getId(); + Mockito.doReturn(catalog).when(catalogs).getCatalog(catalogId); + currentEnv = Mockito.mockStatic(Env.class); + currentEnv.when(Env::getCurrentEnv).thenReturn(env); + } + + private LanceNamespaceClient access() throws Exception { + Field field = LanceCatalogClient.class.getDeclaredField("namespaceClient"); + field.setAccessible(true); + return (LanceNamespaceClient) field.get(client); + } + + @Override + public void close() { + currentEnv.close(); + catalog.onClose(); + } + } + private static LanceCatalogClient client(Session session) { return new LanceCatalogClient(Mockito.mock(LanceNamespace.class), Mockito.mock(BufferAllocator.class), session, "filesystem", "default", Collections.emptyList(), Collections.emptyList(), diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceTableAccessCacheTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceTableAccessCacheTest.java new file mode 100644 index 00000000000..b4e3603f927 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceTableAccessCacheTest.java @@ -0,0 +1,279 @@ +// 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.doris.datasource.lance; + +import org.apache.doris.datasource.lance.metadata.LanceTableAccess; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.lance.namespace.LanceNamespace; +import org.lance.namespace.model.DescribeTableResponse; +import org.mockito.Mockito; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +public class LanceTableAccessCacheTest { + @Test + public void testRepeatedFilesystemReadDescribesOnce() { + LanceNamespace namespace = Mockito.mock(LanceNamespace.class); + Mockito.when(namespace.describeTable(Mockito.any())).thenReturn( + new DescribeTableResponse().tableUri("file:///warehouse/items.lance")); + LanceNamespaceClient client = new LanceNamespaceClient(namespace, "filesystem", "default", + Collections.emptyList(), Collections.emptyList()); + Assertions.assertEquals("file:///warehouse/items.lance", + client.resolveTableAccess("default", "items").getDatasetUri()); + client.resolveTableAccess("default", "items"); + Mockito.verify(namespace, Mockito.times(1)).describeTable(Mockito.any()); + } + + @Test + public void testRepeatedRestReadDescribesOnce() { + LanceNamespace namespace = Mockito.mock(LanceNamespace.class); + Mockito.when(namespace.describeTable(Mockito.any())).thenReturn( + new DescribeTableResponse().tableUri("s3://example-bucket/items.lance")); + LanceNamespaceClient client = new LanceNamespaceClient(namespace, "rest", "default", + Collections.emptyList(), Collections.emptyList()); + client.resolveTableAccess("default", "items"); + client.resolveTableAccess("default", "items"); + Mockito.verify(namespace, Mockito.times(1)).describeTable(Mockito.argThat( + request -> Boolean.TRUE.equals(request.getVendCredentials()))); + } + + @Test + public void testExpiryIsNotExtendedByHits() { + LanceNamespace namespace = namespace(); + AtomicLong millis = new AtomicLong(1_000_000); + LanceNamespaceClient client = client(namespace, "filesystem", 60, millis); + client.resolveTableAccess("default", "items"); + millis.addAndGet(59_000); + client.resolveTableAccess("default", "items"); + Mockito.verify(namespace).describeTable(Mockito.any()); + millis.addAndGet(1_000); + client.resolveTableAccess("default", "items"); + Mockito.verify(namespace, Mockito.times(2)).describeTable(Mockito.any()); + } + + @Test + public void testZeroTtlDisablesCaching() { + LanceNamespace namespace = namespace(); + LanceNamespaceClient client = client(namespace, "rest", 0, new AtomicLong(1_000_000)); + client.resolveTableAccess("default", "items"); + client.resolveTableAccess("default", "items"); + Mockito.verify(namespace, Mockito.times(2)).describeTable(Mockito.any()); + } + + @Test + public void testVendedCredentialsAreResolvedForEveryRead() { + LanceNamespace namespace = namespace(); + AtomicLong millis = new AtomicLong(1_000_000); + Map<String, String> options = new HashMap<>(); + options.put("aws_session_token", "example-token"); + options.put("expires_at_millis", "1040000"); + Mockito.when(namespace.describeTable(Mockito.any())).thenReturn( + new DescribeTableResponse().tableUri("s3://example-bucket/items.lance").storageOptions(options)); + LanceNamespaceClient client = client(namespace, "rest", 60, millis); + LanceTableAccess first = client.resolveTableAccess("default", "items"); + Assertions.assertThrows(UnsupportedOperationException.class, + () -> first.getStorageOptions().put("aws_session_token", "modified")); + millis.addAndGet(9_000); + // A new scan may outlive the remaining credential lifetime, regardless of the cache TTL. + Assertions.assertNotSame(first, client.resolveTableAccess("default", "items")); + Mockito.verify(namespace, Mockito.times(2)).describeTable(Mockito.any()); + } + + @Test + public void testSignedUrisWithoutStorageOptionsAreNotCached() { + for (String uri : new String[] { + "s3://example-bucket/items.lance?X-Amz-Signature=example", + "az://container/items.lance?sig=example&se=example", + "s3://example:password@example-bucket/items.lance", + "s3://example@example_bucket/items.lance", + "s3://example-bucket/items.lance#example"}) { + LanceNamespace namespace = namespace(); + Mockito.when(namespace.describeTable(Mockito.any())).thenReturn( + new DescribeTableResponse().tableUri(uri)); + LanceNamespaceClient client = client(namespace, "rest", 60, new AtomicLong(1_000_000)); + client.resolveTableAccess("default", "items"); + client.resolveTableAccess("default", "items"); + Mockito.verify(namespace, Mockito.times(2)).describeTable(Mockito.any()); + } + } + + @Test + public void testUnsafeVendedExpiryIsNotCached() { + for (String expiry : new String[] {null, "invalid", "-9223372036854775808", "999999", "1029999"}) { + LanceNamespace namespace = namespace(); + Map<String, String> options = new HashMap<>(); + options.put("aws_session_token", "example-token"); + if (expiry != null) { + options.put("expires_at_millis", expiry); + } + Mockito.when(namespace.describeTable(Mockito.any())).thenReturn( + new DescribeTableResponse().tableUri("s3://example-bucket/items.lance").storageOptions(options)); + LanceNamespaceClient client = client(namespace, "rest", 60, new AtomicLong(1_000_000)); + client.resolveTableAccess("default", "items"); + client.resolveTableAccess("default", "items"); + Mockito.verify(namespace, Mockito.times(2)).describeTable(Mockito.any()); + } + } + + @Test + public void testCacheKeyIncludesNamespaceAndTable() { + LanceNamespace namespace = namespace(); + LanceNamespaceClient client = new LanceNamespaceClient(namespace, "filesystem", "default", + Collections.singletonList("parent"), Collections.emptyList()); + client.resolveTableAccess("default", "items"); + client.resolveTableAccess("sales", "items"); + client.resolveTableAccess("default", "other"); + client.resolveTableAccess("sales", "items"); + Mockito.verify(namespace).describeTable(Mockito.argThat( + request -> request.getId().equals(Arrays.asList("parent", "items")))); + Mockito.verify(namespace).describeTable(Mockito.argThat( + request -> request.getId().equals(Arrays.asList("parent", "sales", "items")))); + Mockito.verify(namespace).describeTable(Mockito.argThat( + request -> request.getId().equals(Arrays.asList("parent", "other")))); + } + + @Test + public void testFailedResolutionIsRetried() { + LanceNamespace namespace = namespace(); + Mockito.when(namespace.describeTable(Mockito.any())).thenThrow(new IllegalStateException("unavailable")) + .thenReturn(new DescribeTableResponse().tableUri("file:///warehouse/items.lance")); + LanceNamespaceClient client = client(namespace, "filesystem", 60, new AtomicLong(1_000_000)); + Assertions.assertThrows(IllegalStateException.class, () -> client.resolveTableAccess("default", "items")); + client.resolveTableAccess("default", "items"); + client.resolveTableAccess("default", "items"); + Mockito.verify(namespace, Mockito.times(2)).describeTable(Mockito.any()); + } + + @Test + public void testRefreshAndUncachedResolutionSeeChangedTarget() { + LanceNamespace namespace = namespace(); + LanceNamespaceClient client = client(namespace, "filesystem", 60, new AtomicLong(1_000_000)); + String original = client.resolveTableAccess("default", "items").getDatasetUri(); + Mockito.when(namespace.describeTable(Mockito.any())).thenReturn( + new DescribeTableResponse().tableUri("file:///warehouse/replacement.lance")); + Assertions.assertEquals(original, client.resolveTableAccess("default", "items").getDatasetUri()); + Assertions.assertEquals("file:///warehouse/replacement.lance", + client.resolveTableAccessUncached("default", "items").getDatasetUri()); + client.invalidateTableAccessCache(); + Assertions.assertEquals("file:///warehouse/replacement.lance", + client.resolveTableAccess("default", "items").getDatasetUri()); + } + + @Test + public void testConcurrentMissesShareOneDescribe() throws Exception { + LanceNamespace namespace = namespace(); + LanceNamespaceClient client = client(namespace, "filesystem", 60, new AtomicLong(1_000_000)); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + Mockito.when(namespace.describeTable(Mockito.any())).thenAnswer(invocation -> { + entered.countDown(); + Assertions.assertTrue(release.await(10, TimeUnit.SECONDS)); + return new DescribeTableResponse().tableUri("file:///warehouse/items.lance"); + }); + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future<LanceTableAccess> first = executor.submit(() -> client.resolveTableAccess("default", "items")); + Assertions.assertTrue(entered.await(10, TimeUnit.SECONDS)); + Future<LanceTableAccess> second = executor.submit(() -> client.resolveTableAccess("default", "items")); + release.countDown(); + Assertions.assertSame(first.get(10, TimeUnit.SECONDS), second.get(10, TimeUnit.SECONDS)); + Mockito.verify(namespace).describeTable(Mockito.any()); + } finally { + release.countDown(); + executor.shutdownNow(); + Assertions.assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + } + } + + @Test + public void testRefreshDoesNotWaitForOrRetainAnInFlightDescribe() throws Exception { + LanceNamespace namespace = namespace(); + LanceNamespaceClient client = client(namespace, "filesystem", 60, new AtomicLong(1_000_000)); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + Mockito.when(namespace.describeTable(Mockito.any())).thenAnswer(invocation -> { + entered.countDown(); + Assertions.assertTrue(release.await(10, TimeUnit.SECONDS)); + return new DescribeTableResponse().tableUri("file:///warehouse/original.lance"); + }).thenReturn(new DescribeTableResponse().tableUri("file:///warehouse/replacement.lance")); + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future<LanceTableAccess> first = executor.submit(() -> client.resolveTableAccess("default", "items")); + Assertions.assertTrue(entered.await(10, TimeUnit.SECONDS)); + executor.submit(client::invalidateTableAccessCache).get(10, TimeUnit.SECONDS); + release.countDown(); + Assertions.assertEquals("file:///warehouse/original.lance", first.get(10, TimeUnit.SECONDS).getDatasetUri()); + Assertions.assertEquals("file:///warehouse/replacement.lance", + client.resolveTableAccess("default", "items").getDatasetUri()); + } finally { + release.countDown(); + executor.shutdownNow(); + Assertions.assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + } + } + + @Test + public void testCacheHitDoesNotWaitForAnotherNamespaceRequest() throws Exception { + LanceNamespace namespace = namespace(); + LanceNamespaceClient client = client(namespace, "rest", 60, new AtomicLong(1_000_000)); + LanceTableAccess cached = client.resolveTableAccess("default", "items"); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + Mockito.doAnswer(invocation -> { + entered.countDown(); + Assertions.assertTrue(release.await(10, TimeUnit.SECONDS)); + return null; + }).when(namespace).tableExists(Mockito.any()); + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future<Boolean> exists = executor.submit(() -> client.tableExists("default", "other")); + Assertions.assertTrue(entered.await(10, TimeUnit.SECONDS)); + Assertions.assertSame(cached, + executor.submit(() -> client.resolveTableAccess("default", "items")).get(10, TimeUnit.SECONDS)); + release.countDown(); + Assertions.assertTrue(exists.get(10, TimeUnit.SECONDS)); + } finally { + release.countDown(); + executor.shutdownNow(); + Assertions.assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + } + } + + private static LanceNamespace namespace() { + LanceNamespace namespace = Mockito.mock(LanceNamespace.class); + Mockito.when(namespace.describeTable(Mockito.any())).thenReturn( + new DescribeTableResponse().tableUri("file:///warehouse/items.lance")); + return namespace; + } + + private static LanceNamespaceClient client(LanceNamespace namespace, String type, int ttl, AtomicLong millis) { + return new LanceNamespaceClient(namespace, type, "default", Collections.emptyList(), Collections.emptyList(), + ttl, () -> TimeUnit.MILLISECONDS.toNanos(millis.get())); + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/property/metastore/LancePropertiesTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/property/metastore/LancePropertiesTest.java index b079060ead2..2cae23fa5e5 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/property/metastore/LancePropertiesTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/property/metastore/LancePropertiesTest.java @@ -24,6 +24,22 @@ import java.util.HashMap; import java.util.Map; public class LancePropertiesTest { + @Test + public void testTableAccessCacheTtl() throws Exception { + Map<String, String> properties = new HashMap<>(); + properties.put("type", "lance"); + properties.put(LanceFileSystemMetastoreProperties.WAREHOUSE, "/tmp/lance"); + Assertions.assertEquals(60, ((AbstractLanceProperties) MetastoreProperties.create(properties)) + .getTableAccessCacheTtlSeconds()); + for (String ttl : new String[] {"0", "120"}) { + properties.put(AbstractLanceProperties.TABLE_ACCESS_CACHE_TTL_SECONDS, ttl); + Assertions.assertEquals(Integer.parseInt(ttl), + ((AbstractLanceProperties) MetastoreProperties.create(properties)).getTableAccessCacheTtlSeconds()); + } + properties.put(AbstractLanceProperties.TABLE_ACCESS_CACHE_TTL_SECONDS, "-1"); + Assertions.assertThrows(IllegalArgumentException.class, () -> MetastoreProperties.create(properties)); + } + @Test public void testDefaultFilesystemProperties() throws Exception { Map<String, String> properties = new HashMap<>(); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
