catalog/table_source/
dummy_catalog.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
15//! Dummy catalog for region server.
16
17use std::fmt;
18use std::sync::Arc;
19
20use async_trait::async_trait;
21use common_catalog::format_full_table_name;
22use datafusion::catalog::{CatalogProvider, CatalogProviderList, SchemaProvider};
23use datafusion::datasource::TableProvider;
24use session::context::QueryContextRef;
25use snafu::OptionExt;
26use table::table::adapter::DfTableProviderAdapter;
27
28use crate::CatalogManagerRef;
29use crate::error::TableNotExistSnafu;
30
31/// Delegate the resolving requests to the `[CatalogManager]` unconditionally.
32#[derive(Clone)]
33pub struct DummyCatalogList {
34    catalog_manager: CatalogManagerRef,
35    query_ctx: Option<QueryContextRef>,
36}
37
38impl DummyCatalogList {
39    /// Creates a new catalog list with the given catalog manager (no query context).
40    pub fn new(catalog_manager: CatalogManagerRef) -> Self {
41        Self {
42            catalog_manager,
43            query_ctx: None,
44        }
45    }
46
47    /// Creates a new catalog list with the given catalog manager and query context.
48    pub fn new_with_query_ctx(
49        catalog_manager: CatalogManagerRef,
50        query_ctx: QueryContextRef,
51    ) -> Self {
52        Self {
53            catalog_manager,
54            query_ctx: Some(query_ctx),
55        }
56    }
57}
58
59impl fmt::Debug for DummyCatalogList {
60    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
61        f.debug_struct("DummyCatalogList").finish()
62    }
63}
64
65impl CatalogProviderList for DummyCatalogList {
66    fn register_catalog(
67        &self,
68        _name: String,
69        _catalog: Arc<dyn CatalogProvider>,
70    ) -> Option<Arc<dyn CatalogProvider>> {
71        None
72    }
73
74    fn catalog_names(&self) -> Vec<String> {
75        vec![]
76    }
77
78    fn catalog(&self, catalog_name: &str) -> Option<Arc<dyn CatalogProvider>> {
79        Some(Arc::new(DummyCatalogProvider {
80            catalog_name: catalog_name.to_string(),
81            catalog_manager: self.catalog_manager.clone(),
82            query_ctx: self.query_ctx.clone(),
83        }))
84    }
85}
86
87/// A dummy catalog provider for [DummyCatalogList].
88#[derive(Clone)]
89struct DummyCatalogProvider {
90    catalog_name: String,
91    catalog_manager: CatalogManagerRef,
92    query_ctx: Option<QueryContextRef>,
93}
94
95impl CatalogProvider for DummyCatalogProvider {
96    fn schema_names(&self) -> Vec<String> {
97        vec![]
98    }
99
100    fn schema(&self, schema_name: &str) -> Option<Arc<dyn SchemaProvider>> {
101        Some(Arc::new(DummySchemaProvider {
102            catalog_name: self.catalog_name.clone(),
103            schema_name: schema_name.to_string(),
104            catalog_manager: self.catalog_manager.clone(),
105            query_ctx: self.query_ctx.clone(),
106        }))
107    }
108}
109
110impl fmt::Debug for DummyCatalogProvider {
111    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
112        f.debug_struct("DummyCatalogProvider")
113            .field("catalog_name", &self.catalog_name)
114            .finish()
115    }
116}
117
118/// A dummy schema provider for [DummyCatalogList].
119#[derive(Clone)]
120struct DummySchemaProvider {
121    catalog_name: String,
122    schema_name: String,
123    catalog_manager: CatalogManagerRef,
124    query_ctx: Option<QueryContextRef>,
125}
126
127#[async_trait]
128impl SchemaProvider for DummySchemaProvider {
129    fn table_names(&self) -> Vec<String> {
130        vec![]
131    }
132
133    async fn table(&self, name: &str) -> datafusion::error::Result<Option<Arc<dyn TableProvider>>> {
134        let table = self
135            .catalog_manager
136            .table(
137                &self.catalog_name,
138                &self.schema_name,
139                name,
140                self.query_ctx.as_deref(),
141            )
142            .await?
143            .with_context(|| TableNotExistSnafu {
144                table: format_full_table_name(&self.catalog_name, &self.schema_name, name),
145            })?;
146
147        let table_provider: Arc<dyn TableProvider> = Arc::new(DfTableProviderAdapter::new(table));
148
149        Ok(Some(table_provider))
150    }
151
152    fn table_exist(&self, _name: &str) -> bool {
153        true
154    }
155}
156
157impl fmt::Debug for DummySchemaProvider {
158    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
159        f.debug_struct("DummySchemaProvider")
160            .field("catalog_name", &self.catalog_name)
161            .field("schema_name", &self.schema_name)
162            .finish()
163    }
164}
catalog/table_source/dummy_catalog.rs

catalog/table_source/
dummy_catalog.rs
