risingwave_stream/from_proto/source/
mod.rs

1// Copyright 2023 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
15mod trad_source;
16
17mod fs_fetch;
18
19use std::collections::BTreeMap;
20
21use risingwave_connector::WithPropertiesExt;
22use risingwave_connector::source::UPSTREAM_SOURCE_KEY;
23use risingwave_pb::catalog::PbStreamSourceInfo;
24use risingwave_pb::plan_common::SourceRefreshMode;
25use risingwave_pb::plan_common::source_refresh_mode::RefreshMode;
26use risingwave_pb::stream_plan::stream_node::StreamKind;
27pub use trad_source::create_source_desc_builder;
28
29use super::*;
30
31fn get_connector_name(with_props: &BTreeMap<String, String>) -> String {
32    with_props
33        .get(UPSTREAM_SOURCE_KEY)
34        .map(|s| s.to_lowercase())
35        .unwrap_or_default()
36}
37
38fn is_full_reload_refresh(refresh_mode: &Option<SourceRefreshMode>) -> bool {
39    refresh_mode
40        .as_ref()
41        .map(|refresh_mode| matches!(refresh_mode.refresh_mode, Some(RefreshMode::FullReload(_))))
42        .unwrap_or(false)
43}
44
45/// Whether the List and Fetch executors of an Iceberg source run in streaming update mode.
46/// Both read the same plaintext option, so they cannot disagree on the mode.
47fn iceberg_update_mode(
48    with_properties: &BTreeMap<String, String>,
49    refresh_mode: &Option<SourceRefreshMode>,
50    rate_limit: Option<u32>,
51    row_id_index: Option<u32>,
52    stream_kind: StreamKind,
53) -> StreamResult<bool> {
54    if !with_properties.is_iceberg_update_source() {
55        return Ok(false);
56    }
57    // Retract is also the protobuf default for legacy plans, so it does not imply opt-in.
58    if is_full_reload_refresh(refresh_mode) || stream_kind != StreamKind::Retract {
59        return Err(anyhow::anyhow!(
60            "Iceberg streaming_updates requires a retracting, non-refresh graph"
61        )
62        .into());
63    }
64    if rate_limit.is_some() || row_id_index.is_some() {
65        return Err(anyhow::anyhow!(
66            "Iceberg streaming_updates cannot use generated row IDs or rate limiting"
67        )
68        .into());
69    }
70    Ok(true)
71}
risingwave_stream/from_proto/source/mod.rs

risingwave_stream/from_proto/source/
mod.rs
