[fix](fe) Avoid database object loading for schemata scans (#70010)

### What problem does this PR solve?

Issue Number: N/A

Problem Summary:

Queries such as:

```sql
SELECT COUNT(*) FROM example_catalog.information_schema.schemata;
```

only need database and catalog names. However, `SchemaSchemataScanner`
uses the default `getDbNames` RPC, whose FE handler calls
`catalog.getAllDbs()` to also return database IDs. For external
catalogs, this resolves every database object even though SCHEMATA never
reads those IDs. A large namespace list can therefore churn a much
smaller database-object cache and allocate substantial temporary
metadata while waiting inside a synchronous BE-to-FE RPC. A warm
database-name cache does not remove this work.

This PR adds the optional `include_db_ids` request field, defaulting to
`true`, and lets SCHEMATA set it to `false`. The FE then calls
`catalog.getDbNames()` and omits the unused database IDs. Both paths
share the existing privilege checks, pattern matching, and schema-name
formatting. Callers that need IDs keep the object-enumeration path
without additional intermediate lists.

The new field uses ID 8; field 7 remains reserved. Definitions were
checked on master and branch-3.1/4.0/4.1/4.2. Older FE nodes ignore the
optional field and keep serving the existing response; older BE callers
retain the default ID-inclusive behavior on a new FE.

### Validation

- Added a 100,000-namespace test that rejects any database-object
enumeration and verifies the result-array lengths and omitted IDs.
- Covered privilege/pattern filtering, qualified schema names,
empty/unavailable catalogs, existing ID-returning callers, and binary
deserialization of requests without the new field.
- Confirmed the new behavior tests fail on the original handler before
implementing the change. All 7 targeted FE tests pass, including the
existing `FrontendServiceImplTest#testGetDBNames`.

```bash
bash run-fe-ut.sh --run 'org.apache.doris.service.FrontendServiceGetDbNamesTest,org.apache.doris.service.FrontendServiceImplTest#testGetDBNames'
```

- FE Checkstyle: passed with zero violations.
- Changed BE scanner: clang-format 16 and C++ syntax-only compilation
with regenerated Thrift headers passed.

An in-process FE-handler benchmark uses the actual `ExternalCatalog` /
`TestExternalCatalog` name and object-cache implementations, a synthetic
metadata provider, and stubbed catalog routing and privilege checks. It
compares the original handler source at master commit
`bdb90856e6e468eb495e5bfd67010de90570bfbc` with the patched handler,
using the same dependencies and fixture. Each trial starts with a warm
name cache and cold object cache, whose capacity is 1,000. There are
100,000 synthetic namespaces plus the two system databases. Settings:
JDK 17, 2 GiB heap, 4 active processors; discard one warm-up pair and
report medians from three measured trials.

| Metric | Original master | Patched SCHEMATA path |
| --- | ---: | ---: |
| FE handler elapsed time | 2,984.2 ms | 1,331.1 ms |
| Cumulative current-thread allocation | 2,117.2 MiB | 726.2 MiB |
| Database objects constructed | 100,002 | 0 |
| Name-provider invocations | 1 | 1 |
| Returned rows | 100,002 | 100,002 |

In this fixture, handler time falls by 55.4% (2.24x faster) and
cumulative allocation falls by 65.7%. Allocation is total bytes
allocated during the call, not peak live heap. This benchmark excludes
network transport, Thrift serialization, and SQL execution; it does not
measure a deployed PostgreSQL/Iceberg environment. Full BE linking and a
deployed end-to-end SQL test were not run locally.

### Release note

Improve `information_schema.schemata` enumeration for large catalogs by
avoiding unnecessary database-object loading.

### Check List (For Author)

- Test
    - [ ] Regression test
    - [x] Unit Test
    - [x] Manual test (FE-handler benchmark described above)
    - [ ] No need to test or manual test.
- Behavior changed:
- [x] No. SQL result semantics and existing ID-returning requests are
preserved.
    - [ ] Yes.
- Does this need documentation?
    - [x] No.
    - [ ] Yes.

### Check List (For Reviewer who merge this PR)

- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label

be/src/information_schema/schema_schemata_scanner.cpp

@@ -52,6 +52,8 @@ Status SchemaSchemataScanner::start(RuntimeState* state) {
         return Status::InternalError("used before initial.");
     }
     TGetDbsParams db_params;
+    // SCHEMATA never reads database IDs; avoid creating one database object per namespace in FE.
+    db_params.__set_include_db_ids(false);
     if (nullptr != _param->common_param->wild) {
         db_params.__set_pattern(*(_param->common_param->wild));
     }

fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java

@@ -345,7 +345,6 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -363,6 +362,7 @@ import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.ObjLongConsumer;
 import java.util.stream.Collectors;
 
 // Frontend service used to serve all request for this frontend through
@@ -496,7 +496,7 @@ public class FrontendServiceImpl implements FrontendService.Iface {
         List<Long> dbIds = Lists.newArrayList();
         List<Long> catalogIds = Lists.newArrayList();
 
-        PatternMatcher matcher = null;
+        PatternMatcher matcher;
         if (params.isSetPattern()) {
             try {
                 matcher = PatternMatcher.createMysqlPattern(params.getPattern(),
@@ -504,6 +504,8 @@ public class FrontendServiceImpl implements FrontendService.Iface {
             } catch (PatternMatcherException e) {
                 throw new TException("Pattern is in bad format: " + params.getPattern());
             }
+        } else {
+            matcher = null;
         }
 
         Env env = Env.getCurrentEnv();
@@ -517,55 +519,77 @@ public class FrontendServiceImpl implements FrontendService.Iface {
                             catalog -> new TException("Unknown catalog " + catalog)));
         }
 
+        boolean includeDbIds = params.isIncludeDbIds();
         for (CatalogIf catalog : catalogIfs) {
-            Collection<DatabaseIf> dbs = new HashSet<DatabaseIf>();
+            Collection<DatabaseIf> dbs = Collections.emptyList();
+            List<String> names = Collections.emptyList();
             try {
-                dbs = catalog.getAllDbs();
+                if (includeDbIds) {
+                    dbs = catalog.getAllDbs();
+                } else {
+                    // SCHEMATA only needs names; loading every database churns the object cache on large catalogs.
+                    names = catalog.getDbNames();
+                }
             } catch (Exception e) {
                 LOG.warn("failed to get database names for catalog {}", catalog.getName(), e);
                 // Some external catalog may fail to get databases due to wrong connection info.
             }
+            int dbCount = includeDbIds ? dbs.size() : names.size();
             if (LOG.isDebugEnabled()) {
-                LOG.debug("get db size: {}, in catalog: {}", dbs.size(), catalog.getName());
+                LOG.debug("get db size: {}, in catalog: {}", dbCount, catalog.getName());
             }
-            if (dbs.isEmpty() && params.isSetGetNullCatalog() && params.get_null_catalog) {
+            if (dbCount == 0 && params.isSetGetNullCatalog() && params.get_null_catalog) {
                 catalogNames.add(catalog.getName());
                 dbNames.add("NULL");
                 catalogIds.add(catalog.getId());
-                dbIds.add(-1L);
+                if (includeDbIds) {
+                    dbIds.add(-1L);
+                }
                 continue;
             }
-            if (dbs.isEmpty()) {
+            if (dbCount == 0) {
                 continue;
             }
-            UserIdentity currentUser = null;
+            UserIdentity currentUser;
             if (params.isSetCurrentUserIdent()) {
                 currentUser = UserIdentity.fromThrift(params.current_user_ident);
             } else {
                 currentUser = UserIdentity.createAnalyzedUserIdentWithIp(params.user, params.user_ip);
             }
-            for (DatabaseIf db : dbs) {
-                String dbName = db.getFullName();
+            ObjLongConsumer<String> addDatabase = (dbName, dbId) -> {
                 if (!env.getAccessManager()
                         .checkDbPriv(currentUser, catalog.getName(), dbName, PrivPredicate.SHOW)) {
-                    continue;
+                    return;
                 }
 
                 if (matcher != null && !matcher.match(getMysqlTableSchema(catalog.getName(), dbName))) {
-                    continue;
+                    return;
                 }
 
                 catalogNames.add(catalog.getName());
                 dbNames.add(getMysqlTableSchema(catalog.getName(), dbName));
                 catalogIds.add(catalog.getId());
-                dbIds.add(db.getId());
+                if (includeDbIds) {
+                    dbIds.add(dbId);
+                }
+            };
+            if (includeDbIds) {
+                for (DatabaseIf db : dbs) {
+                    addDatabase.accept(db.getFullName(), db.getId());
+                }
+            } else {
+                for (String dbName : names) {
+                    addDatabase.accept(dbName, -1L);
+                }
             }
         }
 
         result.setDbs(dbNames);
         result.setCatalogs(catalogNames);
         result.setCatalogIds(catalogIds);
-        result.setDbIds(dbIds);
+        if (includeDbIds) {
+            result.setDbIds(dbIds);
+        }
         return result;
     }
 

fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceGetDbNamesTest.java

@@ -0,0 +1,186 @@
+// 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.service;
+
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.DatabaseIf;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.InfoSchemaDb;
+import org.apache.doris.datasource.CatalogIf;
+import org.apache.doris.datasource.CatalogMgr;
+import org.apache.doris.mysql.privilege.AccessControllerManager;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.qe.GlobalVariable;
+import org.apache.doris.thrift.TGetDbsParams;
+import org.apache.doris.thrift.TGetDbsResult;
+
+import org.apache.thrift.TDeserializer;
+import org.apache.thrift.TSerializer;
+import org.apache.thrift.protocol.TBinaryProtocol;
+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.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+public class FrontendServiceGetDbNamesTest {
+    private final Env env = Mockito.mock(Env.class);
+    private final CatalogMgr catalogMgr = Mockito.mock(CatalogMgr.class);
+    private final AccessControllerManager accessManager = Mockito.mock(AccessControllerManager.class);
+    private final CatalogIf catalog = Mockito.mock(CatalogIf.class);
+    private final FrontendServiceImpl service = new FrontendServiceImpl(Mockito.mock(ExecuteEnv.class));
+    private MockedStatic<Env> envMock;
+
+    @BeforeEach
+    public void setUp() {
+        envMock = Mockito.mockStatic(Env.class);
+        envMock.when(Env::getCurrentEnv).thenReturn(env);
+        Mockito.when(env.getCatalogMgr()).thenReturn(catalogMgr);
+        Mockito.when(env.getAccessManager()).thenReturn(accessManager);
+        Mockito.when(catalogMgr.listCatalogs()).thenReturn(Collections.singletonList(catalog));
+        Mockito.when(catalog.getName()).thenReturn("example_catalog");
+        Mockito.when(catalog.getId()).thenReturn(42L);
+        Mockito.when(accessManager.checkDbPriv(Mockito.any(UserIdentity.class), Mockito.anyString(),
+                Mockito.anyString(), Mockito.eq(PrivPredicate.SHOW))).thenReturn(true);
+    }
+
+    @AfterEach
+    public void tearDown() {
+        envMock.close();
+    }
+
+    @Test
+    public void testNamesOnlyDoesNotMaterializeDatabases() throws Exception {
+        List<String> names = new ArrayList<>();
+        for (int i = 0; i < 100_000; i++) {
+            names.add("namespace_" + i);
+        }
+        names.add(InfoSchemaDb.DATABASE_NAME);
+        names.add("mysql");
+        Mockito.when(catalog.getDbNames()).thenReturn(names);
+        Mockito.when(catalog.getAllDbs()).thenThrow(new AssertionError("Database objects must not be loaded"));
+
+        TGetDbsParams params = params();
+        params.setIncludeDbIds(false);
+        TGetDbsResult result = service.getDbNames(params);
+
+        Assertions.assertEquals(names.size(), result.getDbsSize());
+        Assertions.assertEquals(names.size(), result.getCatalogsSize());
+        Assertions.assertEquals(names.size(), result.getCatalogIdsSize());
+        Assertions.assertTrue(result.getCatalogs().stream().allMatch("example_catalog"::equals));
+        Assertions.assertTrue(result.getCatalogIds().stream().allMatch(id -> id == 42L));
+        Assertions.assertFalse(result.isSetDbIds());
+        Mockito.verify(catalog, Mockito.times(1)).getDbNames();
+        Mockito.verify(catalog, Mockito.never()).getAllDbs();
+    }
+
+    @Test
+    public void testNamesOnlyPreservesPrivilegesPatternAndDisplayNames() throws Exception {
+        Mockito.when(catalog.getDbNames()).thenReturn(Arrays.asList("visible_a", "visible_hidden", "other"));
+        Mockito.when(accessManager.checkDbPriv(Mockito.any(UserIdentity.class), Mockito.eq("example_catalog"),
+                Mockito.eq("visible_hidden"), Mockito.eq(PrivPredicate.SHOW))).thenReturn(false);
+
+        boolean original = GlobalVariable.showFullDbNameInInfoSchemaDb;
+        try {
+            for (boolean showFullName : new boolean[] {false, true}) {
+                GlobalVariable.showFullDbNameInInfoSchemaDb = showFullName;
+                TGetDbsParams params = params();
+                params.setIncludeDbIds(false);
+                params.setPattern(showFullName ? "example_catalog.visible%" : "visible%");
+                TGetDbsResult result = service.getDbNames(params);
+                Assertions.assertEquals(Collections.singletonList(
+                        showFullName ? "example_catalog.visible_a" : "visible_a"), result.getDbs());
+                Assertions.assertFalse(result.isSetDbIds());
+            }
+        } finally {
+            GlobalVariable.showFullDbNameInInfoSchemaDb = original;
+        }
+    }
+
+    @Test
+    public void testExistingCallersKeepDatabaseIds() throws Exception {
+        DatabaseIf database = Mockito.mock(DatabaseIf.class);
+        Mockito.when(database.getFullName()).thenReturn("visible_a");
+        Mockito.when(database.getId()).thenReturn(123L);
+        Mockito.when(catalog.getAllDbs()).thenReturn(Collections.singletonList(database));
+
+        for (boolean explicitlySet : new boolean[] {false, true}) {
+            TGetDbsParams params = params();
+            if (explicitlySet) {
+                params.setIncludeDbIds(true);
+            }
+            TGetDbsResult result = service.getDbNames(params);
+            Assertions.assertEquals(Collections.singletonList(123L), result.getDbIds());
+            Assertions.assertEquals(Collections.singletonList(42L), result.getCatalogIds());
+        }
+        Mockito.verify(catalog, Mockito.never()).getDbNames();
+    }
+
+    @Test
+    public void testAbsentIncludeDbIdsOnWireKeepsLegacyDefault() throws Exception {
+        TGetDbsParams legacy = params();
+        legacy.unsetIncludeDbIds();
+        TSerializer serializer = new TSerializer(new TBinaryProtocol.Factory());
+        TDeserializer deserializer = new TDeserializer(new TBinaryProtocol.Factory());
+        TGetDbsParams decoded = new TGetDbsParams();
+        deserializer.deserialize(decoded, serializer.serialize(legacy));
+        Assertions.assertTrue(decoded.isIncludeDbIds());
+
+        legacy.setIncludeDbIds(false);
+        deserializer.deserialize(decoded, serializer.serialize(legacy));
+        Assertions.assertFalse(decoded.isIncludeDbIds());
+    }
+
+    @Test
+    public void testNamesOnlyPreservesEmptyCatalogPlaceholder() throws Exception {
+        Mockito.when(catalog.getDbNames()).thenReturn(Collections.emptyList());
+        TGetDbsParams params = params();
+        params.setIncludeDbIds(false);
+        params.setGetNullCatalog(true);
+
+        TGetDbsResult result = service.getDbNames(params);
+        Assertions.assertEquals(Collections.singletonList("NULL"), result.getDbs());
+        Assertions.assertEquals(Collections.singletonList("example_catalog"), result.getCatalogs());
+        Assertions.assertEquals(Collections.singletonList(42L), result.getCatalogIds());
+        Assertions.assertFalse(result.isSetDbIds());
+    }
+
+    @Test
+    public void testNamesOnlySkipsUnavailableCatalog() throws Exception {
+        Mockito.when(catalog.getDbNames()).thenThrow(new IllegalStateException("Metadata service unavailable"));
+        TGetDbsParams params = params();
+        params.setIncludeDbIds(false);
+
+        TGetDbsResult result = service.getDbNames(params);
+        Assertions.assertTrue(result.getDbs().isEmpty());
+        Assertions.assertTrue(result.getCatalogs().isEmpty());
+        Assertions.assertFalse(result.isSetDbIds());
+    }
+
+    private TGetDbsParams params() {
+        TGetDbsParams params = new TGetDbsParams();
+        params.setCurrentUserIdent(UserIdentity.ROOT.toThrift());
+        return params;
+    }
+}

gensrc/thrift/FrontendService.thrift

@@ -101,6 +101,8 @@ struct TGetDbsParams {
   // Reserved for downstream field `current_roles` to keep thrift field ids
   // wire-compatible across maintained branches. Do not reuse this id.
   7: optional set<string> reserved_field_7
+  // Name-only scanners can avoid materializing database objects and omit db_ids in the response.
+  8: optional bool include_db_ids = true
 }
 
 // getDbNames returns a list of database names , database ids and catalog names ,catalog ids