common_catalog/
lib.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
15use consts::{DEFAULT_CATALOG_NAME, system_schema_name};
16
17pub mod consts;
18
19#[inline]
20pub fn format_schema_name(catalog: &str, schema: &str) -> String {
21    format!("{catalog}.{schema}")
22}
23
24/// Formats table fully-qualified name
25#[inline]
26pub fn format_full_table_name(catalog: &str, schema: &str, table: &str) -> String {
27    format!("{catalog}.{schema}.{table}")
28}
29
30/// Formats flow fully-qualified name
31#[inline]
32pub fn format_full_flow_name(catalog: &str, flow: &str) -> String {
33    format!("{catalog}.{flow}")
34}
35
36/// Build db name from catalog and schema string
37pub fn build_db_string(catalog: &str, schema: &str) -> String {
38    if catalog == DEFAULT_CATALOG_NAME {
39        schema.to_string()
40    } else {
41        format!("{catalog}-{schema}")
42    }
43}
44
45/// Attempt to parse catalog and schema from given database name
46///
47/// The database name may come from different sources:
48///
49/// - MySQL `schema` name in MySQL protocol login request: it's optional and user
50///   and switch database using `USE` command
51/// - Postgres `database` parameter in Postgres wire protocol, required
52/// - HTTP RESTful API: the database parameter, optional
53/// - gRPC: the dbname field in header, optional but has a higher priority than
54///   original catalog/schema
55///
56/// When database name is provided, we attempt to parse catalog and schema from
57/// it. We assume the format `[<catalog>-]<schema>`:
58///
59/// - If `[<catalog>-]` part is not provided, we use whole database name as
60///   schema name
61/// - if `[<catalog>-]` is provided, we split database name with `-` and use
62///   `<catalog>` and `<schema>`.
63pub fn parse_catalog_and_schema_from_db_string(db: &str) -> (String, String) {
64    match parse_optional_catalog_and_schema_from_db_string(db) {
65        (Some(catalog), schema) => (catalog, schema),
66        (None, schema) => (DEFAULT_CATALOG_NAME.to_string(), schema),
67    }
68}
69
70/// Attempt to parse catalog and schema from given database name
71///
72/// Similar to [`parse_catalog_and_schema_from_db_string`] but returns an optional
73/// catalog if it's not provided in the database name.
74pub fn parse_optional_catalog_and_schema_from_db_string(db: &str) -> (Option<String>, String) {
75    let parts = db.splitn(2, '-').collect::<Vec<&str>>();
76    if parts.len() == 2 {
77        (
78            Some(parts[0].to_string()),
79            canonicalize_schema_name(parts[1]),
80        )
81    } else {
82        (None, canonicalize_schema_name(db))
83    }
84}
85
86/// A database name taken from a protocol never reaches the SQL parser, which is what
87/// lowercases unquoted identifiers, so system schemas are folded here instead.
88fn canonicalize_schema_name(schema: &str) -> String {
89    system_schema_name(schema).unwrap_or(schema).to_string()
90}
91
92#[cfg(test)]
93mod tests {
94    use consts::{INFORMATION_SCHEMA_NAME, PG_CATALOG_NAME};
95
96    use super::*;
97
98    #[test]
99    fn test_db_string() {
100        assert_eq!("test", build_db_string(DEFAULT_CATALOG_NAME, "test"));
101        assert_eq!("a0b1c2d3-test", build_db_string("a0b1c2d3", "test"));
102    }
103
104    #[test]
105    fn test_parse_catalog_and_schema() {
106        assert_eq!(
107            (DEFAULT_CATALOG_NAME.to_string(), "fullschema".to_string()),
108            parse_catalog_and_schema_from_db_string("fullschema")
109        );
110
111        assert_eq!(
112            ("catalog".to_string(), "schema".to_string()),
113            parse_catalog_and_schema_from_db_string("catalog-schema")
114        );
115
116        assert_eq!(
117            ("catalog".to_string(), "schema1-schema2".to_string()),
118            parse_catalog_and_schema_from_db_string("catalog-schema1-schema2")
119        );
120
121        assert_eq!(
122            (None, "fullschema".to_string()),
123            parse_optional_catalog_and_schema_from_db_string("fullschema")
124        );
125
126        assert_eq!(
127            (Some("catalog".to_string()), "schema".to_string()),
128            parse_optional_catalog_and_schema_from_db_string("catalog-schema")
129        );
130
131        assert_eq!(
132            (Some("CATALOG".to_string()), "SCHEMA".to_string()),
133            parse_optional_catalog_and_schema_from_db_string("CATALOG-SCHEMA")
134        );
135
136        assert_eq!(
137            (Some("catalog".to_string()), "schema1-schema2".to_string()),
138            parse_optional_catalog_and_schema_from_db_string("catalog-schema1-schema2")
139        );
140    }
141
142    #[test]
143    fn test_parse_system_schema_ignores_case() {
144        assert_eq!(
145            (None, INFORMATION_SCHEMA_NAME.to_string()),
146            parse_optional_catalog_and_schema_from_db_string("INFORMATION_SCHEMA")
147        );
148
149        assert_eq!(
150            (Some("CATALOG".to_string()), PG_CATALOG_NAME.to_string()),
151            parse_optional_catalog_and_schema_from_db_string("CATALOG-Pg_Catalog")
152        );
153    }
154}
common_catalog/lib.rs

common_catalog/
lib.rs
