This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 8ce54eec191 [fix](fe) Preserve external database discovery errors
(#69252)
8ce54eec191 is described below
commit 8ce54eec1918bc080d45a283f625ac5905114968
Author: Gabriel <[email protected]>
AuthorDate: Thu Oct 22 11:27:52 2026 +0800
[fix](fe) Preserve external database discovery errors (#69252)
### What problem does this PR solve?
Problem Summary:
An external catalog initialization or database-discovery failure could
be reported as `Unknown database` because two lookup paths swallowed the
original exception. Preserve authentication, connection and discovery
errors; continue returning null when a successful lookup proves that the
database is absent.
Adapt the callers to the new error behavior:
- Return a MySQL ERR packet for COM_FIELD_LIST metadata failures and
keep the session usable.
- Retain the name of an unavailable forwarded default database so
catalog repair commands can run. When lookup succeeds, preserve the
database privilege check and reuse the resolved database instead of
repeating metadata lookup across a concurrent catalog reset.
- Preserve the previous database name/ID after a failed USE, and restore
the complete namespace after a failed remembered-database SWITCH.
- Restore external-view namespaces without remote metadata IO, and
restore thread-local context when materialized-view setup fails.
- Treat HTTP URL namespaces as defaults for name resolution. Both
SQL/schema handlers retain the catalog/database names without eagerly
loading external metadata; statements that actually depend on the
unavailable database still fail.
For a fresh Iceberg Hadoop catalog, its warehouse directory does not
exist until the first namespace is created. Treat only a missing
implicit Hadoop root as empty; missing named/REST roots and other
storage errors continue to propagate.
Kerberos realm mapping remains deployment configuration. This PR exposes
the underlying error and does not change Hadoop's shared authentication
state.
### Release note
Report external catalog initialization/discovery errors accurately,
preserve session state after failed namespace changes, and avoid
blocking independent queries or catalog repair commands on optional
default-database lookup.
### Check List (For Author)
- [x] Unit Test
- [x] Manual test
- Behavior changed: Yes; lookup failures retain their causes and callers
handle optional namespace setup and failure cleanup.
- Documentation needed: No new configuration or API.
Validation of the 2026-10-21 rebase:
- Rebased all five commits onto master
`02cb614b1ed5e11f21cd782a8af5b9432820a4e8`.
- The only conflict was the import block in `BindRelation.java`:
retained upstream `OlapChangelogTableWrapper` and this PR's
`NameSpaceContext`. `git range-diff` confirms all five patches are
preserved apart from that upstream import context.
- `git diff --check` and FE Checkstyle passed.
- Attempted the related tests through `run-fe-ut.sh`; the build stopped
before tests at the new Thrift compiler-version gate: master requires
0.24.0, while the configured local compiler is 0.16.0. No unit-test pass
is claimed for this rebased head; full validation is requested through
`run buildall`.
Validation before the 2026-10-21 rebase:
- Previous revision: `run-fe-ut.sh` passed 182 tests across nine related
FE test classes, with zero failures, errors or skips.
- Before the caller fixes, six deterministic assertions reproduced the
reviewed SWITCH, forwarded REFRESH/ALTER, HTTP schema, and HTTP
SQL-submission failures. Fault injection invalidates the catalog
immediately after the first successful forwarded lookup.
- Controls verify successful retry after failed SWITCH, errors on actual
unavailable-database access, forwarded database/catalog privilege
denial, and HTTP authentication rejection. HTTP schema tests exercise
real authentication and planning. HTTP execution tests cover the
controller-to-submitter boundary with a mocked StatementSubmitter; they
do not claim JDBC integration coverage.
- FE Checkstyle: zero violations, BUILD SUCCESS.
- Rebased only to resolve the actual conflict against master
`f6359ae80659`. The field-list handler retains both upstream
display-schema behavior and this PR's exception handling. The Paimon
CHAR test correction is already upstream and was dropped as an identical
patch.
Earlier local integration validation:
- Reproduced the original swallowed-error behavior and confirmed genuine
absence still reports `Unknown database`. Live Kerberos checks showed
the original realm-mapping error, and succeeded after compatible
configuration was applied.
- The complete Iceberg Hadoop/Kerberos regression passed with the
corrected connector, including reads, writes, rollback and snapshot/file
operations. Iceberg connector packaging passed 1446 tests with zero
failures/errors and four external-connectivity skips on the previous
base.
- Those live connector tests were not rerun on the current rebased head.
Full integration validation is delegated to CI.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
.../doris/connector/iceberg/IcebergCatalogOps.java | 13 +-
.../iceberg/IcebergHadoopCatalogNamespaceTest.java | 79 +++++
.../main/java/org/apache/doris/catalog/Env.java | 37 ++-
.../apache/doris/datasource/ExternalCatalog.java | 57 +---
.../doris/httpv2/rest/StmtExecutionAction.java | 13 +-
.../java/org/apache/doris/mtmv/MTMVPlanUtil.java | 43 ++-
.../doris/nereids/rules/analysis/BindRelation.java | 13 +-
.../mv/InitMaterializationContextHook.java | 4 +
.../java/org/apache/doris/qe/ConnectContext.java | 19 +-
.../java/org/apache/doris/qe/ConnectProcessor.java | 26 +-
.../org/apache/doris/qe/MysqlConnectProcessor.java | 41 +--
.../doris/datasource/ExternalCatalogTest.java | 40 +++
.../doris/qe/ExternalCatalogLookupFailureTest.java | 337 +++++++++++++++++++++
.../qe/ExternalCatalogNamespaceFailureTest.java | 266 ++++++++++++++++
14 files changed, 874 insertions(+), 114 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogOps.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogOps.java
index 8659fc77340..4968a141ab2 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogOps.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogOps.java
@@ -44,6 +44,7 @@ import org.apache.iceberg.catalog.ViewCatalog;
import org.apache.iceberg.exceptions.NoSuchNamespaceException;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.expressions.Term;
+import org.apache.iceberg.hadoop.HadoopCatalog;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.view.View;
@@ -318,7 +319,17 @@ public interface IcebergCatalogOps {
LOG.warn("Iceberg catalog does not support namespaces");
return Collections.emptyList();
}
- return listNestedNamespaces(rootNamespace());
+ Namespace root = rootNamespace();
+ try {
+ return listNestedNamespaces(root);
+ } catch (NoSuchNamespaceException e) {
+ // Hadoop creates the warehouse lazily with the first
namespace. Its absent implicit
+ // root is an empty catalog, unlike a missing named or REST
root; I/O errors still fail.
+ if (catalog instanceof HadoopCatalog && root.isEmpty()) {
+ return Collections.emptyList();
+ }
+ throw e;
+ }
}
/**
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergHadoopCatalogNamespaceTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergHadoopCatalogNamespaceTest.java
new file mode 100644
index 00000000000..d7ac72a5217
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergHadoopCatalogNamespaceTest.java
@@ -0,0 +1,79 @@
+// 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.connector.iceberg;
+
+import
org.apache.doris.connector.iceberg.IcebergCatalogOps.CatalogBackedIcebergCatalogOps;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.exceptions.NoSuchNamespaceException;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+public class IcebergHadoopCatalogNamespaceTest {
+ @TempDir
+ Path tempDir;
+
+ @Test
+ public void freshWarehouseCanListAndCreateFirstDatabase() throws
IOException {
+ Path warehouse = tempDir.resolve("warehouse");
+ try (HadoopCatalog catalog = new HadoopCatalog(new
Configuration(false), warehouse.toString())) {
+ CatalogBackedIcebergCatalogOps ops = new
CatalogBackedIcebergCatalogOps(catalog);
+ Assertions.assertFalse(Files.exists(warehouse));
+ Assertions.assertTrue(ops.listDatabaseNames().isEmpty());
+ Assertions.assertFalse(Files.exists(warehouse), "Listing must not
create the warehouse");
+ ops.createDatabase("first_db", Collections.emptyMap());
+ Assertions.assertEquals(Collections.singletonList("first_db"),
ops.listDatabaseNames());
+ }
+ }
+
+ @Test
+ public void missingNamedHadoopRootStillFails() throws IOException {
+ try (HadoopCatalog catalog = new HadoopCatalog(new
Configuration(false),
+ tempDir.resolve("warehouse").toString())) {
+ CatalogBackedIcebergCatalogOps ops = new
CatalogBackedIcebergCatalogOps(
+ catalog, false, false, false, Optional.of("missing"));
+ Assertions.assertThrows(NoSuchNamespaceException.class,
ops::listDatabaseNames);
+ }
+ }
+
+ @Test
+ public void hadoopIoFailureIsNotAnEmptyCatalog() throws IOException {
+ RuntimeIOException failure = new RuntimeIOException(new
IOException("Storage unavailable"),
+ "Cannot list warehouse");
+ try (HadoopCatalog catalog = new HadoopCatalog() {
+ @Override
+ public List<Namespace> listNamespaces(Namespace namespace) {
+ throw failure;
+ }
+ }) {
+ Assertions.assertSame(failure,
Assertions.assertThrows(RuntimeIOException.class,
+ () -> new
CatalogBackedIcebergCatalogOps(catalog).listDatabaseNames()));
+ }
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
index 19efb354a55..1b8c36a751e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
@@ -6653,21 +6653,28 @@ public class Env {
ctx.addLastDBOfCatalog(ctx.getCurrentCatalog().getName(),
currentDB);
}
}
- ctx.changeDefaultCatalog(catalogName);
- String lastDb = ctx.getLastDBOfCatalog(catalogName);
- if (StringUtils.isNotEmpty(lastDb)) {
- ctx.setDatabase(lastDb);
- }
- // A data source whose metadata model has no database layer can name
the single database Doris
- // presents for it; the connector owns that name. Asked of the
provider, not the connector, because
- // switching to a catalog must not force it to initialize. Applied
after the remembered database on
- // purpose: such a catalog has exactly one database, so there is
nothing else to return to.
- Map<String, String> catalogProps = catalogIf.getProperties();
- String catalogType = catalogProps.get(CatalogMgr.CATALOG_TYPE_PROP);
- if (catalogType != null) {
- ConnectorFactory.findProvider(catalogType, catalogProps)
- .flatMap(ConnectorProvider::defaultDatabaseOnUse)
- .ifPresent(ctx::setDatabase);
+ NameSpaceContext previousNamespace = ctx.getNameSpaceContext();
+ try {
+ ctx.changeDefaultCatalog(catalogName);
+ String lastDb = ctx.getLastDBOfCatalog(catalogName);
+ if (StringUtils.isNotEmpty(lastDb)) {
+ ctx.setDatabase(lastDb);
+ }
+ // A data source whose metadata model has no database layer can
name the single database Doris
+ // presents for it; the connector owns that name. Asked of the
provider, not the connector, because
+ // switching to a catalog must not force it to initialize. Applied
after the remembered database on
+ // purpose: such a catalog has exactly one database, so there is
nothing else to return to.
+ Map<String, String> catalogProps = catalogIf.getProperties();
+ String catalogType =
catalogProps.get(CatalogMgr.CATALOG_TYPE_PROP);
+ if (catalogType != null) {
+ ConnectorFactory.findProvider(catalogType, catalogProps)
+ .flatMap(ConnectorProvider::defaultDatabaseOnUse)
+ .ifPresent(ctx::setDatabase);
+ }
+ } catch (RuntimeException | Error e) {
+ // A failed remembered/default database lookup must not leave a
partially switched session.
+ ctx.restoreNameSpaceContext(previousNamespace);
+ throw e;
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java
index 917d67a8de8..3934483b97b 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java
@@ -779,17 +779,8 @@ public abstract class ExternalCatalog
if (StringUtils.isEmpty(dbName)) {
return null;
}
- try {
- makeSureInitialized();
- } catch (Exception e) {
- String diagnosticTrace = initErrorStackTrace(e);
- if (diagnosticTrace != null) {
- LOG.warn("failed to get db {} in catalog {}: {}", dbName,
name, diagnosticTrace);
- } else {
- LOG.warn("failed to get db {} in catalog {}", dbName, name, e);
- }
- return null;
- }
+ // Initialization errors do not establish that the requested database
is absent.
+ makeSureInitialized();
SessionContext sessionContext = SessionContext.current();
if (shouldBypassDbNameCache(sessionContext)) {
@@ -1098,42 +1089,14 @@ public abstract class ExternalCatalog
// When running ut, disable this check to make ut pass.
// Because in ut, the database is not created in remote system.
if (checkExists && (!FeConstants.runningUnitTest || this instanceof
TestExternalCatalog)) {
- try {
- final String lookupLocalDbName = localDbName;
- // Reuse the shared names lookup helper so existence checks
follow the same miss-refresh policy.
- Boolean exists =
resolveDatabaseNameFromSnapshot(lookupLocalDbName, false,
- namesValue ->
namesValue.containsLocalName(lookupLocalDbName) ? Boolean.TRUE : null);
- if (!Boolean.TRUE.equals(exists)) {
- LOG.warn("Database {} does not exist in the remote system.
Skipping initialization.",
- localDbName);
- return null;
- }
- } catch (RuntimeException e) {
- // Handle "Found conflicting" exception explicitly
- if (e.getMessage().contains(FOUND_CONFLICTING)) {
- LOG.error(e.getMessage());
- throw e; // Rethrow to let the caller handle this critical
issue
- } else {
- // Any errors other than name conflicts, we default to not
finding the database
- String diagnosticTrace = initErrorStackTrace(e);
- if (diagnosticTrace != null) {
- LOG.warn("Failed to check db {} exist in remote
system, ignore it: {}",
- localDbName, diagnosticTrace);
- } else {
- LOG.warn("Failed to check db {} exist in remote
system, ignore it.", localDbName, e);
- }
- return null;
- }
- } catch (Exception e) {
- // If connection failed, it will throw exception.
- // ignore it and treat it as not exist.
- String diagnosticTrace = initErrorStackTrace(e);
- if (diagnosticTrace != null) {
- LOG.warn("Failed to check db {} exist in remote system,
ignore it: {}",
- localDbName, diagnosticTrace);
- } else {
- LOG.warn("Failed to check db {} exist in remote system,
ignore it.", localDbName, e);
- }
+ // Authentication and connection failures must propagate; only a
successful lookup can prove absence.
+ final String lookupLocalDbName = localDbName;
+ // Reuse the shared names lookup helper so existence checks follow
the same miss-refresh policy.
+ Boolean exists =
resolveDatabaseNameFromSnapshot(lookupLocalDbName, false,
+ namesValue ->
namesValue.containsLocalName(lookupLocalDbName) ? Boolean.TRUE : null);
+ if (!Boolean.TRUE.equals(exists)) {
+ LOG.warn("Database {} does not exist in the remote system.
Skipping initialization.",
+ localDbName);
return null;
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/StmtExecutionAction.java
b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/StmtExecutionAction.java
index 254ae250dab..e343fdf79fb 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/StmtExecutionAction.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/StmtExecutionAction.java
@@ -18,6 +18,7 @@
package org.apache.doris.httpv2.rest;
import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.NameSpaceContext;
import org.apache.doris.catalog.TableIf;
import org.apache.doris.datasource.InternalCatalog;
import org.apache.doris.httpv2.entity.ResponseEntityBuilder;
@@ -110,8 +111,7 @@ public class StmtExecutionAction extends RestBaseController
{
LOG.info("stmt: {}, isSync:{}, limit: {}", stmtRequestBody.stmt,
stmtRequestBody.is_sync,
stmtRequestBody.limit);
- ConnectContext.get().changeDefaultCatalog(ns);
- ConnectContext.get().setDatabase(dbName);
+ setDefaultNamespace(ns, dbName);
String streamHeader = request.getHeader("X-Doris-Stream");
boolean isStream = !("false".equalsIgnoreCase(streamHeader));
@@ -145,8 +145,7 @@ public class StmtExecutionAction extends RestBaseController
{
sql = sql.replaceAll(NEW_LINE_PATTERN, NEW_LINE_REPLACEMENT);
}
LOG.info("sql: {}", sql);
- ConnectContext.get().changeDefaultCatalog(ns);
- ConnectContext.get().setDatabase(dbName);
+ setDefaultNamespace(ns, dbName);
return getSchema(sql);
}
@@ -212,6 +211,12 @@ public class StmtExecutionAction extends
RestBaseController {
}
}
+ private void setDefaultNamespace(String catalog, String database) {
+ // The URL supplies a default for name resolution, not a dependency of
every SQL statement.
+ // Resolve metadata only when the statement uses it, so an outage
cannot block catalog repair.
+ ConnectContext.get().restoreNameSpaceContext(new
NameSpaceContext(catalog, database, -1));
+ }
+
private static class StmtRequestBody {
public Boolean is_sync = true; // CHECKSTYLE IGNORE THIS LINE
public Long limit = DEFAULT_ROW_LIMIT;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPlanUtil.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPlanUtil.java
index 9620883030b..3b4d2efb06f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPlanUtil.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPlanUtil.java
@@ -215,23 +215,34 @@ public class MTMVPlanUtil {
ctx.setCurrentUserIdentity(UserIdentity.ADMIN);
ctx.getState().reset();
ctx.getState().setInternal(true);
- ctx.setThreadLocalInfo();
-
ctx.getSessionVariable().setAffectQueryResultInPlanSessionVariables(sessionVariables);
- // Debug session variable should be disabled when refreshed
- ctx.getSessionVariable().skipDeletePredicate = false;
- ctx.getSessionVariable().skipDeleteBitmap = false;
- ctx.getSessionVariable().skipDeleteSign = false;
- ctx.getSessionVariable().skipStorageEngineMerge = false;
- ctx.getSessionVariable().showHiddenColumns = false;
- ctx.getSessionVariable().allowModifyMaterializedViewData = true;
- ctx.getSessionVariable().setDisableNereidsRules(
-
disableRules.stream().map(RuleType::name).collect(Collectors.joining(",")));
- ctx.setStartTime();
- if (parentContext != null) {
- ctx.changeDefaultCatalog(parentContext.getDefaultCatalog());
- ctx.setDatabase(parentContext.getDatabase());
+ ConnectContext previousContext = ConnectContext.get();
+ try {
+ ctx.setThreadLocalInfo();
+
ctx.getSessionVariable().setAffectQueryResultInPlanSessionVariables(sessionVariables);
+ // Debug session variable should be disabled when refreshed
+ ctx.getSessionVariable().skipDeletePredicate = false;
+ ctx.getSessionVariable().skipDeleteBitmap = false;
+ ctx.getSessionVariable().skipDeleteSign = false;
+ ctx.getSessionVariable().skipStorageEngineMerge = false;
+ ctx.getSessionVariable().showHiddenColumns = false;
+ ctx.getSessionVariable().allowModifyMaterializedViewData = true;
+ ctx.getSessionVariable().setDisableNereidsRules(
+
disableRules.stream().map(RuleType::name).collect(Collectors.joining(",")));
+ ctx.setStartTime();
+ if (parentContext != null) {
+ ctx.changeDefaultCatalog(parentContext.getDefaultCatalog());
+ ctx.setDatabase(parentContext.getDatabase());
+ }
+ return ctx;
+ } catch (RuntimeException | Error e) {
+ // Initialization can fail before MTMVCache takes responsibility
for context cleanup.
+ if (previousContext == null) {
+ ConnectContext.remove();
+ } else {
+ previousContext.setThreadLocalInfo();
+ }
+ throw e;
}
- return ctx;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java
index 012847288dc..87c4d800486 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java
@@ -26,6 +26,7 @@ import org.apache.doris.catalog.DistributionInfo;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.FunctionRegistry;
import org.apache.doris.catalog.KeysType;
+import org.apache.doris.catalog.NameSpaceContext;
import org.apache.doris.catalog.OlapChangelogTableWrapper;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Partition;
@@ -977,20 +978,18 @@ public class BindRelation extends OneAnalysisRuleFactory {
ExternalTable table, String externalCatalog, String externalDb,
String ddlSql, CascadesContext cascadesContext) {
ConnectContext ctx = cascadesContext.getConnectContext();
- String previousCatalog = ctx.getCurrentCatalog().getName();
- String previousDb = ctx.getDatabase();
+ NameSpaceContext previousNamespace = ctx.getNameSpaceContext();
String convertedSql = SqlDialectHelper.convertSqlByDialect(ddlSql,
ctx.getSessionVariable());
// change catalog and db to external catalog and db,
// so that we can parse and analyze the view sql in external context.
- ctx.changeDefaultCatalog(externalCatalog);
- ctx.setDatabase(externalDb);
try {
+ ctx.changeDefaultCatalog(externalCatalog);
+ ctx.setDatabase(externalDb);
return new LogicalView<>(new ExternalView(table, ddlSql),
parseAndAnalyzeView(table, convertedSql, cascadesContext));
} finally {
- // restore catalog and db in connect context
- ctx.changeDefaultCatalog(previousCatalog);
- ctx.setDatabase(previousDb);
+ // Also restore failed namespace entry without masking analysis
errors with metadata IO.
+ ctx.restoreNameSpaceContext(previousNamespace);
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/InitMaterializationContextHook.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/InitMaterializationContextHook.java
index c1642d6f433..0af2f843691 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/InitMaterializationContextHook.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/InitMaterializationContextHook.java
@@ -245,6 +245,7 @@ public class InitMaterializationContextHook implements
PlannerHook {
private List<MaterializationContext> createSyncMvContexts(OlapTable
olapTable,
CascadesContext cascadesContext) {
+ ConnectContext previousContext = ConnectContext.get();
int indexNumber = olapTable.getIndexNumber();
List<MaterializationContext> contexts = new ArrayList<>(indexNumber);
long baseIndexId = olapTable.getBaseIndexId();
@@ -312,6 +313,9 @@ public class InitMaterializationContextHook implements
PlannerHook {
LOG.warn(String.format("createSyncMvContexts exception, index
id is %s, index name is %s, "
+ "table name is %s", entry.getValue(),
indexName, olapTable.getQualifiedName()),
exception);
+ } finally {
+ // Database selection can fail after context creation but
before MTMVCache cleanup.
+ previousContext.setThreadLocalInfo();
}
}
return getMaterializationContextByHint(contexts);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java
index 89b63ccadbf..8d93e63de81 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java
@@ -1002,10 +1002,25 @@ public class ConnectContext {
return new NameSpaceContext(defaultCatalog, currentDb, currentDbId);
}
+ public void restoreNameSpaceContext(NameSpaceContext namespace) {
+ // Restoring a captured namespace must not contact a catalog that may
now be unavailable.
+ defaultCatalog = namespace.getDefaultCatalog();
+ currentDb = namespace.getCurrentDb();
+ currentDbId = namespace.getCurrentDbId();
+ }
+
public void setDatabase(String db) {
- currentDb = db;
Optional<DatabaseIf> dbInstance = getCurrentCatalog().getDb(db);
- currentDbId = dbInstance.map(DatabaseIf::getId).orElse(-1L);
+ long dbId = dbInstance.map(DatabaseIf::getId).orElse(-1L);
+ // A failed lookup must leave the previously selected database name
and ID together.
+ currentDb = db;
+ currentDbId = dbId;
+ }
+
+ void restoreDatabaseName(String db) {
+ // A forwarded session can retain its namespace even when the
metastore cannot resolve the database ID.
+ currentDb = db;
+ currentDbId = -1;
}
public void setExecutor(StmtExecutor executor) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java
index 9190797ab34..b5699152df7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java
@@ -35,6 +35,7 @@ import org.apache.doris.analysis.UserIdentity;
import org.apache.doris.analysis.UuidLiteral;
import org.apache.doris.catalog.DatabaseIf;
import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.NameSpaceContext;
import org.apache.doris.catalog.Type;
import org.apache.doris.cloud.proto.Cloud;
import org.apache.doris.cloud.qe.ComputeGroupException;
@@ -43,6 +44,7 @@ import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.ConnectionException;
import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
import org.apache.doris.common.NotImplementedException;
import org.apache.doris.common.Pair;
import org.apache.doris.common.UserException;
@@ -54,6 +56,7 @@ import org.apache.doris.datasource.SessionContext;
import org.apache.doris.metric.MetricRepo;
import org.apache.doris.mysql.MysqlCommand;
import org.apache.doris.mysql.MysqlServerStatusFlag;
+import org.apache.doris.mysql.privilege.PrivPredicate;
import org.apache.doris.mysql.protocol.MysqlProtocolAdapter;
import org.apache.doris.nereids.SqlCacheContext;
import org.apache.doris.nereids.SqlCacheContext.CacheKeyType;
@@ -591,11 +594,28 @@ public abstract class ConnectProcessor {
CatalogIf catalog =
ctx.getEnv().getCatalogMgr().getCatalog(request.getDefaultCatalog());
if (catalog != null) {
ctx.getEnv().changeCatalog(ctx,
request.getDefaultCatalog());
- // Set default db only when the default catalog is set and
the dbname exists in default catalog.
+ // Restore the selected database after its catalog,
checking existence when metadata is available.
if (request.isSetDefaultDatabase()) {
- DatabaseIf db =
ctx.getCurrentCatalog().getDbNullable(request.getDefaultDatabase());
+ DatabaseIf db = null;
+ try {
+ db =
ctx.getCurrentCatalog().getDbNullable(request.getDefaultDatabase());
+ } catch (RuntimeException e) {
+ // Optional context restoration must not block
catalog repair commands. Retain the name
+ // so statements that actually use the database
still resolve it and report failures.
+
ctx.restoreDatabaseName(request.getDefaultDatabase());
+ LOG.debug("Deferring lookup of forwarded default
database {}.{}",
+ request.getDefaultCatalog(),
request.getDefaultDatabase(), e);
+ }
if (db != null) {
- ctx.getEnv().changeDb(ctx,
request.getDefaultDatabase());
+ if
(!ctx.getEnv().getAccessManager().checkDbPriv(ctx, ctx.getDefaultCatalog(),
+ request.getDefaultDatabase(),
PrivPredicate.SHOW)) {
+
ErrorReport.reportDdlException(ErrorCode.ERR_DBACCESS_DENIED_ERROR,
+ ctx.getQualifiedUser(),
request.getDefaultDatabase());
+ }
+ // Reuse the resolved database: a concurrent
refresh can make a second lookup fail
+ // before a catalog repair statement gets a chance
to execute.
+ ctx.restoreNameSpaceContext(new
NameSpaceContext(ctx.getDefaultCatalog(),
+ request.getDefaultDatabase(), db.getId()));
}
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/qe/MysqlConnectProcessor.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/MysqlConnectProcessor.java
index 1486d9c210e..79e57b2316f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/MysqlConnectProcessor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/MysqlConnectProcessor.java
@@ -431,29 +431,32 @@ public class MysqlConnectProcessor extends
ConnectProcessor {
ctx.getState().setError(ErrorCode.ERR_UNKNOWN_TABLE, "Empty
tableName");
return;
}
- DatabaseIf db =
ctx.getCurrentCatalog().getDbNullable(ctx.getDatabase());
- if (db == null) {
- ctx.getState().setError(ErrorCode.ERR_BAD_DB_ERROR, "Unknown
database(" + ctx.getDatabase() + ")");
- return;
- }
- TableIf table = db.getTableNullable(tableName);
- if (table == null) {
- ctx.getState().setError(ErrorCode.ERR_UNKNOWN_TABLE, "Unknown
table(" + tableName + ")");
- return;
- }
-
- table.readLock();
try {
- MysqlProtocolAdapter.of(ctx).resultSender(ctx)
- // COM_FIELD_LIST is a metadata statement: it must agree
with SHOW COLUMNS, so it serves
- // the display schema (the nullability the source
declares), not the read schema.
- .sendFieldList(db.getFullName(), table.getName(),
table.getBaseSchemaForDisplay());
+ // Metadata lookup can fail before a table is locked; return an
ERR packet instead of losing the session.
+ DatabaseIf db =
ctx.getCurrentCatalog().getDbNullable(ctx.getDatabase());
+ if (db == null) {
+ ctx.getState().setError(ErrorCode.ERR_BAD_DB_ERROR, "Unknown
database(" + ctx.getDatabase() + ")");
+ return;
+ }
+ TableIf table = db.getTableNullable(tableName);
+ if (table == null) {
+ ctx.getState().setError(ErrorCode.ERR_UNKNOWN_TABLE, "Unknown
table(" + tableName + ")");
+ return;
+ }
+
+ table.readLock();
+ try {
+ MysqlProtocolAdapter.of(ctx).resultSender(ctx)
+ // COM_FIELD_LIST is a metadata statement: it must
agree with SHOW COLUMNS, so it serves
+ // the display schema (the nullability the source
declares), not the read schema.
+ .sendFieldList(db.getFullName(), table.getName(),
table.getBaseSchemaForDisplay());
+ } finally {
+ table.readUnlock();
+ }
+ ctx.getState().setEof();
} catch (Throwable throwable) {
handleQueryException(throwable, "", null, null);
- } finally {
- table.readUnlock();
}
- ctx.getState().setEof();
}
private void handleChangeUser() throws IOException {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalCatalogTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalCatalogTest.java
index 511d881c811..bd6b22466af 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalCatalogTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/ExternalCatalogTest.java
@@ -66,6 +66,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
public class ExternalCatalogTest extends TestWithFeService {
@@ -73,6 +74,45 @@ public class ExternalCatalogTest extends TestWithFeService {
private CatalogMgr mgr;
private ConnectContext rootCtx;
+ @Test
+ public void testDatabaseLookupPreservesInitializationFailure() {
+ IllegalStateException failure = new IllegalStateException();
+ IncrementalUpdateCatalog catalog = new IncrementalUpdateCatalog() {
+ @Override
+ protected void initLocalObjectsImpl() {
+ throw failure;
+ }
+ };
+ RuntimeException thrown =
Assertions.assertThrows(RuntimeException.class,
+ () -> catalog.getDbNullable("db_by_id"));
+ Assertions.assertSame(failure, thrown.getCause());
+ }
+
+ @Test
+ public void testDatabaseDiscoveryFailureIsNotMissingAndCanRecover() {
+ AtomicBoolean unavailable = new AtomicBoolean(true);
+ IllegalStateException failure = new IllegalStateException("Kerberos
principal has no matching rule");
+ IncrementalUpdateCatalog catalog = new IncrementalUpdateCatalog() {
+ @Override
+ protected List<String> listDatabaseNames() {
+ if (unavailable.get()) {
+ throw failure;
+ }
+ return super.listDatabaseNames();
+ }
+ };
+ RuntimeException thrown =
Assertions.assertThrows(RuntimeException.class,
+ () -> catalog.getDbNullable("db_by_id"));
+ Throwable cause = thrown;
+ while (cause.getCause() != null) {
+ cause = cause.getCause();
+ }
+ Assertions.assertSame(failure, cause);
+ unavailable.set(false);
+ Assertions.assertNotNull(catalog.getDbNullable("db_by_id"));
+ Assertions.assertNull(catalog.getDbNullable("missing_database"));
+ }
+
@Test
public void
testScopedWeightAlterStillInvalidatesRemoteBackendsAndRowCounts() throws
Exception {
Map<String, String> properties = Maps.newHashMap();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/ExternalCatalogLookupFailureTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/ExternalCatalogLookupFailureTest.java
new file mode 100644
index 00000000000..5c95ed07405
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/qe/ExternalCatalogLookupFailureTest.java
@@ -0,0 +1,337 @@
+// 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.qe;
+
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.datasource.ExternalDatabase;
+import org.apache.doris.datasource.ExternalTable;
+import org.apache.doris.datasource.test.TestExternalCatalog;
+import org.apache.doris.httpv2.entity.ResponseBody;
+import org.apache.doris.httpv2.exception.UnauthorizedException;
+import org.apache.doris.httpv2.rest.StmtExecutionAction;
+import org.apache.doris.httpv2.util.ExecutionResultSet;
+import org.apache.doris.httpv2.util.StatementSubmitter;
+import org.apache.doris.mysql.DummyMysqlChannel;
+import org.apache.doris.mysql.protocol.MysqlProtocolAdapter;
+import org.apache.doris.qe.QueryState.MysqlStateType;
+import org.apache.doris.thrift.TMasterOpRequest;
+import org.apache.doris.thrift.TMasterOpResult;
+import org.apache.doris.utframe.TestWithFeService;
+
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+import org.springframework.http.ResponseEntity;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+public class ExternalCatalogLookupFailureTest extends TestWithFeService {
+ private static final String CATALOG = "lookup_failure_catalog";
+ private static final String DATABASE = "example_db";
+ private static final String FAILURE = "Metastore authentication failed";
+ private TestExternalCatalog catalog;
+ private boolean initializationUnavailable;
+ private boolean failAfterLookup;
+
+ @Override
+ protected void runBeforeAll() throws Exception {
+ executeSql("CREATE USER 'lookup_unprivileged_user'");
+ createDatabase(DATABASE);
+ connectContext.setDatabase(DATABASE);
+ createTable("CREATE TABLE healthy_table (id INT) DISTRIBUTED BY
HASH(id) BUCKETS 1 "
+ + "PROPERTIES ('replication_num'='1')");
+ }
+
+ @BeforeEach
+ public void prepareCatalog() {
+ FailingProvider.unavailable = false;
+ FailingProvider.lookupCount = 0;
+ initializationUnavailable = false;
+ failAfterLookup = false;
+ catalog = new TestExternalCatalog(Env.getCurrentEnv().getNextId(),
CATALOG, "",
+ new HashMap<>(Map.of("catalog_provider.class",
FailingProvider.class.getName())), "") {
+ @Override
+ public ExternalDatabase<? extends ExternalTable>
getDbNullable(String dbName) {
+ ExternalDatabase<? extends ExternalTable> db =
super.getDbNullable(dbName);
+ if (failAfterLookup) {
+ // Invalidate after the first successful probe, before
forwarding can publish its default database.
+ failAfterLookup = false;
+ resetToUninitialized(false);
+ initializationUnavailable = true;
+ }
+ return db;
+ }
+
+ @Override
+ protected void initLocalObjectsImpl() {
+ if (initializationUnavailable) {
+ throw new IllegalStateException(FAILURE);
+ }
+ }
+ };
+ Deencapsulation.invoke(Env.getCurrentEnv().getCatalogMgr(),
"addCatalog", catalog);
+ }
+
+ @AfterEach
+ public void removeCatalog() {
+ FailingProvider.unavailable = false;
+ Deencapsulation.invoke(Env.getCurrentEnv().getCatalogMgr(),
"removeCatalog", catalog.getId());
+ catalog.onClose();
+
Env.getCurrentEnv().getExtMetaCacheMgr().removeCatalog(catalog.getId());
+ connectContext.setThreadLocalInfo();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testFieldListReturnsErrorAndKeepsSessionUsable(boolean
failInitialization) throws Exception {
+ RecordingChannel channel = new RecordingChannel();
+ ConnectContext context = new ConnectContext(new
MysqlProtocolAdapter(channel));
+ context.setEnv(Env.getCurrentEnv());
+ context.setCurrentUserIdentity(UserIdentity.ROOT);
+ context.changeDefaultCatalog(CATALOG);
+ context.setDatabase(DATABASE);
+ context.setThreadLocalInfo();
+ catalog.resetToUninitialized(false);
+ initializationUnavailable = failInitialization;
+ FailingProvider.unavailable = true;
+ channel.request = ByteBuffer.wrap(new byte[] {4, 't', 0, 0});
+ MysqlConnectProcessor processor = new MysqlConnectProcessor(context);
+
+ Assertions.assertDoesNotThrow(processor::processOnce);
+ Assertions.assertEquals(MysqlStateType.ERR,
context.getState().getStateType());
+ Assertions.assertEquals(0xFF,
Byte.toUnsignedInt(channel.response.get(0)));
+
Assertions.assertTrue(StandardCharsets.UTF_8.decode(channel.response.duplicate()).toString().contains(FAILURE));
+ Assertions.assertFalse(context.isKilled());
+
+ channel.request = ByteBuffer.wrap(new byte[] {14});
+ processor.processOnce();
+ Assertions.assertEquals(0, channel.response.get(0));
+ Assertions.assertFalse(context.isKilled());
+ }
+
+ @Test
+ public void testForwardedRefreshSurvivesUnavailableDefaultDatabase()
throws Exception {
+ FailingProvider.unavailable = true;
+ TMasterOpResult result = forward("REFRESH CATALOG " + CATALOG);
+ Assertions.assertEquals(MysqlStateType.OK.name(), result.getStatus(),
result.getErrMessage());
+ }
+
+ @Test
+ public void testForwardedAlterSurvivesUnavailableDefaultDatabase() throws
Exception {
+ FailingProvider.unavailable = true;
+ TMasterOpResult result = forward("ALTER CATALOG " + CATALOG
+ + " SET PROPERTIES ('metadata_refresh_interval_sec' = '123')");
+ Assertions.assertEquals(MysqlStateType.OK.name(), result.getStatus(),
result.getErrMessage());
+ Assertions.assertEquals("123",
catalog.getProperties().get("metadata_refresh_interval_sec"));
+ }
+
+ @Test
+ public void testForwardedDatabaseAccessStillReportsFailure() throws
Exception {
+ FailingProvider.unavailable = true;
+ TMasterOpResult result = forward("SHOW TABLES");
+ Assertions.assertEquals(MysqlStateType.ERR.name(), result.getStatus());
+ Assertions.assertTrue(result.getErrMessage().contains(FAILURE),
result.getErrMessage());
+ Assertions.assertTrue(FailingProvider.lookupCount > 1, "The statement
must attempt database access");
+ }
+
+ @Test
+ public void testForwardedDatabaseAccessRestoresHealthyDefault() throws
Exception {
+ TMasterOpResult result = forward("SHOW TABLES");
+ Assertions.assertNotEquals(MysqlStateType.ERR.name(),
result.getStatus(), result.getErrMessage());
+ }
+
+ @Test
+ public void testForwardedAlterStillChecksCatalogPrivilege() throws
Exception {
+ FailingProvider.unavailable = true;
+ TMasterOpResult result = forward("ALTER CATALOG " + CATALOG
+ + " SET PROPERTIES ('metadata_refresh_interval_sec' = '123')",
+
UserIdentity.createAnalyzedUserIdentWithIp("lookup_unprivileged_user", "%"));
+ Assertions.assertEquals(MysqlStateType.ERR.name(), result.getStatus());
+ Assertions.assertTrue(result.getErrMessage().contains("Access denied
for user"), result.getErrMessage());
+
Assertions.assertNull(catalog.getProperties().get("metadata_refresh_interval_sec"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testForwardedRepairSurvivesResetAfterSuccessfulLookup(boolean
alter) throws Exception {
+ failAfterLookup = true;
+ String sql = alter ? "ALTER CATALOG " + CATALOG + " SET PROPERTIES
('metadata_refresh_interval_sec' = '123')"
+ : "REFRESH CATALOG " + CATALOG;
+ TMasterOpResult result = forward(sql);
+ Assertions.assertTrue(initializationUnavailable, "The successful
lookup must trigger the outage");
+ Assertions.assertEquals(MysqlStateType.OK.name(), result.getStatus(),
result.getErrMessage());
+ if (alter) {
+ Assertions.assertEquals("123",
catalog.getProperties().get("metadata_refresh_interval_sec"));
+ }
+ }
+
+ @Test
+ public void testForwardedActualAccessFailsAfterSuccessfulLookup() throws
Exception {
+ failAfterLookup = true;
+ TMasterOpResult result = forward("SHOW TABLES");
+ Assertions.assertTrue(initializationUnavailable);
+ Assertions.assertEquals(MysqlStateType.ERR.name(), result.getStatus());
+ Assertions.assertTrue(result.getErrMessage().contains(FAILURE),
result.getErrMessage());
+ }
+
+ @Test
+ public void testForwardedHealthyDefaultStillChecksDatabasePrivilege()
throws Exception {
+ TMasterOpResult result = forward("SELECT 1",
+
UserIdentity.createAnalyzedUserIdentWithIp("lookup_unprivileged_user", "%"));
+ Assertions.assertEquals(MysqlStateType.ERR.name(), result.getStatus());
+ Assertions.assertTrue(result.getErrMessage().contains("Access
denied"), result.getErrMessage());
+ }
+
+ @Test
+ public void testHttpSchemaIgnoresUnavailableDefaultDatabase() {
+ FailingProvider.unavailable = true;
+ Object schema = Assertions.assertDoesNotThrow(() -> new
StmtExecutionAction().querySchema(CATALOG, DATABASE,
+ httpRequest("root"), Mockito.mock(HttpServletResponse.class),
+ "SELECT * FROM internal." + DATABASE + ".healthy_table"));
+ Assertions.assertTrue(schema.toString().contains("CREATE TABLE"),
schema.toString());
+ Assertions.assertTrue(schema.toString().contains("healthy_table"),
schema.toString());
+ Assertions.assertEquals(CATALOG,
ConnectContext.get().getDefaultCatalog());
+ Assertions.assertEquals(DATABASE, ConnectContext.get().getDatabase());
+ }
+
+ @Test
+ public void testHttpSchemaActualDatabaseAccessStillFails() {
+ FailingProvider.unavailable = true;
+ RuntimeException error =
Assertions.assertThrows(RuntimeException.class,
+ () -> new StmtExecutionAction().querySchema(CATALOG, DATABASE,
+ httpRequest("root"),
Mockito.mock(HttpServletResponse.class), "SELECT * FROM t"));
+ Assertions.assertTrue(error.toString().contains(FAILURE),
error.toString());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testHttpEndpointsRejectUnauthenticatedRequests(boolean schema)
{
+ FailingProvider.unavailable = true;
+ StmtExecutionAction action = new StmtExecutionAction();
+ HttpServletRequest request = httpRequest("missing_http_user");
+ HttpServletResponse response = Mockito.mock(HttpServletResponse.class);
+ Assertions.assertThrows(UnauthorizedException.class, () -> {
+ if (schema) {
+ action.querySchema(CATALOG, DATABASE, request, response,
+ "SELECT * FROM internal." + DATABASE +
".healthy_table");
+ } else {
+ action.executeSQL(CATALOG, DATABASE, request, response,
+ "{\"stmt\":\"REFRESH CATALOG " + CATALOG + "\"}");
+ }
+ });
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void
testHttpRepairReachesSubmissionWithUnavailableDefaultDatabase(boolean alter)
throws Exception {
+ FailingProvider.unavailable = true;
+ StatementSubmitter original =
Deencapsulation.getField(StmtExecutionAction.class, StatementSubmitter.class);
+ StatementSubmitter submitter = Mockito.mock(StatementSubmitter.class);
+ Mockito.when(submitter.submit(Mockito.any())).thenReturn(
+ CompletableFuture.completedFuture(new
ExecutionResultSet(Map.of())));
+ Deencapsulation.setField(StmtExecutionAction.class, "stmtSubmitter",
submitter);
+ String sql = alter ? "ALTER CATALOG " + CATALOG + " SET PROPERTIES
('metadata_refresh_interval_sec' = '123')"
+ : "REFRESH CATALOG " + CATALOG;
+ try {
+ ResponseEntity<?> response = (ResponseEntity<?>)
Assertions.assertDoesNotThrow(
+ () -> new StmtExecutionAction().executeSQL(CATALOG,
DATABASE, httpRequest("root"),
+ Mockito.mock(HttpServletResponse.class),
"{\"stmt\":\"" + sql + "\"}"));
+ Assertions.assertEquals(0, ((ResponseBody<?>)
response.getBody()).getCode());
+ ArgumentCaptor<StatementSubmitter.StmtContext> submitted =
+
ArgumentCaptor.forClass(StatementSubmitter.StmtContext.class);
+ Mockito.verify(submitter).submit(submitted.capture());
+ Assertions.assertEquals(sql, submitted.getValue().stmt);
+ Assertions.assertEquals("root", submitted.getValue().user);
+ Assertions.assertEquals(CATALOG,
ConnectContext.get().getDefaultCatalog());
+ Assertions.assertEquals(DATABASE,
ConnectContext.get().getDatabase());
+ } finally {
+ Deencapsulation.setField(StmtExecutionAction.class,
"stmtSubmitter", original);
+ }
+ }
+
+ private HttpServletRequest httpRequest(String user) {
+ HttpServletRequest request = Mockito.mock(HttpServletRequest.class);
+ Mockito.when(request.getScheme()).thenReturn("https");
+ Mockito.when(request.getRemoteAddr()).thenReturn("127.0.0.1");
+ Mockito.when(request.getHeader("Authorization")).thenReturn("Basic "
+ + Base64.getEncoder().encodeToString((user +
":").getBytes(StandardCharsets.UTF_8)));
+ Mockito.when(request.getHeader("X-Doris-Stream")).thenReturn("false");
+ return request;
+ }
+
+ private TMasterOpResult forward(String sql) throws Exception {
+ return forward(sql, UserIdentity.ROOT);
+ }
+
+ private TMasterOpResult forward(String sql, UserIdentity user) throws
Exception {
+ TMasterOpRequest request = new TMasterOpRequest();
+ request.setDb(DATABASE);
+ request.setDefaultCatalog(CATALOG);
+ request.setDefaultDatabase(DATABASE);
+ request.setUser(user.getQualifiedUser());
+ request.setCurrentUserIdent(user.toThrift());
+ request.setSql(sql);
+ return new
MysqlConnectProcessor(ConnectContext.forMysqlProxy("lookup-failure-session")).proxyExecute(request);
+ }
+
+ public static class FailingProvider implements
TestExternalCatalog.TestCatalogProvider {
+ private static boolean unavailable;
+ private static int lookupCount;
+
+ @Override
+ public Map<String, Map<String, List<Column>>> getMetadata() {
+ lookupCount++;
+ if (unavailable) {
+ throw new IllegalStateException(FAILURE);
+ }
+ return Map.of(DATABASE, Map.of("t", List.of(new Column("id",
Type.INT))));
+ }
+ }
+
+ private static class RecordingChannel extends DummyMysqlChannel {
+ private ByteBuffer request;
+ private ByteBuffer response;
+
+ @Override
+ public ByteBuffer fetchOnePacket() {
+ return request;
+ }
+
+ @Override
+ public void sendAndFlush(ByteBuffer packet) {
+ response = packet.duplicate();
+ }
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/ExternalCatalogNamespaceFailureTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/ExternalCatalogNamespaceFailureTest.java
new file mode 100644
index 00000000000..e308f97b89a
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/qe/ExternalCatalogNamespaceFailureTest.java
@@ -0,0 +1,266 @@
+// 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.qe;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.NameSpaceContext;
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.datasource.ExternalDatabase;
+import org.apache.doris.datasource.ExternalTable;
+import org.apache.doris.datasource.test.TestExternalCatalog;
+import org.apache.doris.mtmv.MTMVPlanUtil;
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.StatementContext;
+import org.apache.doris.nereids.parser.NereidsParser;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.rules.analysis.BindRelation;
+import
org.apache.doris.nereids.rules.exploration.mv.InitMaterializationContextHook;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.util.PlanChecker;
+import org.apache.doris.utframe.TestWithFeService;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class ExternalCatalogNamespaceFailureTest extends TestWithFeService {
+ private static final String DATABASE = "db_old";
+ private static final String INTERNAL_DB = "namespace_internal";
+ private static final String FAILURE = "Remote catalog unavailable";
+ private FailingCatalog previousCatalog;
+ private FailingCatalog viewCatalog;
+ private ExternalTable viewTable;
+
+ @Override
+ protected void runBeforeAll() throws Exception {
+ createDatabase(INTERNAL_DB);
+ connectContext.setDatabase(INTERNAL_DB);
+ createTable("CREATE TABLE t (k INT NOT NULL, v INT NOT NULL) "
+ + "DISTRIBUTED BY HASH(k) BUCKETS 1 PROPERTIES
('replication_num'='1')");
+ createMv("CREATE MATERIALIZED VIEW mv AS SELECT k AS mv_k, SUM(v) AS
mv_v FROM t GROUP BY k");
+ }
+
+ @BeforeEach
+ public void prepareCatalogs() {
+ previousCatalog = new FailingCatalog("namespace_previous");
+ viewCatalog = new FailingCatalog("namespace_view");
+ for (FailingCatalog catalog : List.of(previousCatalog, viewCatalog)) {
+ Deencapsulation.invoke(Env.getCurrentEnv().getCatalogMgr(),
"addCatalog", catalog);
+ }
+ connectContext.changeDefaultCatalog(previousCatalog.getName());
+ connectContext.setDatabase(DATABASE);
+ connectContext.setThreadLocalInfo();
+ viewTable = viewCatalog.getDbNullable(DATABASE).getTableNullable("t");
+ }
+
+ @AfterEach
+ public void removeCatalogs() {
+ connectContext.setThreadLocalInfo();
+ connectContext.changeDefaultCatalog("internal");
+ connectContext.setDatabase(INTERNAL_DB);
+ for (FailingCatalog catalog : List.of(previousCatalog, viewCatalog)) {
+ Deencapsulation.invoke(Env.getCurrentEnv().getCatalogMgr(),
"removeCatalog", catalog.getId());
+ catalog.onClose();
+
Env.getCurrentEnv().getExtMetaCacheMgr().removeCatalog(catalog.getId());
+ }
+ }
+
+ @Test
+ public void testFailedUseKeepsDatabaseNameAndIdAfterConcurrentRefresh() {
+ NameSpaceContext saved = connectContext.getNameSpaceContext();
+ // Invalidate the cache exactly after Env.changeDb validates db_new,
before its setter's lookup.
+ previousCatalog.failAfterLookup = true;
+ RuntimeException error =
Assertions.assertThrows(RuntimeException.class,
+ () -> Env.getCurrentEnv().changeDb(connectContext, "db_new"));
+ Assertions.assertTrue(error.getMessage().contains(FAILURE),
error.getMessage());
+ assertNamespace(saved);
+ previousCatalog.unavailable = false;
+ Assertions.assertDoesNotThrow(() ->
Env.getCurrentEnv().changeDb(connectContext, "db_new"));
+ Assertions.assertEquals("db_new", connectContext.getDatabase());
+
Assertions.assertEquals(previousCatalog.getDbNullable("db_new").getId(),
connectContext.getCurrentDbId());
+ }
+
+ @Test
+ public void testFailedSwitchRestoresRememberedNamespace() throws Exception
{
+ Env.getCurrentEnv().changeCatalog(connectContext, "internal");
+ connectContext.setDatabase(INTERNAL_DB);
+ NameSpaceContext saved = connectContext.getNameSpaceContext();
+ previousCatalog.makeUnavailable();
+
+ RuntimeException error =
Assertions.assertThrows(RuntimeException.class,
+ () -> Env.getCurrentEnv().changeCatalog(connectContext,
previousCatalog.getName()));
+ Assertions.assertTrue(error.getMessage().contains(FAILURE),
error.getMessage());
+ assertNamespace(saved);
+
+ previousCatalog.unavailable = false;
+ Env.getCurrentEnv().changeCatalog(connectContext,
previousCatalog.getName());
+ Assertions.assertEquals(previousCatalog.getName(),
connectContext.getDefaultCatalog());
+ Assertions.assertEquals(DATABASE, connectContext.getDatabase());
+
Assertions.assertEquals(previousCatalog.getDbNullable(DATABASE).getId(),
connectContext.getCurrentDbId());
+ }
+
+ @Test
+ public void testHealthyExternalViewRestoresUnavailablePreviousCatalog() {
+ NameSpaceContext saved = connectContext.getNameSpaceContext();
+ previousCatalog.makeUnavailable();
+ Assertions.assertDoesNotThrow(() -> analyzeExternalView("SELECT id
FROM t"));
+ assertNamespace(saved);
+ }
+
+ @Test
+ public void testExternalViewEntryFailureRestoresPreviousNamespace() {
+ NameSpaceContext saved = connectContext.getNameSpaceContext();
+ viewCatalog.makeUnavailable();
+ RuntimeException error =
Assertions.assertThrows(RuntimeException.class,
+ () -> analyzeExternalView("SELECT id FROM t"));
+ Assertions.assertTrue(error.getMessage().contains(FAILURE),
error.getMessage());
+ assertNamespace(saved);
+ }
+
+ @Test
+ public void testExternalViewAnalysisFailureIsNotMaskedByRestoration() {
+ NameSpaceContext saved = connectContext.getNameSpaceContext();
+ previousCatalog.makeUnavailable();
+ RuntimeException error =
Assertions.assertThrows(RuntimeException.class,
+ () -> analyzeExternalView("SELECT * FROM missing_view_table"));
+
Assertions.assertTrue(error.getMessage().contains("missing_view_table"),
error.getMessage());
+ assertNamespace(saved);
+ }
+
+ @Test
+ public void testExternalViewPreservesEmptyNamespace() {
+ connectContext.changeDefaultCatalog("");
+ NameSpaceContext saved = connectContext.getNameSpaceContext();
+ Assertions.assertDoesNotThrow(() -> analyzeExternalView("SELECT id
FROM t"));
+ assertNamespace(saved);
+ }
+
+ @Test
+ public void testHealthyMvContextCopiesNamespaceAndBecomesCurrent() {
+ NameSpaceContext saved = connectContext.getNameSpaceContext();
+ ConnectContext mv = MTMVPlanUtil.createBasicMvContext(connectContext,
List.of(), Map.of());
+ Assertions.assertNotSame(connectContext, mv);
+ Assertions.assertSame(mv, ConnectContext.get());
+ Assertions.assertEquals(saved.getDefaultCatalog(),
mv.getDefaultCatalog());
+ Assertions.assertEquals(saved.getCurrentDb(), mv.getDatabase());
+ Assertions.assertEquals(saved.getCurrentDbId(), mv.getCurrentDbId());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testFailedMvContextConstructionPreservesThreadLocal(boolean
noPreviousContext) {
+ if (noPreviousContext) {
+ ConnectContext.remove();
+ }
+ previousCatalog.makeUnavailable();
+ Assertions.assertThrows(RuntimeException.class,
+ () -> MTMVPlanUtil.createBasicMvContext(connectContext,
List.of(), Map.of()));
+ Assertions.assertSame(noPreviousContext ? null : connectContext,
ConnectContext.get());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testSyncMvSetupFailureDoesNotBreakInternalQuery(boolean
failAfterContextCreation) {
+ connectContext.getSessionVariable().enableMaterializedViewRewrite =
true;
+ PlanChecker checker = PlanChecker.from(connectContext)
+ .analyze("SELECT k, SUM(v) FROM internal." + INTERNAL_DB + ".t
GROUP BY k");
+ CascadesContext cascadesContext = checker.getCascadesContext();
+ StatementContext statement = cascadesContext.getStatementContext();
+ OlapTable table = (OlapTable)
Env.getCurrentInternalCatalog().getDbNullable(INTERNAL_DB).getTableNullable("t");
+ Assertions.assertTrue(table.getVisibleIndexIdToMeta().size() > 1,
"Must exercise a visible sync MV");
+ NameSpaceContext saved = connectContext.getNameSpaceContext();
+ if (failAfterContextCreation) {
+ previousCatalog.failAfterLookup = true;
+ } else {
+ previousCatalog.makeUnavailable();
+ }
+ Assertions.assertDoesNotThrow(() ->
InitMaterializationContextHook.INSTANCE
+ .initMaterializationContext(cascadesContext));
+ Assertions.assertTrue(previousCatalog.unavailable, "MV setup must
attempt the default catalog lookup");
+ Assertions.assertSame(connectContext, ConnectContext.get());
+ Assertions.assertSame(statement,
ConnectContext.get().getStatementContext());
+ assertNamespace(saved);
+ Assertions.assertDoesNotThrow(() -> checker.rewrite().optimize());
+ }
+
+ private Plan analyzeExternalView(String sql) {
+ // The real binder parses the view under the second catalog and
restores the first catalog afterwards.
+ StatementContext statement = new StatementContext(connectContext, new
OriginStatement(sql, 0));
+ connectContext.setStatementContext(statement);
+ CascadesContext cascadesContext =
CascadesContext.initContext(statement,
+ new NereidsParser().parseSingle("SELECT 1"),
PhysicalProperties.ANY);
+ return Deencapsulation.invoke(new BindRelation(),
"parseAndAnalyzeExternalView", viewTable,
+ viewCatalog.getName(), DATABASE, sql, cascadesContext);
+ }
+
+ private void assertNamespace(NameSpaceContext saved) {
+ Assertions.assertEquals(saved.getDefaultCatalog(),
connectContext.getDefaultCatalog());
+ Assertions.assertEquals(saved.getCurrentDb(),
connectContext.getDatabase());
+ Assertions.assertEquals(saved.getCurrentDbId(),
connectContext.getCurrentDbId());
+ }
+
+ private static class FailingCatalog extends TestExternalCatalog {
+ private boolean unavailable;
+ private boolean failAfterLookup;
+
+ FailingCatalog(String name) {
+ super(Env.getCurrentEnv().getNextId(), name, "",
+ new HashMap<>(Map.of("catalog_provider.class",
Provider.class.getName())), "");
+ }
+
+ @Override
+ protected void initLocalObjectsImpl() {
+ if (unavailable) {
+ throw new IllegalStateException(FAILURE);
+ }
+ }
+
+ @Override
+ public ExternalDatabase<? extends ExternalTable> getDbNullable(String
dbName) {
+ ExternalDatabase<? extends ExternalTable> db =
super.getDbNullable(dbName);
+ if (failAfterLookup) {
+ failAfterLookup = false;
+ makeUnavailable();
+ }
+ return db;
+ }
+
+ void makeUnavailable() {
+ resetToUninitialized(false);
+ unavailable = true;
+ }
+ }
+
+ public static class Provider implements
TestExternalCatalog.TestCatalogProvider {
+ @Override
+ public Map<String, Map<String, List<Column>>> getMetadata() {
+ return Map.of(DATABASE, Map.of("t", List.of(new Column("id",
Type.INT))),
+ "db_new", Map.of("t", List.of(new Column("id",
Type.INT))));
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
