risingwave_stream/executor/source/batch_source/
mod.rs

1// Copyright 2025 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 batch_posix_fs_list;
16pub use batch_posix_fs_list::*;
17mod batch_posix_fs_fetch;
18pub use batch_posix_fs_fetch::*;
19mod batch_opendal_fs_list;
20pub(crate) use batch_opendal_fs_list::*;
21mod batch_opendal_fs_fetch;
22pub use batch_opendal_fs_fetch::*;
23mod batch_iceberg_list;
24pub use batch_iceberg_list::*;
25mod batch_iceberg_fetch;
26pub use batch_iceberg_fetch::*;
27
28/// Define a stream executor module that is gated by a feature.
29///
30/// This is similar to `feature_gated_source_mod` in the connector crate, allowing heavy or
31/// unpopular source implementations (and their dependencies) to be disabled at compile time
32/// to decrease compilation time and binary size.
33///
34/// When the feature is disabled, this macro generates a dummy executor implementation that
35/// returns an error indicating the feature is not enabled.
36///
37/// # Example
38/// ```ignore
39/// feature_gated_executor_mod!(
40///     batch_adbc_snowflake_list,
41///     BatchAdbcSnowflakeListExecutor<S: StateStore>,
42///     "adbc_snowflake",
43///     (
44///         _actor_ctx: ActorContextRef,
45///         _stream_source_core: StreamSourceCore<S>,
46///         _metrics: Arc<StreamingMetrics>,
47///         _barrier_receiver: UnboundedReceiver<Barrier>,
48///         _barrier_manager: LocalBarrierManager,
49///         _associated_table_id: Option<TableId>,
50///     )
51/// );
52/// ```
53macro_rules! feature_gated_executor_mod {
54    (
55        $mod_name:ident,
56        $executor_name:ident <S: StateStore>,
57        $source_name:literal,
58        ( $( $param_name:ident : $param_type:ty ),* $(,)? )
59    ) => {
60        paste::paste! {
61            #[cfg(feature = "source-" $source_name)]
62            mod $mod_name;
63            #[cfg(feature = "source-" $source_name)]
64            pub use $mod_name::*;
65
66            #[cfg(not(feature = "source-" $source_name))]
67            #[doc = "Dummy implementation for executor when the feature `source-" $source_name "` is not enabled."]
68            mod [<$mod_name _stub>] {
69                #![allow(unused_imports)]
70                use std::sync::Arc;
71
72                use risingwave_common::id::TableId;
73                use risingwave_storage::StateStore;
74                use tokio::sync::mpsc::UnboundedReceiver;
75
76                use crate::executor::prelude::*;
77                use crate::executor::source::StreamSourceCore;
78                use crate::task::LocalBarrierManager;
79
80                fn err_feature_not_enabled() -> StreamExecutorError {
81                    StreamExecutorError::from(anyhow::anyhow!(
82                        "Feature `source-{}` is not enabled at compile time. \
83                        Please enable it in `Cargo.toml` and rebuild.",
84                        $source_name
85                    ))
86                }
87
88                #[doc = "A dummy executor that returns an error, as the feature `source-" $source_name "` is currently not enabled."]
89                pub struct $executor_name<S: StateStore> {
90                    _marker: std::marker::PhantomData<S>,
91                }
92
93                impl<S: StateStore> $executor_name<S> {
94                    #[allow(clippy::too_many_arguments)]
95                    pub fn new( $( $param_name : $param_type ),* ) -> Self {
96                        // Suppress unused variable warnings
97                        $( let _ = $param_name; )*
98                        Self {
99                            _marker: std::marker::PhantomData,
100                        }
101                    }
102                }
103
104                impl<S: StateStore> Execute for $executor_name<S> {
105                    fn execute(self: Box<Self>) -> BoxedMessageStream {
106                        futures::stream::once(async { Err(err_feature_not_enabled()) }).boxed()
107                    }
108                }
109
110                impl<S: StateStore> std::fmt::Debug for $executor_name<S> {
111                    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
112                        f.debug_struct(concat!(stringify!($executor_name), " (stub)"))
113                            .finish()
114                    }
115                }
116            }
117            #[cfg(not(feature = "source-" $source_name))]
118            pub use [<$mod_name _stub>]::*;
119        }
120    };
121}
122
123feature_gated_executor_mod!(
124    batch_adbc_snowflake_list,
125    BatchAdbcSnowflakeListExecutor<S: StateStore>,
126    "adbc_snowflake",
127    (
128        _actor_ctx: ActorContextRef,
129        _stream_source_core: StreamSourceCore<S>,
130        _metrics: Arc<StreamingMetrics>,
131        _barrier_receiver: UnboundedReceiver<Barrier>,
132        _barrier_manager: LocalBarrierManager,
133        _associated_table_id: Option<TableId>,
134    )
135);
136
137feature_gated_executor_mod!(
138    batch_adbc_snowflake_fetch,
139    BatchAdbcSnowflakeFetchExecutor<S: StateStore>,
140    "adbc_snowflake",
141    (
142        _actor_ctx: ActorContextRef,
143        _stream_source_core: StreamSourceCore<S>,
144        _upstream: Executor,
145        _barrier_manager: LocalBarrierManager,
146        _associated_table_id: Option<TableId>,
147    )
148);
risingwave_stream/executor/source/batch_source/mod.rs

risingwave_stream/executor/source/batch_source/
mod.rs
