1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
//! RuntimeConf implementation mirroring `pyspark.sql.conf.RuntimeConfig`.
//!
//! Provides runtime configuration management for Spark sessions through
//! the Spark Connect protocol.
use spark_connect_core::client::SparkConnectClient;
use spark_connect_core::error::Result;
use spark_connect_core::runtime::block_on;
use std::collections::HashMap;
/// Runtime configuration for a Spark session.
///
/// Allows getting and setting runtime configuration parameters on the Spark server.
/// Mirrors `pyspark.sql.conf.RuntimeConfig`.
pub struct RuntimeConf {
/// Shared gRPC client
client: std::sync::Arc<SparkConnectClient>,
}
impl RuntimeConf {
/// Create a new RuntimeConf with a client.
pub(crate) fn new(client: std::sync::Arc<SparkConnectClient>) -> Self {
RuntimeConf { client }
}
/// Set a configuration key-value pair.
///
/// Converts boolean and integer values to strings as required by the server.
///
/// # Arguments
///
/// * `key` - The configuration key (e.g., "spark.sql.shuffle.partitions")
/// * `value` - The configuration value as a string, integer, or boolean
///
/// # Example
///
/// ```ignore
/// conf.set("spark.sql.shuffle.partitions", "200")?;
/// conf.set("spark.sql.adaptive.enabled", true)?;
/// conf.set("spark.sql.maxMetadataStringLength", 100)?;
/// ```
pub fn set(&self, key: &str, value: &str) -> Result<()> {
block_on(self.client.set_config(key, value))
}
/// Get a configuration value by key.
///
/// Returns `Ok(None)` if the key does not exist on the server.
///
/// # Arguments
///
/// * `key` - The configuration key to retrieve
///
/// # Example
///
/// ```ignore
/// let value = conf.get("spark.sql.shuffle.partitions")?;
/// println!("Shuffle partitions: {:?}", value);
/// ```
pub fn get(&self, key: &str) -> Result<Option<String>> {
let results = block_on(self.client.get_configs(&[key]))?;
Ok(results.into_iter().next().flatten())
}
/// Get a configuration value, returning `default` when the key is unset — instead of
/// erroring as `get` does for an unknown key. Uses the server's `GetWithDefault`
/// operation, mirroring `RuntimeConfig.get(key, default)`.
pub fn get_with_default(&self, key: &str, default: Option<&str>) -> Result<Option<String>> {
let results = block_on(self.client.get_config_with_defaults(&[(key, default)]))?;
Ok(results.into_iter().next().flatten())
}
/// Get all configuration values as a HashMap.
///
/// # Example
///
/// ```ignore
/// let all_configs = conf.get_all()?;
/// for (key, value) in all_configs.iter() {
/// println!("{}: {}", key, value);
/// }
/// ```
pub fn get_all(&self) -> Result<HashMap<String, String>> {
let results = block_on(self.client.get_configs_all())?;
Ok(results)
}
/// Unset a configuration key (reset to default).
///
/// # Arguments
///
/// * `key` - The configuration key to unset
///
/// # Example
///
/// ```ignore
/// conf.unset("spark.sql.shuffle.partitions")?;
/// ```
pub fn unset(&self, key: &str) -> Result<()> {
block_on(self.client.unset_config(key))
}
/// Check if a configuration key is modifiable.
///
/// Returns `true` if the configuration can be changed at runtime,
/// `false` if it's read-only or cannot be modified.
///
/// # Arguments
///
/// * `key` - The configuration key to check
///
/// # Example
///
/// ```ignore
/// if conf.is_modifiable("spark.sql.shuffle.partitions")? {
/// println!("This config can be modified");
/// }
/// ```
pub fn is_modifiable(&self, key: &str) -> Result<bool> {
let results = block_on(self.client.is_config_modifiable(key))?;
Ok(results)
}
}
impl Clone for RuntimeConf {
fn clone(&self) -> Self {
RuntimeConf {
client: std::sync::Arc::clone(&self.client),
}
}
}
#[cfg(test)]
mod tests {
use crate::session::SparkSession;
fn session() -> SparkSession {
SparkSession::builder()
.remote("sc://localhost:15002")
.get_or_create()
.expect("failed to build session")
}
#[test]
fn runtime_conf_creation() {
let spark = session();
let conf = spark.conf();
// Verify that conf was created successfully (no network call)
let _ = conf.clone();
}
#[test]
fn runtime_conf_clone() {
let spark = session();
let conf1 = spark.conf();
let conf2 = conf1.clone();
// Both should share the same underlying client
let _ = (conf1, conf2);
}
}