You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by "wangzhaobo957-cloud (via GitHub)" <gi...@apache.org> on 2026/10/21 08:23:00 UTC
[PR] Feature/lance namespace table ddl branch 4.2 [doris]
wangzhaobo957-cloud opened a new pull request, #69384:
URL: https://github.com/apache/doris/pull/69384
### What problem does this PR solve?
Issue Number: close #xxx
Related PR: #xxx
Problem Summary:
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg: https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR should merge into -->
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Feature/lance namespace table ddl branch 4.2 [doris]
Posted by "zhangstar333 (via GitHub)" <gi...@apache.org>.
zhangstar333 commented on PR #69384:
URL: https://github.com/apache/doris/pull/69384#issuecomment-6410773498
run buildall
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Feature/lance namespace table ddl branch 4.2 [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #69384:
URL: https://github.com/apache/doris/pull/69384#issuecomment-6518318005
# FE UT Coverage Report
Increment line coverage ` 63.09% (523/829)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/69384_d6e8236ca9f76e905d3cb4cc090ca2e616d400bd/fe_increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/69384_d6e8236ca9f76e905d3cb4cc090ca2e616d400bd/fe_report/index.html)
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Feature/lance namespace table ddl branch 4.2 [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #69384:
URL: https://github.com/apache/doris/pull/69384#issuecomment-6368745728
# FE Regression Coverage Report
Increment line coverage ` 15.60% (129/827)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/69384_fc50b302caded9c86c056202da0f1d07fe7ac7c7_merge_fe/increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/69384_fc50b302caded9c86c056202da0f1d07fe7ac7c7_merge_fe/report/index.html)
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Feature/lance namespace table ddl branch 4.2 [doris]
Posted by "github-actions[bot] (via GitHub) via commits" <co...@doris.apache.org>.
github-actions[bot] commented on PR #69384:
URL: https://github.com/apache/doris/pull/69384#issuecomment-6568499342
PR approved by anyone and no changes requested.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Feature/lance namespace table ddl branch 4.2 [doris]
Posted by "Gabriel39 (via GitHub)" <gi...@apache.org>.
Gabriel39 commented on PR #69384:
URL: https://github.com/apache/doris/pull/69384#issuecomment-6331775102
run buildall
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Feature/lance namespace table ddl branch 4.2 [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #69384:
URL: https://github.com/apache/doris/pull/69384#issuecomment-6367645660
# FE Regression Coverage Report
Increment line coverage ` 15.60% (129/827)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/69384_fc50b302caded9c86c056202da0f1d07fe7ac7c7_merge_fe/increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/69384_fc50b302caded9c86c056202da0f1d07fe7ac7c7_merge_fe/report/index.html)
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Feature/lance namespace table ddl branch 4.2 [doris]
Posted by "Gabriel39 (via GitHub)" <gi...@apache.org>.
Gabriel39 commented on PR #69384:
URL: https://github.com/apache/doris/pull/69384#issuecomment-6363939964
/review
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Feature/lance namespace table ddl branch 4.2 [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #69384:
URL: https://github.com/apache/doris/pull/69384#issuecomment-6414734772
# FE Regression Coverage Report
Increment line coverage ` 19.78% (164/829)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/69384_94ba0cf6350d889144e9df56c8293d669196c124_merge_fe/increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/69384_94ba0cf6350d889144e9df56c8293d669196c124_merge_fe/report/index.html)
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Feature/lance namespace table ddl branch 4.2 [doris]
Posted by "Gabriel39 (via GitHub)" <gi...@apache.org>.
Gabriel39 commented on PR #69384:
URL: https://github.com/apache/doris/pull/69384#issuecomment-6331772575
/review
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Feature/lance namespace table ddl branch 4.2 [doris]
Posted by "zhangstar333 (via GitHub)" <gi...@apache.org>.
zhangstar333 commented on code in PR #69384:
URL: https://github.com/apache/doris/pull/69384#discussion_r4335234631
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/RefreshManager.java:
##########
@@ -263,7 +263,11 @@ public void replayRefreshTable(ExternalObjectLog log) {
}
if (!Strings.isNullOrEmpty(log.getNewTableName())) {
// this is a rename table op
- db.get().unregisterTable(log.getTableName());
+ db.get().unregisterTableForReplay(log.getTableName());
+ if (db.get().getTableForReplay(log.getNewTableName()).isPresent()
Review Comment:
### F-04 ยท Replay renames even when the source table object is cold
- **Severity**: Major
- **Where**: `fe/fe-core/src/main/java/org/apache/doris/catalog/RefreshManager.java:266-270`
```java
db.get().unregisterTableForReplay(log.getTableName());
if (db.get().getTableForReplay(log.getNewTableName()).isPresent()
|| db.get().hasLocalTableName(log.getNewTableName())) {
db.get().unregisterTableForReplay(log.getNewTableName());
}
```
**What is wrong.** A follower with cached table names but no cached source table object skips the rename-specific names invalidation.
**Why it happens.** `fe/fe-core/src/main/java/org/apache/doris/catalog/RefreshManager.java:235-264` looks up the old object and returns on a miss before reaching this branch and resetMetaCacheNames. ExternalDatabase.getTableForReplay only probes the object cache; getTableNamesWithLock separately serves cached names. The fallback invalidateTableByNameOrWider clears engine caches and row counts, not ExternalDatabase metadata names.
**When it bites.** A follower runs SHOW TABLES and caches events, then the events table object is evicted while the names snapshot remains valid. SHOW TABLES itself constructs table objects through getTables, so the concrete trigger requires subsequent object eviction rather than assuming SHOW TABLES never loads the object. The leader successfully renames events to renamed_events in a REST namespace. The follower replays the journal but continues listing events and omitting renamed_events until a separate names refresh. This needs no concurrent external mutation. Broad engine invalidation means this scenario alone does not prove stale engine schema/data.
**Suggested fix.** Move the whole existing rename branch immediately after successful database resolution, before looking up the old table object, and return after handling it. Keep both-name retirement and resetMetaCacheNames on this path. Remove the later duplicate branch; leave the generic cold-source fallback for non-rename refreshes. The essential early dispatch is:
```diff
--- 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
@@ before resolving the old table object
+ if (!Strings.isNullOrEmpty(log.getNewTableName())) {
+ db.get().unregisterTableForReplay(log.getTableName());
+ if (db.get().getTableForReplay(log.getNewTableName()).isPresent()
+ || db.get().hasLocalTableName(log.getNewTableName())) {
+ db.get().unregisterTableForReplay(log.getNewTableName());
+ }
+ db.get().resetMetaCacheNames();
+ return;
+ }
Optional<? extends ExternalTable> table;
```
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Feature/lance namespace table ddl branch 4.2 [doris]
Posted by "zhangstar333 (via GitHub)" <gi...@apache.org>.
zhangstar333 commented on PR #69384:
URL: https://github.com/apache/doris/pull/69384#issuecomment-6514623082
run buildall
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Feature/lance namespace table ddl branch 4.2 [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #69384:
URL: https://github.com/apache/doris/pull/69384#issuecomment-6362089176
# FE UT Coverage Report
Increment line coverage ` 63.80% (520/815)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/69384_0e771f8e2345ef3e573de7363709623257227d60/fe_increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/69384_0e771f8e2345ef3e573de7363709623257227d60/fe_report/index.html)
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Feature/lance namespace table ddl branch 4.2 [doris]
Posted by "zhangstar333 (via GitHub)" <gi...@apache.org>.
zhangstar333 commented on PR #69384:
URL: https://github.com/apache/doris/pull/69384#issuecomment-6490852034
run buildall
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Feature/lance namespace table ddl branch 4.2 [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #69384:
URL: https://github.com/apache/doris/pull/69384#discussion_r4344273837
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceMetadataOps.java:
##########
@@ -0,0 +1,673 @@
+// 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.ColumnPosition;
+import org.apache.doris.catalog.ArrayType;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.MapType;
+import org.apache.doris.catalog.PrimitiveType;
+import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.catalog.StructField;
+import org.apache.doris.catalog.StructType;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.common.UserException;
+import org.apache.doris.datasource.ExternalDatabase;
+import org.apache.doris.datasource.ExternalTable;
+import org.apache.doris.datasource.lance.metadata.LanceTypeConverter;
+import org.apache.doris.datasource.operations.ExternalMetadataOps;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateOrReplaceBranchInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateOrReplaceTagInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateTableInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.DropBranchInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.DropTagInfo;
+
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.lance.namespace.errors.NamespaceAlreadyExistsException;
+import org.lance.namespace.errors.NamespaceNotFoundException;
+import org.lance.namespace.errors.TableAlreadyExistsException;
+import org.lance.namespace.errors.TableNotFoundException;
+import org.lance.namespace.model.AddColumnsEntry;
+import org.lance.namespace.model.AlterColumnsEntry;
+
+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.Set;
+import java.util.TreeSet;
+
+/** Doris external metadata operations backed by the Lance Namespace API. */
+public class LanceMetadataOps implements ExternalMetadataOps {
+ private static final Logger LOG = LogManager.getLogger(LanceMetadataOps.class);
+ private static final String TABLE_COMMENT_PROPERTY = "comment";
+
+ private final LanceExternalCatalog catalog;
+
+ public LanceMetadataOps(LanceExternalCatalog catalog) {
+ this.catalog = catalog;
+ }
+
+ @Override
+ public boolean createDbImpl(String dbName, boolean ifNotExists, Map<String, String> properties)
+ throws DdlException {
+ return execute("Failed to create Lance database " + dbName, client -> {
+ if (client.isRootDatabase(dbName)) {
+ throw new DdlException("Cannot create the configured Lance root database: " + dbName);
+ }
+ ExternalDatabase<?> cachedDb = catalog.getDbNullable(dbName);
+ if (cachedDb != null) {
+ if (client.databaseExists(cachedDb.getRemoteName())) {
+ if (ifNotExists) {
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_CREATE_EXISTS, dbName);
+ }
+ catalog.unregisterDatabase(cachedDb.getFullName());
+ catalog.resetMetaCacheNames();
+ }
+ if (client.databaseExists(dbName)) {
+ if (ifNotExists) {
+ catalog.resetMetaCacheNames();
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_CREATE_EXISTS, dbName);
+ }
+ try {
+ client.createDatabase(dbName, new HashMap<>(
+ Optional.ofNullable(properties).orElse(Collections.emptyMap())));
+ return false;
+ } catch (NamespaceAlreadyExistsException e) {
+ if (ifNotExists) {
+ catalog.resetMetaCacheNames();
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_CREATE_EXISTS, dbName);
+ throw new IllegalStateException("unreachable");
+ }
+ });
+ }
+
+ @Override
+ public void afterCreateDb(String dbName) {
+ catalog.retireCachedDatabaseForNoOp(dbName);
+ catalog.resetMetaCacheNames();
+ }
+
+ @Override
+ public boolean dropDbImpl(String dbName, boolean ifExists, boolean force) throws DdlException {
+ if (force && LanceExternalCatalog.LANCE_FILESYSTEM.equals(catalog.getLanceCatalogType())) {
+ throw new DdlException(
+ "DROP DATABASE FORCE is not supported for Lance filesystem catalogs");
+ }
+ ExternalDatabase<?> db = catalog.getDbNullable(dbName);
+ return execute("Failed to drop Lance database " + dbName, client -> {
+ if (client.isRootDatabase(dbName)) {
+ throw new DdlException("Cannot drop the configured Lance root database: " + dbName);
+ }
+ if (db == null) {
+ if (ifExists) {
+ return false;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_DROP_EXISTS, dbName);
+ return false;
+ }
+ String remoteDbName = db.getRemoteName();
+ if (client.isRootDatabase(remoteDbName)) {
+ throw new DdlException("Cannot drop the configured Lance root database: " + dbName);
+ }
+ if (!client.databaseExists(remoteDbName)) {
+ if (ifExists) {
+ return false;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_DROP_EXISTS, dbName);
+ }
+ try {
+ client.dropDatabase(remoteDbName, ifExists, force);
+ } catch (NamespaceNotFoundException e) {
+ if (ifExists) {
+ return false;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_DROP_EXISTS, dbName);
+ }
+ return true;
+ });
+ }
+
+ @Override
+ public void afterDropDb(String dbName) {
+ Optional<ExternalDatabase<? extends ExternalTable>> db = catalog.getDbForReplay(dbName);
+ if (db.isPresent()) {
+ catalog.unregisterDatabase(db.get().getFullName());
+ return;
+ }
+ catalog.unregisterDatabase(dbName);
+ catalog.retireAllDatabaseObjectsWithoutEngineInvalidation();
+ }
+
+ @Override
+ public void afterDropDbNoOp(String dbName) {
+ catalog.invalidateTableAccessCache();
+ catalog.retireAllDatabaseObjectsWithoutEngineInvalidation();
+ }
+
+ @Override
+ public boolean shouldJournalDropDbNoOp() {
+ return true;
+ }
+
+ @Override
+ public boolean createTableImpl(CreateTableInfo createTableInfo) throws UserException {
+ String dbName = createTableInfo.getDbName();
+ String tableName = createTableInfo.getTableName();
+ ExternalDatabase<?> db = catalog.getDbNullable(dbName);
+ boolean filesystemCatalog =
+ LanceExternalCatalog.LANCE_FILESYSTEM.equals(catalog.getLanceCatalogType());
+ if (filesystemCatalog && tableName.contains("$")) {
+ throw new DdlException(
+ "Lance filesystem table names must not contain the manifest delimiter '$'");
+ }
+ if (db == null) {
+ throw new DdlException("Failed to get database: '" + dbName
+ + "' in catalog: " + catalog.getName());
+ }
+ List<Column> columns = createTableInfo.getColumns();
+ validateCreateColumns(columns);
+ Schema schema = LanceTypeConverter.toArrowSchema(columns);
+ Map<String, String> properties = new HashMap<>(
+ Optional.ofNullable(createTableInfo.getProperties()).orElse(Collections.emptyMap()));
+ if (StringUtils.isNotBlank(createTableInfo.getComment())) {
+ properties.put(TABLE_COMMENT_PROPERTY, createTableInfo.getComment());
+ }
+
+ return execute("Failed to create Lance table " + dbName + "." + tableName, client -> {
+ if (client.tableExists(db.getRemoteName(), tableName)) {
+ if (createTableInfo.isIfNotExists()) {
+ resetTableNameCache(dbName);
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_TABLE_EXISTS_ERROR, tableName);
+ }
+ ExternalTable cachedTable = db.getTableNullable(tableName);
+ if (cachedTable != null) {
+ if (client.tableExists(db.getRemoteName(), cachedTable.getRemoteName())) {
+ db.resetMetaCacheNames();
+ if (createTableInfo.isIfNotExists()) {
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_TABLE_EXISTS_ERROR, tableName);
+ }
+ catalog.invalidateTableAccessCache();
+ db.unregisterTable(cachedTable.getName());
+ db.resetMetaCacheNames();
+ }
+ try {
+ client.createTable(db.getRemoteName(), tableName, schema, properties);
+ return false;
+ } catch (TableAlreadyExistsException e) {
+ if (createTableInfo.isIfNotExists()) {
+ resetTableNameCache(dbName);
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_TABLE_EXISTS_ERROR, tableName);
+ throw new IllegalStateException("unreachable");
+ }
+ });
+ }
+
+ private static void validateCreateColumns(List<Column> columns) throws UserException {
+ for (Column column : columns) {
+ validateLosslessType(column.getType(), "CREATE TABLE");
+ if (column.isAggregated()) {
+ throw new UserException("Lance columns do not support aggregation: " + column.getName());
+ }
+ if (column.isAutoInc()) {
+ throw new UserException("Lance columns do not support AUTO_INCREMENT: " + column.getName());
+ }
+ if (column.isGeneratedColumn()) {
+ throw new UserException("Lance columns do not support generated columns: " + column.getName());
+ }
+ if (column.getDefaultValue() != null) {
+ throw new UserException("Lance table creation does not support column defaults: "
+ + column.getName());
+ }
+ }
+ }
+
+ @Override
+ public void afterCreateTable(String dbName, String tblName) {
+ catalog.invalidateTableAccessCache();
+ Optional<ExternalDatabase<?>> db = catalog.getDbForReplay(dbName);
+ if (db.isPresent()) {
+ db.get().unregisterTable(tblName);
+ db.get().resetMetaCacheNames();
+ }
+ }
+
+ @Override
+ public void dropTableImpl(ExternalTable dorisTable, boolean ifExists) throws DdlException {
+ String dbName = dorisTable.getRemoteDbName();
+ String tableName = dorisTable.getRemoteName();
+ execute("Failed to drop Lance table " + dbName + "." + tableName, client -> {
+ try {
+ client.dropTable(dbName, tableName);
+ } catch (TableNotFoundException e) {
+ if (!ifExists) {
+ ErrorReport.reportDdlException(ErrorCode.ERR_UNKNOWN_TABLE, tableName, dbName);
+ }
+ }
+ return null;
+ });
+ }
+
+ @Override
+ public void afterDropTable(String dbName, String tblName) {
+ catalog.invalidateTableAccessCache();
+ Optional<ExternalDatabase<?>> db = catalog.getDbForReplay(dbName);
+ if (db.isPresent()) {
+ boolean invalidated = db.get().unregisterTableForReplay(tblName);
Review Comment:
[P2] Reset table names after DROP replay on a cold object
A follower can retain a complete SHOW TABLES names snapshot after the source table object is evicted. `unregisterTableForReplay` handles that cold miss by invalidating engine caches and returning false without changing the names snapshot; `hasLocalTableName` remains true, so this hook does nothing further. Replaying the successful DROP then leaves the dropped table listed until names expiry. Reset the database names generation regardless of object residency, as the rename hook does. This is a separate DROP TABLE path from the existing rename-replay thread, which is fixed at this head.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceMetadataOps.java:
##########
@@ -0,0 +1,673 @@
+// 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.ColumnPosition;
+import org.apache.doris.catalog.ArrayType;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.MapType;
+import org.apache.doris.catalog.PrimitiveType;
+import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.catalog.StructField;
+import org.apache.doris.catalog.StructType;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.common.UserException;
+import org.apache.doris.datasource.ExternalDatabase;
+import org.apache.doris.datasource.ExternalTable;
+import org.apache.doris.datasource.lance.metadata.LanceTypeConverter;
+import org.apache.doris.datasource.operations.ExternalMetadataOps;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateOrReplaceBranchInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateOrReplaceTagInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateTableInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.DropBranchInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.DropTagInfo;
+
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.lance.namespace.errors.NamespaceAlreadyExistsException;
+import org.lance.namespace.errors.NamespaceNotFoundException;
+import org.lance.namespace.errors.TableAlreadyExistsException;
+import org.lance.namespace.errors.TableNotFoundException;
+import org.lance.namespace.model.AddColumnsEntry;
+import org.lance.namespace.model.AlterColumnsEntry;
+
+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.Set;
+import java.util.TreeSet;
+
+/** Doris external metadata operations backed by the Lance Namespace API. */
+public class LanceMetadataOps implements ExternalMetadataOps {
+ private static final Logger LOG = LogManager.getLogger(LanceMetadataOps.class);
+ private static final String TABLE_COMMENT_PROPERTY = "comment";
+
+ private final LanceExternalCatalog catalog;
+
+ public LanceMetadataOps(LanceExternalCatalog catalog) {
+ this.catalog = catalog;
+ }
+
+ @Override
+ public boolean createDbImpl(String dbName, boolean ifNotExists, Map<String, String> properties)
+ throws DdlException {
+ return execute("Failed to create Lance database " + dbName, client -> {
+ if (client.isRootDatabase(dbName)) {
+ throw new DdlException("Cannot create the configured Lance root database: " + dbName);
+ }
+ ExternalDatabase<?> cachedDb = catalog.getDbNullable(dbName);
+ if (cachedDb != null) {
+ if (client.databaseExists(cachedDb.getRemoteName())) {
+ if (ifNotExists) {
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_CREATE_EXISTS, dbName);
+ }
+ catalog.unregisterDatabase(cachedDb.getFullName());
+ catalog.resetMetaCacheNames();
+ }
+ if (client.databaseExists(dbName)) {
+ if (ifNotExists) {
+ catalog.resetMetaCacheNames();
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_CREATE_EXISTS, dbName);
+ }
+ try {
+ client.createDatabase(dbName, new HashMap<>(
+ Optional.ofNullable(properties).orElse(Collections.emptyMap())));
+ return false;
+ } catch (NamespaceAlreadyExistsException e) {
+ if (ifNotExists) {
+ catalog.resetMetaCacheNames();
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_CREATE_EXISTS, dbName);
+ throw new IllegalStateException("unreachable");
+ }
+ });
+ }
+
+ @Override
+ public void afterCreateDb(String dbName) {
+ catalog.retireCachedDatabaseForNoOp(dbName);
+ catalog.resetMetaCacheNames();
+ }
+
+ @Override
+ public boolean dropDbImpl(String dbName, boolean ifExists, boolean force) throws DdlException {
+ if (force && LanceExternalCatalog.LANCE_FILESYSTEM.equals(catalog.getLanceCatalogType())) {
+ throw new DdlException(
+ "DROP DATABASE FORCE is not supported for Lance filesystem catalogs");
+ }
+ ExternalDatabase<?> db = catalog.getDbNullable(dbName);
+ return execute("Failed to drop Lance database " + dbName, client -> {
+ if (client.isRootDatabase(dbName)) {
+ throw new DdlException("Cannot drop the configured Lance root database: " + dbName);
+ }
+ if (db == null) {
+ if (ifExists) {
+ return false;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_DROP_EXISTS, dbName);
+ return false;
+ }
+ String remoteDbName = db.getRemoteName();
+ if (client.isRootDatabase(remoteDbName)) {
+ throw new DdlException("Cannot drop the configured Lance root database: " + dbName);
+ }
+ if (!client.databaseExists(remoteDbName)) {
+ if (ifExists) {
+ return false;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_DROP_EXISTS, dbName);
+ }
+ try {
+ client.dropDatabase(remoteDbName, ifExists, force);
+ } catch (NamespaceNotFoundException e) {
+ if (ifExists) {
+ return false;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_DROP_EXISTS, dbName);
+ }
+ return true;
+ });
+ }
+
+ @Override
+ public void afterDropDb(String dbName) {
+ Optional<ExternalDatabase<? extends ExternalTable>> db = catalog.getDbForReplay(dbName);
+ if (db.isPresent()) {
+ catalog.unregisterDatabase(db.get().getFullName());
+ return;
+ }
+ catalog.unregisterDatabase(dbName);
+ catalog.retireAllDatabaseObjectsWithoutEngineInvalidation();
+ }
+
+ @Override
+ public void afterDropDbNoOp(String dbName) {
+ catalog.invalidateTableAccessCache();
+ catalog.retireAllDatabaseObjectsWithoutEngineInvalidation();
+ }
+
+ @Override
+ public boolean shouldJournalDropDbNoOp() {
+ return true;
+ }
+
+ @Override
+ public boolean createTableImpl(CreateTableInfo createTableInfo) throws UserException {
+ String dbName = createTableInfo.getDbName();
+ String tableName = createTableInfo.getTableName();
+ ExternalDatabase<?> db = catalog.getDbNullable(dbName);
+ boolean filesystemCatalog =
+ LanceExternalCatalog.LANCE_FILESYSTEM.equals(catalog.getLanceCatalogType());
+ if (filesystemCatalog && tableName.contains("$")) {
+ throw new DdlException(
+ "Lance filesystem table names must not contain the manifest delimiter '$'");
+ }
+ if (db == null) {
+ throw new DdlException("Failed to get database: '" + dbName
+ + "' in catalog: " + catalog.getName());
+ }
+ List<Column> columns = createTableInfo.getColumns();
+ validateCreateColumns(columns);
+ Schema schema = LanceTypeConverter.toArrowSchema(columns);
+ Map<String, String> properties = new HashMap<>(
+ Optional.ofNullable(createTableInfo.getProperties()).orElse(Collections.emptyMap()));
+ if (StringUtils.isNotBlank(createTableInfo.getComment())) {
+ properties.put(TABLE_COMMENT_PROPERTY, createTableInfo.getComment());
+ }
+
+ return execute("Failed to create Lance table " + dbName + "." + tableName, client -> {
+ if (client.tableExists(db.getRemoteName(), tableName)) {
+ if (createTableInfo.isIfNotExists()) {
+ resetTableNameCache(dbName);
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_TABLE_EXISTS_ERROR, tableName);
+ }
+ ExternalTable cachedTable = db.getTableNullable(tableName);
+ if (cachedTable != null) {
+ if (client.tableExists(db.getRemoteName(), cachedTable.getRemoteName())) {
+ db.resetMetaCacheNames();
+ if (createTableInfo.isIfNotExists()) {
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_TABLE_EXISTS_ERROR, tableName);
+ }
+ catalog.invalidateTableAccessCache();
+ db.unregisterTable(cachedTable.getName());
+ db.resetMetaCacheNames();
+ }
+ try {
+ client.createTable(db.getRemoteName(), tableName, schema, properties);
+ return false;
+ } catch (TableAlreadyExistsException e) {
+ if (createTableInfo.isIfNotExists()) {
+ resetTableNameCache(dbName);
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_TABLE_EXISTS_ERROR, tableName);
+ throw new IllegalStateException("unreachable");
+ }
+ });
+ }
+
+ private static void validateCreateColumns(List<Column> columns) throws UserException {
+ for (Column column : columns) {
+ validateLosslessType(column.getType(), "CREATE TABLE");
+ if (column.isAggregated()) {
+ throw new UserException("Lance columns do not support aggregation: " + column.getName());
+ }
+ if (column.isAutoInc()) {
+ throw new UserException("Lance columns do not support AUTO_INCREMENT: " + column.getName());
+ }
+ if (column.isGeneratedColumn()) {
+ throw new UserException("Lance columns do not support generated columns: " + column.getName());
+ }
+ if (column.getDefaultValue() != null) {
+ throw new UserException("Lance table creation does not support column defaults: "
+ + column.getName());
+ }
+ }
+ }
+
+ @Override
+ public void afterCreateTable(String dbName, String tblName) {
+ catalog.invalidateTableAccessCache();
+ Optional<ExternalDatabase<?>> db = catalog.getDbForReplay(dbName);
+ if (db.isPresent()) {
+ db.get().unregisterTable(tblName);
+ db.get().resetMetaCacheNames();
+ }
+ }
+
+ @Override
+ public void dropTableImpl(ExternalTable dorisTable, boolean ifExists) throws DdlException {
+ String dbName = dorisTable.getRemoteDbName();
+ String tableName = dorisTable.getRemoteName();
+ execute("Failed to drop Lance table " + dbName + "." + tableName, client -> {
+ try {
+ client.dropTable(dbName, tableName);
+ } catch (TableNotFoundException e) {
+ if (!ifExists) {
+ ErrorReport.reportDdlException(ErrorCode.ERR_UNKNOWN_TABLE, tableName, dbName);
+ }
+ }
+ return null;
+ });
+ }
+
+ @Override
+ public void afterDropTable(String dbName, String tblName) {
+ catalog.invalidateTableAccessCache();
+ Optional<ExternalDatabase<?>> db = catalog.getDbForReplay(dbName);
+ if (db.isPresent()) {
+ boolean invalidated = db.get().unregisterTableForReplay(tblName);
+ if (!invalidated && !db.get().hasLocalTableName(tblName)) {
+ db.get().retireAllTableObjectsWithoutEngineInvalidation();
+ }
+ }
+ }
+
+ @Override
+ public void renameTableImpl(String dbName, String oldName, String newName) throws DdlException {
+ if (!catalog.isRestCatalogConfigured()) {
+ throw new DdlException("RENAME TABLE is not supported for Lance filesystem catalogs");
+ }
+ ExternalDatabase<?> db = catalog.getDbNullable(dbName);
+ ExternalTable source = db == null ? null : db.getTableNullable(oldName);
+ if (source == null) {
+ throw new DdlException("Failed to resolve Lance rename source: " + dbName + "." + oldName);
+ }
+ execute("Failed to rename Lance table " + dbName + "." + oldName + " to " + newName,
+ client -> {
+ client.renameTable(source.getRemoteDbName(), source.getRemoteName(), newName);
+ return null;
+ });
+ }
+
+ @Override
+ public void afterRenameTable(String dbName, String oldName, String newName) {
+ catalog.invalidateTableAccessCache();
+ Optional<ExternalDatabase<?>> db = catalog.getDbForReplay(dbName);
+ if (db.isPresent()) {
+ db.get().unregisterTableForReplay(oldName);
+ if (db.get().getTableForReplay(newName).isPresent() || db.get().hasLocalTableName(newName)) {
+ db.get().unregisterTableForReplay(newName);
+ }
+ db.get().resetMetaCacheNames();
+ }
+ }
+
+ @Override
+ public void addColumn(ExternalTable dorisTable, Column column, ColumnPosition position, long updateTime)
+ throws UserException {
+ validateAddColumn(column, position);
+ ensureNewColumnNames(dorisTable, Collections.singletonList(column));
+ AddColumnsEntry entry = new AddColumnsEntry()
+ .name(column.getName())
+ .expression(LanceTypeConverter.toAddColumnExpression(column.getType()));
+ execute("Failed to add column " + column.getName() + " to Lance table "
+ + tableName(dorisTable),
+ client -> {
+ client.addColumns(dorisTable.getRemoteDbName(), dorisTable.getRemoteName(),
+ Collections.singletonList(entry));
+ return null;
+ });
+ refreshTable(dorisTable, updateTime);
+ }
+
+ @Override
+ public void addColumns(ExternalTable dorisTable, List<Column> columns, long updateTime)
+ throws UserException {
+ if (columns.isEmpty()) {
+ return;
+ }
+ List<AddColumnsEntry> entries = new ArrayList<>(columns.size());
+ for (Column column : columns) {
+ validateAddColumn(column, null);
+ entries.add(new AddColumnsEntry()
+ .name(column.getName())
+ .expression(LanceTypeConverter.toAddColumnExpression(column.getType())));
+ }
+ ensureNewColumnNames(dorisTable, columns);
+ execute("Failed to add columns to Lance table " + tableName(dorisTable), client -> {
+ client.addColumns(dorisTable.getRemoteDbName(), dorisTable.getRemoteName(), entries);
+ return null;
+ });
+ refreshTable(dorisTable, updateTime);
+ }
+
+ @Override
+ public void dropColumn(ExternalTable dorisTable, String columnName, long updateTime)
+ throws UserException {
+ Column currentColumn = requireColumn(dorisTable, columnName);
+ execute("Failed to drop column " + currentColumn.getName() + " from Lance table "
+ + tableName(dorisTable),
+ client -> {
+ client.dropColumns(dorisTable.getRemoteDbName(), dorisTable.getRemoteName(),
+ Collections.singletonList(currentColumn.getName()));
+ return null;
+ });
+ refreshTable(dorisTable, updateTime);
+ }
+
+ @Override
+ public void renameColumn(ExternalTable dorisTable, String oldName, String newName, long updateTime)
+ throws UserException {
+ Column currentColumn = requireColumn(dorisTable, oldName);
+ Column conflictingColumn = dorisTable.getColumn(newName);
+ if (conflictingColumn != null && conflictingColumn != currentColumn) {
+ throw new UserException("Column " + newName
+ + " conflicts with an existing Lance column (case-insensitive)");
+ }
+ if (currentColumn.getName().equals(newName)) {
+ return;
+ }
+ AlterColumnsEntry alteration = new AlterColumnsEntry()
+ .path(currentColumn.getName())
+ .rename(newName);
+ execute("Failed to rename column " + currentColumn.getName() + " to " + newName
+ + " in Lance table " + tableName(dorisTable),
+ client -> {
+ client.alterColumns(dorisTable.getRemoteDbName(), dorisTable.getRemoteName(),
+ Collections.singletonList(alteration));
+ return null;
+ });
+ refreshTable(dorisTable, updateTime);
+ }
+
+ @Override
+ public void modifyColumn(ExternalTable dorisTable, Column column, ColumnPosition position, long updateTime)
+ throws UserException {
+ validateModifyColumn(column, position);
+ Column currentColumn = requireColumn(dorisTable, column.getName());
+ List<AlterColumnsEntry> alterations = new ArrayList<>();
+ boolean typeChanged = !currentColumn.getType().equals(column.getType());
+ if (typeChanged && StringUtils.isNotEmpty(currentColumn.getComment())) {
+ throw new UserException("Lance MODIFY COLUMN does not support changing the type "
+ + "of a column with a comment");
+ }
+ if (typeChanged) {
+ alterations.add(new AlterColumnsEntry()
+ .path(currentColumn.getName())
+ .dataType(LanceTypeConverter.toAlterColumnType(column.getType())));
+ }
+ boolean nullabilityChanged = column.isNullableSpecified()
+ && currentColumn.isAllowNull() != column.isAllowNull();
+ // Lance rejects casting and tightening nullability in the same alter_columns call.
+ if (typeChanged && nullabilityChanged && !column.isAllowNull()) {
+ throw new UserException("Lance MODIFY COLUMN does not support changing type and "
+ + "making the column NOT NULL in one operation. Apply the type change first, "
+ + "then change nullability.");
+ }
+ if (nullabilityChanged) {
+ alterations.add(new AlterColumnsEntry()
+ .path(currentColumn.getName())
+ .nullable(column.isAllowNull()));
+ }
+ if (alterations.isEmpty()) {
+ return;
+ }
+
+ boolean modified = false;
+ DdlException mutationFailure = null;
+ try {
+ for (AlterColumnsEntry alteration : alterations) {
+ execute("Failed to modify column " + currentColumn.getName() + " in Lance table "
+ + tableName(dorisTable),
+ client -> {
+ client.alterColumns(dorisTable.getRemoteDbName(), dorisTable.getRemoteName(),
+ Collections.singletonList(alteration));
+ return null;
+ });
+ modified = true;
+ }
+ } catch (DdlException e) {
+ mutationFailure = e;
+ throw e;
Review Comment:
[P2] Journal refreshes after a partially committed MODIFY
For a comment-free `score INT NOT NULL`, `MODIFY COLUMN score BIGINT NULL` passes the guard and sends separate type and nullability requests. If the type change commits but the second request fails, this rethrow skips `ExternalCatalog.modifyColumn`'s subsequent `logRefreshExternalTable` call. The finally block refreshes only the leader, so followers with cached schema keep describing `score` as INT after BIGINT was committed remotely. Guarantee the refresh journal whenever any mutation committed, including later mutation or local-refresh failures, while preserving the original error.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceMetadataOps.java:
##########
@@ -0,0 +1,673 @@
+// 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.ColumnPosition;
+import org.apache.doris.catalog.ArrayType;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.MapType;
+import org.apache.doris.catalog.PrimitiveType;
+import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.catalog.StructField;
+import org.apache.doris.catalog.StructType;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.common.UserException;
+import org.apache.doris.datasource.ExternalDatabase;
+import org.apache.doris.datasource.ExternalTable;
+import org.apache.doris.datasource.lance.metadata.LanceTypeConverter;
+import org.apache.doris.datasource.operations.ExternalMetadataOps;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateOrReplaceBranchInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateOrReplaceTagInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateTableInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.DropBranchInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.DropTagInfo;
+
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.lance.namespace.errors.NamespaceAlreadyExistsException;
+import org.lance.namespace.errors.NamespaceNotFoundException;
+import org.lance.namespace.errors.TableAlreadyExistsException;
+import org.lance.namespace.errors.TableNotFoundException;
+import org.lance.namespace.model.AddColumnsEntry;
+import org.lance.namespace.model.AlterColumnsEntry;
+
+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.Set;
+import java.util.TreeSet;
+
+/** Doris external metadata operations backed by the Lance Namespace API. */
+public class LanceMetadataOps implements ExternalMetadataOps {
+ private static final Logger LOG = LogManager.getLogger(LanceMetadataOps.class);
+ private static final String TABLE_COMMENT_PROPERTY = "comment";
+
+ private final LanceExternalCatalog catalog;
+
+ public LanceMetadataOps(LanceExternalCatalog catalog) {
+ this.catalog = catalog;
+ }
+
+ @Override
+ public boolean createDbImpl(String dbName, boolean ifNotExists, Map<String, String> properties)
+ throws DdlException {
+ return execute("Failed to create Lance database " + dbName, client -> {
+ if (client.isRootDatabase(dbName)) {
+ throw new DdlException("Cannot create the configured Lance root database: " + dbName);
+ }
+ ExternalDatabase<?> cachedDb = catalog.getDbNullable(dbName);
+ if (cachedDb != null) {
+ if (client.databaseExists(cachedDb.getRemoteName())) {
+ if (ifNotExists) {
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_CREATE_EXISTS, dbName);
+ }
+ catalog.unregisterDatabase(cachedDb.getFullName());
+ catalog.resetMetaCacheNames();
+ }
+ if (client.databaseExists(dbName)) {
+ if (ifNotExists) {
+ catalog.resetMetaCacheNames();
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_CREATE_EXISTS, dbName);
+ }
+ try {
+ client.createDatabase(dbName, new HashMap<>(
+ Optional.ofNullable(properties).orElse(Collections.emptyMap())));
+ return false;
+ } catch (NamespaceAlreadyExistsException e) {
+ if (ifNotExists) {
+ catalog.resetMetaCacheNames();
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_CREATE_EXISTS, dbName);
+ throw new IllegalStateException("unreachable");
+ }
+ });
+ }
+
+ @Override
+ public void afterCreateDb(String dbName) {
+ catalog.retireCachedDatabaseForNoOp(dbName);
+ catalog.resetMetaCacheNames();
+ }
+
+ @Override
+ public boolean dropDbImpl(String dbName, boolean ifExists, boolean force) throws DdlException {
+ if (force && LanceExternalCatalog.LANCE_FILESYSTEM.equals(catalog.getLanceCatalogType())) {
+ throw new DdlException(
+ "DROP DATABASE FORCE is not supported for Lance filesystem catalogs");
+ }
+ ExternalDatabase<?> db = catalog.getDbNullable(dbName);
+ return execute("Failed to drop Lance database " + dbName, client -> {
+ if (client.isRootDatabase(dbName)) {
+ throw new DdlException("Cannot drop the configured Lance root database: " + dbName);
+ }
+ if (db == null) {
+ if (ifExists) {
+ return false;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_DROP_EXISTS, dbName);
+ return false;
+ }
+ String remoteDbName = db.getRemoteName();
+ if (client.isRootDatabase(remoteDbName)) {
+ throw new DdlException("Cannot drop the configured Lance root database: " + dbName);
+ }
+ if (!client.databaseExists(remoteDbName)) {
+ if (ifExists) {
+ return false;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_DROP_EXISTS, dbName);
+ }
+ try {
+ client.dropDatabase(remoteDbName, ifExists, force);
+ } catch (NamespaceNotFoundException e) {
+ if (ifExists) {
+ return false;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_DROP_EXISTS, dbName);
+ }
+ return true;
+ });
+ }
+
+ @Override
+ public void afterDropDb(String dbName) {
+ Optional<ExternalDatabase<? extends ExternalTable>> db = catalog.getDbForReplay(dbName);
+ if (db.isPresent()) {
+ catalog.unregisterDatabase(db.get().getFullName());
+ return;
+ }
+ catalog.unregisterDatabase(dbName);
+ catalog.retireAllDatabaseObjectsWithoutEngineInvalidation();
+ }
+
+ @Override
+ public void afterDropDbNoOp(String dbName) {
+ catalog.invalidateTableAccessCache();
+ catalog.retireAllDatabaseObjectsWithoutEngineInvalidation();
Review Comment:
[P2] Clear database names during no-op DROP reconciliation
If a cached namespace was deleted externally, `DROP DATABASE IF EXISTS` resolves the cached database, observes that the remote namespace is absent, and reaches this hook. `retireAllDatabaseObjectsWithoutEngineInvalidation` swaps only the object cache generation; it preserves the independent names snapshot. Consequently the leader still lists the dropped database after the command, while followers replay `afterDropDb` and remove that name. Reset the catalog names cache here as part of reconciliation.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceMetadataOps.java:
##########
@@ -0,0 +1,673 @@
+// 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.ColumnPosition;
+import org.apache.doris.catalog.ArrayType;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.MapType;
+import org.apache.doris.catalog.PrimitiveType;
+import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.catalog.StructField;
+import org.apache.doris.catalog.StructType;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.common.UserException;
+import org.apache.doris.datasource.ExternalDatabase;
+import org.apache.doris.datasource.ExternalTable;
+import org.apache.doris.datasource.lance.metadata.LanceTypeConverter;
+import org.apache.doris.datasource.operations.ExternalMetadataOps;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateOrReplaceBranchInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateOrReplaceTagInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateTableInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.DropBranchInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.DropTagInfo;
+
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.lance.namespace.errors.NamespaceAlreadyExistsException;
+import org.lance.namespace.errors.NamespaceNotFoundException;
+import org.lance.namespace.errors.TableAlreadyExistsException;
+import org.lance.namespace.errors.TableNotFoundException;
+import org.lance.namespace.model.AddColumnsEntry;
+import org.lance.namespace.model.AlterColumnsEntry;
+
+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.Set;
+import java.util.TreeSet;
+
+/** Doris external metadata operations backed by the Lance Namespace API. */
+public class LanceMetadataOps implements ExternalMetadataOps {
+ private static final Logger LOG = LogManager.getLogger(LanceMetadataOps.class);
+ private static final String TABLE_COMMENT_PROPERTY = "comment";
+
+ private final LanceExternalCatalog catalog;
+
+ public LanceMetadataOps(LanceExternalCatalog catalog) {
+ this.catalog = catalog;
+ }
+
+ @Override
+ public boolean createDbImpl(String dbName, boolean ifNotExists, Map<String, String> properties)
+ throws DdlException {
+ return execute("Failed to create Lance database " + dbName, client -> {
+ if (client.isRootDatabase(dbName)) {
+ throw new DdlException("Cannot create the configured Lance root database: " + dbName);
+ }
+ ExternalDatabase<?> cachedDb = catalog.getDbNullable(dbName);
+ if (cachedDb != null) {
+ if (client.databaseExists(cachedDb.getRemoteName())) {
+ if (ifNotExists) {
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_CREATE_EXISTS, dbName);
+ }
+ catalog.unregisterDatabase(cachedDb.getFullName());
+ catalog.resetMetaCacheNames();
+ }
+ if (client.databaseExists(dbName)) {
+ if (ifNotExists) {
+ catalog.resetMetaCacheNames();
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_CREATE_EXISTS, dbName);
+ }
+ try {
+ client.createDatabase(dbName, new HashMap<>(
+ Optional.ofNullable(properties).orElse(Collections.emptyMap())));
+ return false;
+ } catch (NamespaceAlreadyExistsException e) {
+ if (ifNotExists) {
+ catalog.resetMetaCacheNames();
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_CREATE_EXISTS, dbName);
+ throw new IllegalStateException("unreachable");
+ }
+ });
+ }
+
+ @Override
+ public void afterCreateDb(String dbName) {
+ catalog.retireCachedDatabaseForNoOp(dbName);
+ catalog.resetMetaCacheNames();
+ }
+
+ @Override
+ public boolean dropDbImpl(String dbName, boolean ifExists, boolean force) throws DdlException {
+ if (force && LanceExternalCatalog.LANCE_FILESYSTEM.equals(catalog.getLanceCatalogType())) {
+ throw new DdlException(
+ "DROP DATABASE FORCE is not supported for Lance filesystem catalogs");
+ }
+ ExternalDatabase<?> db = catalog.getDbNullable(dbName);
+ return execute("Failed to drop Lance database " + dbName, client -> {
+ if (client.isRootDatabase(dbName)) {
+ throw new DdlException("Cannot drop the configured Lance root database: " + dbName);
+ }
+ if (db == null) {
+ if (ifExists) {
+ return false;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_DROP_EXISTS, dbName);
+ return false;
+ }
+ String remoteDbName = db.getRemoteName();
+ if (client.isRootDatabase(remoteDbName)) {
+ throw new DdlException("Cannot drop the configured Lance root database: " + dbName);
+ }
+ if (!client.databaseExists(remoteDbName)) {
+ if (ifExists) {
+ return false;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_DROP_EXISTS, dbName);
+ }
+ try {
+ client.dropDatabase(remoteDbName, ifExists, force);
+ } catch (NamespaceNotFoundException e) {
+ if (ifExists) {
+ return false;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_DROP_EXISTS, dbName);
+ }
+ return true;
+ });
+ }
+
+ @Override
+ public void afterDropDb(String dbName) {
+ Optional<ExternalDatabase<? extends ExternalTable>> db = catalog.getDbForReplay(dbName);
+ if (db.isPresent()) {
+ catalog.unregisterDatabase(db.get().getFullName());
+ return;
+ }
+ catalog.unregisterDatabase(dbName);
+ catalog.retireAllDatabaseObjectsWithoutEngineInvalidation();
+ }
+
+ @Override
+ public void afterDropDbNoOp(String dbName) {
+ catalog.invalidateTableAccessCache();
+ catalog.retireAllDatabaseObjectsWithoutEngineInvalidation();
+ }
+
+ @Override
+ public boolean shouldJournalDropDbNoOp() {
+ return true;
+ }
+
+ @Override
+ public boolean createTableImpl(CreateTableInfo createTableInfo) throws UserException {
+ String dbName = createTableInfo.getDbName();
+ String tableName = createTableInfo.getTableName();
+ ExternalDatabase<?> db = catalog.getDbNullable(dbName);
+ boolean filesystemCatalog =
+ LanceExternalCatalog.LANCE_FILESYSTEM.equals(catalog.getLanceCatalogType());
+ if (filesystemCatalog && tableName.contains("$")) {
+ throw new DdlException(
+ "Lance filesystem table names must not contain the manifest delimiter '$'");
+ }
+ if (db == null) {
+ throw new DdlException("Failed to get database: '" + dbName
+ + "' in catalog: " + catalog.getName());
+ }
+ List<Column> columns = createTableInfo.getColumns();
+ validateCreateColumns(columns);
+ Schema schema = LanceTypeConverter.toArrowSchema(columns);
+ Map<String, String> properties = new HashMap<>(
+ Optional.ofNullable(createTableInfo.getProperties()).orElse(Collections.emptyMap()));
+ if (StringUtils.isNotBlank(createTableInfo.getComment())) {
+ properties.put(TABLE_COMMENT_PROPERTY, createTableInfo.getComment());
+ }
+
+ return execute("Failed to create Lance table " + dbName + "." + tableName, client -> {
+ if (client.tableExists(db.getRemoteName(), tableName)) {
+ if (createTableInfo.isIfNotExists()) {
+ resetTableNameCache(dbName);
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_TABLE_EXISTS_ERROR, tableName);
+ }
+ ExternalTable cachedTable = db.getTableNullable(tableName);
+ if (cachedTable != null) {
+ if (client.tableExists(db.getRemoteName(), cachedTable.getRemoteName())) {
+ db.resetMetaCacheNames();
+ if (createTableInfo.isIfNotExists()) {
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_TABLE_EXISTS_ERROR, tableName);
+ }
+ catalog.invalidateTableAccessCache();
+ db.unregisterTable(cachedTable.getName());
+ db.resetMetaCacheNames();
+ }
+ try {
+ client.createTable(db.getRemoteName(), tableName, schema, properties);
+ return false;
+ } catch (TableAlreadyExistsException e) {
+ if (createTableInfo.isIfNotExists()) {
+ resetTableNameCache(dbName);
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_TABLE_EXISTS_ERROR, tableName);
+ throw new IllegalStateException("unreachable");
+ }
+ });
+ }
+
+ private static void validateCreateColumns(List<Column> columns) throws UserException {
+ for (Column column : columns) {
+ validateLosslessType(column.getType(), "CREATE TABLE");
+ if (column.isAggregated()) {
+ throw new UserException("Lance columns do not support aggregation: " + column.getName());
+ }
+ if (column.isAutoInc()) {
+ throw new UserException("Lance columns do not support AUTO_INCREMENT: " + column.getName());
+ }
+ if (column.isGeneratedColumn()) {
+ throw new UserException("Lance columns do not support generated columns: " + column.getName());
+ }
+ if (column.getDefaultValue() != null) {
+ throw new UserException("Lance table creation does not support column defaults: "
+ + column.getName());
+ }
+ }
+ }
+
+ @Override
+ public void afterCreateTable(String dbName, String tblName) {
+ catalog.invalidateTableAccessCache();
+ Optional<ExternalDatabase<?>> db = catalog.getDbForReplay(dbName);
+ if (db.isPresent()) {
+ db.get().unregisterTable(tblName);
+ db.get().resetMetaCacheNames();
+ }
+ }
+
+ @Override
+ public void dropTableImpl(ExternalTable dorisTable, boolean ifExists) throws DdlException {
+ String dbName = dorisTable.getRemoteDbName();
+ String tableName = dorisTable.getRemoteName();
+ execute("Failed to drop Lance table " + dbName + "." + tableName, client -> {
+ try {
+ client.dropTable(dbName, tableName);
+ } catch (TableNotFoundException e) {
+ if (!ifExists) {
+ ErrorReport.reportDdlException(ErrorCode.ERR_UNKNOWN_TABLE, tableName, dbName);
+ }
+ }
+ return null;
+ });
+ }
+
+ @Override
+ public void afterDropTable(String dbName, String tblName) {
+ catalog.invalidateTableAccessCache();
+ Optional<ExternalDatabase<?>> db = catalog.getDbForReplay(dbName);
+ if (db.isPresent()) {
+ boolean invalidated = db.get().unregisterTableForReplay(tblName);
+ if (!invalidated && !db.get().hasLocalTableName(tblName)) {
+ db.get().retireAllTableObjectsWithoutEngineInvalidation();
+ }
+ }
+ }
+
+ @Override
+ public void renameTableImpl(String dbName, String oldName, String newName) throws DdlException {
+ if (!catalog.isRestCatalogConfigured()) {
+ throw new DdlException("RENAME TABLE is not supported for Lance filesystem catalogs");
+ }
+ ExternalDatabase<?> db = catalog.getDbNullable(dbName);
+ ExternalTable source = db == null ? null : db.getTableNullable(oldName);
+ if (source == null) {
+ throw new DdlException("Failed to resolve Lance rename source: " + dbName + "." + oldName);
+ }
+ execute("Failed to rename Lance table " + dbName + "." + oldName + " to " + newName,
+ client -> {
+ client.renameTable(source.getRemoteDbName(), source.getRemoteName(), newName);
+ return null;
+ });
+ }
+
+ @Override
+ public void afterRenameTable(String dbName, String oldName, String newName) {
+ catalog.invalidateTableAccessCache();
+ Optional<ExternalDatabase<?>> db = catalog.getDbForReplay(dbName);
+ if (db.isPresent()) {
+ db.get().unregisterTableForReplay(oldName);
+ if (db.get().getTableForReplay(newName).isPresent() || db.get().hasLocalTableName(newName)) {
+ db.get().unregisterTableForReplay(newName);
+ }
+ db.get().resetMetaCacheNames();
+ }
+ }
+
+ @Override
+ public void addColumn(ExternalTable dorisTable, Column column, ColumnPosition position, long updateTime)
+ throws UserException {
+ validateAddColumn(column, position);
+ ensureNewColumnNames(dorisTable, Collections.singletonList(column));
+ AddColumnsEntry entry = new AddColumnsEntry()
+ .name(column.getName())
+ .expression(LanceTypeConverter.toAddColumnExpression(column.getType()));
+ execute("Failed to add column " + column.getName() + " to Lance table "
+ + tableName(dorisTable),
+ client -> {
+ client.addColumns(dorisTable.getRemoteDbName(), dorisTable.getRemoteName(),
+ Collections.singletonList(entry));
+ return null;
+ });
+ refreshTable(dorisTable, updateTime);
+ }
+
+ @Override
+ public void addColumns(ExternalTable dorisTable, List<Column> columns, long updateTime)
+ throws UserException {
+ if (columns.isEmpty()) {
+ return;
+ }
+ List<AddColumnsEntry> entries = new ArrayList<>(columns.size());
+ for (Column column : columns) {
+ validateAddColumn(column, null);
+ entries.add(new AddColumnsEntry()
+ .name(column.getName())
+ .expression(LanceTypeConverter.toAddColumnExpression(column.getType())));
+ }
+ ensureNewColumnNames(dorisTable, columns);
+ execute("Failed to add columns to Lance table " + tableName(dorisTable), client -> {
+ client.addColumns(dorisTable.getRemoteDbName(), dorisTable.getRemoteName(), entries);
+ return null;
+ });
+ refreshTable(dorisTable, updateTime);
+ }
+
+ @Override
+ public void dropColumn(ExternalTable dorisTable, String columnName, long updateTime)
+ throws UserException {
+ Column currentColumn = requireColumn(dorisTable, columnName);
+ execute("Failed to drop column " + currentColumn.getName() + " from Lance table "
+ + tableName(dorisTable),
+ client -> {
+ client.dropColumns(dorisTable.getRemoteDbName(), dorisTable.getRemoteName(),
+ Collections.singletonList(currentColumn.getName()));
+ return null;
+ });
+ refreshTable(dorisTable, updateTime);
+ }
+
+ @Override
+ public void renameColumn(ExternalTable dorisTable, String oldName, String newName, long updateTime)
+ throws UserException {
+ Column currentColumn = requireColumn(dorisTable, oldName);
+ Column conflictingColumn = dorisTable.getColumn(newName);
+ if (conflictingColumn != null && conflictingColumn != currentColumn) {
+ throw new UserException("Column " + newName
+ + " conflicts with an existing Lance column (case-insensitive)");
+ }
+ if (currentColumn.getName().equals(newName)) {
+ return;
+ }
+ AlterColumnsEntry alteration = new AlterColumnsEntry()
+ .path(currentColumn.getName())
+ .rename(newName);
+ execute("Failed to rename column " + currentColumn.getName() + " to " + newName
+ + " in Lance table " + tableName(dorisTable),
+ client -> {
+ client.alterColumns(dorisTable.getRemoteDbName(), dorisTable.getRemoteName(),
+ Collections.singletonList(alteration));
+ return null;
+ });
+ refreshTable(dorisTable, updateTime);
+ }
+
+ @Override
+ public void modifyColumn(ExternalTable dorisTable, Column column, ColumnPosition position, long updateTime)
+ throws UserException {
+ validateModifyColumn(column, position);
+ Column currentColumn = requireColumn(dorisTable, column.getName());
+ List<AlterColumnsEntry> alterations = new ArrayList<>();
+ boolean typeChanged = !currentColumn.getType().equals(column.getType());
+ if (typeChanged && StringUtils.isNotEmpty(currentColumn.getComment())) {
+ throw new UserException("Lance MODIFY COLUMN does not support changing the type "
+ + "of a column with a comment");
+ }
+ if (typeChanged) {
+ alterations.add(new AlterColumnsEntry()
+ .path(currentColumn.getName())
+ .dataType(LanceTypeConverter.toAlterColumnType(column.getType())));
+ }
+ boolean nullabilityChanged = column.isNullableSpecified()
+ && currentColumn.isAllowNull() != column.isAllowNull();
+ // Lance rejects casting and tightening nullability in the same alter_columns call.
+ if (typeChanged && nullabilityChanged && !column.isAllowNull()) {
+ throw new UserException("Lance MODIFY COLUMN does not support changing type and "
+ + "making the column NOT NULL in one operation. Apply the type change first, "
+ + "then change nullability.");
+ }
+ if (nullabilityChanged) {
+ alterations.add(new AlterColumnsEntry()
+ .path(currentColumn.getName())
+ .nullable(column.isAllowNull()));
+ }
+ if (alterations.isEmpty()) {
+ return;
+ }
+
+ boolean modified = false;
+ DdlException mutationFailure = null;
+ try {
+ for (AlterColumnsEntry alteration : alterations) {
+ execute("Failed to modify column " + currentColumn.getName() + " in Lance table "
+ + tableName(dorisTable),
+ client -> {
+ client.alterColumns(dorisTable.getRemoteDbName(), dorisTable.getRemoteName(),
+ Collections.singletonList(alteration));
+ return null;
+ });
+ modified = true;
+ }
+ } catch (DdlException e) {
+ mutationFailure = e;
+ throw e;
+ } finally {
+ if (modified) {
+ try {
+ refreshTable(dorisTable, updateTime);
+ } catch (RuntimeException refreshFailure) {
+ if (mutationFailure == null) {
+ throw refreshFailure;
+ }
+ mutationFailure.addSuppressed(refreshFailure);
+ }
+ }
+ }
+ }
+
+ static void validateAddColumn(Column column, ColumnPosition position) throws UserException {
+ validateColumnAttributes(column, "ADD COLUMN");
+ if (column.getType().getPrimitiveType() == PrimitiveType.DECIMAL256) {
+ throw new UserException("Lance ADD COLUMN does not support DECIMAL256: "
+ + column.getType().toSql());
+ }
+ if (column.hasDefaultValue() || column.hasOnUpdateDefaultValue()) {
+ throw new UserException("Lance ADD COLUMN does not support default values");
+ }
+ if (!column.isAllowNull()) {
+ throw new UserException("Lance ADD COLUMN only supports nullable columns");
+ }
+ if (column.isCommentSpecified()) {
+ throw new UserException("Lance ADD COLUMN does not support column comments");
+ }
+ if (position != null) {
+ throw new UserException("Lance ADD COLUMN does not support column positions");
+ }
+ }
+
+ static void validateModifyColumn(Column column, ColumnPosition position) throws UserException {
+ validateColumnAttributes(column, "MODIFY COLUMN");
+ if (column.hasDefaultValue() || column.hasOnUpdateDefaultValue()) {
+ throw new UserException("Lance MODIFY COLUMN does not support default values");
+ }
+ if (column.isCommentSpecified()) {
+ throw new UserException("Lance MODIFY COLUMN does not support column comments");
+ }
+ if (position != null) {
+ throw new UserException("Lance MODIFY COLUMN does not support column positions");
+ }
+ }
+
+ private static void validateColumnAttributes(Column column, String operation) throws UserException {
+ validateLosslessType(column.getType(), operation);
+ if (column.isKey()) {
+ throw new UserException("Lance " + operation + " does not support key columns");
+ }
+ if (column.isAggregated()) {
+ throw new UserException("Lance " + operation + " does not support aggregation: "
+ + column.getName());
+ }
+ if (column.isAutoInc()) {
+ throw new UserException("Lance " + operation + " does not support AUTO_INCREMENT: "
+ + column.getName());
+ }
+ if (column.isGeneratedColumn()) {
+ throw new UserException("Lance " + operation + " does not support generated columns: "
+ + column.getName());
+ }
+ }
+
+ private static void validateLosslessType(Type type, String operation) throws UserException {
+ PrimitiveType primitiveType = type.getPrimitiveType();
+ if (primitiveType == PrimitiveType.CHAR || primitiveType == PrimitiveType.VARCHAR
+ || (primitiveType == PrimitiveType.VARBINARY
+ && type.getLength() != ScalarType.MAX_VARBINARY_LENGTH)) {
+ throw new UserException("Lance " + operation
+ + " does not support length-bounded types: " + type.toSql());
+ }
+ switch (primitiveType) {
+ case ARRAY:
+ validateLosslessType(((ArrayType) type).getItemType(), operation);
+ break;
+ case MAP:
+ MapType mapType = (MapType) type;
+ validateLosslessType(mapType.getKeyType(), operation);
+ validateLosslessType(mapType.getValueType(), operation);
+ break;
+ case STRUCT:
+ for (StructField field : ((StructType) type).getFields()) {
+ validateLosslessType(field.getType(), operation);
+ }
+ break;
+ default:
+ break;
+ }
+ }
+
+ private static void ensureNewColumnNames(ExternalTable dorisTable, List<Column> columns)
+ throws UserException {
+ Set<String> names = new TreeSet<>(String.CASE_INSENSITIVE_ORDER);
+ for (Column column : dorisTable.getFullSchema()) {
+ names.add(column.getName());
+ }
+ for (Column column : columns) {
+ if (!names.add(column.getName())) {
+ throw new UserException("Column " + column.getName()
+ + " conflicts with an existing Lance column (case-insensitive)");
+ }
+ }
+ }
+
+ private static Column requireColumn(ExternalTable dorisTable, String columnName)
+ throws UserException {
+ Column column = dorisTable.getColumn(columnName);
+ if (column == null) {
+ throw new UserException("Column " + columnName + " does not exist in Lance table "
+ + tableName(dorisTable));
+ }
+ return column;
+ }
+
+ private static String tableName(ExternalTable dorisTable) {
+ return dorisTable.getRemoteDbName() + "." + dorisTable.getRemoteName();
+ }
+
+ private void refreshTable(ExternalTable dorisTable, long updateTime) {
+ catalog.invalidateTableAccessCache();
+ Optional<ExternalDatabase<?>> db = catalog.getDbForReplay(dorisTable.getDbName());
+ if (db.isPresent()) {
+ Optional<?> table = db.get().getTableForReplay(dorisTable.getName());
Review Comment:
[P2] Refresh the held table even when its cache entry was evicted
Another query can evict this table object from the bounded database object cache while the namespace ALTER call is running, even though the DDL still holds `dorisTable`. On that miss this helper skips `refreshTableInternal`, leaving the independent schema cache intact: the database object cache has no removal listener, and schema entries are keyed by table identity rather than object instance. A successful ADD/DROP/RENAME/MODIFY can therefore return while the leader's next DESC still uses the old schema. Refresh using the held table/database, or invalidate its schema and row-count identity unconditionally.
##########
fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceMetadataOpsTest.java:
##########
@@ -0,0 +1,796 @@
+// 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.ColumnPosition;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.RefreshManager;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.UserException;
+import org.apache.doris.datasource.ExternalDatabase;
+import org.apache.doris.datasource.ExternalTable;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateTableInfo;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.lance.Session;
+import org.lance.namespace.LanceNamespace;
+import org.lance.namespace.errors.NamespaceAlreadyExistsException;
+import org.lance.namespace.errors.NamespaceNotFoundException;
+import org.lance.namespace.errors.TableAlreadyExistsException;
+import org.lance.namespace.errors.TableNotFoundException;
+import org.lance.namespace.model.AlterTableAddColumnsRequest;
+import org.lance.namespace.model.AlterTableAlterColumnsRequest;
+import org.lance.namespace.model.CreateNamespaceRequest;
+import org.lance.namespace.model.CreateTableRequest;
+import org.lance.namespace.model.DropNamespaceRequest;
+import org.lance.namespace.model.RenameTableRequest;
+import org.mockito.ArgumentCaptor;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Optional;
+
+public class LanceMetadataOpsTest {
+ @Test
+ public void testAddColumnValidation() throws UserException {
+ Column nullable = new Column("score", Type.INT, true);
+ Assertions.assertDoesNotThrow(
+ () -> LanceMetadataOps.validateAddColumn(nullable, null));
+
+ Column required = new Column("required", Type.INT, false);
+ assertRejected(() -> LanceMetadataOps.validateAddColumn(required, null),
+ "only supports nullable columns");
+
+ Column defaulted = new Column("defaulted", Type.INT, false, null, true, "1", "");
+ assertRejected(() -> LanceMetadataOps.validateAddColumn(defaulted, null),
+ "does not support default values");
+
+ Column commented = new Column("commented", Type.INT, true, "comment");
+ commented.setCommentSpecified(true);
+ assertRejected(() -> LanceMetadataOps.validateAddColumn(commented, null),
+ "does not support column comments");
+
+ assertRejected(() -> LanceMetadataOps.validateAddColumn(
+ nullable, new ColumnPosition("id")),
+ "does not support column positions");
+ }
+
+ @Test
+ public void testModifyColumnValidation() throws UserException {
+ Column column = new Column("score", Type.BIGINT, true);
+ column.setNullableSpecified(true);
+ Assertions.assertDoesNotThrow(
+ () -> LanceMetadataOps.validateModifyColumn(column, null));
+
+ column.setCommentSpecified(true);
+ assertRejected(() -> LanceMetadataOps.validateModifyColumn(column, null),
+ "does not support column comments");
+ }
+
+ @Test
+ public void testCreateDatabaseIfNotExistsSkipsExistingDatabase() throws DdlException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+
+ try {
+ Assertions.assertTrue(new LanceMetadataOps(catalog).createDb(
+ "analytics", true, Collections.singletonMap("owner", "doris")));
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(namespace, Mockito.never()).createNamespace(Mockito.any());
+ Mockito.verify(catalog).resetMetaCacheNames();
+ }
+
+ @Test
+ public void testCreateRootDatabaseIsRejected() {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ LanceMetadataOps ops = new LanceMetadataOps(catalog);
+
+ try {
+ for (boolean ifNotExists : Arrays.asList(false, true)) {
+ DdlException exception = Assertions.assertThrows(DdlException.class,
+ () -> ops.createDb("default", ifNotExists, Collections.emptyMap()));
+ Assertions.assertTrue(exception.getMessage().contains("root database"));
+ }
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(namespace, Mockito.never()).namespaceExists(Mockito.any());
+ Mockito.verify(namespace, Mockito.never()).createNamespace(Mockito.any());
+ Mockito.verify(catalog, Mockito.never()).resetMetaCacheNames();
+ }
+
+ @Test
+ public void testCreateDatabaseRejectsMappedLocalNameConflict() throws DdlException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.doReturn(database).when(catalog).getDbNullable("sales_db");
+ Mockito.when(database.getRemoteName()).thenReturn("Sales");
+ LanceMetadataOps ops = new LanceMetadataOps(catalog);
+
+ try {
+ Assertions.assertTrue(ops.createDb("sales_db", true, Collections.emptyMap()));
+ DdlException exception = Assertions.assertThrows(DdlException.class,
+ () -> ops.createDb("sales_db", false, Collections.emptyMap()));
+ Assertions.assertTrue(exception.getMessage().contains("exist"));
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(namespace, Mockito.never()).namespaceExists(Mockito.any());
+ Mockito.verify(namespace, Mockito.never()).createNamespace(Mockito.any());
+ }
+
+ @Test
+ public void testCreateDatabaseHandlesConcurrentCreate() throws DdlException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ Mockito.doThrow(new NamespaceNotFoundException("missing"))
+ .when(namespace).namespaceExists(Mockito.any());
+ Mockito.doThrow(new NamespaceAlreadyExistsException("created concurrently"))
+ .when(namespace).createNamespace(Mockito.any());
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ LanceMetadataOps ops = new LanceMetadataOps(catalog);
+
+ try {
+ Assertions.assertTrue(ops.createDb("analytics", true, Collections.emptyMap()));
+ DdlException exception = Assertions.assertThrows(DdlException.class,
+ () -> ops.createDb("analytics", false, Collections.emptyMap()));
+ Assertions.assertTrue(exception.getMessage().contains("exist"));
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(namespace, Mockito.times(2)).createNamespace(Mockito.any());
+ Mockito.verify(catalog).resetMetaCacheNames();
+ }
+
+ @Test
+ public void testFilesystemCreateDatabaseCreatesChildNamespace() throws DdlException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ Mockito.doThrow(new NamespaceNotFoundException("missing"))
+ .when(namespace).namespaceExists(Mockito.any());
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ Mockito.when(catalog.getLanceCatalogType()).thenReturn(LanceExternalCatalog.LANCE_FILESYSTEM);
+
+ try {
+ Assertions.assertFalse(new LanceMetadataOps(catalog).createDb(
+ "analytics", false, Collections.singletonMap("owner", "doris")));
+ } finally {
+ client.close();
+ }
+
+ ArgumentCaptor<CreateNamespaceRequest> request =
+ ArgumentCaptor.forClass(CreateNamespaceRequest.class);
+ Mockito.verify(namespace).createNamespace(request.capture());
+ Assertions.assertEquals(Arrays.asList("tenant", "analytics"), request.getValue().getId());
+ Assertions.assertEquals("doris", request.getValue().getProperties().get("owner"));
+ Mockito.verify(catalog).resetMetaCacheNames();
+ }
+
+ @Test
+ public void testAfterCreateDatabaseRetiresStaleDatabaseObject() {
+ LanceExternalCatalog catalog = Mockito.mock(LanceExternalCatalog.class);
+
+ new LanceMetadataOps(catalog).afterCreateDb("analytics");
+
+ Mockito.verify(catalog).retireCachedDatabaseForNoOp("analytics");
+ Mockito.verify(catalog).resetMetaCacheNames();
+ }
+
+ @Test
+ public void testCreateTableIfNotExistsSkipsExistingRemoteTable() throws UserException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.doReturn(database).when(catalog).getDbNullable("local_db");
+ Mockito.when(catalog.getDbForReplay("local_db")).thenReturn(Optional.of(database));
+ Mockito.when(database.getRemoteName()).thenReturn("analytics");
+ Mockito.when(database.getTableNullable("events")).thenReturn(null);
+ CreateTableInfo createTableInfo = createTableInfo(true);
+
+ try {
+ Assertions.assertTrue(new LanceMetadataOps(catalog).createTable(createTableInfo));
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(namespace, Mockito.never())
+ .createTable(Mockito.any(), Mockito.any(byte[].class));
+ Mockito.verify(database).resetMetaCacheNames();
+ }
+
+ @Test
+ public void testFilesystemCreateTableInRootDatabaseWithParentNamespace() throws UserException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ Mockito.doThrow(new TableNotFoundException("missing"))
+ .when(namespace).tableExists(Mockito.any());
+ LanceCatalogClient client = newClient(namespace, new RootAllocator(1024 * 1024));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ Mockito.when(catalog.getLanceCatalogType()).thenReturn(LanceExternalCatalog.LANCE_FILESYSTEM);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.doReturn(database).when(catalog).getDbNullable("default");
+ Mockito.when(catalog.getDbForReplay("default")).thenReturn(Optional.of(database));
+ Mockito.when(database.getRemoteName()).thenReturn("default");
+ Mockito.when(database.getTableNullable("events")).thenReturn(null);
+
+ try {
+ Assertions.assertFalse(new LanceMetadataOps(catalog).createTable(
+ createTableInfo("default", "events", false)));
+ } finally {
+ client.close();
+ }
+
+ ArgumentCaptor<CreateTableRequest> request = ArgumentCaptor.forClass(CreateTableRequest.class);
+ Mockito.verify(namespace).createTable(request.capture(), Mockito.any(byte[].class));
+ Assertions.assertEquals(Arrays.asList("tenant", "events"), request.getValue().getId());
+ Mockito.verify(catalog).invalidateTableAccessCache();
+ Mockito.verify(database).unregisterTable("events");
+ Mockito.verify(database).resetMetaCacheNames();
+ }
+
+ @Test
+ public void testAfterCreateTableRetiresStaleTableObject() {
+ LanceExternalCatalog catalog = Mockito.mock(LanceExternalCatalog.class);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.when(catalog.getDbForReplay("analytics")).thenReturn(Optional.of(database));
+
+ new LanceMetadataOps(catalog).afterCreateTable("analytics", "events");
+
+ Mockito.verify(catalog).invalidateTableAccessCache();
+ Mockito.verify(database).unregisterTable("events");
+ Mockito.verify(database).resetMetaCacheNames();
+ }
+
+ @Test
+ public void testCreateTableHandlesConcurrentCreate() throws UserException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ Mockito.doThrow(new TableNotFoundException("missing"))
+ .when(namespace).tableExists(Mockito.any());
+ Mockito.doThrow(new TableAlreadyExistsException("created concurrently"))
+ .when(namespace).createTable(Mockito.any(), Mockito.any(byte[].class));
+ LanceCatalogClient client = newClient(namespace, new RootAllocator(1024 * 1024));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.doReturn(database).when(catalog).getDbNullable("local_db");
+ Mockito.when(catalog.getDbForReplay("local_db")).thenReturn(Optional.of(database));
+ Mockito.when(database.getRemoteName()).thenReturn("analytics");
+ Mockito.when(database.getTableNullable("events")).thenReturn(null);
+ LanceMetadataOps ops = new LanceMetadataOps(catalog);
+
+ try {
+ Assertions.assertTrue(ops.createTable(createTableInfo(true)));
+ DdlException exception = Assertions.assertThrows(DdlException.class,
+ () -> ops.createTable(createTableInfo(false)));
+ Assertions.assertTrue(exception.getMessage().contains("already exists"));
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(namespace, Mockito.times(2))
+ .createTable(Mockito.any(), Mockito.any(byte[].class));
+ Mockito.verify(database).resetMetaCacheNames();
+ }
+
+ @Test
+ public void testCreateTableIgnoresStaleLocalCacheWhenRemoteTableIsGone() throws UserException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ Mockito.doThrow(new TableNotFoundException("missing"))
+ .when(namespace).tableExists(Mockito.any());
+ LanceCatalogClient client = newClient(namespace, new RootAllocator(1024 * 1024));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.doReturn(database).when(catalog).getDbNullable("local_db");
+ Mockito.when(catalog.getDbForReplay("local_db")).thenReturn(Optional.of(database));
+ Mockito.when(database.getRemoteName()).thenReturn("analytics");
+ ExternalTable staleTable = table("local_db", "events", "analytics", "stale_events");
+ Mockito.when(database.getTableNullable("events"))
+ .thenReturn(staleTable, null);
+
+ try {
+ Assertions.assertFalse(new LanceMetadataOps(catalog).createTable(createTableInfo(true)));
+ } finally {
+ client.close();
+ }
+
+ ArgumentCaptor<CreateTableRequest> request = ArgumentCaptor.forClass(CreateTableRequest.class);
+ Mockito.verify(namespace).createTable(request.capture(), Mockito.any(byte[].class));
+ Assertions.assertEquals(Arrays.asList("tenant", "analytics", "events"),
+ request.getValue().getId());
+ Mockito.verify(database).unregisterTable("events");
+ Mockito.verify(database, Mockito.atLeastOnce()).resetMetaCacheNames();
+ Mockito.verify(catalog).invalidateTableAccessCache();
+ }
+
+ @Test
+ public void testCreateTableRejectsLocalNameConflictAfterRefresh() throws UserException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ Mockito.doAnswer(invocation -> {
+ if (((org.lance.namespace.model.TableExistsRequest) invocation.getArgument(0))
+ .getId().get(2).equals("events")) {
+ throw new TableNotFoundException("missing");
+ }
+ return null;
+ }).when(namespace).tableExists(Mockito.any());
+ LanceCatalogClient client = newClient(namespace, new RootAllocator(1024 * 1024));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.doReturn(database).when(catalog).getDbNullable("local_db");
+ Mockito.when(catalog.getDbForReplay("local_db")).thenReturn(Optional.of(database));
+ Mockito.when(database.getRemoteName()).thenReturn("analytics");
+ ExternalTable conflictingTable = table("local_db", "events", "analytics", "Events");
+ Mockito.when(database.getTableNullable("events")).thenReturn(conflictingTable);
+ LanceMetadataOps ops = new LanceMetadataOps(catalog);
+
+ try {
+ Assertions.assertTrue(ops.createTable(createTableInfo(true)));
+ DdlException exception = Assertions.assertThrows(DdlException.class,
+ () -> ops.createTable(createTableInfo(false)));
+ Assertions.assertTrue(exception.getMessage().contains("already exists"));
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(namespace, Mockito.never()).createTable(Mockito.any(), Mockito.any(byte[].class));
+ Mockito.verify(database, Mockito.times(2)).resetMetaCacheNames();
+ }
+
+ @Test
+ public void testIfExistsHandlesConcurrentDropAndRefreshesLocalNames() throws DdlException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ Mockito.doThrow(new NamespaceNotFoundException("dropped concurrently"))
+ .when(namespace).dropNamespace(Mockito.any());
+ Mockito.doThrow(new TableNotFoundException("dropped concurrently"))
+ .when(namespace).dropTable(Mockito.any());
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.doReturn(database).when(catalog).getDbNullable("analytics");
+ Mockito.when(catalog.getDbForReplay("local_db")).thenReturn(Optional.of(database));
+ Mockito.when(database.getRemoteName()).thenReturn("analytics");
+ ExternalTable table = table("local_db", "local_table", "analytics", "events");
+ Mockito.when(database.unregisterTableForReplay("local_table")).thenReturn(true);
+ LanceMetadataOps ops = new LanceMetadataOps(catalog);
+
+ try {
+ Assertions.assertDoesNotThrow(() -> ops.dropDb("analytics", true, false));
+ Assertions.assertThrows(DdlException.class,
+ () -> ops.dropDb("analytics", false, false));
+ Assertions.assertDoesNotThrow(() -> ops.dropTable(table, true));
+ Assertions.assertThrows(DdlException.class,
+ () -> ops.dropTable(table, false));
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(catalog, Mockito.never()).unregisterDatabase(Mockito.anyString());
+ Mockito.verify(catalog).retireAllDatabaseObjectsWithoutEngineInvalidation();
+ Mockito.verify(database).unregisterTableForReplay("local_table");
+ }
+
+ @Test
+ public void testDropDatabaseReturnsFalseForIfExistsNoOp() throws DdlException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ LanceMetadataOps ops = new LanceMetadataOps(catalog);
+
+ try {
+ Assertions.assertFalse(ops.dropDb("missing_db", true, false));
+ Assertions.assertThrows(DdlException.class,
+ () -> ops.dropDb("missing_db", false, false));
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(namespace, Mockito.never()).namespaceExists(Mockito.any());
+ Mockito.verify(namespace, Mockito.never()).dropNamespace(Mockito.any());
+ Mockito.verify(catalog, Mockito.never()).unregisterDatabase(Mockito.anyString());
+ Mockito.verify(catalog).retireAllDatabaseObjectsWithoutEngineInvalidation();
+ }
+
+ @Test
+ public void testDropDatabaseUsesRemoteDatabaseName() throws DdlException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.doReturn(database).when(catalog).getDbNullable("sales_db");
+ Mockito.doReturn(Optional.of(database)).when(catalog).getDbForReplay("sales_db");
+ Mockito.when(database.getFullName()).thenReturn("sales_db");
+ Mockito.when(database.getRemoteName()).thenReturn("Sales");
+ LanceMetadataOps ops = new LanceMetadataOps(catalog);
+
+ try {
+ Assertions.assertTrue(ops.dropDb("sales_db", false, true));
+ } finally {
+ client.close();
+ }
+
+ ArgumentCaptor<DropNamespaceRequest> request = ArgumentCaptor.forClass(DropNamespaceRequest.class);
+ Mockito.verify(namespace).dropNamespace(request.capture());
+ Assertions.assertEquals(Arrays.asList("tenant", "Sales"), request.getValue().getId());
+ Assertions.assertEquals("Cascade", request.getValue().getBehavior());
+ Mockito.verify(catalog).unregisterDatabase("sales_db");
+ }
+
+ @Test
+ public void testFilesystemDropDatabaseAllowsEmptyDropButRejectsForce() throws DdlException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ Mockito.when(catalog.getLanceCatalogType()).thenReturn(LanceExternalCatalog.LANCE_FILESYSTEM);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.doReturn(database).when(catalog).getDbNullable("sales_db");
+ Mockito.doReturn(Optional.of(database)).when(catalog).getDbForReplay("sales_db");
+ Mockito.when(database.getFullName()).thenReturn("sales_db");
+ Mockito.when(database.getRemoteName()).thenReturn("Sales");
+ LanceMetadataOps ops = new LanceMetadataOps(catalog);
+
+ try {
+ Assertions.assertTrue(ops.dropDb("sales_db", false, false));
+ DdlException exception = Assertions.assertThrows(DdlException.class,
+ () -> ops.dropDb("sales_db", false, true));
+ Assertions.assertTrue(exception.getMessage().contains("DROP DATABASE FORCE"));
+ } finally {
+ client.close();
+ }
+
+ ArgumentCaptor<DropNamespaceRequest> request = ArgumentCaptor.forClass(DropNamespaceRequest.class);
+ Mockito.verify(namespace).dropNamespace(request.capture());
+ Assertions.assertEquals(Arrays.asList("tenant", "Sales"), request.getValue().getId());
+ Assertions.assertEquals("Restrict", request.getValue().getBehavior());
+ Mockito.verify(catalog).unregisterDatabase("sales_db");
+ }
+
+ @Test
+ public void testAfterDropDatabaseUsesCanonicalLocalName() {
+ LanceExternalCatalog catalog = Mockito.mock(LanceExternalCatalog.class);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.doReturn(Optional.of(database)).when(catalog).getDbForReplay("sales");
+ Mockito.when(database.getFullName()).thenReturn("Sales");
+
+ new LanceMetadataOps(catalog).afterDropDb("sales");
+
+ Mockito.verify(catalog).unregisterDatabase("Sales");
+ Mockito.verify(catalog, Mockito.never()).retireAllDatabaseObjectsWithoutEngineInvalidation();
+ }
+
+ @Test
+ public void testAfterDropDatabaseRetiresAllObjectsWhenCanonicalNameIsUnavailable() {
+ LanceExternalCatalog catalog = Mockito.mock(LanceExternalCatalog.class);
+ Mockito.doReturn(Optional.empty()).when(catalog).getDbForReplay("sales");
+
+ new LanceMetadataOps(catalog).afterDropDb("sales");
+
+ Mockito.verify(catalog).unregisterDatabase("sales");
+ Mockito.verify(catalog).retireAllDatabaseObjectsWithoutEngineInvalidation();
+ }
+
+ @Test
+ public void testDropAndRenameInvalidateTableAccessCacheEvenWhenReplayCacheMisses() throws DdlException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ Mockito.when(catalog.getDbForReplay("local_db")).thenReturn(Optional.empty());
+ ExternalTable table = table("local_db", "local_table", "analytics", "events");
+ LanceMetadataOps ops = new LanceMetadataOps(catalog);
+
+ try {
+ ops.dropTable(table, false);
+ ops.afterRenameTable("local_db", "local_table", "renamed_events");
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(catalog, Mockito.times(2)).invalidateTableAccessCache();
+ }
+
+ @Test
+ public void testSuccessfulNamespaceAndTableMutationsRefreshLocalNames() throws UserException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ Mockito.doThrow(new NamespaceNotFoundException("missing")).doReturn(null)
+ .when(namespace).namespaceExists(Mockito.any());
+ Mockito.doThrow(new TableNotFoundException("missing"))
+ .when(namespace).tableExists(Mockito.any());
+ LanceCatalogClient client = newClient(namespace, new RootAllocator(1024 * 1024));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.doReturn(database).when(catalog).getDbNullable("local_db");
+ Mockito.doReturn(null, database).when(catalog).getDbNullable("analytics");
+ Mockito.when(catalog.getDbForReplay("local_db")).thenReturn(Optional.of(database));
+ Mockito.when(catalog.getDbForReplay("analytics")).thenReturn(Optional.of(database));
+ Mockito.when(database.getFullName()).thenReturn("analytics");
+ Mockito.when(database.getRemoteName()).thenReturn("analytics");
+ Mockito.when(database.getTableNullable("events")).thenReturn(null);
+ Mockito.when(database.unregisterTableForReplay("events")).thenReturn(true);
+ ExternalTable table = table("local_db", "events", "analytics", "events");
+ LanceMetadataOps ops = new LanceMetadataOps(catalog);
+
+ try {
+ Assertions.assertFalse(ops.createDb("analytics", false, Collections.emptyMap()));
+ Assertions.assertFalse(ops.createTable(createTableInfo(false)));
+ ops.dropTable(table, false);
+ ops.dropDb("analytics", false, false);
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(catalog).resetMetaCacheNames();
+ Mockito.verify(catalog).unregisterDatabase("analytics");
+ Mockito.verify(database).resetMetaCacheNames();
+ Mockito.verify(database).unregisterTableForReplay("events");
+ }
+
+ @Test
+ public void testFilesystemRenameTableIsRejected() {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+
+ try {
+ DdlException exception = Assertions.assertThrows(DdlException.class,
+ () -> new LanceMetadataOps(catalog).renameTable("local_db", "events", "renamed_events"));
+ Assertions.assertTrue(exception.getMessage().contains("not supported"));
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(namespace, Mockito.never()).renameTable(Mockito.any());
+ Mockito.verify(catalog, Mockito.never()).invalidateTableAccessCache();
+ }
+
+ @Test
+ public void testRestRenameTableCallsNamespaceAndInvalidatesCache() throws DdlException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ Mockito.when(catalog.isRestCatalogConfigured()).thenReturn(true);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.when(catalog.getDbForReplay("analytics")).thenReturn(Optional.of(database));
+
+ try {
+ new LanceMetadataOps(catalog).renameTable("analytics", "events", "renamed_events");
+ } finally {
+ client.close();
+ }
+
+ ArgumentCaptor<RenameTableRequest> request = ArgumentCaptor.forClass(RenameTableRequest.class);
+ Mockito.verify(namespace).renameTable(request.capture());
+ Assertions.assertEquals(Arrays.asList("tenant", "analytics", "events"),
+ request.getValue().getId());
+ Assertions.assertEquals("renamed_events", request.getValue().getNewTableName());
+ Mockito.verify(catalog).invalidateTableAccessCache();
+ Mockito.verify(database).unregisterTable("events");
+ Mockito.verify(database).resetMetaCacheNames();
+ }
+
+ @Test
+ public void testFailedColumnMutationDoesNotRefreshCache() {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ Mockito.doThrow(new IllegalStateException("mutation failed"))
+ .when(namespace).alterTableAddColumns(Mockito.any());
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ ExternalTable table = table("local_db", "local_table", "analytics", "events");
+ Mockito.when(table.getFullSchema()).thenReturn(
+ Collections.singletonList(new Column("id", Type.INT, false)));
+
+ try {
+ Assertions.assertThrows(DdlException.class, () -> new LanceMetadataOps(catalog).addColumn(
+ table, new Column("score", Type.BIGINT, true), null, 123L));
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(catalog, Mockito.never()).getDbForReplay(Mockito.anyString());
+ }
+
+ @Test
+ public void testModifyColumnSplitsTypeAndNullabilityChanges() throws UserException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ ExternalTable table = table("local_db", "local_table", "analytics", "events");
+ Column currentColumn = new Column("score", Type.INT, true);
+ Mockito.when(table.getColumn("score")).thenReturn(currentColumn);
+
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.when(catalog.getDbForReplay("local_db")).thenReturn(Optional.of(database));
+ Mockito.doReturn(Optional.of(table)).when(database).getTableForReplay("local_table");
+ RefreshManager refreshManager = Mockito.mock(RefreshManager.class);
+ Env env = Mockito.mock(Env.class);
+ Mockito.when(env.getRefreshManager()).thenReturn(refreshManager);
+
+ Column modifiedColumn = new Column("score", Type.BIGINT, false);
+ modifiedColumn.setNullableSpecified(true);
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+ new LanceMetadataOps(catalog).modifyColumn(table, modifiedColumn, null, 123L);
Review Comment:
[P2] Exercise a supported type and nullability combination
This fixture changes nullable INT to BIGINT NOT NULL, but `LanceMetadataOps.modifyColumn` explicitly rejects a simultaneous type change and nullability tightening before either namespace call. Thus this success test throws immediately; `testModifyColumnRefreshesCacheWhenNullabilityChangeFails` repeats the input and receives the earlier UserException rather than its expected DdlException. The new regression's `MODIFY score BIGINT NOT NULL` has the same problem. Use a supported relaxation for the two-request unit scenarios, test the tightening rejection separately, and split the regression's type change and NOT NULL change into two statements.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceNamespaceClient.java:
##########
@@ -213,6 +314,102 @@ boolean tableExists(String dbName, String tblName) {
}
}
+ void createTable(String dbName, String tableName, Map<String, String> properties,
+ byte[] arrowStream) {
+ try {
+ CreateTableRequest request = new CreateTableRequest()
+ .id(buildTableId(dbName, tableName))
+ .mode("Create")
+ .properties(properties == null ? Collections.emptyMap() : properties)
+ .storageOptions(namespaceStorageOptions);
+ synchronized (namespaceLock) {
+ namespace.createTable(request, arrowStream);
+ }
+ } catch (DdlException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ void dropTable(String dbName, String tableName) {
+ try {
+ DropTableRequest request = new DropTableRequest().id(buildTableId(dbName, tableName));
+ executeTableMutation(dbName, tableName, () -> namespace.dropTable(request));
+ } catch (DdlException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ void addColumns(String dbName, String tableName, List<AddColumnsEntry> columns) {
+ try {
+ AlterTableAddColumnsRequest request = new AlterTableAddColumnsRequest()
+ .id(buildTableId(dbName, tableName))
+ .newColumns(columns);
+ executeTableMutation(dbName, tableName, () -> namespace.alterTableAddColumns(request));
+ } catch (DdlException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ void alterColumns(String dbName, String tableName, List<AlterColumnsEntry> alterations) {
+ try {
+ AlterTableAlterColumnsRequest request = new AlterTableAlterColumnsRequest()
+ .id(buildTableId(dbName, tableName))
+ .alterations(alterations);
+ executeTableMutation(dbName, tableName, () -> namespace.alterTableAlterColumns(request));
+ } catch (DdlException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ void dropColumns(String dbName, String tableName, List<String> columns) {
+ try {
+ AlterTableDropColumnsRequest request = new AlterTableDropColumnsRequest()
+ .id(buildTableId(dbName, tableName))
+ .columns(columns);
+ executeTableMutation(dbName, tableName, () -> namespace.alterTableDropColumns(request));
+ } catch (DdlException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void executeTableMutation(String dbName, String tableName, Runnable mutation)
+ throws DdlException {
+ List<String> namespaceId = buildNamespaceId(dbName);
+ List<String> tableId = new ArrayList<>(namespaceId);
+ tableId.add(tableName);
+ synchronized (namespaceLock) {
+ try {
+ mutation.run();
+ return;
+ } catch (TableNotFoundException originalException) {
+ if (!LANCE_FILESYSTEM.equals(catalogType) || !namespaceId.isEmpty()) {
+ throw originalException;
+ }
+ try {
+ namespace.tableExists(new TableExistsRequest().id(tableId));
+ } catch (TableNotFoundException | NamespaceNotFoundException e) {
+ throw originalException;
+ }
+ String expectedLocation = tableName + ".lance";
+ try {
+ // DirectoryNamespace only accepts locations relative to its warehouse root.
+ namespace.registerTable(new RegisterTableRequest()
+ .id(tableId)
+ .location(expectedLocation)
+ .mode("Create"));
+ } catch (TableAlreadyExistsException e) {
+ String registeredLocation = describeTable(tableId).getLocation();
+ if (!expectedLocation.equals(registeredLocation)) {
Review Comment:
[P2] Compare registration locations in the same coordinate system
When another client registers the same discovered filesystem-root table between the existence probe and `registerTable`, the pinned Lance 12.0.0 DirectoryNamespace returns a full warehouse URI from `describeTable().location` (for example `s3://warehouse/lance_ddl/events.lance`), while `expectedLocation` is the relative registration path `events.lance`. This equality check therefore rejects the identical dataset and never retries the mutation. Resolve the expected path against the warehouse before comparing full locations, and make the concurrency test return the SDK's full URI; retain rejection for genuinely different datasets.
##########
fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceMetadataOpsTest.java:
##########
@@ -0,0 +1,796 @@
+// 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.ColumnPosition;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.RefreshManager;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.UserException;
+import org.apache.doris.datasource.ExternalDatabase;
+import org.apache.doris.datasource.ExternalTable;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateTableInfo;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.lance.Session;
+import org.lance.namespace.LanceNamespace;
+import org.lance.namespace.errors.NamespaceAlreadyExistsException;
+import org.lance.namespace.errors.NamespaceNotFoundException;
+import org.lance.namespace.errors.TableAlreadyExistsException;
+import org.lance.namespace.errors.TableNotFoundException;
+import org.lance.namespace.model.AlterTableAddColumnsRequest;
+import org.lance.namespace.model.AlterTableAlterColumnsRequest;
+import org.lance.namespace.model.CreateNamespaceRequest;
+import org.lance.namespace.model.CreateTableRequest;
+import org.lance.namespace.model.DropNamespaceRequest;
+import org.lance.namespace.model.RenameTableRequest;
+import org.mockito.ArgumentCaptor;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Optional;
+
+public class LanceMetadataOpsTest {
+ @Test
+ public void testAddColumnValidation() throws UserException {
+ Column nullable = new Column("score", Type.INT, true);
+ Assertions.assertDoesNotThrow(
+ () -> LanceMetadataOps.validateAddColumn(nullable, null));
+
+ Column required = new Column("required", Type.INT, false);
+ assertRejected(() -> LanceMetadataOps.validateAddColumn(required, null),
+ "only supports nullable columns");
+
+ Column defaulted = new Column("defaulted", Type.INT, false, null, true, "1", "");
+ assertRejected(() -> LanceMetadataOps.validateAddColumn(defaulted, null),
+ "does not support default values");
+
+ Column commented = new Column("commented", Type.INT, true, "comment");
+ commented.setCommentSpecified(true);
+ assertRejected(() -> LanceMetadataOps.validateAddColumn(commented, null),
+ "does not support column comments");
+
+ assertRejected(() -> LanceMetadataOps.validateAddColumn(
+ nullable, new ColumnPosition("id")),
+ "does not support column positions");
+ }
+
+ @Test
+ public void testModifyColumnValidation() throws UserException {
+ Column column = new Column("score", Type.BIGINT, true);
+ column.setNullableSpecified(true);
+ Assertions.assertDoesNotThrow(
+ () -> LanceMetadataOps.validateModifyColumn(column, null));
+
+ column.setCommentSpecified(true);
+ assertRejected(() -> LanceMetadataOps.validateModifyColumn(column, null),
+ "does not support column comments");
+ }
+
+ @Test
+ public void testCreateDatabaseIfNotExistsSkipsExistingDatabase() throws DdlException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+
+ try {
+ Assertions.assertTrue(new LanceMetadataOps(catalog).createDb(
+ "analytics", true, Collections.singletonMap("owner", "doris")));
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(namespace, Mockito.never()).createNamespace(Mockito.any());
+ Mockito.verify(catalog).resetMetaCacheNames();
+ }
+
+ @Test
+ public void testCreateRootDatabaseIsRejected() {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ LanceMetadataOps ops = new LanceMetadataOps(catalog);
+
+ try {
+ for (boolean ifNotExists : Arrays.asList(false, true)) {
+ DdlException exception = Assertions.assertThrows(DdlException.class,
+ () -> ops.createDb("default", ifNotExists, Collections.emptyMap()));
+ Assertions.assertTrue(exception.getMessage().contains("root database"));
+ }
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(namespace, Mockito.never()).namespaceExists(Mockito.any());
+ Mockito.verify(namespace, Mockito.never()).createNamespace(Mockito.any());
+ Mockito.verify(catalog, Mockito.never()).resetMetaCacheNames();
+ }
+
+ @Test
+ public void testCreateDatabaseRejectsMappedLocalNameConflict() throws DdlException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.doReturn(database).when(catalog).getDbNullable("sales_db");
+ Mockito.when(database.getRemoteName()).thenReturn("Sales");
+ LanceMetadataOps ops = new LanceMetadataOps(catalog);
+
+ try {
+ Assertions.assertTrue(ops.createDb("sales_db", true, Collections.emptyMap()));
+ DdlException exception = Assertions.assertThrows(DdlException.class,
+ () -> ops.createDb("sales_db", false, Collections.emptyMap()));
+ Assertions.assertTrue(exception.getMessage().contains("exist"));
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(namespace, Mockito.never()).namespaceExists(Mockito.any());
+ Mockito.verify(namespace, Mockito.never()).createNamespace(Mockito.any());
+ }
+
+ @Test
+ public void testCreateDatabaseHandlesConcurrentCreate() throws DdlException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ Mockito.doThrow(new NamespaceNotFoundException("missing"))
+ .when(namespace).namespaceExists(Mockito.any());
+ Mockito.doThrow(new NamespaceAlreadyExistsException("created concurrently"))
+ .when(namespace).createNamespace(Mockito.any());
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ LanceMetadataOps ops = new LanceMetadataOps(catalog);
+
+ try {
+ Assertions.assertTrue(ops.createDb("analytics", true, Collections.emptyMap()));
+ DdlException exception = Assertions.assertThrows(DdlException.class,
+ () -> ops.createDb("analytics", false, Collections.emptyMap()));
+ Assertions.assertTrue(exception.getMessage().contains("exist"));
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(namespace, Mockito.times(2)).createNamespace(Mockito.any());
+ Mockito.verify(catalog).resetMetaCacheNames();
+ }
+
+ @Test
+ public void testFilesystemCreateDatabaseCreatesChildNamespace() throws DdlException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ Mockito.doThrow(new NamespaceNotFoundException("missing"))
+ .when(namespace).namespaceExists(Mockito.any());
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ Mockito.when(catalog.getLanceCatalogType()).thenReturn(LanceExternalCatalog.LANCE_FILESYSTEM);
+
+ try {
+ Assertions.assertFalse(new LanceMetadataOps(catalog).createDb(
+ "analytics", false, Collections.singletonMap("owner", "doris")));
+ } finally {
+ client.close();
+ }
+
+ ArgumentCaptor<CreateNamespaceRequest> request =
+ ArgumentCaptor.forClass(CreateNamespaceRequest.class);
+ Mockito.verify(namespace).createNamespace(request.capture());
+ Assertions.assertEquals(Arrays.asList("tenant", "analytics"), request.getValue().getId());
+ Assertions.assertEquals("doris", request.getValue().getProperties().get("owner"));
+ Mockito.verify(catalog).resetMetaCacheNames();
+ }
+
+ @Test
+ public void testAfterCreateDatabaseRetiresStaleDatabaseObject() {
+ LanceExternalCatalog catalog = Mockito.mock(LanceExternalCatalog.class);
+
+ new LanceMetadataOps(catalog).afterCreateDb("analytics");
+
+ Mockito.verify(catalog).retireCachedDatabaseForNoOp("analytics");
+ Mockito.verify(catalog).resetMetaCacheNames();
+ }
+
+ @Test
+ public void testCreateTableIfNotExistsSkipsExistingRemoteTable() throws UserException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.doReturn(database).when(catalog).getDbNullable("local_db");
+ Mockito.when(catalog.getDbForReplay("local_db")).thenReturn(Optional.of(database));
+ Mockito.when(database.getRemoteName()).thenReturn("analytics");
+ Mockito.when(database.getTableNullable("events")).thenReturn(null);
+ CreateTableInfo createTableInfo = createTableInfo(true);
+
+ try {
+ Assertions.assertTrue(new LanceMetadataOps(catalog).createTable(createTableInfo));
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(namespace, Mockito.never())
+ .createTable(Mockito.any(), Mockito.any(byte[].class));
+ Mockito.verify(database).resetMetaCacheNames();
+ }
+
+ @Test
+ public void testFilesystemCreateTableInRootDatabaseWithParentNamespace() throws UserException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ Mockito.doThrow(new TableNotFoundException("missing"))
+ .when(namespace).tableExists(Mockito.any());
+ LanceCatalogClient client = newClient(namespace, new RootAllocator(1024 * 1024));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ Mockito.when(catalog.getLanceCatalogType()).thenReturn(LanceExternalCatalog.LANCE_FILESYSTEM);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.doReturn(database).when(catalog).getDbNullable("default");
+ Mockito.when(catalog.getDbForReplay("default")).thenReturn(Optional.of(database));
+ Mockito.when(database.getRemoteName()).thenReturn("default");
+ Mockito.when(database.getTableNullable("events")).thenReturn(null);
+
+ try {
+ Assertions.assertFalse(new LanceMetadataOps(catalog).createTable(
+ createTableInfo("default", "events", false)));
+ } finally {
+ client.close();
+ }
+
+ ArgumentCaptor<CreateTableRequest> request = ArgumentCaptor.forClass(CreateTableRequest.class);
+ Mockito.verify(namespace).createTable(request.capture(), Mockito.any(byte[].class));
+ Assertions.assertEquals(Arrays.asList("tenant", "events"), request.getValue().getId());
+ Mockito.verify(catalog).invalidateTableAccessCache();
+ Mockito.verify(database).unregisterTable("events");
+ Mockito.verify(database).resetMetaCacheNames();
+ }
+
+ @Test
+ public void testAfterCreateTableRetiresStaleTableObject() {
+ LanceExternalCatalog catalog = Mockito.mock(LanceExternalCatalog.class);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.when(catalog.getDbForReplay("analytics")).thenReturn(Optional.of(database));
+
+ new LanceMetadataOps(catalog).afterCreateTable("analytics", "events");
+
+ Mockito.verify(catalog).invalidateTableAccessCache();
+ Mockito.verify(database).unregisterTable("events");
+ Mockito.verify(database).resetMetaCacheNames();
+ }
+
+ @Test
+ public void testCreateTableHandlesConcurrentCreate() throws UserException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ Mockito.doThrow(new TableNotFoundException("missing"))
+ .when(namespace).tableExists(Mockito.any());
+ Mockito.doThrow(new TableAlreadyExistsException("created concurrently"))
+ .when(namespace).createTable(Mockito.any(), Mockito.any(byte[].class));
+ LanceCatalogClient client = newClient(namespace, new RootAllocator(1024 * 1024));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.doReturn(database).when(catalog).getDbNullable("local_db");
+ Mockito.when(catalog.getDbForReplay("local_db")).thenReturn(Optional.of(database));
+ Mockito.when(database.getRemoteName()).thenReturn("analytics");
+ Mockito.when(database.getTableNullable("events")).thenReturn(null);
+ LanceMetadataOps ops = new LanceMetadataOps(catalog);
+
+ try {
+ Assertions.assertTrue(ops.createTable(createTableInfo(true)));
+ DdlException exception = Assertions.assertThrows(DdlException.class,
+ () -> ops.createTable(createTableInfo(false)));
+ Assertions.assertTrue(exception.getMessage().contains("already exists"));
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(namespace, Mockito.times(2))
+ .createTable(Mockito.any(), Mockito.any(byte[].class));
+ Mockito.verify(database).resetMetaCacheNames();
+ }
+
+ @Test
+ public void testCreateTableIgnoresStaleLocalCacheWhenRemoteTableIsGone() throws UserException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ Mockito.doThrow(new TableNotFoundException("missing"))
+ .when(namespace).tableExists(Mockito.any());
+ LanceCatalogClient client = newClient(namespace, new RootAllocator(1024 * 1024));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.doReturn(database).when(catalog).getDbNullable("local_db");
+ Mockito.when(catalog.getDbForReplay("local_db")).thenReturn(Optional.of(database));
+ Mockito.when(database.getRemoteName()).thenReturn("analytics");
+ ExternalTable staleTable = table("local_db", "events", "analytics", "stale_events");
+ Mockito.when(database.getTableNullable("events"))
+ .thenReturn(staleTable, null);
+
+ try {
+ Assertions.assertFalse(new LanceMetadataOps(catalog).createTable(createTableInfo(true)));
+ } finally {
+ client.close();
+ }
+
+ ArgumentCaptor<CreateTableRequest> request = ArgumentCaptor.forClass(CreateTableRequest.class);
+ Mockito.verify(namespace).createTable(request.capture(), Mockito.any(byte[].class));
+ Assertions.assertEquals(Arrays.asList("tenant", "analytics", "events"),
+ request.getValue().getId());
+ Mockito.verify(database).unregisterTable("events");
+ Mockito.verify(database, Mockito.atLeastOnce()).resetMetaCacheNames();
+ Mockito.verify(catalog).invalidateTableAccessCache();
+ }
+
+ @Test
+ public void testCreateTableRejectsLocalNameConflictAfterRefresh() throws UserException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ Mockito.doAnswer(invocation -> {
+ if (((org.lance.namespace.model.TableExistsRequest) invocation.getArgument(0))
+ .getId().get(2).equals("events")) {
+ throw new TableNotFoundException("missing");
+ }
+ return null;
+ }).when(namespace).tableExists(Mockito.any());
+ LanceCatalogClient client = newClient(namespace, new RootAllocator(1024 * 1024));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.doReturn(database).when(catalog).getDbNullable("local_db");
+ Mockito.when(catalog.getDbForReplay("local_db")).thenReturn(Optional.of(database));
+ Mockito.when(database.getRemoteName()).thenReturn("analytics");
+ ExternalTable conflictingTable = table("local_db", "events", "analytics", "Events");
+ Mockito.when(database.getTableNullable("events")).thenReturn(conflictingTable);
+ LanceMetadataOps ops = new LanceMetadataOps(catalog);
+
+ try {
+ Assertions.assertTrue(ops.createTable(createTableInfo(true)));
+ DdlException exception = Assertions.assertThrows(DdlException.class,
+ () -> ops.createTable(createTableInfo(false)));
+ Assertions.assertTrue(exception.getMessage().contains("already exists"));
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(namespace, Mockito.never()).createTable(Mockito.any(), Mockito.any(byte[].class));
+ Mockito.verify(database, Mockito.times(2)).resetMetaCacheNames();
+ }
+
+ @Test
+ public void testIfExistsHandlesConcurrentDropAndRefreshesLocalNames() throws DdlException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ Mockito.doThrow(new NamespaceNotFoundException("dropped concurrently"))
+ .when(namespace).dropNamespace(Mockito.any());
+ Mockito.doThrow(new TableNotFoundException("dropped concurrently"))
+ .when(namespace).dropTable(Mockito.any());
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.doReturn(database).when(catalog).getDbNullable("analytics");
+ Mockito.when(catalog.getDbForReplay("local_db")).thenReturn(Optional.of(database));
+ Mockito.when(database.getRemoteName()).thenReturn("analytics");
+ ExternalTable table = table("local_db", "local_table", "analytics", "events");
+ Mockito.when(database.unregisterTableForReplay("local_table")).thenReturn(true);
+ LanceMetadataOps ops = new LanceMetadataOps(catalog);
+
+ try {
+ Assertions.assertDoesNotThrow(() -> ops.dropDb("analytics", true, false));
+ Assertions.assertThrows(DdlException.class,
+ () -> ops.dropDb("analytics", false, false));
+ Assertions.assertDoesNotThrow(() -> ops.dropTable(table, true));
+ Assertions.assertThrows(DdlException.class,
+ () -> ops.dropTable(table, false));
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(catalog, Mockito.never()).unregisterDatabase(Mockito.anyString());
+ Mockito.verify(catalog).retireAllDatabaseObjectsWithoutEngineInvalidation();
+ Mockito.verify(database).unregisterTableForReplay("local_table");
+ }
+
+ @Test
+ public void testDropDatabaseReturnsFalseForIfExistsNoOp() throws DdlException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ LanceMetadataOps ops = new LanceMetadataOps(catalog);
+
+ try {
+ Assertions.assertFalse(ops.dropDb("missing_db", true, false));
+ Assertions.assertThrows(DdlException.class,
+ () -> ops.dropDb("missing_db", false, false));
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(namespace, Mockito.never()).namespaceExists(Mockito.any());
+ Mockito.verify(namespace, Mockito.never()).dropNamespace(Mockito.any());
+ Mockito.verify(catalog, Mockito.never()).unregisterDatabase(Mockito.anyString());
+ Mockito.verify(catalog).retireAllDatabaseObjectsWithoutEngineInvalidation();
+ }
+
+ @Test
+ public void testDropDatabaseUsesRemoteDatabaseName() throws DdlException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.doReturn(database).when(catalog).getDbNullable("sales_db");
+ Mockito.doReturn(Optional.of(database)).when(catalog).getDbForReplay("sales_db");
+ Mockito.when(database.getFullName()).thenReturn("sales_db");
+ Mockito.when(database.getRemoteName()).thenReturn("Sales");
+ LanceMetadataOps ops = new LanceMetadataOps(catalog);
+
+ try {
+ Assertions.assertTrue(ops.dropDb("sales_db", false, true));
+ } finally {
+ client.close();
+ }
+
+ ArgumentCaptor<DropNamespaceRequest> request = ArgumentCaptor.forClass(DropNamespaceRequest.class);
+ Mockito.verify(namespace).dropNamespace(request.capture());
+ Assertions.assertEquals(Arrays.asList("tenant", "Sales"), request.getValue().getId());
+ Assertions.assertEquals("Cascade", request.getValue().getBehavior());
+ Mockito.verify(catalog).unregisterDatabase("sales_db");
+ }
+
+ @Test
+ public void testFilesystemDropDatabaseAllowsEmptyDropButRejectsForce() throws DdlException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ Mockito.when(catalog.getLanceCatalogType()).thenReturn(LanceExternalCatalog.LANCE_FILESYSTEM);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.doReturn(database).when(catalog).getDbNullable("sales_db");
+ Mockito.doReturn(Optional.of(database)).when(catalog).getDbForReplay("sales_db");
+ Mockito.when(database.getFullName()).thenReturn("sales_db");
+ Mockito.when(database.getRemoteName()).thenReturn("Sales");
+ LanceMetadataOps ops = new LanceMetadataOps(catalog);
+
+ try {
+ Assertions.assertTrue(ops.dropDb("sales_db", false, false));
+ DdlException exception = Assertions.assertThrows(DdlException.class,
+ () -> ops.dropDb("sales_db", false, true));
+ Assertions.assertTrue(exception.getMessage().contains("DROP DATABASE FORCE"));
+ } finally {
+ client.close();
+ }
+
+ ArgumentCaptor<DropNamespaceRequest> request = ArgumentCaptor.forClass(DropNamespaceRequest.class);
+ Mockito.verify(namespace).dropNamespace(request.capture());
+ Assertions.assertEquals(Arrays.asList("tenant", "Sales"), request.getValue().getId());
+ Assertions.assertEquals("Restrict", request.getValue().getBehavior());
+ Mockito.verify(catalog).unregisterDatabase("sales_db");
+ }
+
+ @Test
+ public void testAfterDropDatabaseUsesCanonicalLocalName() {
+ LanceExternalCatalog catalog = Mockito.mock(LanceExternalCatalog.class);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.doReturn(Optional.of(database)).when(catalog).getDbForReplay("sales");
+ Mockito.when(database.getFullName()).thenReturn("Sales");
+
+ new LanceMetadataOps(catalog).afterDropDb("sales");
+
+ Mockito.verify(catalog).unregisterDatabase("Sales");
+ Mockito.verify(catalog, Mockito.never()).retireAllDatabaseObjectsWithoutEngineInvalidation();
+ }
+
+ @Test
+ public void testAfterDropDatabaseRetiresAllObjectsWhenCanonicalNameIsUnavailable() {
+ LanceExternalCatalog catalog = Mockito.mock(LanceExternalCatalog.class);
+ Mockito.doReturn(Optional.empty()).when(catalog).getDbForReplay("sales");
+
+ new LanceMetadataOps(catalog).afterDropDb("sales");
+
+ Mockito.verify(catalog).unregisterDatabase("sales");
+ Mockito.verify(catalog).retireAllDatabaseObjectsWithoutEngineInvalidation();
+ }
+
+ @Test
+ public void testDropAndRenameInvalidateTableAccessCacheEvenWhenReplayCacheMisses() throws DdlException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ Mockito.when(catalog.getDbForReplay("local_db")).thenReturn(Optional.empty());
+ ExternalTable table = table("local_db", "local_table", "analytics", "events");
+ LanceMetadataOps ops = new LanceMetadataOps(catalog);
+
+ try {
+ ops.dropTable(table, false);
+ ops.afterRenameTable("local_db", "local_table", "renamed_events");
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(catalog, Mockito.times(2)).invalidateTableAccessCache();
+ }
+
+ @Test
+ public void testSuccessfulNamespaceAndTableMutationsRefreshLocalNames() throws UserException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ Mockito.doThrow(new NamespaceNotFoundException("missing")).doReturn(null)
+ .when(namespace).namespaceExists(Mockito.any());
+ Mockito.doThrow(new TableNotFoundException("missing"))
+ .when(namespace).tableExists(Mockito.any());
+ LanceCatalogClient client = newClient(namespace, new RootAllocator(1024 * 1024));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.doReturn(database).when(catalog).getDbNullable("local_db");
+ Mockito.doReturn(null, database).when(catalog).getDbNullable("analytics");
+ Mockito.when(catalog.getDbForReplay("local_db")).thenReturn(Optional.of(database));
+ Mockito.when(catalog.getDbForReplay("analytics")).thenReturn(Optional.of(database));
+ Mockito.when(database.getFullName()).thenReturn("analytics");
+ Mockito.when(database.getRemoteName()).thenReturn("analytics");
+ Mockito.when(database.getTableNullable("events")).thenReturn(null);
+ Mockito.when(database.unregisterTableForReplay("events")).thenReturn(true);
+ ExternalTable table = table("local_db", "events", "analytics", "events");
+ LanceMetadataOps ops = new LanceMetadataOps(catalog);
+
+ try {
+ Assertions.assertFalse(ops.createDb("analytics", false, Collections.emptyMap()));
+ Assertions.assertFalse(ops.createTable(createTableInfo(false)));
+ ops.dropTable(table, false);
+ ops.dropDb("analytics", false, false);
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(catalog).resetMetaCacheNames();
+ Mockito.verify(catalog).unregisterDatabase("analytics");
+ Mockito.verify(database).resetMetaCacheNames();
+ Mockito.verify(database).unregisterTableForReplay("events");
+ }
+
+ @Test
+ public void testFilesystemRenameTableIsRejected() {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+
+ try {
+ DdlException exception = Assertions.assertThrows(DdlException.class,
+ () -> new LanceMetadataOps(catalog).renameTable("local_db", "events", "renamed_events"));
+ Assertions.assertTrue(exception.getMessage().contains("not supported"));
+ } finally {
+ client.close();
+ }
+
+ Mockito.verify(namespace, Mockito.never()).renameTable(Mockito.any());
+ Mockito.verify(catalog, Mockito.never()).invalidateTableAccessCache();
+ }
+
+ @Test
+ public void testRestRenameTableCallsNamespaceAndInvalidatesCache() throws DdlException {
+ LanceNamespace namespace = Mockito.mock(LanceNamespace.class);
+ LanceCatalogClient client = newClient(namespace, Mockito.mock(BufferAllocator.class));
+ LanceExternalCatalog catalog = catalogWithClient(client);
+ Mockito.when(catalog.isRestCatalogConfigured()).thenReturn(true);
+ ExternalDatabase<?> database = Mockito.mock(ExternalDatabase.class);
+ Mockito.when(catalog.getDbForReplay("analytics")).thenReturn(Optional.of(database));
+
+ try {
+ new LanceMetadataOps(catalog).renameTable("analytics", "events", "renamed_events");
Review Comment:
[P2] Stub the actual rename resolution path in the success fixture
This test stubs only `getDbForReplay`, but `renameTableImpl` first calls `catalog.getDbNullable` and then `db.getTableNullable`. The unstubbed catalog mock returns null, so the supposed success case throws before `namespace.renameTable`. After fixing resolution, its `unregisterTable` verification also needs to match the implementation's `unregisterTableForReplay`. Other new fixtures have stale invocation counts too: the mapped-database test forbids `namespaceExists` even though both calls check it, and the stale-table CREATE test expects one cleanup although both the stale branch and success hook perform it. Align these fixtures with the executed paths so the new unit suite can pass.
##########
regression-test/suites/external_table_p0/lance/test_lance_ddl.groovy:
##########
@@ -0,0 +1,131 @@
+// 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.
+
+suite("test_lance_ddl", "p0,external") {
+ String enabled = context.config.otherConfigs.get("enableIcebergTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable Lance DDL test because the Iceberg MinIO environment is disabled.")
+ return
+ }
+
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String catalogName = "test_lance_ddl"
+ String databaseName = "default"
+ String tableName = "events"
+
+ sql """DROP CATALOG IF EXISTS `${catalogName}`"""
+ try {
+ sql """
+ CREATE CATALOG `${catalogName}` PROPERTIES (
+ "type" = "lance",
+ "lance.catalog.type" = "filesystem",
+ "warehouse" = "s3://warehouse/lance_ddl",
+ "s3.endpoint" = "http://${externalEnvIp}:${minioPort}",
+ "s3.access_key" = "admin",
+ "s3.secret_key" = "password",
+ "s3.region" = "us-east-1",
+ "use_path_style" = "true"
+ )
+ """
+
+ sql """DROP TABLE IF EXISTS `${catalogName}`.`${databaseName}`.`${tableName}`"""
+ sql """
+ CREATE TABLE `${catalogName}`.`${databaseName}`.`${tableName}` (
+ id INT NOT NULL COMMENT 'identifier',
+ name STRING NULL,
+ tags ARRAY<STRING> NULL,
+ amount DECIMAL(10, 2) NULL
+ ) ENGINE=LANCE
+ COMMENT 'Lance DDL regression table'
+ PROPERTIES ("owner" = "doris")
+ """
+
+ def columns = sql """DESC `${catalogName}`.`${databaseName}`.`${tableName}`"""
+ assertEquals(["id", "name", "tags", "amount"], columns.collect { it[0] })
+ assertEquals("int", columns[0][1].toString().toLowerCase())
+ assertEquals("array<text>", columns[2][1].toString().toLowerCase())
+
+ sql """
+ ALTER TABLE `${catalogName}`.`${databaseName}`.`${tableName}`
+ ADD COLUMN score INT NULL
+ """
+ sql """
+ ALTER TABLE `${catalogName}`.`${databaseName}`.`${tableName}`
+ ADD COLUMN (
+ active BOOLEAN NULL,
+ note STRING NULL
+ )
+ """
+ sql """
+ ALTER TABLE `${catalogName}`.`${databaseName}`.`${tableName}`
+ MODIFY COLUMN score BIGINT NOT NULL
+ """
+ sql """
+ ALTER TABLE `${catalogName}`.`${databaseName}`.`${tableName}`
+ RENAME COLUMN score TO ranking
+ """
+ sql """
+ ALTER TABLE `${catalogName}`.`${databaseName}`.`${tableName}`
+ DROP COLUMN note
+ """
+
+ columns = sql """DESC `${catalogName}`.`${databaseName}`.`${tableName}`"""
+ assertEquals(
+ ["id", "name", "tags", "amount", "ranking", "active"],
+ columns.collect { it[0] })
+ assertEquals("bigint", columns[4][1].toString().toLowerCase())
+ assertEquals("NO", columns[4][2])
+
+ test {
+ sql """
+ ALTER TABLE `${catalogName}`.`${databaseName}`.`${tableName}`
+ ADD COLUMN positioned INT NULL FIRST
+ """
+ exception "does not support column positions"
+ }
+ test {
+ sql """
+ ALTER TABLE `${catalogName}`.`${databaseName}`.`${tableName}`
+ ADD COLUMN defaulted INT NULL DEFAULT '1'
+ """
+ exception "does not support default values"
+ }
+ test {
+ sql """
+ ALTER TABLE `${catalogName}`.`${databaseName}`.`${tableName}`
+ MODIFY COLUMN ranking BIGINT NOT NULL COMMENT 'rank'
+ """
+ exception "does not support column comments"
+ }
+
+ String showCreate = sql("""SHOW CREATE TABLE `${catalogName}`.`${databaseName}`.`${tableName}`""")[0][1]
+ assertTrue(showCreate.contains("ENGINE=LANCE"))
+ assertTrue(showCreate.contains("COMMENT \"Lance DDL regression table\""))
+ assertTrue(showCreate.contains("\"owner\" = \"doris\""))
+ assertFalse(showCreate.contains("password"))
+
+ test {
+ sql """ALTER TABLE `${catalogName}`.`${databaseName}`.`${tableName}` RENAME `renamed_events`"""
+ exception "Lance table rename is not supported"
Review Comment:
[P2] Assert the filesystem rename error that is actually thrown
Once the earlier MODIFY fixture is repaired, this negative case still fails: `LanceMetadataOps.renameTableImpl` throws `RENAME TABLE is not supported for Lance filesystem catalogs`, which does not contain the expected `Lance table rename is not supported`. The regression framework checks the exception with case-sensitive `String.contains`. Use a substring of the actual error so the intended rejection passes the assertion.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Feature/lance namespace table ddl branch 4.2 [doris]
Posted by "zhangstar333 (via GitHub)" <gi...@apache.org>.
zhangstar333 commented on PR #69384:
URL: https://github.com/apache/doris/pull/69384#issuecomment-6358399479
run buildall
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Feature/lance namespace table ddl branch 4.2 [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #69384:
URL: https://github.com/apache/doris/pull/69384#issuecomment-6368438420
# FE Regression Coverage Report
Increment line coverage ` 15.60% (129/827)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/69384_fc50b302caded9c86c056202da0f1d07fe7ac7c7_merge_fe/increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/69384_fc50b302caded9c86c056202da0f1d07fe7ac7c7_merge_fe/report/index.html)
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Feature/lance namespace table ddl branch 4.2 [doris]
Posted by "github-actions[bot] (via GitHub) via commits" <co...@doris.apache.org>.
github-actions[bot] commented on PR #69384:
URL: https://github.com/apache/doris/pull/69384#issuecomment-6568499817
PR approved by at least one committer and no changes requested.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Feature/lance namespace table ddl branch 4.2 [doris]
Posted by "zhangstar333 (via GitHub) via commits" <co...@doris.apache.org>.
zhangstar333 merged PR #69384:
URL: https://github.com/apache/doris/pull/69384
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Feature/lance namespace table ddl branch 4.2 [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #69384:
URL: https://github.com/apache/doris/pull/69384#issuecomment-6337265392
# FE Regression Coverage Report
Increment line coverage ` 15.18% (126/830)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/69384_7f21393defc6ef9a65f8449f24d9c14da0d2d378_merge_fe/increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/69384_7f21393defc6ef9a65f8449f24d9c14da0d2d378_merge_fe/report/index.html)
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Feature/lance namespace table ddl branch 4.2 [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #69384:
URL: https://github.com/apache/doris/pull/69384#issuecomment-6311470651
Thank you for your contribution to Apache Doris.
Don't know what should be done next? See [How to process your PR](https://doris.apache.org/community/how-to-contribute/review-and-ci).
Please clearly describe your PR:
1. What problem was fixed (it's best to include specific error reporting information). How it was fixed.
2. Which behaviors were modified. What was the previous behavior, what is it now, why was it modified, and what possible impacts might there be.
3. What features were added. Why was this function added?
4. Which code was refactored and why was this part of the code refactored?
5. Which functions were optimized and what is the difference before and after the optimization?
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Feature/lance namespace table ddl branch 4.2 [doris]
Posted by "zhangstar333 (via GitHub)" <gi...@apache.org>.
zhangstar333 commented on PR #69384:
URL: https://github.com/apache/doris/pull/69384#issuecomment-6361436503
run buildall
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Feature/lance namespace table ddl branch 4.2 [doris]
Posted by "Gabriel39 (via GitHub)" <gi...@apache.org>.
Gabriel39 commented on PR #69384:
URL: https://github.com/apache/doris/pull/69384#issuecomment-6363941794
run buildall
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] Feature/lance namespace table ddl branch 4.2 [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #69384:
URL: https://github.com/apache/doris/pull/69384#discussion_r4359334975
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterTableCommand.java:
##########
@@ -270,6 +270,21 @@ static void checkColumnOperationsSupported(TableIf table, List<AlterTableOp> alt
}
return;
}
+ if (table instanceof LanceExternalTable) {
Review Comment:
[P1] Route Lance tables through the external ALTER dispatcher
`ALTER TABLE lance.analytics.events ADD COLUMN score INT NULL` passes these new checks, but `AlterTableCommand.run()` then calls `Env.alterTable()` -> `Alter.processAlterTable()`. That switch has no `LANCE_EXTERNAL_TABLE` case and throws `Do not support alter LANCE_EXTERNAL_TABLE table[events]` instead of invoking `processAlterTableForExternalTable`. Consequently every new ADD/MODIFY/RENAME/DROP COLUMN operation, and REST RENAME TABLE, fails before reaching `LanceMetadataOps`; the new regression stops at its first ADD. Add Lance to the external-table dispatch and cover the full command execution path.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceManifestPaths.java:
##########
@@ -108,15 +111,72 @@ static String objectStorePath(String location) {
throw new IllegalArgumentException("Lance namespace returned no table location");
}
if (!URL_SCHEME.matcher(location).find()) {
- return StringUtils.strip(location, "/");
+ return normalizeObjectPath(StringUtils.strip(location, "/"));
}
String path = StringUtils.substringBefore(StringUtils.substringBefore(location, "?"), "#");
path = path.substring(path.indexOf(':') + 1);
if (path.startsWith("//")) {
int slash = path.indexOf('/', 2);
path = slash < 0 ? "" : path.substring(slash);
}
- return StringUtils.strip(percentDecode(path), "/");
+ return normalizeObjectPath(StringUtils.strip(percentDecode(path), "/"));
+ }
+
+ private static String normalizeObjectPath(String path) {
+ if (path.isEmpty()) {
+ return "";
+ }
+ ArrayDeque<String> segments = new ArrayDeque<>();
+ for (String segment : path.split("/+")) {
Review Comment:
[P2] Preserve URI segments until dot-segment resolution
Collapsing `/+` before resolving `..` differs from the pinned SDK's [URL parser](https://github.com/servo/rust-url/blob/v2.5.8/url/src/parser.rs). For a managed response with `location=s3://b/items.lance` and `table_uri=s3://b/warehouse//../items.lance`, Doris normalizes both to `s3://b/items.lance`, whereas Lance resolves the latter to `s3://b/warehouse/items.lance`: `..` consumes the empty segment. The shared `objectStorePath()` also accepts a namespace version-3 manifest at `items.lance/_versions/3.manifest`. Since `openDataset()` passes the original URI to Lance, a dataset at `warehouse/items.lance` with version 3 is read instead of the namespace's dataset. Match the SDK's segment/decoding order, or reject locators whose equivalence cannot be established, rather than weakening both identity checks.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceMetadataOps.java:
##########
@@ -0,0 +1,668 @@
+// 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.ColumnPosition;
+import org.apache.doris.catalog.ArrayType;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.MapType;
+import org.apache.doris.catalog.PrimitiveType;
+import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.catalog.StructField;
+import org.apache.doris.catalog.StructType;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.common.UserException;
+import org.apache.doris.datasource.ExternalDatabase;
+import org.apache.doris.datasource.ExternalObjectLog;
+import org.apache.doris.datasource.ExternalTable;
+import org.apache.doris.datasource.lance.metadata.LanceTypeConverter;
+import org.apache.doris.datasource.operations.ExternalMetadataOps;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateOrReplaceBranchInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateOrReplaceTagInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateTableInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.DropBranchInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.DropTagInfo;
+
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.lance.namespace.errors.NamespaceAlreadyExistsException;
+import org.lance.namespace.errors.NamespaceNotFoundException;
+import org.lance.namespace.errors.TableAlreadyExistsException;
+import org.lance.namespace.errors.TableNotFoundException;
+import org.lance.namespace.model.AddColumnsEntry;
+import org.lance.namespace.model.AlterColumnsEntry;
+
+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.Set;
+import java.util.TreeSet;
+
+/** Doris external metadata operations backed by the Lance Namespace API. */
+public class LanceMetadataOps implements ExternalMetadataOps {
+ private static final Logger LOG = LogManager.getLogger(LanceMetadataOps.class);
+ private static final String TABLE_COMMENT_PROPERTY = "comment";
+
+ private final LanceExternalCatalog catalog;
+
+ public LanceMetadataOps(LanceExternalCatalog catalog) {
+ this.catalog = catalog;
+ }
+
+ @Override
+ public boolean createDbImpl(String dbName, boolean ifNotExists, Map<String, String> properties)
+ throws DdlException {
+ return execute("Failed to create Lance database " + dbName, client -> {
+ if (client.isRootDatabase(dbName)) {
+ throw new DdlException("Cannot create the configured Lance root database: " + dbName);
+ }
+ ExternalDatabase<?> cachedDb = catalog.getDbNullable(dbName);
+ if (cachedDb != null) {
+ if (!cachedDb.getRemoteName().equals(dbName)) {
+ if (ifNotExists) {
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_CREATE_EXISTS, dbName);
+ }
+ if (client.databaseExists(cachedDb.getRemoteName())) {
+ if (ifNotExists) {
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_CREATE_EXISTS, dbName);
+ }
+ catalog.unregisterDatabase(cachedDb.getFullName());
+ catalog.resetMetaCacheNames();
+ }
+ if (client.databaseExists(dbName)) {
+ if (ifNotExists) {
+ catalog.resetMetaCacheNames();
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_CREATE_EXISTS, dbName);
+ }
+ try {
+ client.createDatabase(dbName, new HashMap<>(
+ Optional.ofNullable(properties).orElse(Collections.emptyMap())));
+ return false;
+ } catch (NamespaceAlreadyExistsException e) {
+ if (ifNotExists) {
+ catalog.resetMetaCacheNames();
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_CREATE_EXISTS, dbName);
+ throw new IllegalStateException("unreachable");
+ }
+ });
+ }
+
+ @Override
+ public void afterCreateDb(String dbName) {
+ catalog.retireCachedDatabaseForNoOp(dbName);
+ catalog.resetMetaCacheNames();
+ }
+
+ @Override
+ public boolean dropDbImpl(String dbName, boolean ifExists, boolean force) throws DdlException {
+ if (force && LanceExternalCatalog.LANCE_FILESYSTEM.equals(catalog.getLanceCatalogType())) {
+ throw new DdlException(
+ "DROP DATABASE FORCE is not supported for Lance filesystem catalogs");
+ }
+ ExternalDatabase<?> db = catalog.getDbNullable(dbName);
+ return execute("Failed to drop Lance database " + dbName, client -> {
+ if (client.isRootDatabase(dbName)) {
+ throw new DdlException("Cannot drop the configured Lance root database: " + dbName);
+ }
+ if (db == null) {
+ if (ifExists) {
+ return false;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_DROP_EXISTS, dbName);
+ return false;
+ }
+ String remoteDbName = db.getRemoteName();
+ if (client.isRootDatabase(remoteDbName)) {
+ throw new DdlException("Cannot drop the configured Lance root database: " + dbName);
+ }
+ if (!client.databaseExists(remoteDbName)) {
+ if (ifExists) {
+ return false;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_DROP_EXISTS, dbName);
+ }
+ try {
+ client.dropDatabase(remoteDbName, ifExists, force);
+ } catch (NamespaceNotFoundException e) {
+ if (ifExists) {
+ return false;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_DROP_EXISTS, dbName);
+ }
+ return true;
+ });
+ }
+
+ @Override
+ public void afterDropDb(String dbName) {
+ Optional<ExternalDatabase<? extends ExternalTable>> db = catalog.getDbForReplay(dbName);
+ if (db.isPresent()) {
+ catalog.unregisterDatabase(db.get().getFullName());
Review Comment:
[P2] Retire descendant databases after a cascading namespace drop
On a REST catalog, warm the `parent` and `parent.child` database objects and the `SHOW DATABASES` names snapshot, then execute `DROP DATABASE catalog.parent FORCE`. The request uses `Cascade`, which [also deletes child namespaces](https://github.com/lance-format/lance-namespace/blob/v0.11.1/java/lance-namespace-apache-client/src/main/java/org/lance/namespace/model/DropNamespaceRequest.java), but this success hook unregisters only `parent` and returns. `listDatabaseNames()` exposes descendants as separate Doris databases, and `MetaCache.invalidate()` removes only the supplied name, so deleted `parent.child` remains listed and cached; replay on warm followers has the same gap. Retire the affected descendant identities and reset names after a successful cascade. This differs from the already-fixed no-op DROP names reconciliation.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceMetadataOps.java:
##########
@@ -0,0 +1,668 @@
+// 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.ColumnPosition;
+import org.apache.doris.catalog.ArrayType;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.MapType;
+import org.apache.doris.catalog.PrimitiveType;
+import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.catalog.StructField;
+import org.apache.doris.catalog.StructType;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.common.UserException;
+import org.apache.doris.datasource.ExternalDatabase;
+import org.apache.doris.datasource.ExternalObjectLog;
+import org.apache.doris.datasource.ExternalTable;
+import org.apache.doris.datasource.lance.metadata.LanceTypeConverter;
+import org.apache.doris.datasource.operations.ExternalMetadataOps;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateOrReplaceBranchInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateOrReplaceTagInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateTableInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.DropBranchInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.DropTagInfo;
+
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.lance.namespace.errors.NamespaceAlreadyExistsException;
+import org.lance.namespace.errors.NamespaceNotFoundException;
+import org.lance.namespace.errors.TableAlreadyExistsException;
+import org.lance.namespace.errors.TableNotFoundException;
+import org.lance.namespace.model.AddColumnsEntry;
+import org.lance.namespace.model.AlterColumnsEntry;
+
+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.Set;
+import java.util.TreeSet;
+
+/** Doris external metadata operations backed by the Lance Namespace API. */
+public class LanceMetadataOps implements ExternalMetadataOps {
+ private static final Logger LOG = LogManager.getLogger(LanceMetadataOps.class);
+ private static final String TABLE_COMMENT_PROPERTY = "comment";
+
+ private final LanceExternalCatalog catalog;
+
+ public LanceMetadataOps(LanceExternalCatalog catalog) {
+ this.catalog = catalog;
+ }
+
+ @Override
+ public boolean createDbImpl(String dbName, boolean ifNotExists, Map<String, String> properties)
+ throws DdlException {
+ return execute("Failed to create Lance database " + dbName, client -> {
+ if (client.isRootDatabase(dbName)) {
+ throw new DdlException("Cannot create the configured Lance root database: " + dbName);
+ }
+ ExternalDatabase<?> cachedDb = catalog.getDbNullable(dbName);
+ if (cachedDb != null) {
+ if (!cachedDb.getRemoteName().equals(dbName)) {
+ if (ifNotExists) {
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_CREATE_EXISTS, dbName);
+ }
+ if (client.databaseExists(cachedDb.getRemoteName())) {
+ if (ifNotExists) {
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_CREATE_EXISTS, dbName);
+ }
+ catalog.unregisterDatabase(cachedDb.getFullName());
+ catalog.resetMetaCacheNames();
+ }
+ if (client.databaseExists(dbName)) {
+ if (ifNotExists) {
+ catalog.resetMetaCacheNames();
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_CREATE_EXISTS, dbName);
+ }
+ try {
+ client.createDatabase(dbName, new HashMap<>(
+ Optional.ofNullable(properties).orElse(Collections.emptyMap())));
+ return false;
+ } catch (NamespaceAlreadyExistsException e) {
+ if (ifNotExists) {
+ catalog.resetMetaCacheNames();
+ return true;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_CREATE_EXISTS, dbName);
+ throw new IllegalStateException("unreachable");
+ }
+ });
+ }
+
+ @Override
+ public void afterCreateDb(String dbName) {
+ catalog.retireCachedDatabaseForNoOp(dbName);
+ catalog.resetMetaCacheNames();
+ }
+
+ @Override
+ public boolean dropDbImpl(String dbName, boolean ifExists, boolean force) throws DdlException {
+ if (force && LanceExternalCatalog.LANCE_FILESYSTEM.equals(catalog.getLanceCatalogType())) {
+ throw new DdlException(
+ "DROP DATABASE FORCE is not supported for Lance filesystem catalogs");
+ }
+ ExternalDatabase<?> db = catalog.getDbNullable(dbName);
+ return execute("Failed to drop Lance database " + dbName, client -> {
+ if (client.isRootDatabase(dbName)) {
+ throw new DdlException("Cannot drop the configured Lance root database: " + dbName);
+ }
+ if (db == null) {
+ if (ifExists) {
+ return false;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_DROP_EXISTS, dbName);
+ return false;
+ }
+ String remoteDbName = db.getRemoteName();
+ if (client.isRootDatabase(remoteDbName)) {
+ throw new DdlException("Cannot drop the configured Lance root database: " + dbName);
+ }
+ if (!client.databaseExists(remoteDbName)) {
+ if (ifExists) {
+ return false;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_DROP_EXISTS, dbName);
+ }
+ try {
+ client.dropDatabase(remoteDbName, ifExists, force);
+ } catch (NamespaceNotFoundException e) {
+ if (ifExists) {
+ return false;
+ }
+ ErrorReport.reportDdlException(ErrorCode.ERR_DB_DROP_EXISTS, dbName);
+ }
+ return true;
+ });
+ }
+
+ @Override
+ public void afterDropDb(String dbName) {
+ Optional<ExternalDatabase<? extends ExternalTable>> db = catalog.getDbForReplay(dbName);
+ if (db.isPresent()) {
+ catalog.unregisterDatabase(db.get().getFullName());
+ return;
+ }
+ catalog.unregisterDatabase(dbName);
+ catalog.retireAllDatabaseObjectsWithoutEngineInvalidation();
+ }
+
+ @Override
+ public void afterDropDbNoOp(String dbName) {
+ catalog.invalidateTableAccessCache();
+ catalog.retireAllDatabaseObjectsWithoutEngineInvalidation();
Review Comment:
[P2] Invalidate schema entries during no-op database reconciliation
After caching the schema of `analytics.events`, delete `analytics` externally and run `DROP DATABASE IF EXISTS lance.analytics`. This hook retires database objects with engine invalidation explicitly disabled and resets names/access, leaving the independent default schema cache intact. If `analytics.events` is recreated externally under the same names with a different schema, replacement objects use the same name-based `SchemaCacheKey`, so the leader's next `DESC` still returns the old columns until expiry or refresh. A follower replaying `afterDropDb()` performs engine invalidation instead. Invalidate the affected schema entries as part of no-op cleanup; the names-only fix requested in existing thread 4344273843 does not clear this separate cache.
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org