spark_connect/
conf.rs

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

spark_connect/
conf.rs
