delta_funnel/table_formats/delta/
protocol.rs

1//! Delta protocol preflight.
2
3use crate::support::sanitize_uri_for_display;
4use crate::{
5    DeltaFunnelError, DeltaProtocolReport, error::DeltaProtocolCompatibilitySnafu, observability,
6};
7
8use super::PlannedDeltaSource;
9use super::kernel::{
10    DeltaKernelProtocol, TABLE_FEATURES_MIN_READER_VERSION, Version, snapshot_protocol_report,
11};
12
13// Reader features are Delta correctness requirements. Add features only after
14// the provider path proves the relevant semantics before rows reach DataFusion.
15const SUPPORTED_READER_FEATURES: &[&str] = &["timestampNtz", "deletionVectors", "columnMapping"];
16
17/// Successful protocol preflight for one source.
18#[derive(Debug, Clone, PartialEq, Eq)]
19pub struct ProtocolPreflight {
20    /// Protocol report captured during preflight.
21    protocol: DeltaProtocolReport,
22}
23
24impl ProtocolPreflight {
25    /// Protocol report captured during preflight.
26    #[must_use]
27    pub fn protocol(&self) -> &DeltaProtocolReport {
28        &self.protocol
29    }
30
31    pub(crate) fn into_protocol(self) -> DeltaProtocolReport {
32        self.protocol
33    }
34}
35
36/// Runs conservative Delta protocol preflight for one loaded source.
37///
38/// # Errors
39///
40/// Returns [`DeltaFunnelError::DeltaProtocolCompatibility`] when the source
41/// requires a reader protocol version or reader feature that DeltaFunnel does
42/// not support yet.
43pub fn preflight_delta_protocol(
44    source: &PlannedDeltaSource,
45) -> Result<ProtocolPreflight, DeltaFunnelError> {
46    let protocol = delta_protocol_report(source);
47
48    ensure_protocol_supported(&protocol)?;
49
50    Ok(ProtocolPreflight { protocol })
51}
52
53pub(crate) fn preflight_delta_protocol_with_tracing(
54    source: &PlannedDeltaSource,
55) -> Result<ProtocolPreflight, DeltaFunnelError> {
56    observability::protocol_preflight_started(source.name(), source.version());
57
58    let result = preflight_delta_protocol(source);
59    match &result {
60        Ok(preflight) => observability::protocol_preflight_completed(
61            &preflight.protocol.source_name,
62            preflight.protocol.snapshot_version,
63        ),
64        Err(error) => {
65            observability::protocol_preflight_failed(source.name(), source.version(), error)
66        }
67    }
68
69    result
70}
71
72/// Runs conservative Delta protocol preflight for loaded sources.
73///
74/// # Errors
75///
76/// Returns the first source-specific protocol compatibility error.
77pub fn preflight_delta_sources(
78    sources: &[PlannedDeltaSource],
79) -> Result<Vec<ProtocolPreflight>, DeltaFunnelError> {
80    sources.iter().map(preflight_delta_protocol).collect()
81}
82
83/// Extracts protocol details for one loaded source without applying policy.
84#[must_use]
85pub fn delta_protocol_report(source: &PlannedDeltaSource) -> DeltaProtocolReport {
86    let kernel = snapshot_protocol_report(source.loaded_snapshot().kernel_snapshot());
87
88    report_from_kernel(source, kernel)
89}
90
91fn report_from_kernel(
92    source: &PlannedDeltaSource,
93    kernel: DeltaKernelProtocol,
94) -> DeltaProtocolReport {
95    build_protocol_report(source.name(), source.table_uri(), source.version(), kernel)
96}
97
98fn build_protocol_report(
99    source_name: &str,
100    table_uri: &str,
101    snapshot_version: Version,
102    kernel: DeltaKernelProtocol,
103) -> DeltaProtocolReport {
104    DeltaProtocolReport {
105        source_name: source_name.to_owned(),
106        table_uri: sanitize_uri_for_display(table_uri),
107        snapshot_version,
108        min_reader_version: kernel.min_reader_version,
109        min_writer_version: kernel.min_writer_version,
110        reader_features: kernel.reader_features,
111        writer_features: kernel.writer_features,
112    }
113}
114
115fn ensure_protocol_supported(protocol: &DeltaProtocolReport) -> Result<(), DeltaFunnelError> {
116    if !is_supported_reader_version(protocol.min_reader_version) {
117        return compatibility_error(
118            protocol,
119            format!(
120                "unsupported Delta minReaderVersion {}",
121                protocol.min_reader_version
122            ),
123        );
124    }
125
126    if let Some(feature) = unsupported_reader_feature(&protocol.reader_features) {
127        return compatibility_error(
128            protocol,
129            format!(
130                "unsupported Delta reader feature `{}`",
131                sanitize_value_for_display(feature)
132            ),
133        );
134    }
135
136    Ok(())
137}
138
139fn unsupported_reader_feature(features: &[String]) -> Option<&str> {
140    features
141        .iter()
142        .map(String::as_str)
143        .find(|feature| !SUPPORTED_READER_FEATURES.contains(feature))
144}
145
146fn is_supported_reader_version(version: i32) -> bool {
147    // Version 1 is the basic legacy read protocol. Version 3 is Delta's
148    // table-feature protocol; the concrete reader requirements are then
149    // expressed by `readerFeatures` and checked separately above.
150    // Legacy reader version 2 implies column mapping support.
151    matches!(version, 1 | 2) || version == TABLE_FEATURES_MIN_READER_VERSION
152}
153
154fn compatibility_error<T>(
155    protocol: &DeltaProtocolReport,
156    reason: String,
157) -> Result<T, DeltaFunnelError> {
158    DeltaProtocolCompatibilitySnafu {
159        source_name: protocol.source_name.clone(),
160        table_uri: protocol.table_uri.clone(),
161        snapshot_version: protocol.snapshot_version,
162        reason,
163    }
164    .fail()
165}
166
167fn sanitize_value_for_display(value: &str) -> String {
168    value.chars().flat_map(char::escape_default).collect()
169}
170
171#[cfg(test)]
172mod tests {
173    use std::fs;
174    use std::path::{Path, PathBuf};
175    use std::time::{SystemTime, UNIX_EPOCH};
176
177    use super::{
178        DeltaProtocolReport, delta_protocol_report, preflight_delta_protocol,
179        preflight_delta_sources,
180    };
181    use crate::{DeltaFunnelError, DeltaSourceConfig, load_delta_source, load_delta_sources};
182
183    struct DeltaLogTable {
184        path: PathBuf,
185    }
186
187    impl Drop for DeltaLogTable {
188        fn drop(&mut self) {
189            let _ = fs::remove_dir_all(&self.path);
190        }
191    }
192
193    impl DeltaLogTable {
194        fn new(name: &str, protocol_json: &str) -> Result<Self, Box<dyn std::error::Error>> {
195            let path = Path::new("target")
196                .join("delta-funnel-protocol-tests")
197                .join(unique_name(name)?);
198            let log_path = path.join("_delta_log");
199            fs::create_dir_all(&log_path)?;
200            fs::write(
201                log_path.join("00000000000000000000.json"),
202                format!("{protocol_json}\n{METADATA_JSON}\n"),
203            )?;
204            fs::write(
205                log_path.join("00000000000000000001.json"),
206                format!("{}\n", add_json("part-00001.parquet")),
207            )?;
208
209            Ok(Self { path })
210        }
211    }
212
213    const LEGACY_PROTOCOL_JSON: &str =
214        r#"{"protocol":{"minReaderVersion":1,"minWriterVersion":2}}"#;
215    const WRITER_ONLY_FEATURE_PROTOCOL_JSON: &str = r#"{"protocol":{"minReaderVersion":3,"minWriterVersion":7,"readerFeatures":[],"writerFeatures":["inCommitTimestamp"]}}"#;
216    const TIMESTAMP_NTZ_PROTOCOL_JSON: &str = r#"{"protocol":{"minReaderVersion":3,"minWriterVersion":7,"readerFeatures":["timestampNtz"],"writerFeatures":["timestampNtz"]}}"#;
217    const DELETION_VECTOR_PROTOCOL_JSON: &str = r#"{"protocol":{"minReaderVersion":3,"minWriterVersion":7,"readerFeatures":["deletionVectors"],"writerFeatures":["deletionVectors"]}}"#;
218    const COLUMN_MAPPING_FEATURE_PROTOCOL_JSON: &str = r#"{"protocol":{"minReaderVersion":3,"minWriterVersion":7,"readerFeatures":["columnMapping"],"writerFeatures":["columnMapping"]}}"#;
219    const UNKNOWN_READER_FEATURE_PROTOCOL_JSON: &str = r#"{"protocol":{"minReaderVersion":3,"minWriterVersion":7,"readerFeatures":["madeUpFeature"],"writerFeatures":["madeUpFeature"]}}"#;
220    const LEGACY_COLUMN_MAPPING_PROTOCOL_JSON: &str =
221        r#"{"protocol":{"minReaderVersion":2,"minWriterVersion":5}}"#;
222    const METADATA_JSON: &str = r#"{"metaData":{"id":"delta-funnel-test","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1587968585495}}"#;
223
224    fn add_json(path: &str) -> String {
225        format!(
226            r#"{{"add":{{"path":"{path}","partitionValues":{{}},"size":0,"modificationTime":1587968586000,"dataChange":true}}}}"#
227        )
228    }
229
230    fn unique_name(name: &str) -> Result<String, Box<dyn std::error::Error>> {
231        let nanos = SystemTime::now().duration_since(UNIX_EPOCH)?.as_nanos();
232
233        Ok(format!("{}-{}-{nanos}", std::process::id(), name))
234    }
235
236    fn load_source(
237        name: &str,
238        protocol_json: &str,
239    ) -> Result<crate::PlannedDeltaSource, Box<dyn std::error::Error>> {
240        let table = DeltaLogTable::new(name, protocol_json)?;
241        let source = load_delta_source(DeltaSourceConfig {
242            name: name.to_owned(),
243            table_uri: table.path.to_string_lossy().to_string(),
244            version: None,
245            storage_options: Default::default(),
246        })?;
247
248        Ok(source)
249    }
250
251    fn source_config(name: &str, table: &DeltaLogTable) -> DeltaSourceConfig {
252        DeltaSourceConfig {
253            name: name.to_owned(),
254            table_uri: table.path.to_string_lossy().to_string(),
255            version: None,
256            storage_options: Default::default(),
257        }
258    }
259
260    #[test]
261    fn reports_legacy_protocol_details() -> Result<(), Box<dyn std::error::Error>> {
262        let source = load_source("orders", LEGACY_PROTOCOL_JSON)?;
263
264        let report = delta_protocol_report(&source);
265
266        assert_eq!(report.source_name, "orders");
267        assert!(report.table_uri.starts_with("file://"));
268        assert_eq!(report.snapshot_version, 1);
269        assert_eq!(report.min_reader_version, 1);
270        assert_eq!(report.min_writer_version, 2);
271        assert!(report.reader_features.is_empty());
272        assert!(report.writer_features.is_empty());
273
274        Ok(())
275    }
276
277    #[test]
278    fn protocol_report_sanitizes_uri_context() {
279        let report = super::build_protocol_report(
280            "orders",
281            "s3://user:password@example.com/table?token=secret#debug",
282            42,
283            super::DeltaKernelProtocol {
284                min_reader_version: 1,
285                min_writer_version: 2,
286                reader_features: Vec::new(),
287                writer_features: Vec::new(),
288            },
289        );
290
291        assert_eq!(report.table_uri, "s3://example.com/table");
292    }
293
294    #[test]
295    fn preflight_allows_writer_only_features() -> Result<(), Box<dyn std::error::Error>> {
296        let source = load_source("orders", WRITER_ONLY_FEATURE_PROTOCOL_JSON)?;
297
298        let preflight = preflight_delta_protocol(&source)?;
299
300        assert_eq!(preflight.protocol.min_reader_version, 3);
301        assert!(preflight.protocol.reader_features.is_empty());
302        assert_eq!(
303            preflight.protocol.writer_features,
304            vec!["inCommitTimestamp"]
305        );
306
307        Ok(())
308    }
309
310    #[test]
311    fn preflight_allows_timestamp_ntz_reader_feature() -> Result<(), Box<dyn std::error::Error>> {
312        let source = load_source("orders", TIMESTAMP_NTZ_PROTOCOL_JSON)?;
313
314        let preflight = preflight_delta_protocol(&source)?;
315
316        assert_eq!(preflight.protocol.min_reader_version, 3);
317        assert_eq!(preflight.protocol.reader_features, vec!["timestampNtz"]);
318        assert_eq!(preflight.protocol.writer_features, vec!["timestampNtz"]);
319
320        Ok(())
321    }
322
323    #[test]
324    fn preflight_allows_deletion_vectors_reader_feature() -> Result<(), Box<dyn std::error::Error>>
325    {
326        let source = load_source("orders", DELETION_VECTOR_PROTOCOL_JSON)?;
327
328        let preflight = preflight_delta_protocol(&source)?;
329
330        assert_eq!(preflight.protocol.min_reader_version, 3);
331        assert_eq!(preflight.protocol.reader_features, vec!["deletionVectors"]);
332        assert_eq!(preflight.protocol.writer_features, vec!["deletionVectors"]);
333
334        Ok(())
335    }
336
337    #[test]
338    fn preflight_rejects_unknown_reader_features() -> Result<(), Box<dyn std::error::Error>> {
339        let source = load_source("orders", UNKNOWN_READER_FEATURE_PROTOCOL_JSON)?;
340
341        let result = preflight_delta_protocol(&source);
342
343        assert!(matches!(
344            result,
345            Err(DeltaFunnelError::DeltaProtocolCompatibility {
346                reason,
347                ..
348            }) if reason.contains("madeUpFeature")
349        ));
350
351        Ok(())
352    }
353
354    #[test]
355    fn preflight_allows_legacy_column_mapping_version() -> Result<(), Box<dyn std::error::Error>> {
356        let source = load_source("orders", LEGACY_COLUMN_MAPPING_PROTOCOL_JSON)?;
357
358        let preflight = preflight_delta_protocol(&source)?;
359
360        assert_eq!(preflight.protocol.min_reader_version, 2);
361        assert!(preflight.protocol.reader_features.is_empty());
362
363        Ok(())
364    }
365
366    #[test]
367    fn preflight_allows_column_mapping_reader_feature() -> Result<(), Box<dyn std::error::Error>> {
368        let source = load_source("orders", COLUMN_MAPPING_FEATURE_PROTOCOL_JSON)?;
369
370        let preflight = preflight_delta_protocol(&source)?;
371
372        assert_eq!(preflight.protocol.min_reader_version, 3);
373        assert_eq!(preflight.protocol.reader_features, vec!["columnMapping"]);
374        assert_eq!(preflight.protocol.writer_features, vec!["columnMapping"]);
375
376        Ok(())
377    }
378
379    #[test]
380    fn preflights_multiple_sources() -> Result<(), Box<dyn std::error::Error>> {
381        let orders = DeltaLogTable::new("orders", LEGACY_PROTOCOL_JSON)?;
382        let customers = DeltaLogTable::new("customers", WRITER_ONLY_FEATURE_PROTOCOL_JSON)?;
383        let sources = load_delta_sources([
384            source_config("orders", &orders),
385            source_config("customers", &customers),
386        ])?;
387
388        let preflights = preflight_delta_sources(&sources)?;
389
390        assert_eq!(preflights.len(), 2);
391        assert_eq!(preflights[0].protocol.source_name, "orders");
392        assert_eq!(preflights[0].protocol.min_reader_version, 1);
393        assert_eq!(preflights[1].protocol.source_name, "customers");
394        assert_eq!(
395            preflights[1].protocol.writer_features,
396            vec!["inCommitTimestamp"]
397        );
398
399        Ok(())
400    }
401
402    #[test]
403    fn multi_source_preflight_allows_deletion_vectors_source()
404    -> Result<(), Box<dyn std::error::Error>> {
405        let orders = DeltaLogTable::new("orders", LEGACY_PROTOCOL_JSON)?;
406        let customers = DeltaLogTable::new("customers", DELETION_VECTOR_PROTOCOL_JSON)?;
407        let sources = load_delta_sources([
408            source_config("orders", &orders),
409            source_config("customers", &customers),
410        ])?;
411
412        let preflights = preflight_delta_sources(&sources)?;
413
414        assert_eq!(preflights.len(), 2);
415        assert_eq!(preflights[0].protocol.source_name, "orders");
416        assert_eq!(preflights[1].protocol.source_name, "customers");
417        assert_eq!(
418            preflights[1].protocol.reader_features,
419            vec!["deletionVectors"]
420        );
421
422        Ok(())
423    }
424
425    #[test]
426    fn protocol_policy_rejects_future_reader_versions() {
427        let report = DeltaProtocolReport {
428            source_name: "orders".to_owned(),
429            table_uri: "s3://bucket/table/".to_owned(),
430            snapshot_version: 42,
431            min_reader_version: 4,
432            min_writer_version: 7,
433            reader_features: Vec::new(),
434            writer_features: Vec::new(),
435        };
436
437        let result = super::ensure_protocol_supported(&report);
438
439        assert!(matches!(
440            result,
441            Err(DeltaFunnelError::DeltaProtocolCompatibility {
442                reason,
443                ..
444            }) if reason.contains("minReaderVersion 4")
445        ));
446    }
447
448    #[test]
449    fn protocol_policy_allows_reader_version_three_without_reader_features() {
450        let report = DeltaProtocolReport {
451            source_name: "orders".to_owned(),
452            table_uri: "s3://bucket/table/".to_owned(),
453            snapshot_version: 42,
454            min_reader_version: super::TABLE_FEATURES_MIN_READER_VERSION,
455            min_writer_version: 7,
456            reader_features: Vec::new(),
457            writer_features: vec!["inCommitTimestamp".to_owned()],
458        };
459
460        assert!(super::ensure_protocol_supported(&report).is_ok());
461    }
462
463    #[test]
464    fn protocol_policy_reports_first_unsupported_reader_feature() {
465        let report = DeltaProtocolReport {
466            source_name: "orders".to_owned(),
467            table_uri: "s3://bucket/table/".to_owned(),
468            snapshot_version: 42,
469            min_reader_version: super::TABLE_FEATURES_MIN_READER_VERSION,
470            min_writer_version: 7,
471            reader_features: vec![
472                "deletionVectors".to_owned(),
473                "columnMapping".to_owned(),
474                "madeUpFeature".to_owned(),
475            ],
476            writer_features: vec![
477                "deletionVectors".to_owned(),
478                "columnMapping".to_owned(),
479                "madeUpFeature".to_owned(),
480            ],
481        };
482
483        let result = super::ensure_protocol_supported(&report);
484
485        assert!(matches!(
486            result,
487            Err(DeltaFunnelError::DeltaProtocolCompatibility {
488                reason,
489                ..
490            }) if reason.contains("madeUpFeature")
491                && !reason.contains("deletionVectors")
492                && !reason.contains("columnMapping")
493        ));
494    }
495
496    #[test]
497    fn compatibility_error_display_redacts_uri_credentials() {
498        let error = DeltaFunnelError::DeltaProtocolCompatibility {
499            source_name: "orders".to_owned(),
500            table_uri: "s3://user:password@example.com/table?token=secret".to_owned(),
501            snapshot_version: 9,
502            reason: "unsupported Delta reader feature `madeUpFeature`".to_owned(),
503        };
504
505        let display = error.to_string();
506
507        assert!(display.contains("orders"));
508        assert!(display.contains("madeUpFeature"));
509        assert!(!display.contains("user"));
510        assert!(!display.contains("password"));
511        assert!(!display.contains("token"));
512        assert!(!display.contains("secret"));
513    }
514}
delta_funnel/table_formats/delta/protocol.rs

delta_funnel/table_formats/delta/
protocol.rs
