delta-funnel 0.6.0

Lightweight, fast Delta Lake to SQL Server loads with DataFusion SQL and native TDS
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108

//! Compile-time coverage for the public execution-profile API.

use std::sync::Arc;

use datafusion::{
    common::DataFusionError,
    execution::TaskContext,
    physical_plan::{ExecutionPlan, SendableRecordBatchStream},
};
use delta_funnel::{
    DeltaDataFusionMetricsSnapshot, DeltaFunnelError, DeltaFunnelRuntime, DeltaFunnelSession,
    ExecutionProfileMode, MssqlWriteReport, OutputWritePlan, QueryExecutionMetric,
    QueryExecutionMetricCategory, QueryExecutionMetricValue, QueryExecutionOperatorProfile,
    QueryExecutionOutcome, QueryExecutionProfile, QueryExecutionScope, WriteAllCacheAliasReport,
    datafusion_query_output_stream, progress::ProgressReporter,
};
use serde_json::Value;

type QueryOutputStreamFn = fn(
    Arc<dyn ExecutionPlan>,
    Arc<TaskContext>,
) -> Result<SendableRecordBatchStream, DataFusionError>;

type RuntimeMssqlProfileFn = fn(
    &DeltaFunnelRuntime,
    &DeltaFunnelSession,
    &OutputWritePlan,
    ExecutionProfileMode,
) -> Result<MssqlWriteReport, DeltaFunnelError>;

type RuntimeMssqlProfileProgressFn = fn(
    &DeltaFunnelRuntime,
    &DeltaFunnelSession,
    &OutputWritePlan,
    ExecutionProfileMode,
    ProgressReporter,
) -> Result<MssqlWriteReport, DeltaFunnelError>;

#[allow(dead_code)]
async fn session_mssql_profile_route_compiles(
    session: &DeltaFunnelSession,
    request: &OutputWritePlan,
) -> Result<MssqlWriteReport, DeltaFunnelError> {
    session
        .write_to_mssql_with_profile_mode(request, ExecutionProfileMode::Detailed)
        .await
}

#[test]
fn execution_profile_types_and_accessors_are_exported_from_the_crate_root() {
    let _: QueryOutputStreamFn = datafusion_query_output_stream;
    let _: RuntimeMssqlProfileFn = DeltaFunnelRuntime::write_to_mssql_with_profile_mode;
    let _: RuntimeMssqlProfileProgressFn =
        DeltaFunnelRuntime::write_to_mssql_with_profile_mode_and_progress;

    let _: ExecutionProfileMode = ExecutionProfileMode::default();
    let _: for<'a> fn(&'a MssqlWriteReport) -> Option<&'a QueryExecutionProfile> =
        MssqlWriteReport::execution_profile;
    let _: for<'a> fn(&'a WriteAllCacheAliasReport) -> Option<&'a QueryExecutionProfile> =
        WriteAllCacheAliasReport::execution_profile;
    let _: fn(&QueryExecutionProfile) -> QueryExecutionScope = QueryExecutionProfile::scope;
    let _: fn(&QueryExecutionProfile) -> QueryExecutionOutcome = QueryExecutionProfile::outcome;
    let _: fn(&QueryExecutionProfile) -> bool = QueryExecutionProfile::partial;
    let _: fn(&QueryExecutionProfile) -> Option<u64> =
        QueryExecutionProfile::delta_funnel_row_limit;
    let _: for<'a> fn(&'a QueryExecutionProfile) -> &'a [QueryExecutionOperatorProfile] =
        QueryExecutionProfile::operators;
    let _: fn(&QueryExecutionProfile) -> Value = QueryExecutionProfile::to_json_value;

    let _: fn(&QueryExecutionOperatorProfile) -> u64 = QueryExecutionOperatorProfile::node_id;
    let _: fn(&QueryExecutionOperatorProfile) -> Option<u64> =
        QueryExecutionOperatorProfile::parent_node_id;
    let _: for<'a> fn(&'a QueryExecutionOperatorProfile) -> &'a str =
        QueryExecutionOperatorProfile::operator_name;
    let _: fn(&QueryExecutionOperatorProfile) -> u64 =
        QueryExecutionOperatorProfile::output_partition_count;
    let _: fn(&QueryExecutionOperatorProfile) -> bool =
        QueryExecutionOperatorProfile::metrics_available;
    let _: for<'a> fn(&'a QueryExecutionOperatorProfile) -> &'a [QueryExecutionMetric] =
        QueryExecutionOperatorProfile::aggregated_metrics;
    let _: for<'a> fn(&'a QueryExecutionOperatorProfile) -> &'a [QueryExecutionMetric] =
        QueryExecutionOperatorProfile::metrics;
    let _: for<'a> fn(
        &'a QueryExecutionOperatorProfile,
    ) -> Option<&'a DeltaDataFusionMetricsSnapshot> =
        QueryExecutionOperatorProfile::delta_provider_read_stats;
    let _: for<'a> fn(&'a QueryExecutionOperatorProfile) -> Option<&'a str> =
        QueryExecutionOperatorProfile::delta_provider_source_name;
    let _: fn(&QueryExecutionOperatorProfile) -> Value =
        QueryExecutionOperatorProfile::to_json_value;

    let _: for<'a> fn(&'a QueryExecutionMetric) -> &'a str = QueryExecutionMetric::name;
    let _: fn(&QueryExecutionMetric) -> QueryExecutionMetricCategory =
        QueryExecutionMetric::category;
    let _: fn(&QueryExecutionMetric) -> Option<u64> = QueryExecutionMetric::partition;
    let _: fn(&QueryExecutionMetric) -> Option<u64> = QueryExecutionMetric::output_partition;
    let _: for<'a> fn(&'a QueryExecutionMetric) -> &'a QueryExecutionMetricValue =
        QueryExecutionMetric::value;
    let _: fn(&QueryExecutionMetric) -> Value = QueryExecutionMetric::to_json_value;
    let _: for<'a> fn(&'a QueryExecutionMetricValue) -> &'static str =
        QueryExecutionMetricValue::value_kind;

    assert_eq!(QueryExecutionScope::Preview.as_str(), "preview");
    assert_eq!(
        QueryExecutionScope::WriteAllCacheAlias.as_str(),
        "write_all_cache_alias"
    );
}