This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new fb60467142a [opt](lance) Cache table access resolution in FE (#68305)
fb60467142a is described below
commit fb60467142a120bba119212d28552acfc8e8f43d
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]
