risingwave_common/config/
streaming.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
15use std::time::Duration;
16
17use risingwave_common_proc_macro::serde_prefix_all;
18
19use super::*;
20
21mod async_stack_trace;
22mod cache_refill;
23mod join_encoding_type;
24mod over_window;
25
26pub use async_stack_trace::*;
27pub use cache_refill::*;
28pub use join_encoding_type::*;
29pub use over_window::*;
30
31/// The section `[streaming]` in `risingwave.toml`.
32#[serde_with::apply(Option => #[serde(with = "none_as_empty_string")])]
33#[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde, ConfigDoc)]
34pub struct StreamingConfig {
35    /// The maximum number of pending barriers in each partial graph. Pending barriers include
36    /// in-flight, collected but not committed, and currently completing barriers.
37    #[serde(default = "default::streaming::in_flight_barrier_nums")]
38    pub in_flight_barrier_nums: usize,
39
40    /// The maximum number of lagged barriers allowed when merging a snapshot backfill job into
41    /// the database graph.
42    #[serde(default = "default::streaming::snapshot_backfill_finish_max_lagged_barriers")]
43    pub snapshot_backfill_finish_max_lagged_barriers: usize,
44
45    /// The multiplier applied to `in_flight_barrier_nums` when limiting pending barriers in a
46    /// snapshot backfill partial graph. A value of 0 is treated as 1.
47    #[serde(default = "default::streaming::snapshot_backfill_barrier_amplification_factor")]
48    pub snapshot_backfill_barrier_amplification_factor: usize,
49
50    /// The thread number of the streaming actor runtime in the compute node. The default value is
51    /// decided by `tokio`.
52    #[serde(default)]
53    pub actor_runtime_worker_threads_num: Option<usize>,
54
55    /// Enable async stack tracing through `await-tree` for risectl.
56    #[serde(default = "default::streaming::async_stack_trace")]
57    pub async_stack_trace: AsyncStackTraceOption,
58
59    #[serde(default)]
60    #[config_doc(nested)]
61    pub developer: StreamingDeveloperConfig,
62
63    /// Max unique user stream errors per actor
64    #[serde(default = "default::streaming::unique_user_stream_errors")]
65    pub unique_user_stream_errors: usize,
66
67    /// Disable strict stream consistency checks.
68    #[serde(default = "default::streaming::unsafe_disable_strict_consistency")]
69    pub unsafe_disable_strict_consistency: bool,
70
71    #[serde(default, flatten)]
72    #[config_doc(omitted)]
73    pub unrecognized: Unrecognized<Self>,
74}
75
76/// The subsections `[streaming.developer]`.
77///
78/// It is put at [`StreamingConfig::developer`].
79#[serde_prefix_all("stream_", mode = "alias")]
80#[serde_with::apply(Option => #[serde(with = "none_as_empty_string")])]
81#[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde, ConfigDoc)]
82pub struct StreamingDeveloperConfig {
83    /// Set to true to enable per-executor row count metrics. This will produce a lot of timeseries
84    /// and might affect the prometheus performance. If you only need actor input and output
85    /// rows data, see `stream_actor_in_record_cnt` and `stream_actor_out_record_cnt` instead.
86    #[serde(default = "default::developer::stream_enable_executor_row_count")]
87    pub enable_executor_row_count: bool,
88
89    /// The capacity of the chunks in the channel that connects between `ConnectorSource` and
90    /// `SourceExecutor`.
91    #[serde(default = "default::developer::connector_message_buffer_size")]
92    pub connector_message_buffer_size: usize,
93
94    /// Limit number of the cached entries in an extreme aggregation call.
95    #[serde(default = "default::developer::unsafe_stream_extreme_cache_size")]
96    pub unsafe_extreme_cache_size: usize,
97
98    /// Minimum cache size for TopN cache per group key.
99    #[serde(default = "default::developer::stream_topn_cache_min_capacity")]
100    pub topn_cache_min_capacity: usize,
101
102    /// The maximum size of the chunk produced by executor at a time.
103    #[serde(default = "default::developer::stream_chunk_size")]
104    pub chunk_size: usize,
105
106    /// The initial permits that a channel holds, i.e., the maximum row count can be buffered in
107    /// the channel.
108    #[serde(default = "default::developer::stream_exchange_initial_permits")]
109    pub exchange_initial_permits: usize,
110
111    /// The permits that are batched to add back, for reducing the backward `AddPermits` messages
112    /// in remote exchange.
113    #[serde(default = "default::developer::stream_exchange_batched_permits")]
114    pub exchange_batched_permits: usize,
115
116    /// The maximum number of concurrent barriers in an exchange channel.
117    #[serde(default = "default::developer::stream_exchange_concurrent_barriers")]
118    pub exchange_concurrent_barriers: usize,
119
120    /// The concurrency for dispatching messages to different downstream jobs.
121    ///
122    /// - `1` means no concurrency, i.e., dispatch messages to downstream jobs one by one.
123    /// - `0` means unlimited concurrency.
124    #[serde(default = "default::developer::stream_exchange_concurrent_dispatchers")]
125    pub exchange_concurrent_dispatchers: usize,
126
127    /// The maximum number of chunks that `ProjectExecutor` evaluates concurrently.
128    ///
129    /// - `1` means no chunk-level concurrency.
130    /// - `0` means unlimited concurrency.
131    #[serde(default = "default::developer::stream_project_expr_concurrency")]
132    pub project_expr_concurrency: usize,
133
134    /// The maximum number of in-flight projection evaluation requests in `ProjectExecutor`.
135    ///
136    /// An in-flight request has started projection evaluation but has not finished yet. A finished
137    /// request no longer counts against this limit even if its result is still waiting to be emitted
138    /// in order.
139    ///
140    /// - `0` means unlimited in-flight requests.
141    #[serde(default = "default::developer::stream_project_expr_inflight_request_concurrency")]
142    pub project_expr_inflight_request_concurrency: usize,
143
144    /// The initial permits for a dml channel, i.e., the maximum row count can be buffered in
145    /// the channel.
146    #[serde(default = "default::developer::stream_dml_channel_initial_permits")]
147    pub dml_channel_initial_permits: usize,
148
149    /// The max heap size of dirty groups of `HashAggExecutor`.
150    #[serde(default = "default::developer::stream_hash_agg_max_dirty_groups_heap_size")]
151    pub hash_agg_max_dirty_groups_heap_size: usize,
152
153    #[serde(default = "default::developer::memory_controller_threshold_aggressive")]
154    pub memory_controller_threshold_aggressive: f64,
155
156    #[serde(default = "default::developer::memory_controller_threshold_graceful")]
157    pub memory_controller_threshold_graceful: f64,
158
159    #[serde(default = "default::developer::memory_controller_threshold_stable")]
160    pub memory_controller_threshold_stable: f64,
161
162    #[serde(default = "default::developer::memory_controller_eviction_factor_aggressive")]
163    pub memory_controller_eviction_factor_aggressive: f64,
164
165    #[serde(default = "default::developer::memory_controller_eviction_factor_graceful")]
166    pub memory_controller_eviction_factor_graceful: f64,
167
168    #[serde(default = "default::developer::memory_controller_eviction_factor_stable")]
169    pub memory_controller_eviction_factor_stable: f64,
170
171    #[serde(default = "default::developer::memory_controller_update_interval_ms")]
172    pub memory_controller_update_interval_ms: usize,
173
174    #[serde(default = "default::developer::memory_controller_sequence_tls_step")]
175    pub memory_controller_sequence_tls_step: u64,
176
177    #[serde(default = "default::developer::memory_controller_sequence_tls_lag")]
178    pub memory_controller_sequence_tls_lag: u64,
179
180    #[serde(default = "default::developer::stream_enable_arrangement_backfill")]
181    /// Deprecated and ignored for new streaming jobs. Arrangement backfill is always used as the
182    /// fallback backfill type.
183    #[deprecated(
184        note = "Deprecated and ignored for new streaming jobs. Arrangement backfill is always used as the fallback backfill type."
185    )]
186    pub enable_arrangement_backfill: bool,
187
188    #[serde(default = "default::developer::stream_enable_snapshot_backfill")]
189    /// Enable snapshot backfill
190    /// If false, the snapshot backfill will be disabled,
191    /// even if session variable set.
192    /// If true, it's decided by session variable `streaming_use_snapshot_backfill` (default true)
193    pub enable_snapshot_backfill: bool,
194
195    #[serde(default = "default::developer::stream_high_join_amplification_threshold")]
196    /// If number of hash join matches exceeds this threshold number,
197    /// it will be logged.
198    pub high_join_amplification_threshold: usize,
199
200    #[serde(default = "default::developer::stream_high_gap_fill_amplification_threshold")]
201    /// If number of rows generated by gap fill between two anchor rows exceeds this threshold
202    /// number, it will be logged.
203    pub high_gap_fill_amplification_threshold: usize,
204
205    /// Actor tokio metrics is enabled if `enable_actor_tokio_metrics` is set or metrics level >= Debug.
206    #[serde(default = "default::developer::enable_actor_tokio_metrics")]
207    pub enable_actor_tokio_metrics: bool,
208
209    /// The number of the connections for streaming remote exchange between two nodes.
210    /// If not specified, the value of `server.connection_pool_size` will be used.
211    #[serde(default = "default::developer::stream_exchange_connection_pool_size")]
212    pub(super) exchange_connection_pool_size: Option<u16>,
213
214    /// A flag to allow disabling the auto schema change handling
215    #[serde(default = "default::developer::stream_enable_auto_schema_change")]
216    pub enable_auto_schema_change: bool,
217
218    #[serde(default = "default::developer::enable_shared_source")]
219    /// Enable shared source
220    /// If false, the shared source will be disabled,
221    /// even if session variable set.
222    /// If true, it's decided by session variable `streaming_use_shared_source` (default true)
223    pub enable_shared_source: bool,
224
225    #[serde(default = "default::developer::switch_jdbc_pg_to_native")]
226    /// When true, all jdbc sinks with connector='jdbc' and jdbc.url="jdbc:postgresql://..."
227    /// will be switched from jdbc postgresql sinks to rust native (connector='postgres') sinks.
228    pub switch_jdbc_pg_to_native: bool,
229
230    /// The maximum number of consecutive barriers allowed in a message when sent between actors.
231    #[serde(default = "default::developer::stream_max_barrier_batch_size")]
232    pub max_barrier_batch_size: u32,
233
234    /// Configure the system-wide cache row cardinality of hash join.
235    /// For example, if this is set to 1000, it means we can have at most 1000 rows in cache.
236    #[serde(default = "default::developer::streaming_hash_join_entry_state_max_rows")]
237    pub hash_join_entry_state_max_rows: usize,
238
239    /// Number of processed rows between periodic join cache evictions.
240    /// Values smaller than 1 will be clamped to 1 by the executor.
241    #[serde(default = "default::developer::streaming_join_hash_map_evict_interval_rows")]
242    pub join_hash_map_evict_interval_rows: u32,
243
244    #[serde(default = "default::developer::streaming_now_progress_ratio")]
245    pub now_progress_ratio: Option<f32>,
246
247    /// Enable / Disable profiling stats used by `EXPLAIN ANALYZE`
248    #[serde(default = "default::developer::enable_explain_analyze_stats")]
249    pub enable_explain_analyze_stats: bool,
250
251    #[serde(default)]
252    pub compute_client_config: RpcClientConfig,
253
254    /// The interval in seconds to rebuild snapshot iterators during snapshot backfill.
255    #[serde(default = "default::developer::stream_snapshot_iter_rebuild_interval_secs")]
256    pub snapshot_iter_rebuild_interval_secs: u64,
257
258    /// `IcebergListExecutor`: The interval in seconds for Iceberg source to list new files.
259    #[serde(default = "default::developer::iceberg_list_interval_sec")]
260    pub iceberg_list_interval_sec: u64,
261
262    /// `IcebergFetchExecutor`: The number of files the executor will fetch concurrently in a batch.
263    #[serde(default = "default::developer::iceberg_fetch_batch_size")]
264    pub iceberg_fetch_batch_size: u64,
265
266    /// `IcebergSink`: The size of the cache for positional delete in the sink.
267    #[serde(default = "default::developer::iceberg_sink_positional_delete_cache_size")]
268    pub iceberg_sink_positional_delete_cache_size: usize,
269
270    /// `IcebergSink`: The maximum number of rows in a row group when writing Parquet files.
271    #[serde(default = "default::developer::iceberg_sink_write_parquet_max_row_group_rows")]
272    pub iceberg_sink_write_parquet_max_row_group_rows: usize,
273
274    /// `IcebergWithPkIndexWriter`: The maximum number of concurrent PK index lookups issued for
275    /// deletes and updates in one batch. Values smaller than 1 will be clamped to 1.
276    #[serde(default = "default::developer::iceberg_pk_index_sink_lookup_concurrency")]
277    pub iceberg_pk_index_sink_lookup_concurrency: usize,
278
279    /// `IcebergWithPkIndexWriter`: The maximum number of input rows to process as one batch.
280    /// The writer only merges chunks that are already buffered in its input, so batching never
281    /// waits for new data. Set to 1 to process chunks one by one.
282    #[serde(default = "default::developer::iceberg_pk_index_sink_max_batch_rows")]
283    pub iceberg_pk_index_sink_max_batch_rows: usize,
284
285    /// When enabled, materialized views using default `NoCheck` conflict behavior will be forced
286    /// to use `Overwrite`. Useful to avoid propagating inconsistent changelog downstream.
287    #[serde(default = "default::developer::materialize_force_overwrite_on_no_check")]
288    pub materialize_force_overwrite_on_no_check: bool,
289
290    /// Whether by default enable preloading all rows in memory for state table.
291    /// If true, all capable state tables will preload its state to memory
292    #[serde(default = "default::streaming::default_enable_mem_preload_state_table")]
293    pub default_enable_mem_preload_state_table: bool,
294
295    /// The list of state table ids to *enable* preloading all rows in memory for state table.
296    /// Only takes effect when `default_enable_mem_preload_state_table` is false.
297    #[serde(default)]
298    pub mem_preload_state_table_ids_whitelist: Vec<u32>,
299
300    /// The list of state table ids to *disable* preloading all rows in memory for state table.
301    /// Only takes effect when `default_enable_mem_preload_state_table` is true.
302    #[serde(default)]
303    pub mem_preload_state_table_ids_blacklist: Vec<u32>,
304
305    /// Eliminate unnecessary updates aggressively, even if it impacts performance. Enable this
306    /// only if it's confirmed that no-op updates are causing significant streaming amplification.
307    #[serde(default)]
308    pub aggressive_noop_update_elimination: bool,
309
310    /// The interval in seconds for the refresh scheduler to check and trigger scheduled refreshes.
311    #[serde(default = "default::developer::refresh_scheduler_interval_sec")]
312    pub refresh_scheduler_interval_sec: u64,
313
314    /// Determine which encoding will be used to encode join rows in operator cache.
315    #[serde(default)]
316    pub join_encoding_type: JoinEncodingType,
317
318    /// The timeout for reading from the buffer of the sync log store on barrier.
319    /// Every epoch we will attempt to read the full buffer of the sync log store.
320    /// If we hit the timeout, we will stop reading and continue.
321    #[serde(default = "default::developer::sync_log_store_pause_duration_ms")]
322    pub sync_log_store_pause_duration_ms: usize,
323
324    /// The max buffer size for sync logstore, before we start flushing.
325    #[serde(default = "default::developer::sync_log_store_buffer_size")]
326    pub sync_log_store_buffer_size: usize,
327
328    /// Disable the optimized dispatcher path for sync log store.
329    #[serde(default = "default::developer::disable_sync_log_store_dispatcher")]
330    pub disable_sync_log_store_dispatcher: bool,
331
332    /// Cache policy for partition cache in streaming over window.
333    /// Can be `full`, `recent`, `recent_first_n` or `recent_last_n`.
334    #[serde(default)]
335    pub over_window_cache_policy: OverWindowCachePolicy,
336
337    /// When enabled, vnode stats pruning is applied in production.
338    /// When disabled, vnode stats pruning is in dry-run mode: we still maintain vnode stats
339    /// and verify that pruning would be correct, but we don't actually use the pruning
340    /// results — we still use cache and storage to fulfill the read. This is useful for
341    /// validating the correctness of vnode stats pruning before enabling it in production.
342    #[serde(default = "default::developer::enable_state_table_vnode_stats_pruning")]
343    pub enable_state_table_vnode_stats_pruning: bool,
344
345    /// Runtime switch for the sort buffer of `LocalityProviderExecutor`. Only takes effect on
346    /// streaming jobs planned with the `enable_locality_sort_buffer` session variable enabled.
347    /// Setting it to false makes such locality providers always pass through input directly.
348    #[serde(default = "default::developer::enable_locality_sort_buffer")]
349    pub enable_locality_sort_buffer: bool,
350
351    /// The number of input rows within one epoch after which the sort buffer of
352    /// `LocalityProviderExecutor` activates. Below this threshold, input is passed through
353    /// directly without buffering, so that no extra latency is introduced for normal workloads.
354    #[serde(default = "default::developer::locality_sort_buffer_activate_threshold")]
355    pub locality_sort_buffer_activate_threshold: u64,
356
357    /// Cache refill policy for streaming cache refill feature.
358    /// Can be `enabled`, `disabled`, `streaming`, `serving` or `both`.
359    #[serde(default = "default::developer::cache_refill_policy")]
360    pub cache_refill_policy: CacheRefillPolicy,
361
362    /// Whether `MaterializeExecutor` enables vnode key stats for its state table.
363    #[serde(default = "default::developer::enable_vnode_key_stats_for_materialize")]
364    pub enable_vnode_key_stats_for_materialize: bool,
365
366    /// The maximum number of kv log store readers that can concurrently read historical data
367    /// (i.e., from the state store) during initialization. A reader is considered "initializing"
368    /// until it has read at least one row from the historical stream or the stream returns empty.
369    /// Set to 0 to disable the limit (unlimited concurrency).
370    #[serde(default = "default::developer::max_concurrent_kv_log_store_historical_read")]
371    pub max_concurrent_kv_log_store_historical_read: usize,
372
373    #[serde(default, flatten)]
374    #[serde_prefix_all(skip)]
375    #[config_doc(omitted)]
376    pub unrecognized: Unrecognized<Self>,
377}
378
379impl StreamingDeveloperConfig {
380    pub fn snapshot_iter_rebuild_interval(&self) -> Duration {
381        let rebuild_interval = if self.snapshot_iter_rebuild_interval_secs < 10 {
382            tracing::warn!(
383                "too small rebuild_interval {} second. rewrite to 10",
384                self.snapshot_iter_rebuild_interval_secs
385            );
386            10
387        } else {
388            self.snapshot_iter_rebuild_interval_secs
389        };
390        Duration::from_secs(rebuild_interval)
391    }
392}
393
394impl StreamingConfig {
395    /// Returns the dot-separated keys of all unrecognized fields, including those in `developer` section.
396    pub fn unrecognized_keys(&self) -> impl Iterator<Item = String> {
397        std::iter::from_coroutine(
398            #[coroutine]
399            || {
400                for k in self.unrecognized.inner().keys() {
401                    yield format!("streaming.{k}");
402                }
403                for k in self.developer.unrecognized.inner().keys() {
404                    yield format!("streaming.developer.{k}");
405                }
406            },
407        )
408    }
409}
410
411pub mod default {
412    pub use crate::config::default::developer;
413
414    pub mod streaming {
415        use tracing::info;
416
417        use crate::config::AsyncStackTraceOption;
418        use crate::util::env_var::env_var_is_true;
419
420        pub fn in_flight_barrier_nums() -> usize {
421            // quick fix
422            // TODO: remove this limitation from code
423            10000
424        }
425
426        pub fn snapshot_backfill_finish_max_lagged_barriers() -> usize {
427            100
428        }
429
430        pub fn snapshot_backfill_barrier_amplification_factor() -> usize {
431            1
432        }
433
434        pub fn async_stack_trace() -> AsyncStackTraceOption {
435            AsyncStackTraceOption::default()
436        }
437
438        pub fn unique_user_stream_errors() -> usize {
439            10
440        }
441
442        pub fn unsafe_disable_strict_consistency() -> bool {
443            false
444        }
445
446        pub fn default_enable_mem_preload_state_table() -> bool {
447            if env_var_is_true("DEFAULT_ENABLE_MEM_PRELOAD_STATE_TABLE") {
448                info!("enabled mem_preload_state_table globally by env var");
449                true
450            } else {
451                false
452            }
453        }
454    }
455}
risingwave_common/config/streaming.rs

risingwave_common/config/
streaming.rs
