risingwave_connector_codec/decoder/json/
mod.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
15// The MIT License (MIT)
16//
17// Copyright (c) 2021 David Raznick
18//
19// Permission is hereby granted, free of charge, to any person obtaining a copy
20// of this software and associated documentation files (the "Software"), to deal
21// in the Software without restriction, including without limitation the rights
22// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
23// copies of the Software, and to permit persons to whom the Software is
24// furnished to do so, subject to the following conditions:
25//
26// The above copyright notice and this permission notice shall be included in all
27// copies or substantial portions of the Software.
28//
29// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
30// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
31// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
32// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
33// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
34// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
35// SOFTWARE.
36
37use std::collections::HashMap;
38use std::fs;
39
40use anyhow::{Context, anyhow};
41use risingwave_common::catalog::Field;
42use risingwave_common::util::panic::rw_catch_unwind;
43use serde_json::Value;
44use thiserror::Error;
45use url::Url;
46
47use super::avro::{MapHandling, avro_schema_to_fields};
48
49#[derive(Debug, Error, thiserror_ext::ContextInto)]
50pub enum Error {
51    #[error("could not open schema from {filename}")]
52    SchemaFromFile {
53        filename: String,
54        source: std::io::Error,
55    },
56    #[error("parse error for url {url}")]
57    UrlParse {
58        url: String,
59        source: url::ParseError,
60    },
61    #[error("schema from {url} not valid JSON")]
62    SchemaNotJson { url: String, source: std::io::Error },
63    #[error("request error")]
64    Request { url: String, source: reqwest::Error },
65    #[error("schema from {url} not valid JSON")]
66    SchemaNotJsonSerde {
67        url: String,
68        source: serde_json::Error,
69    },
70    #[error(
71        "ref `{ref_string}` cannot be resolved as a pointer, and `{ref_fragment}` cannot be found in the schema"
72    )]
73    JsonRefPointerNotFound {
74        ref_string: String,
75        ref_fragment: String,
76    },
77    #[error("json ref error")]
78    JsonRef {
79        #[from]
80        source: std::io::Error,
81    },
82    #[error("need url to be a file or a http based, got {url}")]
83    UnsupportedUrl { url: String },
84    #[error(transparent)]
85    Uncategorized(
86        #[from]
87        #[backtrace]
88        anyhow::Error,
89    ),
90}
91
92type Result<T, E = Error> = std::result::Result<T, E>;
93
94#[derive(Debug)]
95pub struct JsonRef {
96    schema_cache: HashMap<String, Value>,
97}
98
99impl JsonRef {
100    fn new() -> JsonRef {
101        JsonRef {
102            schema_cache: HashMap::new(),
103        }
104    }
105
106    async fn deref_value(&mut self, value: &mut Value, retrieval_url: &Url) -> Result<()> {
107        self.schema_cache
108            .insert(retrieval_url.to_string(), value.clone());
109        self.deref(value, retrieval_url, &vec![]).await?;
110        Ok(())
111    }
112
113    async fn deref(
114        &mut self,
115        value: &mut Value,
116        base_url: &Url,
117        used_refs: &Vec<String>,
118    ) -> Result<()> {
119        if let Some(obj) = value.as_object_mut()
120            && let Some(ref_value) = obj.remove("$ref")
121            && let Some(ref_string) = ref_value.as_str()
122        {
123            let ref_url = base_url.join(ref_string).into_url_parse(ref_string)?;
124            let mut ref_url_no_fragment = ref_url.clone();
125            ref_url_no_fragment.set_fragment(None);
126            let url_schema = ref_url_no_fragment.scheme();
127            let ref_no_fragment = ref_url_no_fragment.to_string();
128
129            let mut schema = match self.schema_cache.get(&ref_no_fragment) {
130                Some(cached_schema) => cached_schema.clone(),
131                None => {
132                    if url_schema == "http" || url_schema == "https" {
133                        reqwest::get(ref_url_no_fragment.clone())
134                            .await
135                            .into_request(&ref_no_fragment)?
136                            .json()
137                            .await
138                            .into_request(&ref_no_fragment)?
139                    } else if url_schema == "file" {
140                        let file_path = ref_url_no_fragment.to_file_path().map_err(|_| {
141                            anyhow::anyhow!(
142                                "could not convert url {} to file path",
143                                ref_url_no_fragment
144                            )
145                        })?;
146                        let file =
147                            fs::File::open(file_path).into_schema_from_file(&ref_no_fragment)?;
148                        serde_json::from_reader(file)
149                            .into_schema_not_json_serde(ref_no_fragment.clone())?
150                    } else {
151                        return Err(Error::UnsupportedUrl {
152                            url: ref_no_fragment,
153                        });
154                    }
155                }
156            };
157
158            if !self.schema_cache.contains_key(&ref_no_fragment) {
159                self.schema_cache
160                    .insert(ref_no_fragment.clone(), schema.clone());
161            }
162
163            let ref_url_string = ref_url.to_string();
164            if let Some(ref_fragment) = ref_url.fragment() {
165                schema = schema
166                    .pointer(ref_fragment)
167                    .ok_or(Error::JsonRefPointerNotFound {
168                        ref_string: ref_string.to_owned(),
169                        ref_fragment: ref_fragment.to_owned(),
170                    })?
171                    .clone();
172            }
173            // Do not deref a url twice to prevent infinite loops
174            if used_refs.contains(&ref_url_string) {
175                return Ok(());
176            }
177            let mut new_used_refs = used_refs.clone();
178            new_used_refs.push(ref_url_string);
179            Box::pin(self.deref(&mut schema, &ref_url_no_fragment, &new_used_refs)).await?;
180
181            *value = schema;
182        }
183
184        if let Some(obj) = value.as_object_mut() {
185            for obj_value in obj.values_mut() {
186                Box::pin(self.deref(obj_value, base_url, used_refs)).await?
187            }
188        }
189        Ok(())
190    }
191}
192
193impl crate::JsonSchema {
194    /// ## Notes on type conversion
195    /// Map will be used when an object doesn't have `properties` but has `additionalProperties`.
196    /// When an object has `properties` and `additionalProperties`, the latter will be ignored.
197    /// <https://github.com/mozilla/jsonschema-transpiler/blob/fb715c7147ebd52427e0aea09b2bba2d539850b1/src/jsonschema.rs#L228-L280>
198    ///
199    /// TODO: examine other stuff like `oneOf`, `patternProperties`, etc.
200    pub async fn json_schema_to_columns(
201        &mut self,
202        retrieval_url: Url,
203    ) -> anyhow::Result<Vec<Field>> {
204        JsonRef::new()
205            .deref_value(&mut self.0, &retrieval_url)
206            .await?;
207        let avro_schema =
208            rw_catch_unwind(|| jst::convert_avro(&self.0, jst::Context::default()).to_string())
209                .map_err(|payload| {
210                    anyhow!(
211                        "failed to convert JSON schema to Avro schema: {}",
212                        panic_message::panic_message(&payload)
213                    )
214                })?;
215        let schema =
216            apache_avro::Schema::parse_str(&avro_schema).context("failed to parse avro schema")?;
217        avro_schema_to_fields(&schema, Some(MapHandling::Jsonb))
218    }
219}
risingwave_connector_codec/decoder/json/mod.rs

risingwave_connector_codec/decoder/json/
mod.rs
