risingwave_frontend/handler/create_source/external_schema/
json.rs

1// Copyright 2024 RisingWave Labs
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 super::*;
16
17pub fn json_schema_infer_use_schema_registry(schema_config: &Option<(AstString, bool)>) -> bool {
18    match schema_config {
19        None => false,
20        Some((_, use_registry)) => *use_registry,
21    }
22}
23
24/// Map a JSON schema to a relational schema
25pub async fn extract_json_table_schema(
26    schema_config: &Option<(AstString, bool)>,
27    with_properties: &BTreeMap<String, String>,
28    format_encode_options: &mut BTreeMap<String, String>,
29) -> Result<Option<Vec<ColumnCatalog>>> {
30    match schema_config {
31        None => Ok(None),
32        Some((schema_location, use_schema_registry)) => {
33            let schema_registry_auth = use_schema_registry.then(|| {
34                let config: SchemaRegistryConfig =
35                    SchemaRegistryConfig::from(&*format_encode_options);
36                try_consume_schema_registry_config_from_options(format_encode_options);
37                config
38            });
39            Ok(Some(
40                fetch_json_schema_and_map_to_columns(
41                    &schema_location.0,
42                    schema_registry_auth,
43                    with_properties,
44                )
45                .await?
46                .into_iter()
47                .map(|col| ColumnCatalog {
48                    column_desc: ColumnDesc::from_field_without_column_id(&col),
49                    is_hidden: false,
50                })
51                .collect_vec(),
52            ))
53        }
54    }
55}
56
57/// Map the latest Pulsar `JSON` schema of the source topic to a relational schema.
58///
59/// Pulsar stores `JSON` schemas as Avro record definitions, while payloads are plain JSON without
60/// a schema-registry header. JSON is self-describing, so only the latest schema is needed.
61pub async fn extract_pulsar_json_table_schema(
62    config: &PulsarSchemaConfig,
63    with_properties: &BTreeMap<String, String>,
64    format_encode_options: &mut BTreeMap<String, String>,
65) -> Result<Vec<ColumnCatalog>> {
66    try_consume_pulsar_schema_config_from_options(format_encode_options);
67    Ok(
68        fetch_pulsar_json_schema_and_map_to_columns(config, with_properties)
69            .await?
70            .into_iter()
71            .map(|col| ColumnCatalog {
72                column_desc: ColumnDesc::from_field_without_column_id(&col),
73                is_hidden: false,
74            })
75            .collect_vec(),
76    )
77}
78
79pub fn get_json_schema_location(
80    format_encode_options: &mut BTreeMap<String, String>,
81) -> Result<Option<(AstString, bool)>> {
82    let schema_location = try_consume_string_from_options(format_encode_options, "schema.location");
83    let schema_registry = try_consume_string_from_options(format_encode_options, "schema.registry");
84    match (schema_location, schema_registry) {
85        (None, None) => Ok(None),
86        (None, Some(schema_registry)) => Ok(Some((schema_registry, true))),
87        (Some(schema_location), None) => Ok(Some((schema_location, false))),
88        (Some(_), Some(_)) => Err(RwError::from(ProtocolError(
89            "only need either the schema location or the schema registry".to_owned(),
90        ))),
91    }
92}
risingwave_frontend/handler/create_source/external_schema/json.rs

risingwave_frontend/handler/create_source/external_schema/
json.rs
