delta_funnel/orchestrator/session/registry/
derived.rs

1use std::{fmt, sync::Arc};
2
3use datafusion::{
4    arrow::datatypes::SchemaRef,
5    datasource::TableProvider,
6    prelude::{DataFrame, SQLOptions},
7};
8
9use crate::{
10    DeltaFunnelError, PhaseTimingReport, SqlTablePhase, report::PhaseTimer,
11    table_formats::validate_table_source_names,
12};
13
14use super::super::{
15    DeltaFunnelSession, LazyTable, LazyTableKind,
16    errors::{sql_table_error, unknown_lazy_table_error},
17};
18use super::{DerivedTableLineage, RegisteredSessionSource};
19
20const LAZY_SQL_PLANNING_PHASE: &str = "lazy_sql_planning";
21
22/// Registered SQL-derived table alias tracked by a query-load session.
23#[derive(Clone, PartialEq, Eq)]
24pub struct RegisteredDerivedTable {
25    table: LazyTable,
26    schema: SchemaRef,
27    pub(in crate::orchestrator::session) sql_text: String,
28    pub(in crate::orchestrator::session) lineage: DerivedTableLineage,
29    phase_timings: Vec<PhaseTimingReport>,
30}
31
32impl RegisteredDerivedTable {
33    pub(super) fn new(
34        table: LazyTable,
35        schema: SchemaRef,
36        sql_text: String,
37        lineage: DerivedTableLineage,
38        phase_timings: Vec<PhaseTimingReport>,
39    ) -> Self {
40        Self {
41            table,
42            schema,
43            sql_text,
44            lineage,
45            phase_timings,
46        }
47    }
48
49    /// Returns the lazy table handle for this registered derived alias.
50    #[must_use]
51    pub const fn table(&self) -> &LazyTable {
52        &self.table
53    }
54
55    /// Returns the DataFusion table name for this derived alias.
56    #[must_use]
57    pub fn name(&self) -> &str {
58        self.table.name()
59    }
60
61    /// Returns the logical Arrow schema exposed to DataFusion.
62    #[must_use]
63    pub fn schema(&self) -> &SchemaRef {
64        &self.schema
65    }
66
67    /// Returns the retained SQL text used to create this derived alias.
68    #[allow(dead_code)]
69    #[must_use]
70    pub(crate) fn sql_text(&self) -> &str {
71        &self.sql_text
72    }
73
74    /// Returns dependency lineage captured from the retained SQL text.
75    #[allow(dead_code)]
76    #[must_use]
77    pub(crate) const fn lineage(&self) -> &DerivedTableLineage {
78        &self.lineage
79    }
80
81    /// Returns durable phase timings captured while planning this derived table.
82    #[must_use]
83    pub fn phase_timings(&self) -> &[PhaseTimingReport] {
84        &self.phase_timings
85    }
86}
87
88impl fmt::Debug for RegisteredDerivedTable {
89    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
90        formatter
91            .debug_struct("RegisteredDerivedTable")
92            .field("table", &self.table)
93            .field("schema", &self.schema)
94            .field("sql_text", &"<redacted>")
95            .field("lineage", &self.lineage)
96            .field("phase_timings", &self.phase_timings)
97            .finish()
98    }
99}
100
101#[derive(Clone)]
102pub(in crate::orchestrator::session) struct PendingDerivedTable {
103    pub(in crate::orchestrator::session) table: LazyTable,
104    pub(in crate::orchestrator::session) provider: Arc<dyn TableProvider>,
105    pub(in crate::orchestrator::session) schema: SchemaRef,
106    pub(in crate::orchestrator::session) sql_text: String,
107    pub(in crate::orchestrator::session) lineage: DerivedTableLineage,
108    pub(in crate::orchestrator::session) phase_timings: Vec<PhaseTimingReport>,
109}
110
111impl DeltaFunnelSession {
112    /// Builds a lazy SQL-derived table without registering a query alias.
113    ///
114    /// The SQL must be one read-only tabular query. Planning uses DataFusion to
115    /// produce a lazy table provider and does not execute rows, contact SQL
116    /// Server, or create an output target.
117    ///
118    /// # Errors
119    ///
120    /// Returns [`DeltaFunnelError::SqlTable`] when the SQL is empty, contains
121    /// an unsupported or non-read-only statement, or cannot be planned against
122    /// the session's registered aliases.
123    pub async fn table_from_sql(&mut self, sql: &str) -> Result<LazyTable, DeltaFunnelError> {
124        let sql = sql.trim();
125        if sql.is_empty() {
126            return sql_table_error(SqlTablePhase::ValidateSql, "SQL text must not be empty");
127        }
128
129        let planning_timer = PhaseTimer::start(LAZY_SQL_PLANNING_PHASE);
130        let dataframe = match self.plan_read_only_sql(sql).await {
131            Ok(dataframe) => dataframe,
132            Err(error) => {
133                let _ = planning_timer.failed();
134                return Err(error);
135            }
136        };
137        let schema = Arc::new(dataframe.schema().as_arrow().clone());
138        let provider = dataframe.into_view();
139        let lineage = self.derive_table_lineage_from_sql(sql);
140        let phase_timings = vec![planning_timer.completed()];
141        let table = self.allocate_derived_sql_table();
142        self.pending_derived_tables.push(PendingDerivedTable {
143            table: table.clone(),
144            provider,
145            schema,
146            sql_text: sql.to_owned(),
147            lineage,
148            phase_timings,
149        });
150        Ok(table)
151    }
152
153    /// Registers a session-owned alias for a lazy SQL-derived table.
154    ///
155    /// Alias names use the same unquoted identifier rules as Delta source
156    /// aliases. The alias is registered into the session's DataFusion catalog
157    /// only after all local validation succeeds.
158    ///
159    /// # Errors
160    ///
161    /// Returns [`DeltaFunnelError::InvalidSourceName`] or
162    /// [`DeltaFunnelError::DuplicateSourceName`] for invalid or ambiguous
163    /// aliases, and [`DeltaFunnelError::SqlTable`] when the table handle is not
164    /// a pending SQL-derived table owned by this session or DataFusion rejects
165    /// the alias registration.
166    pub fn register_alias(
167        &mut self,
168        name: impl Into<String>,
169        table: &LazyTable,
170    ) -> Result<LazyTable, DeltaFunnelError> {
171        let name = name.into();
172        validate_table_source_names([name.as_str()])?;
173        self.reject_registered_alias_name(&name)?;
174
175        let Some(index) = self.find_pending_derived_table(table) else {
176            return sql_table_error(
177                SqlTablePhase::RegisterDerivedAlias,
178                "lazy table is not a pending SQL-derived table owned by this session",
179            );
180        };
181        let pending = self.pending_derived_tables.remove(index);
182
183        if let Err(error) = self
184            .context
185            .register_table(name.as_str(), Arc::clone(&pending.provider))
186        {
187            let message = error.to_string();
188            self.pending_derived_tables.push(pending);
189            return sql_table_error(SqlTablePhase::RegisterDerivedAlias, message);
190        }
191
192        let alias_table = pending.table.with_name(name);
193        self.derived_tables.push(RegisteredDerivedTable::new(
194            alias_table.clone(),
195            pending.schema,
196            pending.sql_text,
197            pending.lineage,
198            pending.phase_timings,
199        ));
200        Ok(alias_table)
201    }
202
203    /// Finds a registered Delta source by alias using unquoted SQL semantics.
204    #[must_use]
205    pub fn registered_source(&self, name: &str) -> Option<&RegisteredSessionSource> {
206        self.sources
207            .iter()
208            .find(|source| source.name().eq_ignore_ascii_case(name))
209    }
210
211    /// Returns registered SQL-derived aliases in registration order.
212    #[must_use]
213    pub fn derived_tables(&self) -> &[RegisteredDerivedTable] {
214        &self.derived_tables
215    }
216
217    /// Finds a registered SQL-derived alias by name using unquoted SQL semantics.
218    #[must_use]
219    pub fn registered_derived_table(&self, name: &str) -> Option<&RegisteredDerivedTable> {
220        self.derived_tables
221            .iter()
222            .find(|table| table.name().eq_ignore_ascii_case(name))
223    }
224
225    pub(in crate::orchestrator::session) fn registered_derived_table_by_id(
226        &self,
227        table_id: u64,
228    ) -> Option<&RegisteredDerivedTable> {
229        self.derived_tables
230            .iter()
231            .find(|table| table.table().id() == table_id)
232    }
233
234    pub(in crate::orchestrator::session) fn phase_timings_for_lazy_table(
235        &self,
236        table: &LazyTable,
237    ) -> Result<Vec<PhaseTimingReport>, DeltaFunnelError> {
238        if table.kind() == LazyTableKind::DeltaSource {
239            return Ok(Vec::new());
240        }
241
242        self.derived_tables
243            .iter()
244            .find(|derived| derived.table().id() == table.id())
245            .map(|derived| derived.phase_timings().to_vec())
246            .or_else(|| {
247                self.pending_derived_tables
248                    .iter()
249                    .find(|pending| pending.table.id() == table.id())
250                    .map(|pending| pending.phase_timings.clone())
251            })
252            .ok_or_else(|| unknown_lazy_table_error(table))
253    }
254
255    /// Resolves the session metadata for an alias that is eligible for scoped caching.
256    ///
257    /// The cache primitive only supports registered SQL-derived aliases. Raw
258    /// sources, pending derived tables, and foreign or stale table handles are
259    /// rejected before any DataFusion catalog mutation can happen.
260    pub(in crate::orchestrator::session) fn registered_derived_for_scoped_cache_alias(
261        &self,
262        table: &LazyTable,
263    ) -> Result<&RegisteredDerivedTable, DeltaFunnelError> {
264        if table.kind() != LazyTableKind::DerivedSql {
265            return Err(unknown_lazy_table_error(table));
266        }
267
268        self.registered_derived_table_by_id(table.id())
269            .ok_or_else(|| unknown_lazy_table_error(table))
270    }
271
272    #[allow(dead_code)]
273    pub(crate) fn sql_text_for_derived_table(
274        &self,
275        table: &LazyTable,
276    ) -> Result<&str, DeltaFunnelError> {
277        if table.kind() != LazyTableKind::DerivedSql {
278            return Err(unknown_lazy_table_error(table));
279        }
280
281        self.derived_tables
282            .iter()
283            .find(|derived| derived.table().id() == table.id())
284            .map(RegisteredDerivedTable::sql_text)
285            .or_else(|| {
286                self.pending_derived_tables
287                    .iter()
288                    .find(|pending| pending.table.id() == table.id())
289                    .map(|pending| pending.sql_text.as_str())
290            })
291            .ok_or_else(|| unknown_lazy_table_error(table))
292    }
293
294    pub(super) async fn plan_read_only_sql(
295        &self,
296        sql: &str,
297    ) -> Result<DataFrame, DeltaFunnelError> {
298        self.context
299            .sql_with_options(sql, read_only_sql_options())
300            .await
301            .map_err(|error| DeltaFunnelError::SqlTable {
302                phase: classify_sql_error_phase(&error.to_string()),
303                message: error.to_string(),
304            })
305    }
306
307    fn allocate_derived_sql_table(&mut self) -> LazyTable {
308        let id = self.next_table_id;
309        self.next_table_id = self.next_table_id.saturating_add(1);
310        LazyTable::derived_sql(id)
311    }
312
313    pub(super) fn reject_registered_alias_name(&self, name: &str) -> Result<(), DeltaFunnelError> {
314        if self.registered_source(name).is_some() || self.registered_derived_table(name).is_some() {
315            return Err(DeltaFunnelError::DuplicateSourceName {
316                name: name.to_owned(),
317            });
318        }
319        Ok(())
320    }
321
322    fn find_pending_derived_table(&self, table: &LazyTable) -> Option<usize> {
323        if table.kind() != LazyTableKind::DerivedSql {
324            return None;
325        }
326
327        self.pending_derived_tables
328            .iter()
329            .position(|pending| pending.table.id() == table.id())
330    }
331}
332
333pub(in crate::orchestrator::session) fn read_only_sql_options() -> SQLOptions {
334    SQLOptions::new()
335        .with_allow_ddl(false)
336        .with_allow_dml(false)
337        .with_allow_statements(false)
338}
339
340fn classify_sql_error_phase(error: &str) -> SqlTablePhase {
341    if error.contains("DDL not supported")
342        || error.contains("DML not supported")
343        || error.contains("Statement not supported")
344        || error.contains("only supports a single SQL statement")
345    {
346        SqlTablePhase::ValidateSql
347    } else {
348        SqlTablePhase::PlanSql
349    }
350}
351
352#[cfg(test)]
353mod tests {
354    use datafusion::{
355        common::tree_node::{TreeNode, TreeNodeRecursion},
356        error::Result as DataFusionResult,
357        logical_expr::LogicalPlan,
358    };
359
360    use crate::{DeltaFunnelError, DeltaSourceConfig, SqlTablePhase};
361
362    use super::super::super::{
363        DeltaFunnelSession, LazyTableKind, SessionOptions, test_support::DeltaLogTable,
364    };
365    use super::LAZY_SQL_PLANNING_PHASE;
366
367    #[derive(Debug, Clone, PartialEq, Eq)]
368    struct TableScanProofReference {
369        table_name: String,
370        nested_table_names: Vec<String>,
371    }
372
373    fn table_scan_proof_references(
374        plan: &LogicalPlan,
375    ) -> DataFusionResult<Vec<TableScanProofReference>> {
376        let mut references = Vec::new();
377
378        plan.apply(|node| {
379            if let LogicalPlan::TableScan(scan) = node {
380                let nested_table_names = scan
381                    .source
382                    .get_logical_plan()
383                    .map(|nested| table_scan_table_names(nested.as_ref()))
384                    .transpose()?
385                    .unwrap_or_default();
386                references.push(TableScanProofReference {
387                    table_name: scan.table_name.table().to_owned(),
388                    nested_table_names,
389                });
390            }
391
392            Ok(TreeNodeRecursion::Continue)
393        })?;
394
395        Ok(references)
396    }
397
398    fn table_scan_table_names(plan: &LogicalPlan) -> DataFusionResult<Vec<String>> {
399        let mut names = Vec::new();
400
401        plan.apply(|node| {
402            if let LogicalPlan::TableScan(scan) = node {
403                names.push(scan.table_name.table().to_owned());
404            }
405
406            Ok(TreeNodeRecursion::Continue)
407        })?;
408
409        Ok(names)
410    }
411
412    #[tokio::test]
413    async fn table_from_sql_builds_lazy_derived_table_without_row_execution()
414    -> Result<(), Box<dyn std::error::Error>> {
415        let table = DeltaLogTable::new("orders")?;
416        let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
417        session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
418
419        let derived = session
420            .table_from_sql("select id, customer_name from orders")
421            .await?;
422
423        assert_eq!(derived.id(), 1);
424        assert_eq!(derived.kind(), LazyTableKind::DerivedSql);
425        assert_eq!(derived.name(), "table_1");
426        assert_eq!(session.next_table_id(), 2);
427        assert!(session.derived_tables().is_empty());
428        let phase_timings = session.phase_timings_for_lazy_table(&derived)?;
429        assert_eq!(
430            phase_timings
431                .iter()
432                .map(crate::PhaseTimingReport::phase_name)
433                .collect::<Vec<_>>(),
434            vec![LAZY_SQL_PLANNING_PHASE]
435        );
436        assert!(
437            phase_timings
438                .iter()
439                .all(|timing| timing.status().is_completed())
440        );
441        assert!(
442            phase_timings
443                .iter()
444                .all(|timing| timing.elapsed_micros().is_some())
445        );
446        Ok(())
447    }
448
449    #[tokio::test]
450    async fn table_from_sql_retains_trimmed_pending_sql_text()
451    -> Result<(), Box<dyn std::error::Error>> {
452        let table = DeltaLogTable::new("orders")?;
453        let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
454        session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
455
456        let derived = session
457            .table_from_sql(" \n\t select id from orders \t ")
458            .await?;
459
460        assert_eq!(
461            session.sql_text_for_derived_table(&derived)?,
462            "select id from orders"
463        );
464        assert!(session.derived_tables().is_empty());
465        Ok(())
466    }
467
468    #[tokio::test]
469    async fn registered_derived_alias_can_be_referenced_by_later_sql()
470    -> Result<(), Box<dyn std::error::Error>> {
471        let table = DeltaLogTable::new("orders")?;
472        let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
473        session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
474        let derived = session
475            .table_from_sql("select id, customer_name from orders")
476            .await?;
477
478        let alias = session.register_alias("recent_orders", &derived)?;
479        let second = session
480            .table_from_sql("select id from recent_orders")
481            .await?;
482
483        assert_eq!(alias.id(), derived.id());
484        assert_eq!(alias.name(), "recent_orders");
485        assert_eq!(
486            session.phase_timings_for_lazy_table(&alias)?,
487            session.phase_timings_for_lazy_table(&derived)?
488        );
489        assert_eq!(second.id(), 2);
490        assert_eq!(second.kind(), LazyTableKind::DerivedSql);
491        assert_eq!(session.derived_tables().len(), 1);
492        let registered = session
493            .registered_derived_table("RECENT_ORDERS")
494            .ok_or("registered derived alias missing")?;
495        assert_eq!(registered.table(), &alias);
496        assert_eq!(registered.schema().fields().len(), 2);
497        Ok(())
498    }
499
500    #[tokio::test]
501    async fn registered_derived_alias_retains_sql_text() -> Result<(), Box<dyn std::error::Error>> {
502        let table = DeltaLogTable::new("orders")?;
503        let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
504        session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
505        let derived = session
506            .table_from_sql("select id, customer_name from orders")
507            .await?;
508
509        let alias = session.register_alias("recent_orders", &derived)?;
510        let registered = session
511            .registered_derived_table("RECENT_ORDERS")
512            .ok_or("registered derived alias missing")?;
513
514        assert_eq!(
515            session.sql_text_for_derived_table(&alias)?,
516            "select id, customer_name from orders"
517        );
518        assert_eq!(
519            registered.sql_text(),
520            "select id, customer_name from orders"
521        );
522        assert_eq!(
523            session.sql_text_for_derived_table(&derived)?,
524            "select id, customer_name from orders"
525        );
526        Ok(())
527    }
528
529    #[tokio::test]
530    async fn empty_sql_fails_before_lazy_table_allocation() -> Result<(), DeltaFunnelError> {
531        let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
532
533        let error = session.table_from_sql(" \n\t ").await;
534
535        assert!(matches!(
536            error,
537            Err(DeltaFunnelError::SqlTable {
538                phase: SqlTablePhase::ValidateSql,
539                message,
540            }) if message.contains("must not be empty")
541        ));
542        assert_eq!(session.next_table_id(), 0);
543        assert!(session.derived_tables().is_empty());
544        Ok(())
545    }
546
547    #[tokio::test]
548    async fn multiple_sql_statements_fail_before_lazy_table_allocation()
549    -> Result<(), Box<dyn std::error::Error>> {
550        let table = DeltaLogTable::new("orders")?;
551        let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
552        session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
553
554        let error = session
555            .table_from_sql("select id from orders; select id from orders")
556            .await;
557
558        assert!(matches!(
559            error,
560            Err(DeltaFunnelError::SqlTable {
561                phase: SqlTablePhase::ValidateSql,
562                ..
563            })
564        ));
565        assert_eq!(session.next_table_id(), 1);
566        assert!(session.derived_tables().is_empty());
567        Ok(())
568    }
569
570    #[tokio::test]
571    async fn ddl_sql_fails_before_alias_registration() -> Result<(), Box<dyn std::error::Error>> {
572        let table = DeltaLogTable::new("orders")?;
573        let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
574        session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
575
576        let error = session
577            .table_from_sql("create table created_orders as select id from orders")
578            .await;
579
580        assert!(matches!(
581            error,
582            Err(DeltaFunnelError::SqlTable {
583                phase: SqlTablePhase::ValidateSql,
584                ..
585            })
586        ));
587        assert_eq!(session.next_table_id(), 1);
588        assert!(session.derived_tables().is_empty());
589        Ok(())
590    }
591
592    #[tokio::test]
593    async fn dml_sql_fails_before_alias_registration() -> Result<(), Box<dyn std::error::Error>> {
594        let table = DeltaLogTable::new("orders")?;
595        let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
596        session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
597
598        let error = session
599            .table_from_sql("insert into orders select id, customer_name from orders")
600            .await;
601
602        assert!(matches!(
603            error,
604            Err(DeltaFunnelError::SqlTable {
605                phase: SqlTablePhase::ValidateSql,
606                ..
607            })
608        ));
609        assert_eq!(session.next_table_id(), 1);
610        assert!(session.derived_tables().is_empty());
611        Ok(())
612    }
613
614    #[tokio::test]
615    async fn missing_table_sql_fails_with_planning_context() -> Result<(), DeltaFunnelError> {
616        let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
617
618        let error = session
619            .table_from_sql("select id from missing_orders")
620            .await;
621
622        assert!(matches!(
623            error,
624            Err(DeltaFunnelError::SqlTable {
625                phase: SqlTablePhase::PlanSql,
626                message,
627            }) if message.contains("missing_orders")
628        ));
629        assert_eq!(session.next_table_id(), 0);
630        assert!(session.derived_tables().is_empty());
631        Ok(())
632    }
633
634    #[tokio::test]
635    async fn derived_alias_duplicate_with_source_fails_before_registration()
636    -> Result<(), Box<dyn std::error::Error>> {
637        let table = DeltaLogTable::new("orders")?;
638        let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
639        session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
640        let derived = session.table_from_sql("select id from orders").await?;
641
642        let error = session.register_alias("ORDERS", &derived);
643
644        assert!(matches!(
645            error,
646            Err(DeltaFunnelError::DuplicateSourceName { name }) if name == "ORDERS"
647        ));
648        assert!(session.derived_tables().is_empty());
649        assert!(session.context().table("ORDERS").await.is_ok());
650        let alias = session.register_alias("recent_orders", &derived)?;
651        assert_eq!(alias.name(), "recent_orders");
652        assert_eq!(session.derived_tables().len(), 1);
653        Ok(())
654    }
655
656    #[tokio::test]
657    async fn invalid_derived_alias_fails_without_consuming_pending_table()
658    -> Result<(), Box<dyn std::error::Error>> {
659        let table = DeltaLogTable::new("orders")?;
660        let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
661        session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
662        let derived = session.table_from_sql("select id from orders").await?;
663
664        let error = session.register_alias("select", &derived);
665
666        assert!(matches!(
667            error,
668            Err(DeltaFunnelError::InvalidSourceName { name, .. }) if name == "select"
669        ));
670        assert!(session.derived_tables().is_empty());
671        let alias = session.register_alias("recent_orders", &derived)?;
672        assert_eq!(alias.name(), "recent_orders");
673        assert_eq!(session.derived_tables().len(), 1);
674        Ok(())
675    }
676
677    #[tokio::test]
678    async fn invalid_derived_alias_preserves_pending_sql_text()
679    -> Result<(), Box<dyn std::error::Error>> {
680        let table = DeltaLogTable::new("orders")?;
681        let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
682        session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
683        let derived = session.table_from_sql("select id from orders").await?;
684
685        let error = session.register_alias("select", &derived);
686
687        assert!(matches!(
688            error,
689            Err(DeltaFunnelError::InvalidSourceName { name, .. }) if name == "select"
690        ));
691        assert_eq!(
692            session.sql_text_for_derived_table(&derived)?,
693            "select id from orders"
694        );
695        let alias = session.register_alias("recent_orders", &derived)?;
696        assert_eq!(
697            session.sql_text_for_derived_table(&alias)?,
698            "select id from orders"
699        );
700        Ok(())
701    }
702
703    #[tokio::test]
704    async fn register_alias_rejects_non_pending_table_handle()
705    -> Result<(), Box<dyn std::error::Error>> {
706        let table = DeltaLogTable::new("orders")?;
707        let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
708        let source = session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
709
710        let error = session.register_alias("recent_orders", &source);
711
712        assert!(matches!(
713            error,
714            Err(DeltaFunnelError::SqlTable {
715                phase: SqlTablePhase::RegisterDerivedAlias,
716                message,
717            }) if message.contains("pending SQL-derived table")
718        ));
719        assert!(session.derived_tables().is_empty());
720        assert!(session.context().table("recent_orders").await.is_err());
721        Ok(())
722    }
723
724    #[tokio::test]
725    async fn derived_alias_duplicate_with_derived_alias_fails_before_registration()
726    -> Result<(), Box<dyn std::error::Error>> {
727        let table = DeltaLogTable::new("orders")?;
728        let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
729        session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
730        let first = session.table_from_sql("select id from orders").await?;
731        session.register_alias("recent_orders", &first)?;
732        let second = session.table_from_sql("select id from orders").await?;
733
734        let error = session.register_alias("RECENT_ORDERS", &second);
735
736        assert!(matches!(
737            error,
738            Err(DeltaFunnelError::DuplicateSourceName { name }) if name == "RECENT_ORDERS"
739        ));
740        assert_eq!(session.derived_tables().len(), 1);
741        assert!(session.context().table("RECENT_ORDERS").await.is_ok());
742        Ok(())
743    }
744
745    #[tokio::test]
746    async fn registered_derived_debug_redacts_retained_sql_text()
747    -> Result<(), Box<dyn std::error::Error>> {
748        let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
749        let derived = session
750            .table_from_sql("select 'super-secret-literal' as marker")
751            .await?;
752        session.register_alias("secret_marker", &derived)?;
753        let registered = session
754            .registered_derived_table("secret_marker")
755            .ok_or("registered derived alias missing")?;
756
757        let debug = format!("{registered:?}");
758
759        assert!(debug.contains("sql_text"));
760        assert!(debug.contains("<redacted>"));
761        assert!(!debug.contains("super-secret-literal"));
762        assert!(!debug.contains("select '"));
763        Ok(())
764    }
765
766    #[tokio::test]
767    async fn source_alias_duplicate_with_derived_alias_fails_before_loading_source()
768    -> Result<(), Box<dyn std::error::Error>> {
769        let table = DeltaLogTable::new("orders")?;
770        let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
771        session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
772        let derived = session.table_from_sql("select id from orders").await?;
773        session.register_alias("recent_orders", &derived)?;
774
775        let error = session.delta_lake(DeltaSourceConfig::new("RECENT_ORDERS", ""));
776
777        assert!(matches!(
778            error,
779            Err(DeltaFunnelError::DuplicateSourceName { name }) if name == "RECENT_ORDERS"
780        ));
781        assert_eq!(session.sources().len(), 1);
782        assert_eq!(session.derived_tables().len(), 1);
783        Ok(())
784    }
785
786    #[tokio::test]
787    async fn planned_downstream_sql_expands_registered_derived_alias_reference()
788    -> Result<(), Box<dyn std::error::Error>> {
789        const MARKER_REGION_SCHEMA_FIELDS_JSON: &str = r#"[{\"name\":\"marker\",\"type\":\"string\",\"nullable\":false,\"metadata\":{}},{\"name\":\"region\",\"type\":\"string\",\"nullable\":false,\"metadata\":{}}]"#;
790
791        let table = DeltaLogTable::new_with_schema("orders", MARKER_REGION_SCHEMA_FIELDS_JSON)?;
792        let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
793        session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
794
795        let source_dataframe = session
796            .plan_read_only_sql("select marker from orders")
797            .await?;
798        let source_references = table_scan_proof_references(source_dataframe.logical_plan())?;
799        assert_eq!(
800            source_references,
801            vec![TableScanProofReference {
802                table_name: "orders".to_owned(),
803                nested_table_names: Vec::new(),
804            }]
805        );
806        assert!(session.registered_source("orders").is_some());
807        assert!(session.registered_derived_table("orders").is_none());
808
809        let pending_big = session
810            .table_from_sql("select marker, region from orders")
811            .await?;
812        let _big = session.register_alias("big", &pending_big)?;
813        let west_dataframe = session
814            .plan_read_only_sql("select marker from BIG where region = 'west'")
815            .await?;
816        let east_dataframe = session
817            .plan_read_only_sql("select marker from big where region = 'east'")
818            .await?;
819        let west_references = table_scan_proof_references(west_dataframe.logical_plan())?;
820        let east_references = table_scan_proof_references(east_dataframe.logical_plan())?;
821
822        for references in [&west_references, &east_references] {
823            assert_eq!(
824                references,
825                &vec![TableScanProofReference {
826                    table_name: "orders".to_owned(),
827                    nested_table_names: Vec::new(),
828                }]
829            );
830            assert!(
831                session
832                    .registered_source(&references[0].table_name)
833                    .is_some()
834            );
835            assert!(
836                session
837                    .registered_derived_table(&references[0].table_name)
838                    .is_none()
839            );
840        }
841
842        // Conclusion for #257: DataFusion expands the registered derived
843        // alias during SQL planning, so planned LogicalPlan table scans do not
844        // preserve a structured west/east -> big dependency for #250.
845        assert!(
846            !west_references
847                .iter()
848                .any(|reference| reference.table_name.eq_ignore_ascii_case("big"))
849        );
850        assert!(
851            !east_references
852                .iter()
853                .any(|reference| reference.table_name.eq_ignore_ascii_case("big"))
854        );
855        assert!(session.registered_derived_table("big").is_some());
856        assert!(session.registered_source("big").is_none());
857        Ok(())
858    }
859}
delta_funnel/orchestrator/session/registry/derived.rs

delta_funnel/orchestrator/session/registry/
derived.rs
