github-actions[bot] commented on code in PR #68078: URL: https://github.com/apache/doris/pull/68078#discussion_r4038563044
########## fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceCatalogClient.java: ########## @@ -0,0 +1,331 @@ +// 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.analysis.TableSnapshot; +import org.apache.doris.common.DdlException; +import org.apache.doris.common.util.TimeUtils; +import org.apache.doris.datasource.lance.index.LanceIndexInspection; +import org.apache.doris.datasource.lance.index.LanceIndexInspectionExecutor; +import org.apache.doris.datasource.lance.index.LancePhysicalIndexEntry; +import org.apache.doris.datasource.lance.index.LanceShowIndexInfo; +import org.apache.doris.datasource.lance.job.LanceIndexDatasetLocator; +import org.apache.doris.datasource.lance.metadata.LanceMetadataLoader; +import org.apache.doris.datasource.lance.metadata.LanceReadOptions; +import org.apache.doris.datasource.lance.metadata.LanceSnapshotResolver; +import org.apache.doris.datasource.lance.metadata.LanceTableAccess; +import org.apache.doris.datasource.lance.metadata.LanceTableMetadata; +import org.apache.doris.datasource.lance.profile.LanceMetadataMetrics.Stage; +import org.apache.doris.datasource.lance.profile.LanceMetadataMetrics; +import org.apache.doris.datasource.property.metastore.AbstractLanceProperties; +import org.apache.doris.datasource.property.storage.StorageProperties; + +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; +import org.apache.arrow.vector.types.pojo.Schema; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.lance.Dataset; +import org.lance.Session; +import org.lance.namespace.LanceNamespace; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.OptionalLong; +import java.util.function.BiFunction; + +/** One catalog generation: Namespace access, snapshot reads and native resource lifetime. */ +final class LanceCatalogClient implements AutoCloseable { + + private static final Logger LOG = LogManager.getLogger(LanceCatalogClient.class); + static final long ALLOCATOR_LIMIT = 256L * 1024 * 1024; + private static final long METADATA_CACHE_SIZE_BYTES = 64L * 1024 * 1024; + private static final long INDEX_CACHE_SIZE_BYTES = 128L * 1024 * 1024; + + private final LanceNamespace namespace; + private final Session session; + private final LanceNamespaceClient namespaceClient; + private final Map<String, String> namespaceStorageOptions; + private final BufferAllocator namespaceAllocator; + private final List<String> catalogSecrets; + private int activeOperations; + private boolean retired; + + static LanceCatalogClient create(AbstractLanceProperties properties, + List<StorageProperties> storageProperties, Map<String, String> namespaceOptions, + List<String> catalogSecrets) throws DdlException { + List<String> parent = LanceNamespaceName.parseParentNamespace( + properties.getNamespaceParent(), properties.getNamespaceDelimiter()); + BufferAllocator allocator = new RootAllocator(ALLOCATOR_LIMIT); + LanceNamespace namespace = null; + Session session = null; + try { + namespace = properties.createNamespace(allocator, namespaceOptions); + session = Session.builder().metadataCacheSizeBytes(METADATA_CACHE_SIZE_BYTES) Review Comment: [P1] Rotate or partition the Session across table incarnations All ordinary reads now reuse this catalog-lifetime Session, but `REFRESH TABLE` invalidates only Doris caches and does not replace it. The pinned Lance 11.0.0 has a known index-metadata cache collision when a dataset is dropped and recreated at the same URI with a restarted version ([upstream #8904](https://github.com/lance-format/lance/pull/8904), merged after v11.0.0). The FE can then serialize the previous incarnation's UUID/coverage, and the BE's fresh snapshot rejects that UUID during scan materialization, so queries keep failing until the Session rotates. Please evict/partition this cache or rotate the appropriate Session on table-incarnation refresh, and add a same-URI recreate regression. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceCatalogClient.java: ########## @@ -0,0 +1,331 @@ +// 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.analysis.TableSnapshot; +import org.apache.doris.common.DdlException; +import org.apache.doris.common.util.TimeUtils; +import org.apache.doris.datasource.lance.index.LanceIndexInspection; +import org.apache.doris.datasource.lance.index.LanceIndexInspectionExecutor; +import org.apache.doris.datasource.lance.index.LancePhysicalIndexEntry; +import org.apache.doris.datasource.lance.index.LanceShowIndexInfo; +import org.apache.doris.datasource.lance.job.LanceIndexDatasetLocator; +import org.apache.doris.datasource.lance.metadata.LanceMetadataLoader; +import org.apache.doris.datasource.lance.metadata.LanceReadOptions; +import org.apache.doris.datasource.lance.metadata.LanceSnapshotResolver; +import org.apache.doris.datasource.lance.metadata.LanceTableAccess; +import org.apache.doris.datasource.lance.metadata.LanceTableMetadata; +import org.apache.doris.datasource.lance.profile.LanceMetadataMetrics.Stage; +import org.apache.doris.datasource.lance.profile.LanceMetadataMetrics; +import org.apache.doris.datasource.property.metastore.AbstractLanceProperties; +import org.apache.doris.datasource.property.storage.StorageProperties; + +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; +import org.apache.arrow.vector.types.pojo.Schema; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.lance.Dataset; +import org.lance.Session; +import org.lance.namespace.LanceNamespace; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.OptionalLong; +import java.util.function.BiFunction; + +/** One catalog generation: Namespace access, snapshot reads and native resource lifetime. */ +final class LanceCatalogClient implements AutoCloseable { + + private static final Logger LOG = LogManager.getLogger(LanceCatalogClient.class); + static final long ALLOCATOR_LIMIT = 256L * 1024 * 1024; + private static final long METADATA_CACHE_SIZE_BYTES = 64L * 1024 * 1024; + private static final long INDEX_CACHE_SIZE_BYTES = 128L * 1024 * 1024; + + private final LanceNamespace namespace; + private final Session session; + private final LanceNamespaceClient namespaceClient; + private final Map<String, String> namespaceStorageOptions; + private final BufferAllocator namespaceAllocator; + private final List<String> catalogSecrets; + private int activeOperations; + private boolean retired; + + static LanceCatalogClient create(AbstractLanceProperties properties, + List<StorageProperties> storageProperties, Map<String, String> namespaceOptions, + List<String> catalogSecrets) throws DdlException { + List<String> parent = LanceNamespaceName.parseParentNamespace( + properties.getNamespaceParent(), properties.getNamespaceDelimiter()); + BufferAllocator allocator = new RootAllocator(ALLOCATOR_LIMIT); + LanceNamespace namespace = null; + Session session = null; + try { + namespace = properties.createNamespace(allocator, namespaceOptions); + 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); + } catch (RuntimeException | Error e) { + closeResource(namespace); + closeResource(session); + closeResource(allocator); + throw e; + } + } + + LanceCatalogClient(LanceNamespace namespace, BufferAllocator allocator, Session session, + String catalogType, String rootDatabase, List<String> parentNamespace, + List<StorageProperties> storageProperties, Map<String, String> namespaceStorageOptions, + List<String> catalogSecrets) { + 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); + this.namespaceStorageOptions = Collections.unmodifiableMap(new HashMap<>(namespaceStorageOptions)); + } + + /** Pins this generation for one operation; the lock does not cover its SDK or JNI calls. */ + private synchronized Lease acquire() { + if (retired) { + throw new IllegalStateException("Lance catalog resources have been closed"); + } + activeOperations++; + return new Lease(this); + } + + /** Retires this generation immediately; its last active operation performs resource cleanup. */ + @Override + public void close() { + boolean release; + synchronized (this) { + if (retired) { + return; + } + retired = true; + release = activeOperations == 0; + } + if (release) { + closeResources(); + } + } + + private void release() { + boolean release; + synchronized (this) { + activeOperations--; + release = retired && activeOperations == 0; + } + if (release) { + closeResources(); + } + } + + private void closeResources() { + closeResource(namespace); + closeResource(session); + closeResource(namespaceAllocator); + } + + private static void closeResource(Object resource) { + if (resource instanceof AutoCloseable) { + try { + ((AutoCloseable) resource).close(); + } catch (Exception e) { + // Provider exception messages may contain credentials. + LOG.warn("Failed to close a Lance catalog resource ({})", resource.getClass().getSimpleName()); + } + } + } + + private static final class Lease implements AutoCloseable { + private final LanceCatalogClient client; + private boolean closed; + + private Lease(LanceCatalogClient client) { + this.client = client; + } + + @Override + public synchronized void close() { + if (!closed) { + closed = true; + client.release(); + } + } + } + + List<String> listDatabaseNames() { + try (Lease operation = acquire()) { + return namespaceClient.listDatabaseNames(); + } + } + + List<String> listTableNames(String dbName) { + try (Lease operation = acquire()) { + return namespaceClient.listTableNames(dbName); + } + } + + boolean tableExists(String dbName, String tableName) { + try (Lease operation = acquire()) { + return namespaceClient.tableExists(dbName, tableName); + } + } + + public LanceTableMetadata loadTableMetadata(String dbName, String tableName) { + return loadTableMetadata(dbName, tableName, Optional.empty()); + } + + public LanceTableMetadata loadTableMetadataForSearch(String dbName, String tableName) { + LanceTableMetadata metadata = loadQueryMetadata(dbName, tableName, Optional.empty(), + LanceMetadataLoader.MetadataScope.WITH_INDEXES); + if (!metadata.getIndexMetadataState().canPlanIndexSegments()) { + throw new IllegalArgumentException("Lance SDK cannot provide field IDs required for search planning"); + } + return metadata; + } + + public LanceTableMetadata loadBasicTableMetadata(String dbName, String tableName) { + return loadQueryMetadata(dbName, tableName, Optional.empty(), LanceMetadataLoader.MetadataScope.BASIC); + } + + public Schema loadTableSchema(String dbName, String tableName) { + return readTableSnapshot(dbName, tableName, Optional.empty(), + (dataset, access, metrics) -> metrics.measure(Stage.SCHEMA, dataset::getSchema)); + } + + public LanceTableMetadata loadTableMetadata(String dbName, String tableName, + Optional<TableSnapshot> tableSnapshot) { + return loadQueryMetadata(dbName, tableName, tableSnapshot, LanceMetadataLoader.MetadataScope.WITH_INDEXES); + } + + private LanceTableMetadata loadQueryMetadata(String dbName, String tableName, + Optional<TableSnapshot> tableSnapshot, LanceMetadataLoader.MetadataScope mode) { + return readTableSnapshot(dbName, tableName, tableSnapshot, + (dataset, access, metrics) -> LanceMetadataLoader.read(dataset, access, mode, metrics)); + } + + /** Pins one resource generation, fresh 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; + LanceMetadataMetrics metrics = LanceMetadataMetrics.startMetadataRead(); + try { + T result; + try (Lease operation = acquire(); + BufferAllocator allocator = new RootAllocator(LanceMetadataLoader.READ_ALLOCATOR_LIMIT)) { Review Comment: [P1] Keep one aggregate Arrow allocation budget Before this change, ordinary schema/metadata/time-travel reads shared the catalog's 256 MiB root. This creates a fresh root with the full 256 MiB limit for every operation, and unlike index inspection, ordinary planning is not bounded by an executor or semaphore. N concurrent plans can therefore allocate up to N times the former catalog-wide limit, plus Session/namespace memory, defeating the cap and potentially OOMing the FE. Please use child allocators beneath a catalog-wide parent that spans live and draining generations, or enforce an equivalent catalog-wide memory/admission bound. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceExternalCatalog.java: ########## @@ -220,374 +193,108 @@ public boolean isRestCatalogConfigured() { @Override protected List<String> listDatabaseNames() { - makeSureInitialized(); - - // The configured root database represents the empty relative Lance namespace. - LinkedHashSet<String> databases = new LinkedHashSet<>(); - databases.add(rootDatabase); - - // Breadth-first traversal starts at the catalog's configured parent namespace. - // Queue entries remain relative so they can be exposed as Doris database names. - Queue<List<String>> queue = new ArrayDeque<>(); - queue.add(Collections.emptyList()); - Set<List<String>> visited = new HashSet<>(); - while (!queue.isEmpty()) { - List<String> relativeParent = queue.remove(); - if (!visited.add(relativeParent)) { - continue; - } - - // The Lance API expects a full namespace, including the configured parent. - List<String> fullParentNamespace = buildFullNamespace(relativeParent); - for (String child : listChildNamespaces(fullParentNamespace)) { - List<String> relativeChild = new ArrayList<>(relativeParent); - relativeChild.add(child); - - // Doris exposes each hierarchical relative namespace as one flat database name. - databases.add(LanceNamespaceName.namespaceToDorisDatabaseName( - relativeChild, DATABASE_NAMESPACE_DELIMITER, rootDatabase)); - - // Visit this child later to discover namespaces nested below it. - queue.add(relativeChild); - } - } - return new ArrayList<>(databases); - } - - /** - * Lists all direct child namespace names under the given full Lance namespace. - * - * <p>Each request asks for at most {@link #PAGE_SIZE} children. If Lance returns a - * page token, this method keeps requesting subsequent pages until all children are collected. - */ - private List<String> listChildNamespaces(List<String> namespaceId) { - List<String> result = new ArrayList<>(); - String pageToken = null; - Set<String> consumedTokens = new HashSet<>(); - do { - ListNamespacesRequest request = new ListNamespacesRequest().id(namespaceId).limit(PAGE_SIZE); - if (pageToken != null) { - request.pageToken(pageToken); - } - ListNamespacesResponse response; - synchronized (namespaceLock) { - response = namespace.listNamespaces(request); - } - if (response.getNamespaces() != null) { - result.addAll(response.getNamespaces()); - } - pageToken = response.getPageToken(); - } while (StringUtils.isNotEmpty(pageToken) && consumedTokens.add(pageToken)); - return result; + return getClient().listDatabaseNames(); } @Override - protected List<String> listTableNamesFromRemote(SessionContext ctx, String dbName) { - makeSureInitialized(); - try { - List<String> relativeNamespace = LanceNamespaceName.dorisDatabaseNameToNamespace( - dbName, DATABASE_NAMESPACE_DELIMITER, rootDatabase); - List<String> namespaceId = buildFullNamespace(relativeNamespace); - List<String> result = new ArrayList<>(); - String pageToken = null; - Set<String> consumedTokens = new HashSet<>(); - do { - ListTablesRequest request = new ListTablesRequest().id(namespaceId).limit(PAGE_SIZE); - if (pageToken != null) { - request.pageToken(pageToken); - } - ListTablesResponse response; - synchronized (namespaceLock) { - response = namespace.listTables(request); - } - if (response.getTables() != null) { - result.addAll(response.getTables()); - } - pageToken = response.getPageToken(); - } while (StringUtils.isNotEmpty(pageToken) && consumedTokens.add(pageToken)); - return result; - } catch (DdlException e) { - throw new RuntimeException(e); - } + protected List<String> listTableNamesFromRemote(SessionContext context, String dbName) { + return getClient().listTableNames(dbName); } @Override - public boolean tableExist(SessionContext ctx, String dbName, String tblName) { - makeSureInitialized(); - try { - List<String> relativeNamespace = LanceNamespaceName.dorisDatabaseNameToNamespace( - dbName, DATABASE_NAMESPACE_DELIMITER, rootDatabase); - List<String> tableId = buildFullNamespace(relativeNamespace); - tableId.add(tblName); - TableExistsRequest request = new TableExistsRequest().id(tableId); - synchronized (namespaceLock) { - namespace.tableExists(request); - } - return true; - } catch (TableNotFoundException | NamespaceNotFoundException e) { - return false; - } catch (DdlException e) { - throw new RuntimeException(e); - } + public boolean tableExist(SessionContext context, String dbName, String tableName) { + return getClient().tableExists(dbName, tableName); } public LanceTableMetadata loadTableMetadata(String dbName, String tableName) { return loadTableMetadata(dbName, tableName, Optional.empty()); } + public LanceTableMetadata loadTableMetadata(String dbName, String tableName, Optional<TableSnapshot> snapshot) { + return getClient().loadTableMetadata(dbName, tableName, snapshot); + } + public LanceTableMetadata loadTableMetadataForSearch(String dbName, String tableName) { - makeSureInitialized(); - ResolvedTableAccess tableAccess = resolveTableAccess(dbName, tableName); - try { - return LanceMetadataLoader.loadLatestForSearch( - tableAccess.datasetUri, tableAccess.storageOptions, allocator); - } catch (Exception e) { - throw new RuntimeException("Failed to load Lance table metadata for " + dbName + "." + tableName - + ": " + sanitizedRootCauseMessage(e), safeCause(e)); - } + return getClient().loadTableMetadataForSearch(dbName, tableName); } - public LanceTableMetadata loadTableMetadata(String dbName, String tableName, - Optional<TableSnapshot> tableSnapshot) { - makeSureInitialized(); - ResolvedTableAccess tableAccess = resolveTableAccess(dbName, tableName); - try { - if (tableSnapshot.isPresent()) { - TableSnapshot snapshot = tableSnapshot.get(); - long version; - if (snapshot.getType() == TableSnapshot.VersionType.VERSION) { - version = LanceSnapshotResolver.parseVersion(snapshot.getValue()); - } else { - long timestamp = TimeUtils.timeStringToLong(snapshot.getValue(), TimeUtils.getTimeZone()); - if (timestamp < 0) { - throw new IllegalArgumentException( - "Cannot parse Lance FOR TIME AS OF value '" + snapshot.getValue() + "'"); - } - version = LanceSnapshotResolver.getVersionAtOrBefore( - tableAccess.datasetUri, tableAccess.storageOptions, timestamp, allocator); - } - return LanceMetadataLoader.loadVersion( - tableAccess.datasetUri, tableAccess.storageOptions, version, allocator); - } - return LanceMetadataLoader.loadLatest(tableAccess.datasetUri, tableAccess.storageOptions, allocator); - } catch (Exception e) { - throw new RuntimeException("Failed to load Lance table metadata for " + dbName + "." + tableName - + ": " + sanitizedRootCauseMessage(e), safeCause(e)); - } + public LanceTableMetadata loadBasicTableMetadata(String dbName, String tableName) { + return getClient().loadBasicTableMetadata(dbName, tableName); } - public List<LanceLogicalIndex> loadTableIndexMetadata( - String dbName, String tableName) throws AnalysisException { + public Schema loadTableSchema(String dbName, String tableName) { + return getClient().loadTableSchema(dbName, tableName); + } + + public List<LanceShowIndexInfo> loadTableIndexesForShow(String dbName, String tableName) throws AnalysisException { if (isRestCatalogConfigured()) { throw new AnalysisException("SHOW INDEX is not supported for Lance REST catalogs"); } - try { - makeSureInitialized(); - } catch (Exception e) { - throw indexMetadataLoadFailure( - dbName, tableName, e, null, namespaceStorageOptions); - } - - ResolvedTableAccess tableAccess = null; - try { - // Keep Directory namespace resolution on the caller while it owns the catalog's - // shared namespace and allocator. Moving that shared owner into a timed task would - // let catalog close release it after the caller returns but before the task ends. - // The deadline below covers the Dataset/JNI index metadata read itself. - tableAccess = resolveTableAccess(dbName, tableName); - String datasetUri = tableAccess.datasetUri; - Map<String, String> storageOptions = tableAccess.storageOptions; - return LanceMetadataReadExecutor.execute(() -> { - // The caller may return on deadline while JNI is still running. A task-owned - // allocator prevents catalog close from releasing native resources prematurely. - try (BufferAllocator readAllocator = new RootAllocator(ALLOCATOR_LIMIT)) { - return LanceIndexMetadataLoader.load(datasetUri, storageOptions, readAllocator); - } - }); - } catch (Exception e) { - String datasetUri = tableAccess == null ? null : tableAccess.datasetUri; - Map<String, String> runtimeStorageOptions = tableAccess == null - ? namespaceStorageOptions : tableAccess.storageOptions; - throw indexMetadataLoadFailure( - dbName, tableName, e, datasetUri, runtimeStorageOptions); - } + return getClient().loadTableIndexesForShow(dbName, tableName); } - public List<LancePhysicalIndexEntry> loadTableIndexEntries( - String dbName, String tableName) throws AnalysisException { + public List<LancePhysicalIndexEntry> loadTableIndexEntries(String dbName, String tableName) + throws AnalysisException { if (isRestCatalogConfigured()) { - throw new AnalysisException( - "Lance index inspection is not supported for Lance REST catalogs"); - } - try { - makeSureInitialized(); - } catch (Exception e) { - throw indexMetadataLoadFailure(dbName, tableName, e, null, namespaceStorageOptions); - } - - ResolvedTableAccess tableAccess = null; - try { - // Keep Directory namespace resolution on the caller while it owns the catalog's - // shared namespace and allocator. Moving that shared owner into a timed task would - // let catalog close release it after the caller returns but before the task ends. - // The deadline below covers the Dataset/JNI index metadata read itself. - tableAccess = resolveTableAccess(dbName, tableName); - String datasetUri = tableAccess.datasetUri; - Map<String, String> storageOptions = tableAccess.storageOptions; - return LanceMetadataReadExecutor.execute(() -> { - // The caller may return on deadline while JNI is still running. A task-owned - // allocator prevents catalog close from releasing native resources prematurely. - try (BufferAllocator readAllocator = new RootAllocator(ALLOCATOR_LIMIT)) { - return LanceIndexMetadataLoader.loadPhysicalEntries( - datasetUri, storageOptions, readAllocator); - } - }); - } catch (Exception e) { - String datasetUri = tableAccess == null ? null : tableAccess.datasetUri; - Map<String, String> runtimeStorageOptions = tableAccess == null - ? namespaceStorageOptions : tableAccess.storageOptions; - throw indexMetadataLoadFailure( - dbName, tableName, e, datasetUri, runtimeStorageOptions); + throw new AnalysisException("Lance index inspection is not supported for Lance REST catalogs"); } + return getClient().loadTableIndexEntries(dbName, tableName); } - /** - * Loads the pinned latest-snapshot admission view (version, schema fields, logical and - * physical indexes) for a Directory table. REST catalogs are rejected before any - * resolution, exactly like {@link #loadTableIndexMetadata}. - */ + /** Loads the pinned admission snapshot; REST catalogs remain unsupported. */ public LanceIndexAdmissionSnapshot loadTableIndexAdmissionSnapshot( String dbName, String tableName) throws Exception { if (isRestCatalogConfigured()) { - throw new AnalysisException( - "Lance index admission is not supported for Lance REST catalogs"); - } - try { - makeSureInitialized(); - } catch (Exception e) { - throw indexAdmissionSnapshotLoadFailure( - dbName, tableName, e, null, namespaceStorageOptions); + throw new AnalysisException("Lance index admission is not supported for Lance REST catalogs"); } - - ResolvedTableAccess tableAccess = null; try { - // Same ownership split as loadTableIndexMetadata: the caller resolves the table - // through the catalog's shared namespace, while the deadline-bound task owns the - // allocator backing its Dataset/JNI read. - tableAccess = resolveTableAccess(dbName, tableName); - String datasetUri = tableAccess.datasetUri; - Map<String, String> storageOptions = tableAccess.storageOptions; - return LanceMetadataReadExecutor.execute(() -> { - try (BufferAllocator readAllocator = new RootAllocator(ALLOCATOR_LIMIT)) { - return LanceIndexMetadataLoader.loadAdmissionSnapshot( - datasetUri, storageOptions, readAllocator); - } - }); + return getClient().loadTableIndexAdmissionSnapshot(dbName, tableName); } catch (Exception e) { - String datasetUri = tableAccess == null ? null : tableAccess.datasetUri; - Map<String, String> runtimeStorageOptions = tableAccess == null - ? namespaceStorageOptions : tableAccess.storageOptions; - throw indexAdmissionSnapshotLoadFailure( - dbName, tableName, e, datasetUri, runtimeStorageOptions); + throw indexAdmissionSnapshotLoadFailure(dbName, tableName, e, null, Collections.emptyMap()); } } @VisibleForTesting RuntimeException indexAdmissionSnapshotLoadFailure(String dbName, String tableName, Throwable throwable, String datasetUri, Map<String, String> runtimeStorageOptions) { - String sanitizedMessage = sanitizedRootCauseMessage( - throwable, datasetUri, runtimeStorageOptions); - Throwable sanitizedCause = throwable instanceof IllegalArgumentException - ? new IllegalArgumentException(sanitizedMessage) - : new RuntimeException(sanitizedMessage); - return new RuntimeException("Failed to load Lance index admission snapshot for " - + dbName + "." + tableName + ": " + sanitizedMessage, sanitizedCause); + return LanceErrorMessages.failure("Failed to load Lance index admission snapshot for " + + dbName + "." + tableName, throwable, datasetUri, runtimeStorageOptions, catalogSecrets()); } - @VisibleForTesting - RuntimeException indexMetadataLoadFailure(String dbName, String tableName, - Throwable throwable, String datasetUri, Map<String, String> runtimeStorageOptions) { - String sanitizedMessage = sanitizedRootCauseMessage( - throwable, datasetUri, runtimeStorageOptions); - Throwable sanitizedCause = throwable instanceof IllegalArgumentException - ? new IllegalArgumentException(sanitizedMessage) - : new RuntimeException(sanitizedMessage); - return new RuntimeException("Failed to load Lance index metadata for " + dbName + "." + tableName - + ": " + sanitizedMessage, sanitizedCause); + public String getLanceCatalogType() { + return getLanceProperties().getLanceCatalogType(); } - private ResolvedTableAccess resolveTableAccess(String dbName, String tableName) { - DescribeTableResponse table = describeTable(dbName, tableName); - 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); + private synchronized LanceCatalogClient getClient() { + makeSureInitialized(); + if (client == null) { + throw new IllegalStateException("Lance catalog resources have been closed"); } - - // One option map serves both readers: the FE opens the dataset through the Lance Java SDK - // and the BE through lance-c, so neither can end up with credentials the other lacks. The - // dataset URL picks the option vocabulary, the same way Lance picks a provider from it. - Map<String, String> storageOptions = LanceStorageOptions.fromDorisAndVendedStorageOptions(datasetUri, - catalogProperty.getOrderedStoragePropertiesList(), table.getStorageOptions()); - return new ResolvedTableAccess(datasetUri, storageOptions); + return client; Review Comment: [P1] Acquire the generation before releasing the catalog lock `getClient()` returns the selected generation before any `LanceCatalogClient.acquire()`. A reader can return the old client, then `onRefreshCache(true)` can publish a replacement and retire/close the old client while `activeOperations == 0`, after which the reader's first `acquire()` throws. Thus an operation already admitted by the catalog fails solely because refresh won this handoff race. Please pin the selected generation atomically under the catalog handoff lock, release it after the delegated call, and add a test that pauses between selection and operation entry. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
