delta-arrow-reader 0.1.2

use std::{error::Error as _, future::Future};

use arrow::{datatypes::SchemaRef, record_batch::RecordBatch};
use delta_arrow_reader::{
    DeltaBatchStream, DeltaComparison, DeltaPredicate, DeltaProtocolInfo, DeltaReadMetrics,
    DeltaReadMetricsSnapshot, DeltaReaderBackend, DeltaReaderError, DeltaReaderExecutionOptions,
    DeltaReaderPhase, DeltaScalar, DeltaScan, DeltaScanBuilder,
    DeltaScanPartitionTargetDiagnosticInput, DeltaScanPartitionTargetDiagnosticOutput,
    DeltaScanPartitionTargetDiagnosticSource, DeltaScanPartitionTargetLocalEnvironmentDiagnostic,
    DeltaScanPartitionTargetLocalUnixFileDescriptorLimitStatus, DeltaSnapshotSelection,
    DeltaStorageOptions, DeltaTable, DeltaTableBuilder, DeltaTableSnapshot,
    delta_scan_partition_target_local_environment_diagnostic,
    derive_delta_scan_partition_target_diagnostic,
};
use futures_util::Stream;

#[test]
fn configuration_and_error_contract_is_public() -> Result<(), DeltaReaderError> {
    let snapshot: fn(&DeltaReadMetrics) -> DeltaReadMetricsSnapshot = DeltaReadMetrics::snapshot;
    let _ = snapshot;
    let snapshot_version: fn(&DeltaProtocolInfo) -> u64 = DeltaProtocolInfo::snapshot_version;
    let min_reader_version: fn(&DeltaProtocolInfo) -> i32 = DeltaProtocolInfo::min_reader_version;
    let min_writer_version: fn(&DeltaProtocolInfo) -> i32 = DeltaProtocolInfo::min_writer_version;
    let reader_features: for<'a> fn(&'a DeltaProtocolInfo) -> &'a [String] =
        DeltaProtocolInfo::reader_features;
    let writer_features: for<'a> fn(&'a DeltaProtocolInfo) -> &'a [String] =
        DeltaProtocolInfo::writer_features;
    let first_unsupported_reader_feature: for<'a> fn(&'a DeltaProtocolInfo) -> Option<&'a str> =
        DeltaProtocolInfo::first_unsupported_reader_feature;
    let _ = (
        snapshot_version,
        min_reader_version,
        min_writer_version,
        reader_features,
        writer_features,
        first_unsupported_reader_feature,
    );

    let mut storage_options = DeltaStorageOptions::new();
    storage_options.insert("region".into(), "example".into());
    assert_eq!(storage_options.len(), 1);

    assert_eq!(
        DeltaSnapshotSelection::Version(3),
        DeltaSnapshotSelection::Version(3)
    );

    let options = DeltaReaderExecutionOptions::new()
        .with_reader_backend(DeltaReaderBackend::OfficialKernel)?
        .with_max_concurrent_file_reads_per_scan(Some(6))?
        .with_max_concurrent_file_reads_per_partition(3)?
        .with_output_buffer_capacity_per_partition(1)?
        .with_native_async_prefetch_file_count_per_partition(2)?
        .with_parquet_metadata_size_hint(Some(65_536))?
        .with_parquet_full_file_read_threshold(None)?;

    assert_eq!(options.reader_backend(), DeltaReaderBackend::OfficialKernel);
    options.validate()?;

    let error = DeltaReaderExecutionOptions::new()
        .with_output_buffer_capacity_per_partition(0)
        .expect_err("zero output capacity must fail");
    assert_eq!(error.phase(), DeltaReaderPhase::Configuration);
    assert_eq!(error.as_str(), "invalid_configuration");
    assert!(error.source().is_none());

    Ok(())
}

#[test]
fn scan_partition_target_diagnostic_contract_is_public() -> Result<(), DeltaReaderError> {
    let _: DeltaScanPartitionTargetDiagnosticInput = Default::default();
    let local: DeltaScanPartitionTargetLocalEnvironmentDiagnostic =
        delta_scan_partition_target_local_environment_diagnostic();
    let _: DeltaScanPartitionTargetDiagnosticInput = local.policy_input;
    let _: Option<u64> = local.memory_total_bytes;
    let _: Option<u64> = local.memory_available_bytes;
    let _: Option<u64> = local.unix_soft_file_descriptor_limit;
    let _: DeltaScanPartitionTargetLocalUnixFileDescriptorLimitStatus =
        local.unix_soft_file_descriptor_limit_status;
    let _ = [
        DeltaScanPartitionTargetLocalUnixFileDescriptorLimitStatus::Unsupported,
        DeltaScanPartitionTargetLocalUnixFileDescriptorLimitStatus::Unknown,
        DeltaScanPartitionTargetLocalUnixFileDescriptorLimitStatus::Finite,
        DeltaScanPartitionTargetLocalUnixFileDescriptorLimitStatus::Unlimited,
    ];
    let input = DeltaScanPartitionTargetDiagnosticInput {
        explicit_target_partitions: None,
        datafusion_target_partitions: Some(8),
        available_parallelism: Some(4),
        available_memory_bytes: None,
        unix_soft_file_descriptor_limit: None,
        min_default_partitions: 1,
        parallelism_multiplier: 1,
        file_descriptors_per_partition: 16,
        available_memory_bytes_per_partition: 256 * 1024 * 1024,
    };
    let output: DeltaScanPartitionTargetDiagnosticOutput =
        derive_delta_scan_partition_target_diagnostic(input)?;

    assert_eq!(output.target_partitions, 4);
    assert_eq!(
        output.source,
        DeltaScanPartitionTargetDiagnosticSource::AvailableParallelismFallback
    );
    let _ = [
        DeltaScanPartitionTargetDiagnosticSource::ExplicitOverride,
        DeltaScanPartitionTargetDiagnosticSource::AvailableParallelismFallback,
        DeltaScanPartitionTargetDiagnosticSource::StaticFallback,
    ];
    assert_eq!(output.explicit_target_partitions, None);
    assert_eq!(output.datafusion_target_partitions, Some(8));
    assert_eq!(output.available_parallelism, Some(4));
    assert_eq!(output.datafusion_target_cap, Some(8));
    assert_eq!(output.unix_file_descriptor_cap, None);
    assert_eq!(output.memory_cap, None);
    Ok(())
}

#[test]
fn exact_predicate_model_is_public() {
    let comparisons = [
        DeltaComparison::Eq,
        DeltaComparison::NotEq,
        DeltaComparison::Lt,
        DeltaComparison::LtEq,
        DeltaComparison::Gt,
        DeltaComparison::GtEq,
    ];
    let copied_comparisons = comparisons;
    assert_eq!(copied_comparisons, comparisons);

    let scalars = vec![
        DeltaScalar::Boolean(true),
        DeltaScalar::Int8(1),
        DeltaScalar::Int16(2),
        DeltaScalar::Int32(3),
        DeltaScalar::Int64(4),
        DeltaScalar::Float32(5.0),
        DeltaScalar::Float64(6.0),
        DeltaScalar::Date32(7),
        DeltaScalar::Decimal128 {
            value: 8,
            precision: 9,
            scale: 1,
        },
        DeltaScalar::Utf8("utf8".into()),
        DeltaScalar::LargeUtf8("large utf8".into()),
        DeltaScalar::Binary(vec![10]),
        DeltaScalar::LargeBinary(vec![11]),
        DeltaScalar::FixedSizeBinary {
            size: 2,
            value: vec![12, 13],
        },
        DeltaScalar::TimestampMicrosecond {
            value: 14,
            timezone: Some("UTC".into()),
        },
    ];
    assert_eq!(scalars, scalars.clone());

    let predicates = vec![
        DeltaPredicate::Boolean(true),
        DeltaPredicate::Compare {
            column: "id".into(),
            op: DeltaComparison::Eq,
            value: DeltaScalar::Int64(1),
        },
        DeltaPredicate::IsNull {
            column: "optional".into(),
        },
        DeltaPredicate::IsNotNull {
            column: "required".into(),
        },
        DeltaPredicate::And(Vec::new()),
        DeltaPredicate::Or(Vec::new()),
        DeltaPredicate::Not(Box::new(DeltaPredicate::Boolean(false))),
    ];
    assert_eq!(predicates, predicates.clone());
    assert!(format!("{predicates:?}").contains("Compare"));
}

#[test]
fn direct_reader_contract_is_public() {
    fn assert_send<T: Send>() {}
    fn assert_send_sync<T: Send + Sync>() {}
    fn assert_clone<T: Clone>() {}
    fn assert_batch_stream<T: Stream<Item = Result<RecordBatch, DeltaReaderError>>>() {}
    fn assert_future<T>(_: impl Future<Output = T>) {}
    const fn table_version(table: &DeltaTable) -> u64 {
        table.version()
    }
    const fn scan_partition_count(scan: &DeltaScan) -> usize {
        scan.partition_count()
    }

    assert_send_sync::<DeltaTable>();
    assert_clone::<DeltaTable>();
    assert_send::<DeltaBatchStream>();
    assert_batch_stream::<DeltaBatchStream>();

    let builder = DeltaTableBuilder::new("file:///tmp/table")
        .with_storage_options(DeltaStorageOptions::new())
        .with_snapshot_selection(DeltaSnapshotSelection::Version(1))
        .with_execution_options(DeltaReaderExecutionOptions::new());
    assert_future::<Result<DeltaTable, DeltaReaderError>>(builder.load_async());
    let load: fn(DeltaTableBuilder) -> Result<DeltaTable, DeltaReaderError> =
        DeltaTableBuilder::load;
    let load_snapshot: fn(DeltaTableBuilder) -> Result<DeltaTableSnapshot, DeltaReaderError> =
        DeltaTableBuilder::load_snapshot;
    let snapshot_builder = DeltaTableBuilder::new("file:///tmp/table");
    assert_future::<Result<DeltaTableSnapshot, DeltaReaderError>>(
        snapshot_builder.load_snapshot_async(),
    );
    let _ = (load, load_snapshot);

    let snapshot_version: fn(&DeltaTableSnapshot) -> u64 = DeltaTableSnapshot::version;
    let snapshot_protocol: for<'a> fn(&'a DeltaTableSnapshot) -> &'a DeltaProtocolInfo =
        DeltaTableSnapshot::protocol;
    let snapshot_table_uri: for<'a> fn(&'a DeltaTableSnapshot) -> &'a str =
        DeltaTableSnapshot::table_uri;
    let validate_snapshot_protocol: fn(&DeltaTableSnapshot) -> Result<(), DeltaReaderError> =
        DeltaTableSnapshot::validate_protocol;
    let into_table: fn(DeltaTableSnapshot) -> Result<DeltaTable, DeltaReaderError> =
        DeltaTableSnapshot::into_table;
    let _ = (
        snapshot_version,
        snapshot_protocol,
        snapshot_table_uri,
        validate_snapshot_protocol,
        into_table,
    );

    let version: fn(&DeltaTable) -> u64 = DeltaTable::version;
    let schema: for<'a> fn(&'a DeltaTable) -> &'a SchemaRef = DeltaTable::schema;
    let protocol: for<'a> fn(&'a DeltaTable) -> &'a DeltaProtocolInfo = DeltaTable::protocol;
    let table_uri: for<'a> fn(&'a DeltaTable) -> &'a str = DeltaTable::table_uri;
    let validate_protocol: fn(&DeltaTable) -> Result<(), DeltaReaderError> =
        DeltaTable::validate_protocol;
    let scan: for<'a> fn(&'a DeltaTable) -> DeltaScanBuilder<'a> = DeltaTable::scan;
    let _ = (
        version,
        table_version,
        schema,
        protocol,
        table_uri,
        validate_protocol,
        scan,
    );

    fn configure_scan<'a>(
        builder: DeltaScanBuilder<'a>,
        predicate: DeltaPredicate,
        options: DeltaReaderExecutionOptions,
    ) -> Result<DeltaScanBuilder<'a>, DeltaReaderError> {
        builder
            .with_projection(vec!["id".into()])
            .with_predicate(predicate)
            .with_limit(1)
            .with_target_partitions(1)?
            .with_execution_options(options)
    }
    fn assert_scan_futures(builder: DeltaScanBuilder<'_>, scan: DeltaScan) {
        assert_future::<Result<DeltaScan, DeltaReaderError>>(builder.build());
        assert_future::<Result<DeltaBatchStream, DeltaReaderError>>(scan.execute());
    }
    let _ = configure_scan;
    let _ = assert_scan_futures;

    let scan_schema: for<'a> fn(&'a DeltaScan) -> &'a SchemaRef = DeltaScan::schema;
    let partition_count: fn(&DeltaScan) -> usize = DeltaScan::partition_count;
    let _ = (scan_schema, partition_count, scan_partition_count);

    let stream_schema: for<'a> fn(&'a DeltaBatchStream) -> &'a SchemaRef = DeltaBatchStream::schema;
    let metrics: fn(&DeltaBatchStream) -> DeltaReadMetrics = DeltaBatchStream::metrics;
    let _ = (stream_schema, metrics);
}

#[cfg(feature = "datafusion")]
#[test]
fn datafusion_metrics_contract_is_public() {
    use delta_arrow_reader::{
        DeltaDataFusionMetrics, DeltaDataFusionMetricsSnapshot, collect_delta_datafusion_metrics,
    };

    fn assert_clone<T: Clone>() {}
    fn assert_snapshot_traits<T: std::fmt::Debug + Clone + PartialEq + Eq>() {}
    fn inspect(snapshot: DeltaDataFusionMetricsSnapshot) {
        let _: DeltaReadMetricsSnapshot = snapshot.reader;
        let _: Option<u64> = snapshot.output_batch_size;
        let _: u64 = snapshot.dynamic_partition_files_pruned;
        let _: u64 = snapshot.dynamic_partition_files_kept;
        let _: u64 = snapshot.dynamic_filters_received;
        let _: u64 = snapshot.dynamic_filters_accepted;
        let _: u64 = snapshot.dynamic_filters_unsupported;
        let _: u64 = snapshot.dynamic_filter_snapshots;
        let _: u64 = snapshot.dynamic_files_not_pruned_missing_metadata;
        let _: u64 = snapshot.dynamic_files_not_pruned_unsupported_expression;
    }

    assert_clone::<DeltaDataFusionMetrics>();
    assert_snapshot_traits::<DeltaDataFusionMetricsSnapshot>();
    let source_name: for<'a> fn(&'a DeltaDataFusionMetrics) -> Option<&'a str> =
        DeltaDataFusionMetrics::source_name;
    let snapshot: fn(&DeltaDataFusionMetrics) -> DeltaDataFusionMetricsSnapshot =
        DeltaDataFusionMetrics::snapshot;
    let same_instance: fn(&DeltaDataFusionMetrics, &DeltaDataFusionMetrics) -> bool =
        DeltaDataFusionMetrics::same_instance;
    let collect: fn(&dyn datafusion::physical_plan::ExecutionPlan) -> Vec<DeltaDataFusionMetrics> =
        collect_delta_datafusion_metrics;
    let _ = (source_name, snapshot, same_instance, collect, inspect);
}

#[cfg(feature = "datafusion")]
#[test]
fn datafusion_provider_contract_is_public() {
    use delta_arrow_reader::{
        DeltaDataFusionScanOptions, DeltaReaderError, DeltaTable, DeltaTableProvider,
        RegisteredDeltaTable, register_delta_table,
    };

    fn assert_clone<T: Clone>() {}
    fn assert_debug_clone<T: std::fmt::Debug + Clone>() {}
    fn assert_result_traits<T: std::fmt::Debug + Clone + PartialEq + Eq>() {}

    assert_debug_clone::<DeltaDataFusionScanOptions>();
    assert_clone::<DeltaTableProvider>();
    assert_result_traits::<RegisteredDeltaTable>();
    let construct: fn(
        DeltaTable,
        DeltaDataFusionScanOptions,
    ) -> Result<DeltaTableProvider, DeltaReaderError> = DeltaTableProvider::try_new;
    fn register(
        context: &datafusion::execution::context::SessionContext,
        name: String,
        table: DeltaTable,
        options: DeltaDataFusionScanOptions,
    ) -> Result<RegisteredDeltaTable, DeltaReaderError> {
        register_delta_table(context, name, table, options)
    }
    let _ = (construct, register);
}

#[cfg(not(feature = "native-async"))]
#[test]
fn disabled_native_backend_fails_before_uri_access() {
    let error = DeltaTableBuilder::new("this URI must never be inspected")
        .load()
        .expect_err("disabled default backend must fail");
    assert_eq!(error.phase(), DeltaReaderPhase::Configuration);
    assert_eq!(error.as_str(), "unsupported_backend");
}

#[cfg(not(feature = "official-kernel"))]
#[test]
fn disabled_official_backend_fails_before_uri_access() -> Result<(), DeltaReaderError> {
    let options = DeltaReaderExecutionOptions::new()
        .with_reader_backend(DeltaReaderBackend::OfficialKernel)?;
    let error = DeltaTableBuilder::new("this URI must never be inspected")
        .with_execution_options(options)
        .load()
        .expect_err("disabled official backend must fail");
    assert_eq!(error.phase(), DeltaReaderPhase::Configuration);
    assert_eq!(error.as_str(), "unsupported_backend");
    Ok(())
}