risingwave_common/session_config/
mod.rs

1// Copyright 2022 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 iceberg_query_storage_mode;
16mod non_zero64;
17mod opt;
18pub mod parallelism;
19mod query_mode;
20mod search_path;
21pub mod sink_decouple;
22mod statement_timeout;
23mod transaction_isolation_level;
24mod visibility_mode;
25
26use chrono_tz::Tz;
27pub use iceberg_query_storage_mode::IcebergQueryStorageMode;
28use itertools::Itertools;
29pub use opt::OptionConfig;
30pub use query_mode::QueryMode;
31use risingwave_common_proc_macro::{ConfigDoc, SessionConfig};
32pub use search_path::{SearchPath, USER_NAME_WILD_CARD};
33use serde::{Deserialize, Serialize};
34pub use statement_timeout::StatementTimeout;
35use thiserror::Error;
36
37use self::non_zero64::ConfigNonZeroU64;
38use crate::config::mutate::TomlTableMutateExt;
39use crate::config::streaming::{JoinEncodingType, OverWindowCachePolicy};
40use crate::config::{ConfigMergeError, StreamingConfig, merge_streaming_config_section};
41use crate::hash::VirtualNode;
42use crate::session_config::parallelism::{ConfigBackfillParallelism, ConfigParallelism};
43use crate::session_config::sink_decouple::SinkDecouple;
44use crate::session_config::transaction_isolation_level::IsolationLevel;
45pub use crate::session_config::visibility_mode::VisibilityMode;
46use crate::{PG_VERSION, SERVER_ENCODING, SERVER_VERSION_NUM, STANDARD_CONFORMING_STRINGS};
47
48pub const SESSION_CONFIG_LIST_SEP: &str = ", ";
49
50#[derive(Error, Debug)]
51pub enum SessionConfigError {
52    #[error("Invalid value `{value}` for `{entry}`")]
53    InvalidValue {
54        entry: &'static str,
55        value: String,
56        source: anyhow::Error,
57    },
58
59    #[error("Unrecognized config entry `{0}`")]
60    UnrecognizedEntry(String),
61}
62
63type SessionConfigResult<T> = std::result::Result<T, SessionConfigError>;
64
65// NOTE(kwannoel): We declare it separately as a constant,
66// otherwise seems like it can't infer the type of -1 when written inline.
67const DISABLE_BACKFILL_RATE_LIMIT: i32 = -1;
68const DISABLE_SOURCE_RATE_LIMIT: i32 = -1;
69const DISABLE_DML_RATE_LIMIT: i32 = -1;
70const DISABLE_SINK_RATE_LIMIT: i32 = -1;
71
72/// Default to bypass cluster limits iff in debug mode.
73const BYPASS_CLUSTER_LIMITS: bool = cfg!(debug_assertions);
74
75/// This is the Session Config of RisingWave.
76///
77/// All config entries implement `Display` and `FromStr` for getter and setter, to be read and
78/// altered within a session.
79///
80/// Users can change the default value of a configuration entry using `ALTER SYSTEM SET`. To
81/// facilitate this, a `serde` implementation is used as the wire format for retrieving initial
82/// configurations and updates from the meta service. It's important to note that the meta
83/// service stores the overridden value of each configuration entry per row with `Display` in
84/// the meta store, rather than using the `serde` format. However, we still delegate the `serde`
85/// impl of all fields to `Display`/`FromStr` to make it consistent.
86#[serde_with::apply(_ => #[serde_as(as = "serde_with::DisplayFromStr")] )]
87#[serde_with::serde_as]
88#[derive(Clone, Debug, Deserialize, Serialize, SessionConfig, ConfigDoc, PartialEq)]
89pub struct SessionConfig {
90    /// If `RW_IMPLICIT_FLUSH` is on, then every INSERT/UPDATE/DELETE statement will block
91    /// until the entire dataflow is refreshed. In other words, every related table & MV will
92    /// be able to see the write.
93    #[parameter(default = false, alias = "rw_implicit_flush")]
94    implicit_flush: bool,
95
96    /// If `DML_WAIT_PERSISTENCE` is on, then every INSERT/UPDATE/DELETE statement waits until
97    /// the transaction is included in a checkpoint. This is ignored when `IMPLICIT_FLUSH` is on.
98    #[parameter(default = false)]
99    dml_wait_persistence: bool,
100
101    /// If `CREATE_COMPACTION_GROUP_FOR_MV` is on, dedicated compaction groups will be created in
102    /// MV creation.
103    #[parameter(default = false)]
104    create_compaction_group_for_mv: bool,
105
106    /// A temporary config variable to force query running in either local or distributed mode.
107    /// The default value is auto which means let the system decide to run batch queries in local
108    /// or distributed mode automatically.
109    #[parameter(default = QueryMode::default())]
110    query_mode: QueryMode,
111
112    /// For Iceberg engine tables, which storage to use for batch SELECT: Iceberg (columnar) or
113    /// Hummock (row). Only affects batch SELECT on tables with ENGINE = ICEBERG.
114    #[parameter(default = IcebergQueryStorageMode::default())]
115    iceberg_query_storage_mode: IcebergQueryStorageMode,
116
117    /// Sets the number of digits displayed for floating-point values.
118    /// See <https://www.postgresql.org/docs/current/runtime-config-client.html#:~:text=for%20more%20information.-,extra_float_digits,-(integer)>
119    #[parameter(default = 1)]
120    extra_float_digits: i32,
121
122    /// Sets the application name to be reported in statistics and logs.
123    /// See <https://www.postgresql.org/docs/14/runtime-config-logging.html#:~:text=What%20to%20Log-,application_name,-(string)>
124    #[parameter(default = "", flags = "REPORT")]
125    application_name: String,
126
127    /// It is typically set by an application upon connection to the server.
128    /// see <https://www.postgresql.org/docs/current/runtime-config-client.html#GUC-DATESTYLE>
129    #[parameter(default = "", rename = "datestyle")]
130    date_style: String,
131
132    /// Force the use of lookup join instead of hash join when possible for local batch execution.
133    #[parameter(default = true, alias = "rw_batch_enable_lookup_join")]
134    batch_enable_lookup_join: bool,
135
136    /// Enable usage of sortAgg instead of hash agg when order property is satisfied in batch
137    /// execution
138    #[parameter(default = true, alias = "rw_batch_enable_sort_agg")]
139    batch_enable_sort_agg: bool,
140
141    /// Enable distributed DML, so an insert, delete, and update statement can be executed in a distributed way (e.g. running in multiple compute nodes).
142    /// No atomicity guarantee in this mode. Its goal is to gain the best ingestion performance for initial batch ingestion where users always can drop their table when failure happens.
143    #[parameter(default = false, rename = "batch_enable_distributed_dml")]
144    batch_enable_distributed_dml: bool,
145
146    /// Evaluate expression in strict mode for batch queries.
147    /// If set to false, an expression failure will not cause an error but leave a null value
148    /// on the result set.
149    #[parameter(default = true)]
150    batch_expr_strict_mode: bool,
151
152    /// The max gap allowed to transform small range scan into multi point lookup.
153    #[parameter(default = 8)]
154    max_split_range_gap: i32,
155
156    /// Sets the order in which schemas are searched when an object (table, data type, function, etc.)
157    /// is referenced by a simple name with no schema specified.
158    /// See <https://www.postgresql.org/docs/14/runtime-config-client.html#GUC-SEARCH-PATH>
159    #[parameter(default = SearchPath::default())]
160    search_path: SearchPath,
161
162    /// If `VISIBILITY_MODE` is all, we will support querying data without checkpoint.
163    #[parameter(default = VisibilityMode::default())]
164    visibility_mode: VisibilityMode,
165
166    /// See <https://www.postgresql.org/docs/current/transaction-iso.html>
167    #[parameter(default = IsolationLevel::default())]
168    transaction_isolation: IsolationLevel,
169
170    /// Select as of specific epoch.
171    /// Sets the historical epoch for querying data. If 0, querying latest data.
172    #[parameter(default = ConfigNonZeroU64::default())]
173    query_epoch: ConfigNonZeroU64,
174
175    /// Session timezone. Defaults to UTC.
176    #[parameter(default = "UTC", check_hook = check_timezone)]
177    timezone: String,
178
179    /// The execution parallelism for streaming queries, including tables, materialized views,
180    /// indexes, and sinks. Defaults to `default`, which preserves the legacy adaptive
181    /// scheduling behavior during effective resolution.
182    #[parameter(default = ConfigParallelism::Default, flags = "SESSION_INIT")]
183    streaming_parallelism: ConfigParallelism,
184
185    /// Specific parallelism for backfill. Only `default` and a fixed positive integer are
186    /// supported here. Adaptive backfill strategies are deferred to a later change.
187    #[parameter(
188        default = ConfigBackfillParallelism::Default,
189        check_hook = check_streaming_parallelism_for_backfill,
190        flags = "SESSION_INIT"
191    )]
192    streaming_parallelism_for_backfill: ConfigBackfillParallelism,
193
194    /// Specific parallelism for table. Defaults to `default`, which preserves the legacy
195    /// bounded adaptive behavior only when the global parallelism itself remains `default`.
196    /// Otherwise it follows the explicit global parallelism.
197    #[parameter(default = ConfigParallelism::Default, flags = "SESSION_INIT")]
198    streaming_parallelism_for_table: ConfigParallelism,
199
200    /// Specific parallelism for sink. By default, it will fall back to `STREAMING_PARALLELISM`.
201    #[parameter(default = ConfigParallelism::Default, flags = "SESSION_INIT")]
202    streaming_parallelism_for_sink: ConfigParallelism,
203
204    /// Specific parallelism for index. By default, it will fall back to `STREAMING_PARALLELISM`.
205    #[parameter(default = ConfigParallelism::Default, flags = "SESSION_INIT")]
206    streaming_parallelism_for_index: ConfigParallelism,
207
208    /// Specific parallelism for source. Defaults to `default`, which preserves the legacy
209    /// bounded adaptive behavior only when the global parallelism itself remains `default`.
210    /// Otherwise it follows the explicit global parallelism.
211    #[parameter(default = ConfigParallelism::Default, flags = "SESSION_INIT")]
212    streaming_parallelism_for_source: ConfigParallelism,
213
214    /// Specific parallelism for materialized view. By default, it will fall back to `STREAMING_PARALLELISM`.
215    #[parameter(default = ConfigParallelism::Default, flags = "SESSION_INIT")]
216    streaming_parallelism_for_materialized_view: ConfigParallelism,
217
218    /// Enable delta join for streaming queries. Defaults to false.
219    #[parameter(default = false, alias = "rw_streaming_enable_delta_join")]
220    streaming_enable_delta_join: bool,
221
222    /// Enable bushy join for streaming queries. Defaults to true.
223    #[parameter(default = true, alias = "rw_streaming_enable_bushy_join")]
224    streaming_enable_bushy_join: bool,
225
226    /// Force filtering to be done inside the join whenever there's a choice between optimizations.
227    /// Defaults to false.
228    #[parameter(default = false, alias = "rw_streaming_force_filter_inside_join")]
229    streaming_force_filter_inside_join: bool,
230
231    /// Deprecated. Arrangement backfill is always used as the fallback backfill type for new
232    /// streaming jobs, and this setting is ignored.
233    #[parameter(
234        default = true,
235        deprecated = "The session variable STREAMING_USE_ARRANGEMENT_BACKFILL has been deprecated and is ignored. Arrangement backfill is always used as the fallback backfill type for new streaming jobs."
236    )]
237    streaming_use_arrangement_backfill: bool,
238
239    #[parameter(default = true)]
240    streaming_use_snapshot_backfill: bool,
241
242    /// Enable serverless backfill for streaming queries. Defaults to false.
243    #[parameter(default = false)]
244    enable_serverless_backfill: bool,
245
246    /// Allow `jsonb` in stream key
247    #[parameter(default = false, alias = "rw_streaming_allow_jsonb_in_stream_key")]
248    streaming_allow_jsonb_in_stream_key: bool,
249
250    /// Unsafe: allow impure expressions on non-append-only streams without materialization.
251    ///
252    /// This may lead to inconsistent results or panics due to re-evaluation on updates/retracts.
253    #[parameter(default = false)]
254    streaming_unsafe_allow_unmaterialized_impure_expr: bool,
255
256    /// Unsafe: allow an upsert sink to use downstream primary-key columns that are not part of
257    /// the upstream stream key.
258    ///
259    /// This may leave stale rows in the downstream system if a downstream primary-key column
260    /// changes without the upsert stream providing its old value.
261    #[parameter(default = false)]
262    streaming_unsafe_allow_upsert_sink_pk_mismatch: bool,
263
264    /// Separate consecutive `StreamHashJoin` by no-shuffle `StreamExchange`
265    #[parameter(default = false)]
266    streaming_separate_consecutive_join: bool,
267
268    /// Separate `StreamSink` by no-shuffle `StreamExchange`
269    #[parameter(default = false)]
270    streaming_separate_sink: bool,
271
272    /// Determine which encoding will be used to encode join rows in operator cache.
273    ///
274    /// This overrides the corresponding entry from the `[streaming.developer]` section in the config file,
275    /// taking effect for new streaming jobs created in the current session.
276    #[parameter(default = None)]
277    streaming_join_encoding: OptionConfig<JoinEncodingType>,
278
279    /// Enable join ordering for streaming and batch queries. Defaults to true.
280    #[parameter(default = true, alias = "rw_enable_join_ordering")]
281    enable_join_ordering: bool,
282
283    /// Enable two phase agg optimization. Defaults to true.
284    /// Setting this to true will always set `FORCE_TWO_PHASE_AGG` to false.
285    #[parameter(default = true, flags = "SETTER", alias = "rw_enable_two_phase_agg")]
286    enable_two_phase_agg: bool,
287
288    /// Force two phase agg optimization whenever there's a choice between
289    /// optimizations. Defaults to false.
290    /// Setting this to true will always set `ENABLE_TWO_PHASE_AGG` to false.
291    #[parameter(default = false, flags = "SETTER", alias = "rw_force_two_phase_agg")]
292    force_two_phase_agg: bool,
293
294    /// Enable sharing of common sub-plans.
295    /// This means that DAG structured query plans can be constructed,
296    #[parameter(default = true, alias = "rw_enable_share_plan")]
297    /// rather than only tree structured query plans.
298    enable_share_plan: bool,
299
300    /// Enable split distinct agg
301    #[parameter(default = false, alias = "rw_force_split_distinct_agg")]
302    force_split_distinct_agg: bool,
303
304    /// See <https://www.postgresql.org/docs/current/runtime-config-client.html#GUC-INTERVALSTYLE>
305    #[parameter(default = "", rename = "intervalstyle")]
306    interval_style: String,
307
308    /// If `BATCH_PARALLELISM` is non-zero, batch queries will use this parallelism.
309    #[parameter(default = ConfigNonZeroU64::default())]
310    batch_parallelism: ConfigNonZeroU64,
311
312    /// The version of PostgreSQL that Risingwave claims to be.
313    #[parameter(default = PG_VERSION)]
314    server_version: String,
315
316    /// The version of PostgreSQL that Risingwave claims to be.
317    #[parameter(default = SERVER_VERSION_NUM)]
318    server_version_num: i32,
319
320    /// see <https://www.postgresql.org/docs/15/runtime-config-client.html#GUC-CLIENT-MIN-MESSAGES>
321    #[parameter(default = "notice")]
322    client_min_messages: String,
323
324    /// see <https://www.postgresql.org/docs/15/runtime-config-client.html#GUC-CLIENT-ENCODING>
325    #[parameter(default = SERVER_ENCODING, check_hook = check_client_encoding)]
326    client_encoding: String,
327
328    /// Enable decoupling sink and internal streaming graph or not
329    #[parameter(default = SinkDecouple::default())]
330    sink_decouple: SinkDecouple,
331
332    /// See <https://www.postgresql.org/docs/current/runtime-config-compatible.html#RUNTIME-CONFIG-COMPATIBLE-VERSION>
333    /// Unused in RisingWave, support for compatibility.
334    #[parameter(default = false)]
335    synchronize_seqscans: bool,
336
337    /// Abort query statement that takes more than the specified amount of time in sec. If
338    /// `log_min_error_statement` is set to ERROR or lower, the statement that timed out will also be
339    /// logged. If this value is specified without units, it is taken as milliseconds. A value of
340    /// zero (the default) disables the timeout.
341    #[parameter(default = StatementTimeout::default())]
342    statement_timeout: StatementTimeout,
343
344    /// Terminate any session that has been idle (that is, waiting for a client query) within an open transaction for longer than the specified amount of time in milliseconds.
345    #[parameter(default = 60000u32)]
346    idle_in_transaction_session_timeout: u32,
347
348    /// See <https://www.postgresql.org/docs/current/runtime-config-client.html#GUC-LOCK-TIMEOUT>
349    /// Unused in RisingWave, support for compatibility.
350    #[parameter(default = 0)]
351    lock_timeout: i32,
352
353    /// For limiting the startup time of a shareable CDC streaming source when the source is being created. Unit: seconds.
354    #[parameter(default = 60)]
355    cdc_source_wait_streaming_start_timeout: i32,
356
357    /// see <https://www.postgresql.org/docs/current/runtime-config-client.html#GUC-ROW-SECURITY>.
358    /// Unused in RisingWave, support for compatibility.
359    #[parameter(default = true)]
360    row_security: bool,
361
362    /// see <https://www.postgresql.org/docs/current/runtime-config-client.html#GUC-STANDARD-CONFORMING-STRINGS>
363    #[parameter(default = STANDARD_CONFORMING_STRINGS)]
364    standard_conforming_strings: String,
365
366    /// Set streaming rate limit (rows per second) for each parallelism for mv / source / sink backfilling
367    /// If set to -1, disable rate limit.
368    /// If set to 0, this pauses the snapshot read / source read.
369    #[parameter(default = DISABLE_BACKFILL_RATE_LIMIT)]
370    backfill_rate_limit: i32,
371
372    /// Set streaming rate limit (rows per second) for each parallelism for mv / source backfilling, source reads.
373    /// If set to -1, disable rate limit.
374    /// If set to 0, this pauses the snapshot read / source read.
375    #[parameter(default = DISABLE_SOURCE_RATE_LIMIT)]
376    source_rate_limit: i32,
377
378    /// Set streaming rate limit (rows per second) for each parallelism for table DML.
379    /// If set to -1, disable rate limit.
380    /// If set to 0, this pauses the DML.
381    #[parameter(default = DISABLE_DML_RATE_LIMIT)]
382    dml_rate_limit: i32,
383
384    /// Set sink rate limit (rows per second) for each parallelism for external sink.
385    /// If set to -1, disable rate limit.
386    /// If set to 0, this pauses the sink.
387    #[parameter(default = DISABLE_SINK_RATE_LIMIT)]
388    sink_rate_limit: i32,
389
390    /// Cache policy for partition cache in streaming over window.
391    /// Can be `full`, `recent`, `recent_first_n` or `recent_last_n`.
392    ///
393    /// This overrides the corresponding entry from the `[streaming.developer]` section in the config file,
394    /// taking effect for new streaming jobs created in the current session.
395    #[parameter(default = None, alias = "rw_streaming_over_window_cache_policy")]
396    streaming_over_window_cache_policy: OptionConfig<OverWindowCachePolicy>,
397
398    /// Run DDL statements in background
399    #[parameter(default = false)]
400    background_ddl: bool,
401
402    /// Enable shared source. Currently only for Kafka.
403    ///
404    /// When enabled, `CREATE SOURCE` will create a source streaming job, and `CREATE MATERIALIZED VIEWS` from the source
405    /// will forward the data from the same source streaming job, and also backfill prior data from the external source.
406    #[parameter(default = true)]
407    streaming_use_shared_source: bool,
408
409    /// Enable in-memory cache for `AsOf` join executor.
410    ///
411    /// When enabled (default), `AsOf` join uses the cache-based implementation.
412    ///
413    /// When disabled, `AsOf` join uses a no-cache implementation that directly queries
414    /// the state table on-demand, reducing unnecessary data fetches for cache.
415    #[parameter(default = true)]
416    streaming_asof_join_use_cache: bool,
417
418    /// Shows the server-side character set encoding. At present, this parameter can be shown but not set, because the encoding is determined at database creation time.
419    #[parameter(default = SERVER_ENCODING)]
420    server_encoding: String,
421
422    #[parameter(default = "hex", check_hook = check_bytea_output)]
423    bytea_output: String,
424
425    /// Bypass checks on cluster limits
426    ///
427    /// When enabled, `CREATE MATERIALIZED VIEW` will not fail if the cluster limit is hit.
428    #[parameter(default = BYPASS_CLUSTER_LIMITS)]
429    bypass_cluster_limits: bool,
430
431    /// The maximum number of parallelism a streaming query can use. Defaults to 256.
432    ///
433    /// Compared to `STREAMING_PARALLELISM`, which configures the initial parallelism, this configures
434    /// the maximum parallelism a streaming query can use in the future, if the cluster size changes or
435    /// users manually change the parallelism with `ALTER .. SET PARALLELISM`.
436    ///
437    /// It's not always a good idea to set this to a very large number, as it may cause performance
438    /// degradation when performing range scans on the table or the materialized view.
439    // a.k.a. vnode count
440    #[parameter(default = VirtualNode::COUNT_FOR_COMPAT, check_hook = check_streaming_max_parallelism)]
441    streaming_max_parallelism: usize,
442
443    /// Used to provide the connection information for the iceberg engine.
444    /// Format: `iceberg_engine_connection` = `schema_name.connection_name`.
445    #[parameter(default = "", check_hook = check_iceberg_engine_connection)]
446    iceberg_engine_connection: String,
447
448    /// Whether the streaming join should be unaligned or not.
449    #[parameter(default = false)]
450    streaming_enable_unaligned_join: bool,
451
452    /// The timeout for reading from the buffer of the sync log store on barrier.
453    /// Every epoch we will attempt to read the full buffer of the sync log store.
454    /// If we hit the timeout, we will stop reading and continue.
455    ///
456    /// This overrides the corresponding entry from the `[streaming.developer]` section in the config file,
457    /// taking effect for new streaming jobs created in the current session.
458    #[parameter(default = None)]
459    streaming_sync_log_store_pause_duration_ms: OptionConfig<usize>,
460
461    /// The max buffer size for sync logstore, before we start flushing.
462    ///
463    /// This overrides the corresponding entry from the `[streaming.developer]` section in the config file,
464    /// taking effect for new streaming jobs created in the current session.
465    #[parameter(default = None)]
466    streaming_sync_log_store_buffer_size: OptionConfig<usize>,
467
468    /// Whether to disable purifying the definition of the table or source upon retrieval.
469    /// Only set this if encountering issues with functionalities like `SHOW` or `ALTER TABLE/SOURCE`.
470    /// This config may be removed in the future.
471    #[parameter(default = false, flags = "NO_ALTER_SYS")]
472    disable_purify_definition: bool,
473
474    /// The `ef_search` used in querying hnsw vector index
475    #[parameter(default = 40_usize)] // default value borrowed from pg_vector
476    batch_hnsw_ef_search: usize,
477
478    /// Enable index selection for queries
479    #[parameter(default = true)]
480    enable_index_selection: bool,
481
482    /// Enable mv selection for queries
483    #[parameter(default = false)]
484    enable_mv_selection: bool,
485
486    /// Enable locality backfill for streaming queries. Defaults to false.
487    #[parameter(default = false)]
488    enable_locality_backfill: bool,
489
490    /// Duration in seconds before notifying the user that a long-running DDL operation (e.g., DROP TABLE, CANCEL JOBS)
491    /// is still running. Set to 0 to disable notifications. Defaults to 30 seconds.
492    #[parameter(default = 30u32)]
493    slow_ddl_notification_secs: u32,
494
495    /// Unsafe: Enable storage retention for non-append-only tables.
496    /// Enabling this can lead to streaming inconsistency and node panic
497    /// if there is any row INSERT/UPDATE/DELETE operation corresponding to the ttled primary key.
498    #[parameter(default = false)]
499    unsafe_enable_storage_retention_for_non_append_only_tables: bool,
500
501    /// Enable DataFusion Engine
502    /// When enabled, queries involving Iceberg tables will be executed using the DataFusion engine.
503    #[parameter(default = true)]
504    enable_datafusion_engine: bool,
505
506    /// Prefer hash join over sort merge join in DataFusion engine
507    /// When enabled, the DataFusion engine will prioritize hash joins for query execution plans,
508    /// potentially improving performance for certain workloads, but may cause OOM for large datasets.
509    #[parameter(default = true)]
510    datafusion_prefer_hash_join: bool,
511
512    /// Emit chunks in upsert format for `UPDATE` and `DELETE` DMLs.
513    /// May lead to undefined behavior if the table is created with `ON CONFLICT DO NOTHING`.
514    ///
515    /// When enabled:
516    /// - `UPDATE` will only emit `Insert` records for new rows, instead of `Update` records.
517    /// - `DELETE` will only include key columns and pad the rest with NULL, instead of emitting complete rows.
518    #[parameter(default = false)]
519    upsert_dml: bool,
520}
521
522fn check_iceberg_engine_connection(val: &str) -> Result<(), String> {
523    if val.is_empty() {
524        return Ok(());
525    }
526
527    let parts: Vec<&str> = val.split('.').collect();
528    if parts.len() != 2 {
529        return Err("Invalid iceberg engine connection format, Should be set to this format: schema_name.connection_name.".to_owned());
530    }
531
532    Ok(())
533}
534
535fn check_timezone(val: &str) -> Result<(), String> {
536    // Check if the provided string is a valid timezone.
537    Tz::from_str_insensitive(val).map_err(|_e| "Not a valid timezone")?;
538    Ok(())
539}
540
541fn check_client_encoding(val: &str) -> Result<(), String> {
542    // https://github.com/postgres/postgres/blob/REL_15_3/src/common/encnames.c#L525
543    let clean = val.replace(|c: char| !c.is_ascii_alphanumeric(), "");
544    if !clean.eq_ignore_ascii_case("UTF8") {
545        Err("Only support 'UTF8' for CLIENT_ENCODING".to_owned())
546    } else {
547        Ok(())
548    }
549}
550
551fn check_bytea_output(val: &str) -> Result<(), String> {
552    if val == "hex" {
553        Ok(())
554    } else {
555        Err("Only support 'hex' for BYTEA_OUTPUT".to_owned())
556    }
557}
558
559/// Check if the provided value is a valid max parallelism.
560fn check_streaming_max_parallelism(val: &usize) -> Result<(), String> {
561    match val {
562        // TODO(var-vnode): this is to prevent confusion with singletons, after we distinguish
563        // them better, we may allow 1 as the max parallelism (though not much point).
564        0 | 1 => Err("STREAMING_MAX_PARALLELISM must be greater than 1".to_owned()),
565        2..=VirtualNode::MAX_COUNT => Ok(()),
566        _ => Err(format!(
567            "STREAMING_MAX_PARALLELISM must be less than or equal to {}",
568            VirtualNode::MAX_COUNT
569        )),
570    }
571}
572
573fn check_streaming_parallelism_for_backfill(val: &ConfigBackfillParallelism) -> Result<(), String> {
574    match val {
575        ConfigBackfillParallelism::Default | ConfigBackfillParallelism::Fixed(_) => Ok(()),
576        ConfigBackfillParallelism::Adaptive
577        | ConfigBackfillParallelism::Bounded(_)
578        | ConfigBackfillParallelism::Ratio(_) => Err(
579            "Only `default` or fixed backfill parallelism is supported here; adaptive backfill strategy is deferred to a later change.".to_owned(),
580        ),
581    }
582}
583
584impl SessionConfig {
585    pub fn set_force_two_phase_agg(
586        &mut self,
587        val: bool,
588        reporter: &mut impl ConfigReporter,
589    ) -> SessionConfigResult<bool> {
590        let set_val = self.set_force_two_phase_agg_inner(val, reporter)?;
591        if self.force_two_phase_agg {
592            self.set_enable_two_phase_agg(true, reporter)
593        } else {
594            Ok(set_val)
595        }
596    }
597
598    pub fn set_enable_two_phase_agg(
599        &mut self,
600        val: bool,
601        reporter: &mut impl ConfigReporter,
602    ) -> SessionConfigResult<bool> {
603        let set_val = self.set_enable_two_phase_agg_inner(val, reporter)?;
604        if !self.force_two_phase_agg {
605            self.set_force_two_phase_agg(false, reporter)
606        } else {
607            Ok(set_val)
608        }
609    }
610}
611
612pub struct VariableInfo {
613    pub name: String,
614    pub setting: String,
615    pub description: String,
616}
617
618/// Report status or notice to caller.
619pub trait ConfigReporter {
620    fn report_status(&mut self, key: &str, new_val: String);
621}
622
623// Report nothing.
624impl ConfigReporter for () {
625    fn report_status(&mut self, _key: &str, _new_val: String) {}
626}
627
628def_anyhow_newtype! {
629    pub SessionConfigToOverrideError,
630    toml::ser::Error => "failed to serialize session config",
631    ConfigMergeError => transparent,
632}
633
634impl SessionConfig {
635    /// Generate an initial override for the streaming config from the session config.
636    pub fn to_initial_streaming_config_override(
637        &self,
638    ) -> Result<String, SessionConfigToOverrideError> {
639        let mut table = toml::Table::new();
640
641        // TODO: make this more type safe.
642        // We `unwrap` here to assert the hard-coded keys are correct.
643        if let Some(v) = self.streaming_join_encoding.as_ref() {
644            table
645                .upsert("streaming.developer.join_encoding_type", v)
646                .unwrap();
647        }
648        if let Some(v) = self.streaming_sync_log_store_pause_duration_ms.as_ref() {
649            table
650                .upsert("streaming.developer.sync_log_store_pause_duration_ms", v)
651                .unwrap();
652        }
653        if let Some(v) = self.streaming_sync_log_store_buffer_size.as_ref() {
654            table
655                .upsert("streaming.developer.sync_log_store_buffer_size", v)
656                .unwrap();
657        }
658        if let Some(v) = self.streaming_over_window_cache_policy.as_ref() {
659            table
660                .upsert("streaming.developer.over_window_cache_policy", v)
661                .unwrap();
662        }
663
664        let res = toml::to_string(&table)?;
665
666        // Validate all fields are valid by trying to merge it to the default config.
667        if !res.is_empty() {
668            let merged =
669                merge_streaming_config_section(&StreamingConfig::default(), res.as_str())?.unwrap();
670
671            let unrecognized_keys = merged.unrecognized_keys().collect_vec();
672            if !unrecognized_keys.is_empty() {
673                bail!("unrecognized configs: {:?}", unrecognized_keys);
674            }
675        }
676
677        Ok(res)
678    }
679}
680
681#[cfg(test)]
682mod test {
683    use expect_test::expect;
684
685    use super::*;
686
687    #[derive(SessionConfig)]
688    struct TestConfig {
689        #[parameter(default = 1, flags = "NO_ALTER_SYS", alias = "test_param_alias" | "alias_param_test")]
690        test_param: i32,
691        #[parameter(default = false, deprecated = "deprecated test notice")]
692        deprecated_test_param: bool,
693    }
694
695    #[test]
696    fn test_session_config_alias() {
697        let mut config = TestConfig::default();
698        config.set("test_param", "2".to_owned(), &mut ()).unwrap();
699        assert_eq!(config.get("test_param_alias").unwrap(), "2");
700        config
701            .set("alias_param_test", "3".to_owned(), &mut ())
702            .unwrap();
703        assert_eq!(config.get("test_param_alias").unwrap(), "3");
704        assert!(TestConfig::check_no_alter_sys("test_param").unwrap());
705        assert_eq!(
706            TestConfig::deprecated_notice("deprecated_test_param").unwrap(),
707            Some("deprecated test notice")
708        );
709        assert_eq!(TestConfig::deprecated_notice("test_param").unwrap(), None);
710    }
711
712    #[test]
713    fn test_initial_streaming_config_override() {
714        let mut config = SessionConfig::default();
715        config
716            .set_streaming_join_encoding(Some(JoinEncodingType::Cpu).into(), &mut ())
717            .unwrap();
718        config
719            .set_streaming_over_window_cache_policy(
720                Some(OverWindowCachePolicy::RecentFirstN).into(),
721                &mut (),
722            )
723            .unwrap();
724
725        // Check the converted config override string.
726        let override_str = config.to_initial_streaming_config_override().unwrap();
727        expect![[r#"
728            [streaming.developer]
729            join_encoding_type = "cpu_optimized"
730            over_window_cache_policy = "recent_first_n"
731        "#]]
732        .assert_eq(&override_str);
733
734        // Try merging it to the default streaming config.
735        let merged = merge_streaming_config_section(&StreamingConfig::default(), &override_str)
736            .unwrap()
737            .unwrap();
738        assert_eq!(merged.developer.join_encoding_type, JoinEncodingType::Cpu);
739        assert_eq!(
740            merged.developer.over_window_cache_policy,
741            OverWindowCachePolicy::RecentFirstN
742        );
743    }
744
745    #[test]
746    fn test_streaming_parallelism_defaults() {
747        let config = SessionConfig::default();
748
749        assert_eq!(config.streaming_parallelism(), ConfigParallelism::Default);
750        assert_eq!(
751            config.streaming_parallelism_for_table(),
752            ConfigParallelism::Default
753        );
754        assert_eq!(
755            config.streaming_parallelism_for_source(),
756            ConfigParallelism::Default
757        );
758        assert_eq!(
759            config.streaming_parallelism_for_sink(),
760            ConfigParallelism::Default
761        );
762        assert_eq!(
763            config.streaming_parallelism_for_index(),
764            ConfigParallelism::Default
765        );
766        assert_eq!(
767            config.streaming_parallelism_for_materialized_view(),
768            ConfigParallelism::Default
769        );
770        assert!(!config.streaming_unsafe_allow_upsert_sink_pk_mismatch());
771    }
772
773    #[test]
774    fn test_streaming_parallelism_default_round_trip() {
775        let mut config = SessionConfig::default();
776
777        assert_eq!(config.get("streaming_parallelism").unwrap(), "default");
778        assert_eq!(
779            config.get("streaming_parallelism_for_table").unwrap(),
780            "default"
781        );
782        assert_eq!(
783            config.get("streaming_parallelism_for_source").unwrap(),
784            "default"
785        );
786
787        config
788            .set("streaming_parallelism", "default".to_owned(), &mut ())
789            .unwrap();
790        assert_eq!(config.get("streaming_parallelism").unwrap(), "default");
791
792        config
793            .set("streaming_parallelism", "bounded(16)".to_owned(), &mut ())
794            .unwrap();
795        config
796            .set(
797                "streaming_parallelism_for_table",
798                "bounded(8)".to_owned(),
799                &mut (),
800            )
801            .unwrap();
802        config
803            .set(
804                "streaming_parallelism_for_source",
805                "bounded(8)".to_owned(),
806                &mut (),
807            )
808            .unwrap();
809
810        assert_eq!(
811            config.reset("streaming_parallelism", &mut ()).unwrap(),
812            "default"
813        );
814        assert_eq!(
815            config
816                .reset("streaming_parallelism_for_table", &mut ())
817                .unwrap(),
818            "default"
819        );
820        assert_eq!(
821            config
822                .reset("streaming_parallelism_for_source", &mut ())
823                .unwrap(),
824            "default"
825        );
826    }
827    #[test]
828    fn test_streaming_parallelism_for_backfill_accepts_default_and_fixed() {
829        let mut config = SessionConfig::default();
830
831        config
832            .set(
833                "streaming_parallelism_for_backfill",
834                "default".to_owned(),
835                &mut (),
836            )
837            .unwrap();
838        assert_eq!(
839            config.get("streaming_parallelism_for_backfill").unwrap(),
840            "default"
841        );
842
843        config
844            .set(
845                "streaming_parallelism_for_backfill",
846                "2".to_owned(),
847                &mut (),
848            )
849            .unwrap();
850        assert_eq!(config.streaming_parallelism_for_backfill().to_string(), "2");
851    }
852
853    #[test]
854    fn test_streaming_parallelism_for_backfill_rejects_adaptive_modes() {
855        let mut config = SessionConfig::default();
856        let expected = "Only `default` or fixed backfill parallelism is supported here; adaptive backfill strategy is deferred to a later change.";
857
858        for value in ["adaptive", "bounded(2)", "ratio(0.5)"] {
859            let err = config
860                .set(
861                    "streaming_parallelism_for_backfill",
862                    value.to_owned(),
863                    &mut (),
864                )
865                .unwrap_err();
866
867            match err {
868                SessionConfigError::InvalidValue {
869                    entry,
870                    value: actual_value,
871                    source,
872                } => {
873                    assert_eq!(entry, "streaming_parallelism_for_backfill");
874                    assert_eq!(actual_value, value);
875                    assert_eq!(source.to_string(), expected);
876                }
877                other => panic!("unexpected error: {other:?}"),
878            }
879        }
880    }
881}
risingwave_common/session_config/mod.rs

risingwave_common/session_config/
mod.rs
