[Fix] (multi catalog)Fix external table couldn't find db bug (#22074) (…

…#22877)

Nereids LogicalCatalogRelation and PhysicalCatalogRelation getDatabase function only try to search InternalCatalog to find a table. This will cause all external table failed to query because it couldn't find the external database in Internal catalog.
```
mysql> explain select count(*) from multi_partition_orc;
ERROR 1105 (HY000): AnalysisException, msg: Database [default_cluster:multi_partition] does not exist.
```

This pr is using catalog name to find the correct catalog first, and then try to get the database in this catalog.

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java

@@ -200,7 +200,6 @@ private LogicalPlan makeOlapScan(TableIf table, UnboundRelation unboundRelation,
 
     private LogicalPlan getLogicalPlan(TableIf table, UnboundRelation unboundRelation, List<String> tableQualifier,
                                        CascadesContext cascadesContext) {
-        String dbName = tableQualifier.get(1); //[catalogName, dbName, tableName]
         switch (table.getType()) {
             case OLAP:
                 return makeOlapScan(table, unboundRelation, tableQualifier);
@@ -223,14 +222,12 @@ private LogicalPlan getLogicalPlan(TableIf table, UnboundRelation unboundRelatio
                 return new LogicalFileScan(StatementScopeIdGenerator.newRelationId(),
                         (ExternalTable) table, tableQualifier);
             case SCHEMA:
-                return new LogicalSchemaScan(unboundRelation.getRelationId(),
-                        table, ImmutableList.of(dbName));
+                return new LogicalSchemaScan(unboundRelation.getRelationId(), table, tableQualifier);
             case JDBC_EXTERNAL_TABLE:
             case JDBC:
-                return new LogicalJdbcScan(unboundRelation.getRelationId(), table, ImmutableList.of(dbName));
+                return new LogicalJdbcScan(unboundRelation.getRelationId(), table, tableQualifier);
             case ES_EXTERNAL_TABLE:
-                return new LogicalEsScan(unboundRelation.getRelationId(),
-                    (EsExternalTable) table, ImmutableList.of(dbName));
+                return new LogicalEsScan(unboundRelation.getRelationId(), (EsExternalTable) table, tableQualifier);
             default:
                 throw new AnalysisException("Unsupported tableType:" + table.getType());
         }

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/algebra/CatalogRelation.java

@@ -17,7 +17,7 @@
 
 package org.apache.doris.nereids.trees.plans.algebra;
 
-import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.DatabaseIf;
 import org.apache.doris.catalog.TableIf;
 import org.apache.doris.nereids.exceptions.AnalysisException;
 
@@ -26,5 +26,5 @@ public interface CatalogRelation extends Relation {
 
     TableIf getTable();
 
-    Database getDatabase() throws AnalysisException;
+    DatabaseIf getDatabase() throws AnalysisException;
 }

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCatalogRelation.java

@@ -17,9 +17,10 @@
 
 package org.apache.doris.nereids.trees.plans.logical;
 
-import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.DatabaseIf;
 import org.apache.doris.catalog.Env;
 import org.apache.doris.catalog.TableIf;
+import org.apache.doris.datasource.CatalogIf;
 import org.apache.doris.nereids.exceptions.AnalysisException;
 import org.apache.doris.nereids.memo.GroupExpression;
 import org.apache.doris.nereids.properties.LogicalProperties;
@@ -43,6 +44,7 @@
 public abstract class LogicalCatalogRelation extends LogicalRelation implements CatalogRelation {
 
     protected final TableIf table;
+    // [catalogName, databaseName, tableName]
     protected final ImmutableList<String> qualifier;
 
     public LogicalCatalogRelation(RelationId relationId, PlanType type, TableIf table, List<String> qualifier) {
@@ -64,10 +66,19 @@ public TableIf getTable() {
     }
 
     @Override
-    public Database getDatabase() throws AnalysisException {
-        Preconditions.checkArgument(!qualifier.isEmpty());
-        return Env.getCurrentInternalCatalog().getDbOrException(qualifier.get(0),
-                s -> new AnalysisException("Database [" + qualifier.get(0) + "] does not exist."));
+    public DatabaseIf getDatabase() throws AnalysisException {
+        Preconditions.checkArgument(!qualifier.isEmpty(), "qualifier can not be empty");
+        try {
+            CatalogIf catalog = qualifier.size() == 3
+                    ? Env.getCurrentEnv().getCatalogMgr().getCatalogOrException(qualifier.get(0),
+                        s -> new Exception("Catalog [" + qualifier.get(0) + "] does not exist."))
+                    : Env.getCurrentEnv().getCurrentCatalog();
+            return catalog.getDbOrException(qualifier.size() == 3 ? qualifier.get(1) : qualifier.get(0),
+                    s -> new Exception("Database [" + qualifier.get(1) + "] does not exist in catalog ["
+                        + qualifier.get(0) + "]."));
+        } catch (Exception e) {
+            throw new AnalysisException(e.getMessage(), e);
+        }
     }
 
     @Override

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalCatalogRelation.java

@@ -17,9 +17,10 @@
 
 package org.apache.doris.nereids.trees.plans.physical;
 
-import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.DatabaseIf;
 import org.apache.doris.catalog.Env;
 import org.apache.doris.catalog.TableIf;
+import org.apache.doris.datasource.CatalogIf;
 import org.apache.doris.nereids.exceptions.AnalysisException;
 import org.apache.doris.nereids.memo.GroupExpression;
 import org.apache.doris.nereids.properties.LogicalProperties;
@@ -81,10 +82,19 @@ public TableIf getTable() {
     }
 
     @Override
-    public Database getDatabase() throws AnalysisException {
-        Preconditions.checkArgument(!qualifier.isEmpty());
-        return Env.getCurrentInternalCatalog().getDbOrException(qualifier.get(0),
-                s -> new AnalysisException("Database [" + qualifier.get(0) + "] does not exist."));
+    public DatabaseIf getDatabase() throws AnalysisException {
+        Preconditions.checkArgument(!qualifier.isEmpty(), "qualifier can not be empty");
+        try {
+            CatalogIf catalog = qualifier.size() == 3
+                    ? Env.getCurrentEnv().getCatalogMgr().getCatalogOrException(qualifier.get(0),
+                        s -> new Exception("Catalog [" + qualifier.get(0) + "] does not exist."))
+                    : Env.getCurrentEnv().getCurrentCatalog();
+            return catalog.getDbOrException(qualifier.size() == 3 ? qualifier.get(1) : qualifier.get(0),
+                    s -> new Exception("Database [" + qualifier.get(1) + "] does not exist in catalog ["
+                        + qualifier.get(0) + "]."));
+        } catch (Exception e) {
+            throw new AnalysisException(e.getMessage(), e);
+        }
     }
 
     @Override