risingwave_connector/schema/
mod.rs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
// Copyright 2025 RisingWave Labs
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use crate::error::ConnectorError;

pub mod avro;
mod loader;
pub mod protobuf;
pub mod schema_registry;

pub use loader::{ConfluentSchemaLoader, SchemaLoader, SchemaVersion};

const MESSAGE_NAME_KEY: &str = "message";
const KEY_MESSAGE_NAME_KEY: &str = "key.message";
const SCHEMA_LOCATION_KEY: &str = "schema.location";
const SCHEMA_REGISTRY_KEY: &str = "schema.registry";
const NAME_STRATEGY_KEY: &str = "schema.registry.name.strategy";
pub const AWS_GLUE_SCHEMA_ARN_KEY: &str = "aws.glue.schema_arn";

#[derive(Debug, thiserror::Error, thiserror_ext::Macro)]
#[error("Invalid option: {message}")]
pub struct InvalidOptionError {
    pub message: String,
    // #[backtrace]
    // source: Option<risingwave_common::error::BoxedError>,
}

#[derive(Debug, thiserror::Error, thiserror_ext::Macro)]
#[error("Malformed response: {message}")]
pub struct MalformedResponseError {
    pub message: String,
}

#[derive(Debug, thiserror::Error)]
pub enum SchemaFetchError {
    #[error(transparent)]
    InvalidOption(#[from] InvalidOptionError),
    #[error(transparent)]
    License(#[from] risingwave_common::license::FeatureNotAvailable),
    #[error(transparent)]
    Request(#[from] schema_registry::ConcurrentRequestError),
    #[error(transparent)]
    AwsGlue(#[from] Box<aws_sdk_glue::operation::get_schema_version::GetSchemaVersionError>),
    #[error(transparent)]
    MalformedResponse(#[from] MalformedResponseError),
    #[error("schema version id invalid: {0}")]
    InvalidUuid(#[from] uuid::Error),
    #[error("schema compilation error: {0}")]
    SchemaCompile(
        #[source]
        #[backtrace]
        risingwave_common::error::BoxedError,
    ),
    #[error("{0}")] // source+{0} is effectively transparent but allows backtrace
    YetToMigrate(
        #[source]
        #[backtrace]
        ConnectorError,
    ),
}