risingwave_common/config/
batch.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 risingwave_common_proc_macro::serde_prefix_all;
16
17use super::*;
18
19/// The section `[batch]` in `risingwave.toml`.
20#[serde_with::apply(Option => #[serde(with = "none_as_empty_string")])]
21#[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde, ConfigDoc)]
22pub struct BatchConfig {
23    /// The thread number of the batch task runtime in the compute node. The default value is
24    /// decided by `tokio`.
25    #[serde(default)]
26    pub worker_threads_num: Option<usize>,
27
28    #[serde(default)]
29    #[config_doc(nested)]
30    pub developer: BatchDeveloperConfig,
31
32    /// This is the max number of queries per sql session.
33    #[serde(default)]
34    pub distributed_query_limit: Option<u64>,
35
36    /// This is the max number of batch queries per frontend node.
37    #[serde(default)]
38    pub max_batch_queries_per_frontend_node: Option<u64>,
39
40    #[serde(default = "default::batch::enable_barrier_read")]
41    pub enable_barrier_read: bool,
42
43    /// Timeout for a batch query in seconds.
44    #[serde(default = "default::batch::statement_timeout_in_sec")]
45    pub statement_timeout_in_sec: u32,
46
47    #[serde(default, flatten)]
48    #[config_doc(omitted)]
49    pub unrecognized: Unrecognized<Self>,
50
51    #[serde(default)]
52    /// frontend compute runtime worker threads
53    pub frontend_compute_runtime_worker_threads: Option<usize>,
54
55    /// This is the secs used to mask a worker unavailable temporarily.
56    #[serde(default = "default::batch::mask_worker_temporary_secs")]
57    pub mask_worker_temporary_secs: usize,
58
59    /// Keywords on which SQL option redaction is based in query logs and diagnostic reports.
60    /// Matching is case-insensitive. A SQL option with a name containing any of these keywords
61    /// will be redacted unless its full name is in `redact_sql_option_allowlist`.
62    #[serde(default = "default::batch::redact_sql_option_keywords")]
63    pub redact_sql_option_keywords: Vec<String>,
64
65    /// SQL option names exempt from keyword-based redaction in query logs and diagnostic reports.
66    /// Matching is exact and case-insensitive. This list takes precedence over
67    /// `redact_sql_option_keywords`. Set to an empty list to disable exemptions.
68    #[serde(default = "default::batch::redact_sql_option_allowlist")]
69    pub redact_sql_option_allowlist: Vec<String>,
70
71    /// Enable the spill out to disk feature for batch queries.
72    #[serde(default = "default::batch::enable_spill")]
73    pub enable_spill: bool,
74}
75
76/// The subsections `[batch.developer]`.
77///
78/// It is put at [`BatchConfig::developer`].
79#[serde_prefix_all("batch_", mode = "alias")]
80#[serde_with::apply(Option => #[serde(with = "none_as_empty_string")])]
81#[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde, ConfigDoc)]
82pub struct BatchDeveloperConfig {
83    /// The capacity of the chunks in the channel that connects between `ConnectorSource` and
84    /// `SourceExecutor`.
85    #[serde(default = "default::developer::connector_message_buffer_size")]
86    pub connector_message_buffer_size: usize,
87
88    /// The size of the channel used for output to exchange/shuffle.
89    #[serde(default = "default::developer::batch_output_channel_size")]
90    pub output_channel_size: usize,
91
92    #[serde(default = "default::developer::batch_receiver_channel_size")]
93    pub receiver_channel_size: usize,
94
95    #[serde(default = "default::developer::batch_root_stage_channel_size")]
96    pub root_stage_channel_size: usize,
97
98    /// The size of a chunk produced by `RowSeqScanExecutor`
99    #[serde(default = "default::developer::batch_chunk_size")]
100    pub chunk_size: usize,
101
102    /// The number of the connections for batch remote exchange between two nodes.
103    /// If not specified, the value of `server.connection_pool_size` will be used.
104    #[serde(default = "default::developer::batch_exchange_connection_pool_size")]
105    pub(super) exchange_connection_pool_size: Option<u16>,
106
107    #[serde(default)]
108    pub compute_client_config: RpcClientConfig,
109
110    #[serde(default)]
111    pub frontend_client_config: RpcClientConfig,
112
113    #[serde(default = "default::developer::batch_local_execute_buffer_size")]
114    pub local_execute_buffer_size: usize,
115}
116
117pub mod default {
118    pub use crate::config::default::developer;
119
120    pub mod batch {
121        pub fn enable_barrier_read() -> bool {
122            false
123        }
124
125        pub fn enable_spill() -> bool {
126            true
127        }
128
129        pub fn statement_timeout_in_sec() -> u32 {
130            // 1 hour
131            60 * 60
132        }
133
134        pub fn mask_worker_temporary_secs() -> usize {
135            30
136        }
137
138        pub fn redact_sql_option_keywords() -> Vec<String> {
139            [
140                "credential",
141                "key",
142                "password",
143                "private",
144                "secret",
145                "token",
146            ]
147            .into_iter()
148            .map(str::to_owned)
149            .collect()
150        }
151
152        pub fn redact_sql_option_allowlist() -> Vec<String> {
153            vec!["primary_key".to_owned()]
154        }
155    }
156}
risingwave_common/config/batch.rs

risingwave_common/config/
batch.rs
