robin_sparkless_core/
runtime_config.rs

1//! Session runtime configuration for PySpark compatibility profiles (3.5 vs 4.0).
2
3use std::collections::HashMap;
4
5/// PySpark compatibility tier (see docs/PYSPARK_COMPAT_PROFILES.md).
6#[derive(Clone, Copy, Debug, PartialEq, Eq, Default)]
7pub enum PysparkCompat {
8    /// PySpark 3.2–3.5-like semantics (Sparkless default).
9    #[default]
10    V3_5,
11    /// PySpark 4.0+ semantics when opted in.
12    V4_0,
13}
14
15impl PysparkCompat {
16    pub fn parse(s: &str) -> Option<Self> {
17        match s.trim() {
18            "3.5" | "3" | "3.x" => Some(Self::V3_5),
19            "4.0" | "4" | "4.x" => Some(Self::V4_0),
20            _ => None,
21        }
22    }
23
24    pub fn as_config_str(self) -> &'static str {
25        match self {
26            Self::V3_5 => "3.5",
27            Self::V4_0 => "4.0",
28        }
29    }
30}
31
32/// Execution semantics derived from `spark.conf` and compat profile.
33#[derive(Clone, Debug, PartialEq, Eq)]
34pub struct SessionRuntimeConfig {
35    pub pyspark_compat: PysparkCompat,
36    pub ansi_enabled: bool,
37    pub disable_map_key_normalization: bool,
38    pub infer_map_type_from_first_pair: bool,
39    pub infer_array_type_from_first_element: bool,
40    /// PySpark 3.5 JDBC datetime mapping for PostgreSQL (legacy restore).
41    pub legacy_postgres_datetime_mapping: bool,
42    /// PySpark 3.5 JDBC type mapping for MySQL/MariaDB (legacy restore).
43    pub legacy_mysql_datetime_mapping: bool,
44    /// PySpark 3.5 JDBC timestamp write mapping for Oracle (legacy restore).
45    pub legacy_oracle_timestamp_mapping: bool,
46    /// PySpark 3.5 JDBC type mapping for SQL Server (legacy restore).
47    pub legacy_mssqlserver_datetime_mapping: bool,
48    /// PySpark 3.5 JDBC type mapping for DB2 (legacy restore).
49    pub legacy_db2_datetime_mapping: bool,
50}
51
52impl Default for SessionRuntimeConfig {
53    fn default() -> Self {
54        Self::for_compat(PysparkCompat::V3_5)
55    }
56}
57
58fn parse_bool(s: &str) -> bool {
59    matches!(s.trim().to_ascii_lowercase().as_str(), "true" | "1" | "yes")
60}
61
62fn bool_str(v: bool) -> String {
63    if v {
64        "true".to_string()
65    } else {
66        "false".to_string()
67    }
68}
69
70impl SessionRuntimeConfig {
71    /// Bundle defaults for a compatibility profile.
72    pub fn for_compat(compat: PysparkCompat) -> Self {
73        match compat {
74            PysparkCompat::V3_5 => Self {
75                pyspark_compat: PysparkCompat::V3_5,
76                ansi_enabled: false,
77                disable_map_key_normalization: true,
78                infer_map_type_from_first_pair: true,
79                infer_array_type_from_first_element: true,
80                legacy_postgres_datetime_mapping: true,
81                legacy_mysql_datetime_mapping: true,
82                legacy_oracle_timestamp_mapping: true,
83                legacy_mssqlserver_datetime_mapping: true,
84                legacy_db2_datetime_mapping: true,
85            },
86            PysparkCompat::V4_0 => Self {
87                pyspark_compat: PysparkCompat::V4_0,
88                ansi_enabled: true,
89                disable_map_key_normalization: false,
90                infer_map_type_from_first_pair: false,
91                infer_array_type_from_first_element: false,
92                legacy_postgres_datetime_mapping: false,
93                legacy_mysql_datetime_mapping: false,
94                legacy_oracle_timestamp_mapping: false,
95                legacy_mssqlserver_datetime_mapping: false,
96                legacy_db2_datetime_mapping: false,
97            },
98        }
99    }
100
101    /// Parse from session config map (and optional env compat before session is built).
102    pub fn from_session_map(map: &HashMap<String, String>) -> Self {
103        let mut cfg = map
104            .get("sparkless.pyspark.compat")
105            .and_then(|v| PysparkCompat::parse(v).map(Self::for_compat))
106            .unwrap_or_default();
107
108        if let Some(v) = map.get("spark.sql.ansi.enabled") {
109            cfg.ansi_enabled = parse_bool(v);
110        }
111        if let Some(v) = map.get("spark.sql.legacy.disableMapKeyNormalization") {
112            cfg.disable_map_key_normalization = parse_bool(v);
113        }
114        if let Some(v) = map.get("spark.sql.pyspark.legacy.inferMapTypeFromFirstPair.enabled") {
115            cfg.infer_map_type_from_first_pair = parse_bool(v);
116        }
117        if let Some(v) = map.get("spark.sql.pyspark.legacy.inferArrayTypeFromFirstElement.enabled")
118        {
119            cfg.infer_array_type_from_first_element = parse_bool(v);
120        }
121        if let Some(v) = map.get("spark.sql.legacy.postgres.datetimeMapping.enabled") {
122            cfg.legacy_postgres_datetime_mapping = parse_bool(v);
123        }
124        if let Some(v) = map.get("spark.sql.legacy.mysql.datetimeMapping.enabled") {
125            cfg.legacy_mysql_datetime_mapping = parse_bool(v);
126        }
127        if let Some(v) = map.get("spark.sql.legacy.oracle.timestampMapping.enabled") {
128            cfg.legacy_oracle_timestamp_mapping = parse_bool(v);
129        }
130        if let Some(v) = map.get("spark.sql.legacy.mssqlserver.datetimeMapping.enabled") {
131            cfg.legacy_mssqlserver_datetime_mapping = parse_bool(v);
132        }
133        if let Some(v) = map.get("spark.sql.legacy.db2.datetimeMapping.enabled") {
134            cfg.legacy_db2_datetime_mapping = parse_bool(v);
135        }
136        cfg
137    }
138
139    /// Apply profile bundle keys into a session config map (used when setting compat).
140    pub fn apply_profile_to_map(&self, map: &mut HashMap<String, String>) {
141        map.insert(
142            "sparkless.pyspark.compat".to_string(),
143            self.pyspark_compat.as_config_str().to_string(),
144        );
145        map.insert(
146            "spark.sql.ansi.enabled".to_string(),
147            bool_str(self.ansi_enabled),
148        );
149        map.insert(
150            "spark.sql.legacy.disableMapKeyNormalization".to_string(),
151            bool_str(self.disable_map_key_normalization),
152        );
153        map.insert(
154            "spark.sql.pyspark.legacy.inferMapTypeFromFirstPair.enabled".to_string(),
155            bool_str(self.infer_map_type_from_first_pair),
156        );
157        map.insert(
158            "spark.sql.pyspark.legacy.inferArrayTypeFromFirstElement.enabled".to_string(),
159            bool_str(self.infer_array_type_from_first_element),
160        );
161        map.insert(
162            "spark.sql.legacy.postgres.datetimeMapping.enabled".to_string(),
163            bool_str(self.legacy_postgres_datetime_mapping),
164        );
165        map.insert(
166            "spark.sql.legacy.mysql.datetimeMapping.enabled".to_string(),
167            bool_str(self.legacy_mysql_datetime_mapping),
168        );
169        map.insert(
170            "spark.sql.legacy.oracle.timestampMapping.enabled".to_string(),
171            bool_str(self.legacy_oracle_timestamp_mapping),
172        );
173        map.insert(
174            "spark.sql.legacy.mssqlserver.datetimeMapping.enabled".to_string(),
175            bool_str(self.legacy_mssqlserver_datetime_mapping),
176        );
177        map.insert(
178            "spark.sql.legacy.db2.datetimeMapping.enabled".to_string(),
179            bool_str(self.legacy_db2_datetime_mapping),
180        );
181    }
182
183    /// Apply compat profile and write bundled keys into the map.
184    pub fn set_compat_profile(map: &mut HashMap<String, String>, compat: PysparkCompat) {
185        let cfg = Self::for_compat(compat);
186        cfg.apply_profile_to_map(map);
187    }
188
189    /// Read compat from environment (SPARKLESS_PYSPARK_COMPAT).
190    pub fn compat_from_env() -> Option<PysparkCompat> {
191        std::env::var("SPARKLESS_PYSPARK_COMPAT")
192            .ok()
193            .and_then(|v| PysparkCompat::parse(&v))
194    }
195}
196
197#[cfg(test)]
198mod tests {
199    use super::*;
200
201    #[test]
202    fn default_is_v3_5_ansi_off() {
203        let c = SessionRuntimeConfig::default();
204        assert_eq!(c.pyspark_compat, PysparkCompat::V3_5);
205        assert!(!c.ansi_enabled);
206        assert!(c.disable_map_key_normalization);
207    }
208
209    #[test]
210    fn v4_profile_ansi_on() {
211        let c = SessionRuntimeConfig::for_compat(PysparkCompat::V4_0);
212        assert!(c.ansi_enabled);
213        assert!(!c.disable_map_key_normalization);
214    }
215
216    #[test]
217    fn from_session_map_compat_4() {
218        let mut m = HashMap::new();
219        m.insert("sparkless.pyspark.compat".to_string(), "4.0".to_string());
220        let c = SessionRuntimeConfig::from_session_map(&m);
221        assert_eq!(c.pyspark_compat, PysparkCompat::V4_0);
222        assert!(c.ansi_enabled);
223    }
224
225    #[test]
226    fn ansi_override_after_profile() {
227        let mut m = HashMap::new();
228        m.insert("sparkless.pyspark.compat".to_string(), "4.0".to_string());
229        m.insert("spark.sql.ansi.enabled".to_string(), "false".to_string());
230        let c = SessionRuntimeConfig::from_session_map(&m);
231        assert!(!c.ansi_enabled);
232    }
233
234    #[test]
235    fn v4_jdbc_legacy_flags_off_by_default() {
236        let c = SessionRuntimeConfig::for_compat(PysparkCompat::V4_0);
237        assert!(!c.legacy_postgres_datetime_mapping);
238        assert!(!c.legacy_mysql_datetime_mapping);
239        assert!(!c.legacy_mssqlserver_datetime_mapping);
240    }
241
242    #[test]
243    fn v3_jdbc_legacy_flags_on_by_default() {
244        let c = SessionRuntimeConfig::for_compat(PysparkCompat::V3_5);
245        assert!(c.legacy_postgres_datetime_mapping);
246        assert!(c.legacy_mysql_datetime_mapping);
247    }
248}
robin_sparkless_core/runtime_config.rs

robin_sparkless_core/
runtime_config.rs
