delta_funnel/query_engine/datafusion/catalog/
registration.rs

1//! DataFusion session registration for Delta sources.
2
3use std::sync::Arc;
4
5use datafusion::arrow::datatypes::SchemaRef;
6use datafusion::datasource::TableProvider;
7use datafusion::prelude::SessionContext;
8
9use crate::{
10    DeltaFunnelError, DeltaProtocolReport, PlannedDeltaSource, ProtocolPreflight,
11    error::DataFusionRegistrationSnafu, observability, support::sanitize_uri_for_display,
12    table_formats::validate_table_source_names,
13};
14
15use super::super::execution::DeltaProviderScanExecutionOptions;
16use super::provider::DeltaTableProvider;
17
18/// Delta source and preflight state used to build a DataFusion table provider.
19pub struct DeltaTableProviderConfig {
20    /// Loaded Delta source.
21    pub source: PlannedDeltaSource,
22    /// Successful Delta protocol preflight for the source.
23    pub protocol: ProtocolPreflight,
24    /// Optional DeltaFunnel scan file task partition target override.
25    ///
26    /// When set, this value wins over DataFusion's session target and automatic
27    /// machine fallback policy for this Delta source only.
28    pub scan_target_partitions: Option<usize>,
29}
30
31/// Registered Delta sources visible to a DataFusion session.
32#[derive(Debug, Clone)]
33pub struct RegisteredDeltaSources {
34    /// Per-source registration reports.
35    pub sources: Vec<RegisteredDeltaSource>,
36}
37
38/// One registered Delta source.
39#[derive(Debug, Clone)]
40pub struct RegisteredDeltaSource {
41    /// DataFusion table name for this source.
42    pub name: String,
43    /// Sanitized normalized Delta table URI context.
44    pub table_uri: String,
45    /// Resolved Delta snapshot version.
46    pub snapshot_version: u64,
47    /// Logical Arrow schema exposed to DataFusion.
48    pub schema: SchemaRef,
49    /// Protocol report captured before registration.
50    pub protocol: DeltaProtocolReport,
51}
52
53/// Registers preflighted Delta sources into a DataFusion session.
54///
55/// # Errors
56///
57/// Returns [`DeltaFunnelError::DeltaSourceSchema`] when a source schema cannot
58/// be converted to Arrow, or [`DeltaFunnelError::DataFusionRegistration`] when
59/// DataFusion rejects a table registration.
60pub fn register_delta_sources(
61    ctx: &SessionContext,
62    configs: Vec<DeltaTableProviderConfig>,
63) -> Result<RegisteredDeltaSources, DeltaFunnelError> {
64    register_delta_sources_with_options(ctx, configs, DeltaProviderScanExecutionOptions::default())
65}
66
67/// Registers preflighted Delta sources with explicit provider execution bounds.
68///
69/// # Errors
70///
71/// Returns [`DeltaFunnelError::Config`] when execution bounds are invalid,
72/// [`DeltaFunnelError::DeltaSourceSchema`] when a source schema cannot be
73/// converted to Arrow, or [`DeltaFunnelError::DataFusionRegistration`] when
74/// DataFusion rejects a table registration.
75pub fn register_delta_sources_with_scan_execution_options(
76    ctx: &SessionContext,
77    configs: Vec<DeltaTableProviderConfig>,
78    execution_options: DeltaProviderScanExecutionOptions,
79) -> Result<RegisteredDeltaSources, DeltaFunnelError> {
80    execution_options.validate()?;
81    register_delta_sources_with_options(ctx, configs, execution_options)
82}
83
84fn register_delta_sources_with_options(
85    ctx: &SessionContext,
86    configs: Vec<DeltaTableProviderConfig>,
87    execution_options: DeltaProviderScanExecutionOptions,
88) -> Result<RegisteredDeltaSources, DeltaFunnelError> {
89    reject_duplicate_registration_names(&configs)?;
90    let providers = configs
91        .into_iter()
92        .map(|config| {
93            DeltaTableProvider::try_new_with_execution_options(
94                config.source,
95                config.protocol,
96                config.scan_target_partitions,
97                execution_options,
98            )
99        })
100        .collect::<Result<Vec<_>, _>>()?;
101
102    reject_existing_registration_names(ctx, &providers)?;
103
104    let sources = register_delta_providers(ctx, providers)?;
105
106    Ok(RegisteredDeltaSources { sources })
107}
108
109fn reject_duplicate_registration_names(
110    configs: &[DeltaTableProviderConfig],
111) -> Result<(), DeltaFunnelError> {
112    validate_table_source_names(configs.iter().map(|config| config.source.name()))
113}
114
115fn reject_existing_registration_names(
116    ctx: &SessionContext,
117    providers: &[DeltaTableProvider],
118) -> Result<(), DeltaFunnelError> {
119    let state = ctx.state();
120    let catalog_options = &state.config_options().catalog;
121    let default_catalog = ctx.catalog(&catalog_options.default_catalog);
122    let default_schema = default_catalog
123        .as_ref()
124        .and_then(|catalog| catalog.schema(&catalog_options.default_schema));
125    let existing_names = default_schema
126        .as_ref()
127        .map_or_else(Vec::new, |schema| schema.table_names());
128
129    for provider in providers {
130        if let Some(existing_name) = existing_names
131            .iter()
132            .find(|existing_name| existing_name.eq_ignore_ascii_case(provider.source_name()))
133        {
134            return DataFusionRegistrationSnafu {
135                source_name: provider.source_name().to_owned(),
136                table_uri: provider.source_table_uri().to_owned(),
137                reason: format!("table already exists: {existing_name}"),
138            }
139            .fail();
140        }
141    }
142
143    Ok(())
144}
145
146fn register_delta_provider(
147    ctx: &SessionContext,
148    provider: DeltaTableProvider,
149) -> Result<RegisteredDeltaSource, DeltaFunnelError> {
150    let registered = RegisteredDeltaSource {
151        name: provider.source_name().to_owned(),
152        table_uri: sanitize_uri_for_display(provider.source_table_uri()),
153        snapshot_version: provider.snapshot_version(),
154        schema: provider.schema(),
155        protocol: provider.protocol().clone(),
156    };
157    let table_uri = provider.source_table_uri().to_owned();
158
159    if let Err(error) = ctx.register_table(registered.name.as_str(), Arc::new(provider)) {
160        return DataFusionRegistrationSnafu {
161            source_name: registered.name.clone(),
162            table_uri,
163            reason: error.to_string(),
164        }
165        .fail();
166    }
167
168    Ok(registered)
169}
170
171fn register_delta_provider_with_tracing(
172    ctx: &SessionContext,
173    provider: DeltaTableProvider,
174) -> Result<RegisteredDeltaSource, DeltaFunnelError> {
175    let source_name = provider.source_name().to_owned();
176    let snapshot_version = provider.snapshot_version();
177    observability::datafusion_registration_started(&source_name, snapshot_version);
178
179    let result = register_delta_provider(ctx, provider);
180    match &result {
181        Ok(registered) => {
182            observability::datafusion_registration_completed(&registered.name, snapshot_version);
183        }
184        Err(error) => {
185            observability::datafusion_registration_failed(&source_name, snapshot_version, error);
186        }
187    }
188    result
189}
190
191fn register_delta_providers(
192    ctx: &SessionContext,
193    providers: Vec<DeltaTableProvider>,
194) -> Result<Vec<RegisteredDeltaSource>, DeltaFunnelError> {
195    let mut registered_sources = Vec::with_capacity(providers.len());
196    let mut registered_names = Vec::with_capacity(providers.len());
197
198    for provider in providers {
199        let registered = match register_delta_provider_with_tracing(ctx, provider) {
200            Ok(registered) => registered,
201            Err(error) => {
202                rollback_registered_delta_sources(ctx, &registered_names);
203                return Err(error);
204            }
205        };
206
207        registered_names.push(registered.name.clone());
208        registered_sources.push(registered);
209    }
210
211    Ok(registered_sources)
212}
213
214fn rollback_registered_delta_sources(ctx: &SessionContext, names: &[String]) {
215    for name in names.iter().rev() {
216        let _ = ctx.deregister_table(name.as_str());
217    }
218}
219
220pub(super) fn reject_mismatched_preflight(
221    source: &PlannedDeltaSource,
222    protocol: &DeltaProtocolReport,
223) -> Result<(), DeltaFunnelError> {
224    let source_table_uri = sanitize_uri_for_display(source.table_uri());
225
226    if protocol.source_name != source.name()
227        || protocol.snapshot_version != source.version()
228        || protocol.table_uri != source_table_uri
229    {
230        return DataFusionRegistrationSnafu {
231            source_name: source.name().to_owned(),
232            table_uri: source.table_uri().to_owned(),
233            reason: format!(
234                "protocol preflight belongs to source `{}` at snapshot version {} ({})",
235                protocol.source_name, protocol.snapshot_version, protocol.table_uri
236            ),
237        }
238        .fail();
239    }
240
241    Ok(())
242}
243
244#[cfg(test)]
245mod tests {
246    use std::sync::Arc;
247
248    use datafusion::arrow::datatypes::{DataType, Field, Schema};
249    use datafusion::datasource::TableType;
250    use datafusion::datasource::empty::EmptyTable;
251    use datafusion::prelude::{SessionConfig, SessionContext};
252
253    use super::*;
254    use crate::query_engine::datafusion::execution::DeltaProviderReaderBackend;
255    use crate::query_engine::datafusion::test_support::{
256        DeltaLogTable, FailsOnCustomersSchemaProvider, INVALID_NESTED_IDS_SCHEMA_FIELDS_JSON,
257        SingleSchemaCatalogProvider, register_fixture_source,
258    };
259    use crate::{DeltaFunnelError, DeltaSourceConfig, load_delta_source, preflight_delta_protocol};
260
261    #[test]
262    fn registers_preflighted_delta_source() -> Result<(), Box<dyn std::error::Error>> {
263        let table = DeltaLogTable::new("registration")?;
264        let source = load_delta_source(DeltaSourceConfig {
265            name: "orders".to_owned(),
266            table_uri: table.path().to_string_lossy().to_string(),
267            version: None,
268            storage_options: Default::default(),
269        })?;
270        let preflight = preflight_delta_protocol(&source)?;
271        let ctx = SessionContext::new();
272
273        let registered = register_delta_sources(
274            &ctx,
275            vec![DeltaTableProviderConfig {
276                source,
277                protocol: preflight,
278                scan_target_partitions: None,
279            }],
280        )?;
281
282        assert_eq!(registered.sources.len(), 1);
283        assert_eq!(registered.sources[0].name, "orders");
284        assert!(registered.sources[0].table_uri.starts_with("file://"));
285        assert_eq!(registered.sources[0].snapshot_version, 1);
286        assert_eq!(registered.sources[0].schema.field(0).name(), "id");
287        assert_eq!(registered.sources[0].protocol.source_name, "orders");
288
289        Ok(())
290    }
291
292    #[test]
293    fn registration_rejects_zero_execution_bounds_before_registration()
294    -> Result<(), Box<dyn std::error::Error>> {
295        let table = DeltaLogTable::new("registration-zero-execution-bounds")?;
296        let source = load_delta_source(DeltaSourceConfig {
297            name: "orders".to_owned(),
298            table_uri: table.path().to_string_lossy().to_string(),
299            version: None,
300            storage_options: Default::default(),
301        })?;
302        let preflight = preflight_delta_protocol(&source)?;
303        let ctx = SessionContext::new();
304
305        let result = register_delta_sources_with_scan_execution_options(
306            &ctx,
307            vec![DeltaTableProviderConfig {
308                source,
309                protocol: preflight,
310                scan_target_partitions: None,
311            }],
312            DeltaProviderScanExecutionOptions {
313                reader_backend: DeltaProviderReaderBackend::OfficialKernel,
314                max_concurrent_file_reads_per_scan: Some(0),
315                max_concurrent_file_reads_per_partition: 1,
316                output_buffer_capacity_per_partition: 1,
317                native_async_prefetch_file_count_per_partition: 0,
318            },
319        );
320
321        assert!(matches!(
322            result,
323            Err(DeltaFunnelError::Config { message })
324                if message == "max_concurrent_file_reads_per_scan must be greater than zero"
325        ));
326        assert!(!ctx.table_exist("orders")?);
327
328        Ok(())
329    }
330
331    #[test]
332    fn registration_accepts_native_async_backend_for_local_file_uri()
333    -> Result<(), Box<dyn std::error::Error>> {
334        let table = DeltaLogTable::new("registration-native-async-local-file")?;
335        let source = load_delta_source(DeltaSourceConfig {
336            name: "orders".to_owned(),
337            table_uri: table.path().to_string_lossy().to_string(),
338            version: None,
339            storage_options: Default::default(),
340        })?;
341        let preflight = preflight_delta_protocol(&source)?;
342        let ctx = SessionContext::new();
343
344        let registered = register_delta_sources_with_scan_execution_options(
345            &ctx,
346            vec![DeltaTableProviderConfig {
347                source,
348                protocol: preflight,
349                scan_target_partitions: None,
350            }],
351            DeltaProviderScanExecutionOptions {
352                reader_backend: DeltaProviderReaderBackend::NativeAsync,
353                max_concurrent_file_reads_per_scan: Some(1),
354                max_concurrent_file_reads_per_partition: 1,
355                output_buffer_capacity_per_partition: 1,
356                native_async_prefetch_file_count_per_partition: 0,
357            },
358        )?;
359
360        assert_eq!(registered.sources.len(), 1);
361        assert_eq!(registered.sources[0].name, "orders");
362        assert!(ctx.table_exist("orders")?);
363
364        Ok(())
365    }
366
367    #[tokio::test]
368    async fn catalog_inspection_exposes_registered_provider_schema()
369    -> Result<(), Box<dyn std::error::Error>> {
370        let ctx = SessionContext::new();
371        let _table = register_fixture_source(&ctx, "orders", "catalog-inspection")?;
372
373        let catalog = ctx.catalog("datafusion").ok_or("missing default catalog")?;
374        let schema = catalog.schema("public").ok_or("missing default schema")?;
375        let provider = schema
376            .table("orders")
377            .await?
378            .ok_or("missing registered table provider")?;
379        let provider_schema = provider.schema();
380
381        assert!(schema.table_names().contains(&"orders".to_owned()));
382        assert_eq!(provider.table_type(), TableType::Base);
383        assert_eq!(provider_schema.fields().len(), 2);
384        assert_eq!(provider_schema.field(0).name(), "id");
385        assert_eq!(provider_schema.field(0).data_type(), &DataType::Int32);
386        assert!(!provider_schema.field(0).is_nullable());
387        assert_eq!(provider_schema.field(1).name(), "customer_name");
388        assert_eq!(provider_schema.field(1).data_type(), &DataType::Utf8);
389        assert!(provider_schema.field(1).is_nullable());
390
391        Ok(())
392    }
393
394    #[test]
395    fn registration_failure_reports_source_context() -> Result<(), Box<dyn std::error::Error>> {
396        let table = DeltaLogTable::new("registration-failure")?;
397        let source = load_delta_source(DeltaSourceConfig {
398            name: "orders".to_owned(),
399            table_uri: table.path().to_string_lossy().to_string(),
400            version: None,
401            storage_options: Default::default(),
402        })?;
403        let preflight = preflight_delta_protocol(&source)?;
404        let ctx = SessionContext::new();
405        let placeholder_schema = Arc::new(Schema::new(vec![Field::new(
406            "existing",
407            DataType::Utf8,
408            true,
409        )]));
410
411        ctx.register_table("orders", Arc::new(EmptyTable::new(placeholder_schema)))?;
412        let result = register_delta_sources(
413            &ctx,
414            vec![DeltaTableProviderConfig {
415                source,
416                protocol: preflight,
417                scan_target_partitions: None,
418            }],
419        );
420
421        assert!(matches!(
422            result,
423            Err(DeltaFunnelError::DataFusionRegistration {
424                source_name,
425                reason,
426                ..
427            }) if source_name == "orders" && reason.contains("already exists")
428        ));
429
430        Ok(())
431    }
432
433    #[test]
434    fn mismatched_preflight_is_rejected_before_registration()
435    -> Result<(), Box<dyn std::error::Error>> {
436        let orders = DeltaLogTable::new("mismatched-preflight-orders")?;
437        let customers = DeltaLogTable::new("mismatched-preflight-customers")?;
438        let orders_source = load_delta_source(DeltaSourceConfig {
439            name: "orders".to_owned(),
440            table_uri: orders.path().to_string_lossy().to_string(),
441            version: None,
442            storage_options: Default::default(),
443        })?;
444        let customers_source = load_delta_source(DeltaSourceConfig {
445            name: "customers".to_owned(),
446            table_uri: customers.path().to_string_lossy().to_string(),
447            version: None,
448            storage_options: Default::default(),
449        })?;
450        let customers_preflight = preflight_delta_protocol(&customers_source)?;
451        let ctx = SessionContext::new();
452
453        let result = register_delta_sources(
454            &ctx,
455            vec![DeltaTableProviderConfig {
456                source: orders_source,
457                protocol: customers_preflight,
458                scan_target_partitions: None,
459            }],
460        );
461
462        assert!(matches!(
463            result,
464            Err(DeltaFunnelError::DataFusionRegistration {
465                source_name,
466                reason,
467                ..
468            }) if source_name == "orders"
469                && reason.contains("protocol preflight belongs to source `customers`")
470        ));
471        assert!(!ctx.table_exist("orders")?);
472
473        Ok(())
474    }
475
476    #[test]
477    fn existing_table_conflict_fails_before_partial_registration()
478    -> Result<(), Box<dyn std::error::Error>> {
479        let orders = DeltaLogTable::new("existing-conflict-orders")?;
480        let customers = DeltaLogTable::new("existing-conflict-customers")?;
481        let orders_source = load_delta_source(DeltaSourceConfig {
482            name: "orders".to_owned(),
483            table_uri: orders.path().to_string_lossy().to_string(),
484            version: None,
485            storage_options: Default::default(),
486        })?;
487        let customers_source = load_delta_source(DeltaSourceConfig {
488            name: "customers".to_owned(),
489            table_uri: customers.path().to_string_lossy().to_string(),
490            version: None,
491            storage_options: Default::default(),
492        })?;
493        let orders_preflight = preflight_delta_protocol(&orders_source)?;
494        let customers_preflight = preflight_delta_protocol(&customers_source)?;
495        let ctx = SessionContext::new();
496        let placeholder_schema = Arc::new(Schema::new(vec![Field::new(
497            "existing",
498            DataType::Utf8,
499            true,
500        )]));
501
502        ctx.register_table("customers", Arc::new(EmptyTable::new(placeholder_schema)))?;
503        let result = register_delta_sources(
504            &ctx,
505            vec![
506                DeltaTableProviderConfig {
507                    source: orders_source,
508                    protocol: orders_preflight,
509                    scan_target_partitions: None,
510                },
511                DeltaTableProviderConfig {
512                    source: customers_source,
513                    protocol: customers_preflight,
514                    scan_target_partitions: None,
515                },
516            ],
517        );
518
519        assert!(matches!(
520            result,
521            Err(DeltaFunnelError::DataFusionRegistration {
522                source_name,
523                reason,
524                ..
525            }) if source_name == "customers" && reason.contains("already exists")
526        ));
527        assert!(!ctx.table_exist("orders")?);
528        assert!(ctx.table_exist("customers")?);
529
530        Ok(())
531    }
532
533    #[test]
534    fn late_registration_failure_rolls_back_prior_sources() -> Result<(), Box<dyn std::error::Error>>
535    {
536        let orders = DeltaLogTable::new("rollback-orders")?;
537        let customers = DeltaLogTable::new("rollback-customers")?;
538        let orders_source = load_delta_source(DeltaSourceConfig {
539            name: "orders".to_owned(),
540            table_uri: orders.path().to_string_lossy().to_string(),
541            version: None,
542            storage_options: Default::default(),
543        })?;
544        let customers_source = load_delta_source(DeltaSourceConfig {
545            name: "customers".to_owned(),
546            table_uri: customers.path().to_string_lossy().to_string(),
547            version: None,
548            storage_options: Default::default(),
549        })?;
550        let orders_preflight = preflight_delta_protocol(&orders_source)?;
551        let customers_preflight = preflight_delta_protocol(&customers_source)?;
552        let ctx = SessionContext::new();
553        let failing_schema: Arc<dyn datafusion::catalog::SchemaProvider> =
554            Arc::new(FailsOnCustomersSchemaProvider::default());
555
556        ctx.register_catalog(
557            "datafusion",
558            Arc::new(SingleSchemaCatalogProvider::new(Arc::clone(
559                &failing_schema,
560            ))),
561        );
562        let result = register_delta_sources(
563            &ctx,
564            vec![
565                DeltaTableProviderConfig {
566                    source: orders_source,
567                    protocol: orders_preflight,
568                    scan_target_partitions: None,
569                },
570                DeltaTableProviderConfig {
571                    source: customers_source,
572                    protocol: customers_preflight,
573                    scan_target_partitions: None,
574                },
575            ],
576        );
577
578        assert!(matches!(
579            result,
580            Err(DeltaFunnelError::DataFusionRegistration {
581                source_name,
582                reason,
583                ..
584            }) if source_name == "customers"
585                && reason.contains("forced customers registration failure")
586        ));
587        assert!(!ctx.table_exist("orders")?);
588        assert!(!ctx.table_exist("customers")?);
589
590        Ok(())
591    }
592
593    #[test]
594    fn existing_table_conflict_uses_configured_default_catalog_and_schema()
595    -> Result<(), Box<dyn std::error::Error>> {
596        let orders = DeltaLogTable::new("custom-default-orders")?;
597        let customers = DeltaLogTable::new("custom-default-customers")?;
598        let orders_source = load_delta_source(DeltaSourceConfig {
599            name: "orders".to_owned(),
600            table_uri: orders.path().to_string_lossy().to_string(),
601            version: None,
602            storage_options: Default::default(),
603        })?;
604        let customers_source = load_delta_source(DeltaSourceConfig {
605            name: "customers".to_owned(),
606            table_uri: customers.path().to_string_lossy().to_string(),
607            version: None,
608            storage_options: Default::default(),
609        })?;
610        let orders_preflight = preflight_delta_protocol(&orders_source)?;
611        let customers_preflight = preflight_delta_protocol(&customers_source)?;
612        let ctx = SessionContext::new_with_config(
613            SessionConfig::new().with_default_catalog_and_schema("custom_catalog", "custom_schema"),
614        );
615        let placeholder_schema = Arc::new(Schema::new(vec![Field::new(
616            "existing",
617            DataType::Utf8,
618            true,
619        )]));
620
621        ctx.register_table("customers", Arc::new(EmptyTable::new(placeholder_schema)))?;
622        let result = register_delta_sources(
623            &ctx,
624            vec![
625                DeltaTableProviderConfig {
626                    source: orders_source,
627                    protocol: orders_preflight,
628                    scan_target_partitions: None,
629                },
630                DeltaTableProviderConfig {
631                    source: customers_source,
632                    protocol: customers_preflight,
633                    scan_target_partitions: None,
634                },
635            ],
636        );
637
638        assert!(matches!(
639            result,
640            Err(DeltaFunnelError::DataFusionRegistration {
641                source_name,
642                reason,
643                ..
644            }) if source_name == "customers" && reason.contains("already exists")
645        ));
646        assert!(!ctx.table_exist("orders")?);
647        assert!(ctx.table_exist("customers")?);
648
649        Ok(())
650    }
651
652    #[test]
653    fn schema_conversion_failure_fails_before_partial_registration()
654    -> Result<(), Box<dyn std::error::Error>> {
655        let orders = DeltaLogTable::new("schema-partial-orders")?;
656        let customers = DeltaLogTable::new_with_schema(
657            "schema-partial-customers",
658            INVALID_NESTED_IDS_SCHEMA_FIELDS_JSON,
659            "[]",
660            r#""partitionValues":{}"#,
661        )?;
662        let orders_source = load_delta_source(DeltaSourceConfig {
663            name: "orders".to_owned(),
664            table_uri: orders.path().to_string_lossy().to_string(),
665            version: None,
666            storage_options: Default::default(),
667        })?;
668        let customers_source = load_delta_source(DeltaSourceConfig {
669            name: "customers".to_owned(),
670            table_uri: customers.path().to_string_lossy().to_string(),
671            version: None,
672            storage_options: Default::default(),
673        })?;
674        let orders_preflight = preflight_delta_protocol(&orders_source)?;
675        let customers_preflight = preflight_delta_protocol(&customers_source)?;
676        let ctx = SessionContext::new();
677
678        let result = register_delta_sources(
679            &ctx,
680            vec![
681                DeltaTableProviderConfig {
682                    source: orders_source,
683                    protocol: orders_preflight,
684                    scan_target_partitions: None,
685                },
686                DeltaTableProviderConfig {
687                    source: customers_source,
688                    protocol: customers_preflight,
689                    scan_target_partitions: None,
690                },
691            ],
692        );
693
694        assert!(matches!(
695            result,
696            Err(DeltaFunnelError::DeltaSourceSchema {
697                source_name,
698                reason,
699                ..
700            }) if source_name == "customers"
701                && reason.contains("bad_array")
702                && reason.contains("delta.columnMapping.nested.ids")
703        ));
704        assert!(!ctx.table_exist("orders")?);
705        assert!(!ctx.table_exist("customers")?);
706
707        Ok(())
708    }
709
710    #[test]
711    fn duplicate_registration_names_fail_before_partial_registration()
712    -> Result<(), Box<dyn std::error::Error>> {
713        let orders = DeltaLogTable::new("duplicate-orders")?;
714        let customers = DeltaLogTable::new("duplicate-customers")?;
715        let orders_source = load_delta_source(DeltaSourceConfig {
716            name: "orders".to_owned(),
717            table_uri: orders.path().to_string_lossy().to_string(),
718            version: None,
719            storage_options: Default::default(),
720        })?;
721        let customers_source = load_delta_source(DeltaSourceConfig {
722            name: "Orders".to_owned(),
723            table_uri: customers.path().to_string_lossy().to_string(),
724            version: None,
725            storage_options: Default::default(),
726        })?;
727        let orders_preflight = preflight_delta_protocol(&orders_source)?;
728        let customers_preflight = preflight_delta_protocol(&customers_source)?;
729        let ctx = SessionContext::new();
730
731        let result = register_delta_sources(
732            &ctx,
733            vec![
734                DeltaTableProviderConfig {
735                    source: orders_source,
736                    protocol: orders_preflight,
737                    scan_target_partitions: None,
738                },
739                DeltaTableProviderConfig {
740                    source: customers_source,
741                    protocol: customers_preflight,
742                    scan_target_partitions: None,
743                },
744            ],
745        );
746
747        assert!(matches!(
748            result,
749            Err(DeltaFunnelError::DuplicateSourceName { name }) if name == "Orders"
750        ));
751        assert!(!ctx.table_exist("orders")?);
752
753        Ok(())
754    }
755}
delta_funnel/query_engine/datafusion/catalog/registration.rs

delta_funnel/query_engine/datafusion/catalog/
registration.rs
