catalog/system_schema/
information_schema.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15mod cluster_info;
16pub mod columns;
17mod compaction_tasks;
18pub mod flow_statistics;
19pub mod flows;
20mod information_memory_table;
21pub mod key_column_usage;
22mod partitions;
23mod procedure_info;
24pub mod process_list;
25#[cfg(feature = "enterprise")]
26mod recycle_bin;
27mod region_info;
28pub mod region_peers;
29mod region_statistics;
30pub mod schemata;
31mod ssts;
32pub mod statistics;
33mod table_constraints;
34mod table_names;
35mod table_semantics;
36pub mod tables;
37mod views;
38
39#[cfg(all(test, feature = "enterprise"))]
40mod recycle_bin_test;
41
42use std::collections::{HashMap, HashSet};
43use std::sync::{Arc, Weak};
44
45use common_catalog::consts::{self, DEFAULT_CATALOG_NAME, INFORMATION_SCHEMA_NAME};
46use common_error::ext::ErrorExt;
47use common_meta::cluster::NodeInfo;
48use common_meta::datanode::RegionStat;
49use common_meta::key::flow::FlowMetadataManager;
50use common_meta::key::flow::flow_state::FlowStat;
51use common_meta::kv_backend::KvBackendRef;
52use common_procedure::ProcedureInfo;
53use common_recordbatch::SendableRecordBatchStream;
54use datafusion::error::DataFusionError;
55use datafusion::logical_expr::LogicalPlan;
56use datafusion::physical_plan::ExecutionPlan;
57use datatypes::schema::SchemaRef;
58use lazy_static::lazy_static;
59use paste::paste;
60use process_list::InformationSchemaProcessList;
61use region_info::InformationSchemaRegionInfo;
62use store_api::compaction_task::CompactionTaskEntry;
63use store_api::metric_engine_consts::{
64    MEMTABLE_PARTITION_TREE_PRIMARY_KEY_ENCODING, PRIMARY_KEY_ENCODING,
65};
66use store_api::region_info::RegionInfoEntry;
67use store_api::sst_entry::{ManifestSstEntry, PuffinIndexMetaEntry, StorageSstEntry};
68use store_api::storage::ScanRequest;
69use table::TableRef;
70use table::requests::{SEMANTIC_TABLE_SCOPE, TableOptions};
71pub use table_names::*;
72use views::InformationSchemaViews;
73
74use self::columns::InformationSchemaColumns;
75use crate::CatalogManager;
76use crate::error::Error;
77use crate::process_manager::ProcessManagerRef;
78use crate::system_schema::information_schema::cluster_info::InformationSchemaClusterInfo;
79use crate::system_schema::information_schema::compaction_tasks::InformationSchemaCompactionTasks;
80use crate::system_schema::information_schema::flow_statistics::InformationSchemaFlowStatistics;
81use crate::system_schema::information_schema::flows::InformationSchemaFlows;
82use crate::system_schema::information_schema::information_memory_table::get_schema_columns;
83use crate::system_schema::information_schema::key_column_usage::InformationSchemaKeyColumnUsage;
84use crate::system_schema::information_schema::partitions::InformationSchemaPartitions;
85#[cfg(feature = "enterprise")]
86use crate::system_schema::information_schema::recycle_bin::InformationSchemaRecycleBin;
87use crate::system_schema::information_schema::region_peers::InformationSchemaRegionPeers;
88use crate::system_schema::information_schema::schemata::InformationSchemaSchemata;
89use crate::system_schema::information_schema::ssts::{
90    InformationSchemaSstsIndexMeta, InformationSchemaSstsManifest, InformationSchemaSstsStorage,
91};
92use crate::system_schema::information_schema::statistics::InformationSchemaStatistics;
93use crate::system_schema::information_schema::table_constraints::InformationSchemaTableConstraints;
94use crate::system_schema::information_schema::table_semantics::InformationSchemaTableSemantics;
95use crate::system_schema::information_schema::tables::InformationSchemaTables;
96use crate::system_schema::memory_table::MemoryTable;
97pub(crate) use crate::system_schema::predicate::Predicates;
98use crate::system_schema::{SystemSchemaProvider, SystemTable, build_system_table};
99
100const DENSE_PRIMARY_KEY_ENCODING: &str = "dense";
101const SPARSE_PRIMARY_KEY_ENCODING: &str = "sparse";
102
103pub(crate) fn primary_key_encoding_index_type(options: &HashMap<String, String>) -> &'static str {
104    options
105        .get(PRIMARY_KEY_ENCODING)
106        .or_else(|| options.get(MEMTABLE_PARTITION_TREE_PRIMARY_KEY_ENCODING))
107        .map(|value| {
108            if value.eq_ignore_ascii_case(SPARSE_PRIMARY_KEY_ENCODING) {
109                SPARSE_PRIMARY_KEY_ENCODING
110            } else {
111                DENSE_PRIMARY_KEY_ENCODING
112            }
113        })
114        .unwrap_or(DENSE_PRIMARY_KEY_ENCODING)
115}
116
117lazy_static! {
118    // Memory tables in `information_schema`.
119    static ref MEMORY_TABLES: &'static [&'static str] = &[
120        ENGINES,
121        COLUMN_PRIVILEGES,
122        COLUMN_STATISTICS,
123        CHARACTER_SETS,
124        COLLATIONS,
125        COLLATION_CHARACTER_SET_APPLICABILITY,
126        CHECK_CONSTRAINTS,
127        EVENTS,
128        FILES,
129        OPTIMIZER_TRACE,
130        PARAMETERS,
131        PROFILING,
132        REFERENTIAL_CONSTRAINTS,
133        ROUTINES,
134        SCHEMA_PRIVILEGES,
135        TABLE_PRIVILEGES,
136        GLOBAL_STATUS,
137        SESSION_STATUS,
138        PLUGINS,
139        USER_PRIVILEGES,
140        PROCESSLIST,
141    ];
142}
143
144macro_rules! setup_memory_table {
145    ($name: expr, $scope: ident) => {
146        paste! {
147            {
148                let (schema, columns) = get_schema_columns($name);
149                Some(Arc::new(MemoryTable::new(
150                    consts::[<INFORMATION_SCHEMA_ $name  _TABLE_ID>],
151                    $name,
152                    schema,
153                    columns,
154                    InformationSchemaTableScope::$scope
155                )) as _)
156            }
157        }
158    };
159}
160
161pub struct MakeInformationTableRequest {
162    pub catalog_name: String,
163    pub catalog_manager: Weak<dyn CatalogManager>,
164    pub kv_backend: KvBackendRef,
165}
166
167/// The visibility guaranteed by an information-schema table provider.
168#[derive(Debug, Clone, Copy, PartialEq, Eq)]
169pub enum InformationSchemaTableScope {
170    /// Every query is filtered to the catalog supplied when constructing the provider.
171    Catalog,
172    /// Catalog-independent data that is safe for authenticated non-admin users to read.
173    PublicGlobal,
174    /// Shared data without a non-admin visibility guarantee, including compatibility stubs.
175    Global,
176}
177
178impl InformationSchemaTableScope {
179    pub const fn as_str(self) -> &'static str {
180        match self {
181            Self::Catalog => "catalog",
182            Self::PublicGlobal => "public_global",
183            Self::Global => "global",
184        }
185    }
186}
187
188/// A factory trait for making information schema tables.
189///
190/// This trait allows for extensibility of the information schema by providing
191/// a way to dynamically create custom information schema tables.
192pub trait InformationSchemaTableFactory {
193    /// Creates a table whose scope is declared by its [`InformationTable`] implementation.
194    fn make_information_table(&self, req: MakeInformationTableRequest) -> InformationTableRef;
195}
196
197pub type InformationSchemaTableFactoryRef = Arc<dyn InformationSchemaTableFactory + Send + Sync>;
198
199/// The `information_schema` tables info provider.
200pub struct InformationSchemaProvider {
201    catalog_name: String,
202    catalog_manager: Weak<dyn CatalogManager>,
203    process_manager: Option<ProcessManagerRef>,
204    flow_metadata_manager: Arc<FlowMetadataManager>,
205    tables: HashMap<String, TableRef>,
206    kv_backend: KvBackendRef,
207    extra_table_factories: HashMap<String, InformationSchemaTableFactoryRef>,
208}
209
210impl SystemSchemaProvider for InformationSchemaProvider {
211    fn tables(&self) -> &HashMap<String, TableRef> {
212        assert!(!self.tables.is_empty());
213
214        &self.tables
215    }
216}
217
218impl InformationSchemaProvider {
219    fn information_table(&self, name: &str) -> Option<InformationTableRef> {
220        if let Some(factory) = self.extra_table_factories.get(name) {
221            let req = MakeInformationTableRequest {
222                catalog_name: self.catalog_name.clone(),
223                catalog_manager: self.catalog_manager.clone(),
224                kv_backend: self.kv_backend.clone(),
225            };
226            return Some(factory.make_information_table(req));
227        }
228
229        match name.to_ascii_lowercase().as_str() {
230            TABLES => Some(Arc::new(InformationSchemaTables::new(
231                self.catalog_name.clone(),
232                self.catalog_manager.clone(),
233            )) as _),
234            COLUMNS => Some(Arc::new(InformationSchemaColumns::new(
235                self.catalog_name.clone(),
236                self.catalog_manager.clone(),
237            )) as _),
238            ENGINES => setup_memory_table!(ENGINES, PublicGlobal),
239            COLUMN_PRIVILEGES => setup_memory_table!(COLUMN_PRIVILEGES, Global),
240            COLUMN_STATISTICS => setup_memory_table!(COLUMN_STATISTICS, Global),
241            BUILD_INFO => setup_memory_table!(BUILD_INFO, PublicGlobal),
242            CHARACTER_SETS => setup_memory_table!(CHARACTER_SETS, PublicGlobal),
243            COLLATIONS => setup_memory_table!(COLLATIONS, PublicGlobal),
244            COLLATION_CHARACTER_SET_APPLICABILITY => {
245                setup_memory_table!(COLLATION_CHARACTER_SET_APPLICABILITY, PublicGlobal)
246            }
247            CHECK_CONSTRAINTS => setup_memory_table!(CHECK_CONSTRAINTS, Global),
248            EVENTS => setup_memory_table!(EVENTS, Global),
249            FILES => setup_memory_table!(FILES, Global),
250            OPTIMIZER_TRACE => setup_memory_table!(OPTIMIZER_TRACE, Global),
251            PARAMETERS => setup_memory_table!(PARAMETERS, Global),
252            PROFILING => setup_memory_table!(PROFILING, Global),
253            REFERENTIAL_CONSTRAINTS => setup_memory_table!(REFERENTIAL_CONSTRAINTS, Global),
254            ROUTINES => setup_memory_table!(ROUTINES, Global),
255            SCHEMA_PRIVILEGES => setup_memory_table!(SCHEMA_PRIVILEGES, Global),
256            TABLE_PRIVILEGES => setup_memory_table!(TABLE_PRIVILEGES, Global),
257            GLOBAL_STATUS => setup_memory_table!(GLOBAL_STATUS, Global),
258            SESSION_STATUS => setup_memory_table!(SESSION_STATUS, Global),
259            PLUGINS => setup_memory_table!(PLUGINS, Global),
260            USER_PRIVILEGES => setup_memory_table!(USER_PRIVILEGES, Global),
261            PROCESSLIST => setup_memory_table!(PROCESSLIST, Global),
262            KEY_COLUMN_USAGE => Some(Arc::new(InformationSchemaKeyColumnUsage::new(
263                self.catalog_name.clone(),
264                self.catalog_manager.clone(),
265            )) as _),
266            SCHEMATA => Some(Arc::new(InformationSchemaSchemata::new(
267                self.catalog_name.clone(),
268                self.catalog_manager.clone(),
269            )) as _),
270            PARTITIONS => Some(Arc::new(InformationSchemaPartitions::new(
271                self.catalog_name.clone(),
272                self.catalog_manager.clone(),
273            )) as _),
274            REGION_PEERS => Some(Arc::new(InformationSchemaRegionPeers::new(
275                self.catalog_name.clone(),
276                self.catalog_manager.clone(),
277            )) as _),
278            TABLE_CONSTRAINTS => Some(Arc::new(InformationSchemaTableConstraints::new(
279                self.catalog_name.clone(),
280                self.catalog_manager.clone(),
281            )) as _),
282            STATISTICS => Some(Arc::new(InformationSchemaStatistics::new(
283                self.catalog_name.clone(),
284                self.catalog_manager.clone(),
285            )) as _),
286            CLUSTER_INFO => Some(Arc::new(InformationSchemaClusterInfo::new(
287                self.catalog_manager.clone(),
288            )) as _),
289            VIEWS => Some(Arc::new(InformationSchemaViews::new(
290                self.catalog_name.clone(),
291                self.catalog_manager.clone(),
292            )) as _),
293            FLOWS => Some(Arc::new(InformationSchemaFlows::new(
294                self.catalog_name.clone(),
295                self.catalog_manager.clone(),
296                self.flow_metadata_manager.clone(),
297            )) as _),
298            FLOW_STATISTICS => Some(Arc::new(InformationSchemaFlowStatistics::new(
299                self.catalog_name.clone(),
300                self.catalog_manager.clone(),
301                self.flow_metadata_manager.clone(),
302            )) as _),
303            PROCEDURE_INFO => Some(
304                Arc::new(procedure_info::InformationSchemaProcedureInfo::new(
305                    self.catalog_manager.clone(),
306                )) as _,
307            ),
308            #[cfg(feature = "enterprise")]
309            RECYCLE_BIN => Some(Arc::new(InformationSchemaRecycleBin::new(
310                self.catalog_name.clone(),
311                self.catalog_manager.clone(),
312            )) as _),
313            REGION_STATISTICS => Some(Arc::new(
314                region_statistics::InformationSchemaRegionStatistics::new(
315                    self.catalog_manager.clone(),
316                ),
317            ) as _),
318            COMPACTION_TASKS => Some(Arc::new(InformationSchemaCompactionTasks::new(
319                self.catalog_manager.clone(),
320            )) as _),
321            REGION_INFO => Some(Arc::new(InformationSchemaRegionInfo::new(
322                self.catalog_manager.clone(),
323            )) as _),
324            PROCESS_LIST => self
325                .process_manager
326                .as_ref()
327                .map(|p| Arc::new(InformationSchemaProcessList::new(p.clone())) as _),
328            SSTS_MANIFEST => Some(Arc::new(InformationSchemaSstsManifest::new(
329                self.catalog_manager.clone(),
330            )) as _),
331            SSTS_STORAGE => Some(Arc::new(InformationSchemaSstsStorage::new(
332                self.catalog_manager.clone(),
333            )) as _),
334            SSTS_INDEX_META => Some(Arc::new(InformationSchemaSstsIndexMeta::new(
335                self.catalog_manager.clone(),
336            )) as _),
337            TABLE_SEMANTICS => Some(Arc::new(InformationSchemaTableSemantics::new(
338                self.catalog_name.clone(),
339                self.catalog_manager.clone(),
340            )) as _),
341            _ => None,
342        }
343    }
344}
345
346impl InformationSchemaProvider {
347    pub fn new(
348        catalog_name: String,
349        catalog_manager: Weak<dyn CatalogManager>,
350        flow_metadata_manager: Arc<FlowMetadataManager>,
351        process_manager: Option<ProcessManagerRef>,
352        kv_backend: KvBackendRef,
353    ) -> Self {
354        let mut provider = Self {
355            catalog_name,
356            catalog_manager,
357            flow_metadata_manager,
358            process_manager,
359            tables: HashMap::new(),
360            kv_backend,
361            extra_table_factories: HashMap::new(),
362        };
363
364        provider.build_tables();
365
366        provider
367    }
368
369    pub(crate) fn with_extra_table_factories(
370        mut self,
371        factories: HashMap<String, InformationSchemaTableFactoryRef>,
372    ) -> Self {
373        self.extra_table_factories = factories;
374        self.build_tables();
375        self
376    }
377
378    fn build_tables(&mut self) {
379        let mut table_names = HashSet::new();
380
381        // SECURITY NOTE:
382        // Carefully consider the tables that may expose sensitive cluster configurations,
383        // authentication details, and other critical information.
384        // Only put these tables under `greptime` catalog to prevent info leak.
385        if self.catalog_name == DEFAULT_CATALOG_NAME {
386            table_names.extend([
387                REGION_PEERS,
388                CLUSTER_INFO,
389                PROCEDURE_INFO,
390                REGION_STATISTICS,
391                COMPACTION_TASKS,
392                REGION_INFO,
393                SSTS_MANIFEST,
394                SSTS_STORAGE,
395                SSTS_INDEX_META,
396            ]);
397        }
398
399        table_names.extend([
400            BUILD_INFO,
401            TABLES,
402            VIEWS,
403            SCHEMATA,
404            COLUMNS,
405            KEY_COLUMN_USAGE,
406            TABLE_CONSTRAINTS,
407            STATISTICS,
408            FLOWS,
409            FLOW_STATISTICS,
410            #[cfg(feature = "enterprise")]
411            RECYCLE_BIN,
412            TABLE_SEMANTICS,
413            PARTITIONS,
414        ]);
415        if self.process_manager.is_some() {
416            table_names.insert(PROCESS_LIST);
417        }
418        table_names.extend(MEMORY_TABLES.iter().copied());
419        table_names.extend(self.extra_table_factories.keys().map(String::as_str));
420        self.tables = table_names
421            .into_iter()
422            .map(|name| {
423                let table = self.information_table(name).expect(name);
424                let options = TableOptions {
425                    extra_options: HashMap::from([(
426                        SEMANTIC_TABLE_SCOPE.to_string(),
427                        table.scope().as_str().to_string(),
428                    )]),
429                    ..Default::default()
430                };
431                let table = build_system_table(
432                    self.catalog_name.clone(),
433                    INFORMATION_SCHEMA_NAME,
434                    table,
435                    options,
436                );
437                (name.to_string(), table)
438            })
439            .collect();
440    }
441}
442
443/// An information-schema table with an explicit visibility guarantee.
444pub trait InformationTable: SystemTable {
445    /// Returns the visibility guaranteed by this table's query implementation.
446    fn scope(&self) -> InformationSchemaTableScope;
447}
448
449pub type InformationTableRef = Arc<dyn InformationTable + Send + Sync>;
450
451pub type InformationExtensionRef = Arc<dyn InformationExtension<Error = Error> + Send + Sync>;
452
453/// The `InformationExtension` trait provides the extension methods for the `information_schema` tables.
454#[async_trait::async_trait]
455pub trait InformationExtension {
456    type Error: ErrorExt;
457
458    /// Gets the nodes information.
459    async fn nodes(&self) -> std::result::Result<Vec<NodeInfo>, Self::Error>;
460
461    /// Gets the procedures information.
462    async fn procedures(&self) -> std::result::Result<Vec<(String, ProcedureInfo)>, Self::Error>;
463
464    /// Gets the region statistics.
465    async fn region_stats(&self) -> std::result::Result<Vec<RegionStat>, Self::Error>;
466
467    /// Get the flow statistics. If no flownode is available, return `None`.
468    async fn flow_stats(&self) -> std::result::Result<Option<FlowStat>, Self::Error>;
469
470    /// Inspects the datanode.
471    async fn inspect_datanode(
472        &self,
473        request: DatanodeInspectRequest,
474    ) -> std::result::Result<SendableRecordBatchStream, Self::Error>;
475
476    /// Whether this extension can execute an inspection query on every datanode.
477    fn supports_inspect_query_plan(&self) -> bool {
478        false
479    }
480
481    /// Builds a plan that executes an inspection query independently on each datanode.
482    /// The caller supplies a plan over reserved inspection tables and retains any
483    /// required global aggregation. The returned schema is the query output schema.
484    fn inspect_datanode_query_plan(
485        &self,
486        _plan: LogicalPlan,
487    ) -> std::result::Result<Option<Arc<dyn ExecutionPlan>>, Self::Error> {
488        Ok(None)
489    }
490
491    /// Builds a physical plan for datanode inspect if the extension can expose
492    /// the distributed fan-in semantics to DataFusion.
493    fn inspect_datanode_plan(
494        &self,
495        _request: DatanodeInspectRequest,
496        _schema: SchemaRef,
497    ) -> std::result::Result<Option<Arc<dyn ExecutionPlan>>, Self::Error> {
498        Ok(None)
499    }
500}
501
502/// The request to inspect the datanode.
503#[derive(Debug, Clone, PartialEq)]
504pub struct DatanodeInspectRequest {
505    /// Kind to fetch from datanode.
506    pub kind: DatanodeInspectKind,
507
508    /// Pushdown scan configuration (projection/predicate/limit) for the returned stream.
509    /// This allows server-side filtering to reduce I/O and network costs.
510    pub scan: ScanRequest,
511}
512
513/// The kind of the datanode inspect request.
514#[derive(Debug, Clone, Copy, PartialEq, Eq)]
515pub enum DatanodeInspectKind {
516    /// Live compaction cycles.
517    CompactionTasks,
518    /// List SST entries recorded in manifest
519    SstManifest,
520    /// List SST entries discovered in storage layer
521    SstStorage,
522    /// List index metadata collected from manifest
523    SstIndexMeta,
524    /// List region runtime and manifest info
525    RegionInfo,
526}
527
528impl DatanodeInspectRequest {
529    /// Builds a logical plan for the datanode inspect request.
530    pub fn build_plan(self) -> std::result::Result<LogicalPlan, DataFusionError> {
531        match self.kind {
532            DatanodeInspectKind::CompactionTasks => CompactionTaskEntry::build_plan(self.scan),
533            DatanodeInspectKind::SstManifest => ManifestSstEntry::build_plan(self.scan),
534            DatanodeInspectKind::SstStorage => StorageSstEntry::build_plan(self.scan),
535            DatanodeInspectKind::SstIndexMeta => PuffinIndexMetaEntry::build_plan(self.scan),
536            DatanodeInspectKind::RegionInfo => RegionInfoEntry::build_plan(self.scan),
537        }
538    }
539}
540pub struct NoopInformationExtension;
541
542#[async_trait::async_trait]
543impl InformationExtension for NoopInformationExtension {
544    type Error = Error;
545
546    async fn nodes(&self) -> std::result::Result<Vec<NodeInfo>, Self::Error> {
547        Ok(vec![])
548    }
549
550    async fn procedures(&self) -> std::result::Result<Vec<(String, ProcedureInfo)>, Self::Error> {
551        Ok(vec![])
552    }
553
554    async fn region_stats(&self) -> std::result::Result<Vec<RegionStat>, Self::Error> {
555        Ok(vec![])
556    }
557
558    async fn flow_stats(&self) -> std::result::Result<Option<FlowStat>, Self::Error> {
559        Ok(None)
560    }
561
562    async fn inspect_datanode(
563        &self,
564        _request: DatanodeInspectRequest,
565    ) -> std::result::Result<SendableRecordBatchStream, Self::Error> {
566        Ok(common_recordbatch::RecordBatches::empty().as_stream())
567    }
568}
569
570#[cfg(test)]
571mod tests {
572    use cache::{build_fundamental_cache_registry, with_default_composite_cache_registry};
573    use common_meta::cache::{CacheRegistryBuilder, LayeredCacheRegistryBuilder};
574    use common_meta::ddl::test_util::create_table::test_create_table_task;
575    use common_meta::key::schema_name::SchemaNameKey;
576    use common_meta::key::table_route::TableRouteValue;
577    use common_meta::kv_backend::memory::MemoryKvBackend;
578    use common_meta::kv_backend::txn::TxnService;
579    use common_meta::wal_provider::RegionWalOptions;
580    use datatypes::schema::Schema;
581    use futures_util::TryStreamExt;
582    use store_api::region_info::RegionInfoEntry;
583    use store_api::storage::TableId;
584
585    use super::*;
586    use crate::error::Result;
587    use crate::information_schema::NoopInformationExtension;
588    use crate::kvbackend::KvBackendCatalogManagerBuilder;
589    use crate::memory::MemoryCatalogManager;
590    use crate::process_manager::ProcessManager;
591
592    struct ScopedFactory {
593        name: &'static str,
594        scope: InformationSchemaTableScope,
595    }
596
597    impl InformationSchemaTableFactory for ScopedFactory {
598        fn make_information_table(&self, _req: MakeInformationTableRequest) -> InformationTableRef {
599            Arc::new(ScopedTable {
600                name: self.name,
601                scope: self.scope,
602            })
603        }
604    }
605
606    struct ScopedTable {
607        name: &'static str,
608        scope: InformationSchemaTableScope,
609    }
610
611    impl InformationTable for ScopedTable {
612        fn scope(&self) -> InformationSchemaTableScope {
613            self.scope
614        }
615    }
616
617    impl SystemTable for ScopedTable {
618        fn table_id(&self) -> TableId {
619            999
620        }
621
622        fn table_name(&self) -> &'static str {
623            self.name
624        }
625
626        fn schema(&self) -> SchemaRef {
627            Arc::new(Schema::new(vec![]))
628        }
629
630        fn to_stream(&self, _request: ScanRequest) -> Result<SendableRecordBatchStream> {
631            Ok(common_recordbatch::RecordBatches::empty().as_stream())
632        }
633    }
634
635    #[test]
636    fn every_registered_information_schema_table_has_scope() {
637        use InformationSchemaTableScope::{Catalog, Global, PublicGlobal};
638
639        let manager: Arc<dyn CatalogManager> = MemoryCatalogManager::new();
640        let backend = Arc::new(MemoryKvBackend::default());
641        let factories: HashMap<_, InformationSchemaTableFactoryRef> = [
642            ("future_catalog_table", Catalog),
643            ("future_global_table", Global),
644            ("future_public_table", PublicGlobal),
645            (BUILD_INFO, Global),
646        ]
647        .into_iter()
648        .map(|(name, scope)| {
649            (
650                name.to_string(),
651                Arc::new(ScopedFactory { name, scope }) as _,
652            )
653        })
654        .collect();
655        let catalog_tables = [
656            TABLES,
657            VIEWS,
658            SCHEMATA,
659            COLUMNS,
660            KEY_COLUMN_USAGE,
661            TABLE_CONSTRAINTS,
662            STATISTICS,
663            FLOWS,
664            FLOW_STATISTICS,
665            TABLE_SEMANTICS,
666            PARTITIONS,
667            REGION_PEERS,
668            #[cfg(feature = "enterprise")]
669            RECYCLE_BIN,
670            "future_catalog_table",
671        ];
672        let public_global_tables = [
673            ENGINES,
674            CHARACTER_SETS,
675            COLLATIONS,
676            COLLATION_CHARACTER_SET_APPLICABILITY,
677            "future_public_table",
678        ];
679
680        for (catalog, with_process_manager) in [
681            (DEFAULT_CATALOG_NAME, true),
682            (DEFAULT_CATALOG_NAME, false),
683            ("tenant", true),
684            ("tenant", false),
685        ] {
686            let provider = InformationSchemaProvider::new(
687                catalog.to_string(),
688                Arc::downgrade(&manager),
689                Arc::new(FlowMetadataManager::new(backend.clone())),
690                with_process_manager.then(|| Arc::new(ProcessManager::new(String::new(), None))),
691                backend.clone(),
692            );
693            for name in [
694                BUILD_INFO,
695                ENGINES,
696                CHARACTER_SETS,
697                COLLATIONS,
698                COLLATION_CHARACTER_SET_APPLICABILITY,
699            ] {
700                assert_eq!(
701                    provider
702                        .table(name)
703                        .unwrap()
704                        .table_info()
705                        .meta
706                        .options
707                        .extra_options
708                        .get(SEMANTIC_TABLE_SCOPE)
709                        .map(String::as_str),
710                    Some(PublicGlobal.as_str()),
711                    "{catalog}.{name}"
712                );
713            }
714            // An extension replacing a built-in supplies the replacement table's scope.
715            let provider = provider.with_extra_table_factories(factories.clone());
716            let expected_count =
717                45 + usize::from(with_process_manager) + usize::from(cfg!(feature = "enterprise"))
718                    - if catalog == DEFAULT_CATALOG_NAME {
719                        0
720                    } else {
721                        9
722                    };
723            assert_eq!(provider.tables().len(), expected_count);
724            for (name, table) in provider.tables() {
725                let scope = if catalog_tables.contains(&name.as_str()) {
726                    Catalog
727                } else if public_global_tables.contains(&name.as_str()) {
728                    PublicGlobal
729                } else {
730                    Global
731                };
732                assert_eq!(
733                    table
734                        .table_info()
735                        .meta
736                        .options
737                        .extra_options
738                        .get(SEMANTIC_TABLE_SCOPE)
739                        .map(String::as_str),
740                    Some(scope.as_str()),
741                    "{catalog}.{name}"
742                );
743            }
744            assert_eq!(
745                provider.table(REGION_PEERS).is_some(),
746                catalog == DEFAULT_CATALOG_NAME
747            );
748            assert_eq!(provider.table(PROCESS_LIST).is_some(), with_process_manager);
749            if let Some(table) = provider.table(PROCESS_LIST) {
750                assert!(table.schema().column_schema_by_name("catalog").is_some());
751            }
752        }
753    }
754
755    #[tokio::test]
756    async fn newly_accessible_information_schema_tables_filter_catalogs() {
757        let backend = Arc::new(MemoryKvBackend::default());
758        let caches = LayeredCacheRegistryBuilder::default()
759            .add_cache_registry(CacheRegistryBuilder::default().build())
760            .add_cache_registry(build_fundamental_cache_registry(backend.clone()));
761        let manager = KvBackendCatalogManagerBuilder::new(
762            Arc::new(NoopInformationExtension),
763            backend.clone(),
764            Arc::new(
765                with_default_composite_cache_registry(caches)
766                    .unwrap()
767                    .build(),
768            ),
769        )
770        .build();
771        let metadata = manager.table_metadata_manager_ref();
772        let flow_metadata = FlowMetadataManager::new(backend.clone());
773        for (catalog, name, id) in [
774            ("tenant_a", "visible_a", 1024),
775            ("tenant_b", "visible_b", 1025),
776        ] {
777            metadata
778                .schema_manager()
779                .create(SchemaNameKey::new(catalog, "public"), None, false)
780                .await
781                .unwrap();
782            let mut info = test_create_table_task(name, id).table_info;
783            info.catalog_name = catalog.to_string();
784            info.meta
785                .options
786                .extra_options
787                .insert(SEMANTIC_TABLE_SCOPE.to_string(), "catalog".to_string());
788            metadata
789                .create_table_metadata(
790                    info,
791                    TableRouteValue::physical(vec![]),
792                    RegionWalOptions::default(),
793                )
794                .await
795                .unwrap();
796            let (txn, _) = flow_metadata
797                .flow_name_manager()
798                .build_create_txn(catalog, name, id)
799                .unwrap();
800            backend.txn(txn).await.unwrap();
801        }
802
803        for (catalog, visible, hidden) in [
804            ("tenant_a", "visible_a", "visible_b"),
805            ("tenant_b", "visible_b", "visible_a"),
806        ] {
807            for name in [TABLE_SEMANTICS, STATISTICS, FLOW_STATISTICS] {
808                let table = manager
809                    .table(catalog, INFORMATION_SCHEMA_NAME, name, None)
810                    .await
811                    .unwrap()
812                    .unwrap();
813                let batches = table
814                    .scan_to_stream(ScanRequest::default())
815                    .await
816                    .unwrap()
817                    .try_collect::<Vec<_>>()
818                    .await
819                    .unwrap();
820                let output = batches
821                    .iter()
822                    .map(|batch| batch.pretty_print())
823                    .collect::<Vec<_>>()
824                    .join("\n");
825                assert!(output.contains(visible), "{catalog}.{name}: {output}");
826                assert!(!output.contains(hidden), "{catalog}.{name}: {output}");
827            }
828        }
829        assert!(
830            !manager
831                .information_schema_provider()
832                .tables()
833                .contains_key("visible_a")
834        );
835    }
836
837    #[test]
838    fn test_compaction_tasks_build_plan() {
839        let request = DatanodeInspectRequest {
840            kind: DatanodeInspectKind::CompactionTasks,
841            scan: ScanRequest {
842                projection: Some(vec![0]),
843                filters: vec![
844                    datafusion::logical_expr::col("state")
845                        .eq(datafusion::logical_expr::lit("running")),
846                ],
847                limit: Some(1),
848                ..Default::default()
849            },
850        };
851        let plan = request.build_plan().unwrap();
852        assert_eq!(plan.schema().fields().len(), 1);
853        assert!(
854            plan.display_indent()
855                .to_string()
856                .contains("__compaction_tasks")
857        );
858    }
859
860    #[test]
861    fn test_datanode_inspect_region_info_build_plan() {
862        let plan = DatanodeInspectRequest {
863            kind: DatanodeInspectKind::RegionInfo,
864            scan: ScanRequest::default(),
865        }
866        .build_plan()
867        .unwrap();
868
869        let LogicalPlan::TableScan(scan) = plan else {
870            panic!("expected table scan");
871        };
872        assert_eq!(
873            scan.table_name.to_string(),
874            RegionInfoEntry::reserved_table_name_for_inspection()
875        );
876    }
877}
catalog/system_schema/information_schema.rs

catalog/system_schema/
information_schema.rs
