use cudf::testing;
use query_planning::source::CredentialMode;
use query_runtime::runtime::{S3SourceStorageProfile, SourceStorageProfile};
use super::*;
use crate::{
native::{ParquetReadStrategyConfig, S3_STORAGE_PROFILE_ID},
native_exec::AlgeonNativeExecutionConfig,
};
fn pushed_down_parquet_predicate_with_filter_is_reader_pushdown() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-pushed-predicate");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_with_pushed_down_predicate(&path);
let plan = aggregate_sort_limit_plan(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1);
assert_eq!(result.not_selected_count(), 0);
assert!(matches!(result.report.outcomes()[0], CandidateOutcome::Selected { .. }));
assert_eq!(result.source_diagnostics.len(), 1);
assert!(result.source_diagnostics[0].direct_device_read);
assert_eq!(result.source_diagnostics[0].predicate_disposition.as_str(), "reader_pushdown");
assert!(result.source_diagnostics[0].rejected_features.is_empty());
let source = native_local_parquet_source(result.selected_plan());
assert!(source.reader_predicate.is_some());
assert!(native_filter_exprs(result.selected_plan()).is_empty());
fs::remove_dir_all(dir).unwrap();
}
#[test]
fn reader_pushdown_covered_filter_keeps_frontend_projection_order_after_reordered_source_read() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-source-order");
let path = dir.join("input.parquet");
write_snap_edge_parquet_file(&path);
let scan = parquet_edge_scan_with_projection_and_pushed_src_predicate(&path);
let scan_schema = scan.schema();
let predicate =
binary(col("src", &scan_schema).unwrap(), Operator::GtEq, lit(ScalarValue::Int64(Some(10))), &scan_schema)
.unwrap();
let plan: Arc<dyn ExecutionPlan> = Arc::new(FilterExec::try_new(predicate, scan).unwrap());
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.source_diagnostics.len(), 1);
assert_eq!(result.source_diagnostics[0].predicate_disposition.as_str(), "reader_pushdown");
let source = native_local_parquet_source(result.selected_plan());
assert_eq!(source.read_columns, vec!["dst".to_string(), "src".to_string(), "weight".to_string()]);
assert_eq!(source.projected_output_columns, vec!["src".to_string(), "dst".to_string(), "weight".to_string()]);
assert!(source.reader_predicate.is_some());
assert!(native_filter_exprs(result.selected_plan()).is_empty());
assert_eq!(
result
.selected_plan()
.root_node()
.output_schema()
.fields()
.iter()
.map(|field| field.name().as_str())
.collect::<Vec<_>>(),
vec!["src", "dst", "weight"],
"reader-pushdown-covered source must emit the frontend scan order"
);
fs::remove_dir_all(dir).unwrap();
}
fn pushed_down_parquet_predicate_subset_of_filter_is_reader_pushdown_with_native_verification() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-pushed-predicate-subset");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_with_pushed_down_predicate(&path);
let scan_schema = scan.schema();
let pushed =
binary(col("payload", &scan_schema).unwrap(), Operator::GtEq, lit(ScalarValue::Int32(Some(10))), &scan_schema)
.unwrap();
let retained =
binary(col("key", &scan_schema).unwrap(), Operator::NotEq, lit(ScalarValue::Int32(Some(2))), &scan_schema)
.unwrap();
let filter_predicate = binary(pushed, Operator::And, retained, &scan_schema).unwrap();
let plan = aggregate_sort_limit_plan_with_filter(scan, filter_predicate, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1);
assert_eq!(result.not_selected_count(), 0);
assert!(matches!(result.report.outcomes()[0], CandidateOutcome::Selected { .. }));
assert_eq!(result.source_diagnostics.len(), 1);
assert!(result.source_diagnostics[0].direct_device_read);
assert_eq!(result.source_diagnostics[0].predicate_disposition.as_str(), "reader_pushdown");
assert!(result.source_diagnostics[0].rejected_features.is_empty());
let source = native_local_parquet_source(result.selected_plan());
assert!(source.reader_predicate.is_some());
assert_eq!(native_filter_exprs(result.selected_plan()).len(), 1);
fs::remove_dir_all(dir).unwrap();
}
fn pushed_down_parquet_predicate_uses_filter_input_schema_for_validation() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-pushed-predicate-filter-schema");
let path = dir.join("input.parquet");
write_parquet_file_with_tpch_like_columns(&path);
let scan = parquet_scan_with_projection_and_pushed_label_predicate(&path);
let scan_schema = scan.schema();
let predicate = binary(
col("label", &scan_schema).unwrap(),
Operator::NotEq,
lit(ScalarValue::Utf8View(Some("z".to_string()))),
&scan_schema,
)
.unwrap();
let plan = aggregate_sort_limit_plan_with_filter(scan, predicate, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.source_diagnostics.len(), 1);
assert!(result.source_diagnostics[0].direct_device_read);
assert_eq!(result.source_diagnostics[0].predicate_disposition.as_str(), "reader_pushdown");
assert!(result.source_diagnostics[0].rejected_features.is_empty());
let source = native_local_parquet_source(result.selected_plan());
assert!(source.reader_predicate.is_some());
assert!(native_filter_exprs(result.selected_plan()).is_empty());
fs::remove_dir_all(dir).unwrap();
}
fn pushed_down_substring_prefix_predicate_is_rewritten_to_reader_pushdown_ranges() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-pushed-prefix-predicate");
let path = dir.join("input.parquet");
write_parquet_file_with_tpch_like_columns(&path);
let scan = parquet_scan_with_pushed_down_label_prefix_predicate(&path);
let plan = aggregate_sort_limit_plan_without_filter(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.source_diagnostics.len(), 1);
assert_eq!(result.source_diagnostics[0].predicate_disposition.as_str(), "reader_pushdown");
assert!(result.source_diagnostics[0].rejected_features.is_empty());
let source = native_local_parquet_source(result.selected_plan());
let reader_predicate = source.reader_predicate.as_ref().expect("prefix predicate should be pushed to the reader");
let reader_predicate_text = reader_predicate.to_string();
assert!(!reader_predicate_text.contains("SUBSTRING"), "{reader_predicate_text}");
assert!(reader_predicate_text.contains("label >= \"a\""), "{reader_predicate_text}");
assert!(reader_predicate_text.contains("label < \"b\""), "{reader_predicate_text}");
assert!(reader_predicate_text.contains("label >= \"b\""), "{reader_predicate_text}");
assert!(reader_predicate_text.contains("label < \"c\""), "{reader_predicate_text}");
assert!(native_filter_exprs(result.selected_plan()).is_empty());
fs::remove_dir_all(dir).unwrap();
}
fn pushed_down_parquet_predicate_without_filter_is_reader_pushdown() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-pushed-predicate-without-filter");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_with_pushed_down_predicate(&path);
let plan = aggregate_sort_limit_plan_without_filter(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1);
assert_eq!(result.not_selected_count(), 0);
assert!(matches!(result.report.outcomes()[0], CandidateOutcome::Selected { .. }));
assert_eq!(result.source_diagnostics.len(), 1);
assert!(result.source_diagnostics[0].direct_device_read);
assert_eq!(result.source_diagnostics[0].predicate_disposition.as_str(), "reader_pushdown");
assert!(result.source_diagnostics[0].rejected_features.is_empty());
let source = native_local_parquet_source(result.selected_plan());
assert!(source.reader_predicate.is_some());
assert!(native_filter_exprs(result.selected_plan()).is_empty());
fs::remove_dir_all(dir).unwrap();
}
#[test]
fn pushed_down_parquet_predicate_with_source_limit_lowers_limit_after_reader_predicate() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-pushed-predicate-source-limit");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_with_source_limit(parquet_scan_with_pushed_down_predicate(&path), 3);
let result = lower_physical_plan_to_native(scan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.source_diagnostics.len(), 1);
assert_eq!(result.source_diagnostics[0].predicate_disposition.as_str(), "reader_pushdown");
assert!(result.source_diagnostics[0].rejected_features.is_empty());
let source = native_local_parquet_source(result.selected_plan());
assert_eq!(source.scan_limit, None, "pushed predicates must be applied before the scan limit");
assert!(source.reader_predicate.is_some());
let limits = result
.selected_plan()
.nodes()
.iter()
.filter_map(|node| match &node.kind {
PlanNode::Limit(limit) => Some(limit),
_ => None,
})
.collect::<Vec<_>>();
assert_eq!(
limits.len(),
1,
"native plan should apply the scan limit after the predicate:\n{}",
result.selected_plan()
);
assert_eq!(limits[0].spec.fetch, Some(3));
assert!(native_filter_exprs(result.selected_plan()).is_empty());
fs::remove_dir_all(dir).unwrap();
}
#[test]
fn pushed_down_post_scan_predicate_with_source_limit_lowers_filter_then_limit() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-pushed-post-scan-predicate-source-limit");
let path = dir.join("input.parquet");
write_household_demographics_like_parquet_file(&path);
let scan = parquet_scan_with_source_limit(parquet_scan_with_pushed_down_case_decimal_divide_predicate(&path), 3);
let result = lower_physical_plan_to_native(scan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.source_diagnostics.len(), 1);
assert_eq!(result.source_diagnostics[0].predicate_disposition.as_str(), "native_post_scan_only");
assert!(result.source_diagnostics[0].rejected_features.is_empty());
let source = native_local_parquet_source(result.selected_plan());
assert_eq!(source.scan_limit, None, "pushed predicates must be applied before the scan limit");
assert!(source.reader_predicate.is_none());
let post_scan_filters = native_filter_exprs(result.selected_plan());
assert_eq!(post_scan_filters.len(), 1);
let limit_node = result
.selected_plan()
.nodes()
.iter()
.find_map(|node| match &node.kind {
PlanNode::Limit(limit) if limit.spec.fetch == Some(3) => Some(limit),
_ => None,
})
.expect("native plan should apply scan limit after post-scan filter");
let limit_input = result.selected_plan().node(limit_node.input).expect("limit input should exist");
let limit_is_after_filter = match &limit_input.kind {
PlanNode::Filter(_) => true,
PlanNode::Projection(projection) => {
result.selected_plan().node(projection.input).is_some_and(|node| matches!(node.kind, PlanNode::Filter(_)))
},
_ => false,
};
assert!(limit_is_after_filter, "limit should be above the predicate filter:\n{}", result.selected_plan());
fs::remove_dir_all(dir).unwrap();
}
fn empty_local_parquet_scan_lowers_to_empty_literal_source() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-empty-local-scan");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_without_files(parquet_scan_with_projection_indices(&path, vec![0, 2]));
let result = lower_physical_plan_to_native(scan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
assert!(result.source_diagnostics.is_empty());
assert!(native_local_parquet_sources(result.selected_plan()).is_empty());
let literal_sources = native_literal_batch_sources(result.selected_plan());
assert_eq!(literal_sources.len(), 1);
assert_eq!(literal_sources[0].row_count(), 0);
let field_names = literal_sources[0].schema.fields().iter().map(|field| field.name().as_str()).collect::<Vec<_>>();
assert_eq!(field_names, vec!["key", "m00"]);
fs::remove_dir_all(dir).unwrap();
}
fn empty_s3_parquet_scan_lowers_to_empty_literal_source() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-empty-s3-scan");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_without_files(parquet_scan_with_object_store_url_and_location(
&path,
ObjectStoreUrl::parse("s3://bucket").unwrap(),
"table/input.parquet",
));
let result = lower_physical_plan_to_native(scan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
assert!(result.source_diagnostics.is_empty());
assert!(native_local_parquet_sources(result.selected_plan()).is_empty());
let literal_sources = native_literal_batch_sources(result.selected_plan());
assert_eq!(literal_sources.len(), 1);
assert_eq!(literal_sources[0].row_count(), 0);
fs::remove_dir_all(dir).unwrap();
}
#[test]
fn source_contracts_tc_empty_cases_1() {
testing::run_named_cases(&[
(
"empty_local_parquet_scan_with_filter_projection_lowers_to_projected_empty_literal_source",
empty_local_parquet_scan_with_filter_projection_lowers_to_projected_empty_literal_source,
),
(
"empty_local_parquet_scan_lowers_to_empty_literal_source",
empty_local_parquet_scan_lowers_to_empty_literal_source,
),
(
"empty_local_parquet_scan_with_source_limit_lowers_to_empty_literal_source",
empty_local_parquet_scan_with_source_limit_lowers_to_empty_literal_source,
),
(
"empty_local_parquet_aggregate_lowers_through_literal_source",
empty_local_parquet_aggregate_lowers_through_literal_source,
),
("empty_s3_parquet_scan_lowers_to_empty_literal_source", empty_s3_parquet_scan_lowers_to_empty_literal_source),
(
"empty_local_parquet_scan_with_pushed_predicate_and_source_limit_lowers_to_empty_literal_source",
empty_local_parquet_scan_with_pushed_predicate_and_source_limit_lowers_to_empty_literal_source,
),
]);
}
fn empty_local_parquet_scan_with_filter_projection_lowers_to_projected_empty_literal_source() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-empty-local-filter-projection");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_without_files(parquet_scan(&path));
let scan_schema = scan.schema();
let predicate =
binary(col("payload", &scan_schema).unwrap(), Operator::GtEq, lit(ScalarValue::Int32(Some(10))), &scan_schema)
.unwrap();
let plan: Arc<dyn ExecutionPlan> =
Arc::new(FilterExecBuilder::new(predicate, scan).apply_projection(Some(vec![0, 2])).unwrap().build().unwrap());
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
assert!(result.source_diagnostics.is_empty());
assert!(native_filter_exprs(result.selected_plan()).is_empty());
assert!(native_local_parquet_sources(result.selected_plan()).is_empty());
let literal_sources = native_literal_batch_sources(result.selected_plan());
assert_eq!(literal_sources.len(), 1);
assert_eq!(literal_sources[0].row_count(), 0);
let field_names = literal_sources[0].schema.fields().iter().map(|field| field.name().as_str()).collect::<Vec<_>>();
assert_eq!(field_names, vec!["key", "m00"]);
fs::remove_dir_all(dir).unwrap();
}
fn empty_local_parquet_scan_with_source_limit_lowers_to_empty_literal_source() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-empty-local-source-limit");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_without_files(parquet_scan_with_source_limit(parquet_scan(&path), 3));
let result = lower_physical_plan_to_native(scan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
assert!(result.source_diagnostics.is_empty());
assert!(native_local_parquet_sources(result.selected_plan()).is_empty());
assert!(result.selected_plan().nodes().iter().all(|node| !matches!(node.kind, PlanNode::Limit(_))));
let literal_sources = native_literal_batch_sources(result.selected_plan());
assert_eq!(literal_sources.len(), 1);
assert_eq!(literal_sources[0].row_count(), 0);
fs::remove_dir_all(dir).unwrap();
}
fn empty_local_parquet_scan_with_pushed_predicate_and_source_limit_lowers_to_empty_literal_source() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-empty-local-predicate-source-limit");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan =
parquet_scan_without_files(parquet_scan_with_source_limit(parquet_scan_with_pushed_down_predicate(&path), 3));
let result = lower_physical_plan_to_native(scan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
assert!(result.source_diagnostics.is_empty());
assert!(native_local_parquet_sources(result.selected_plan()).is_empty());
assert!(native_filter_exprs(result.selected_plan()).is_empty());
assert!(result.selected_plan().nodes().iter().all(|node| !matches!(node.kind, PlanNode::Limit(_))));
let literal_sources = native_literal_batch_sources(result.selected_plan());
assert_eq!(literal_sources.len(), 1);
assert_eq!(literal_sources[0].row_count(), 0);
fs::remove_dir_all(dir).unwrap();
}
fn empty_local_parquet_aggregate_lowers_through_literal_source() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-empty-local-aggregate");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let plan = count_only_sort_limit_plan(parquet_scan_without_files(parquet_scan(&path)));
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
assert!(result.source_diagnostics.is_empty());
assert!(native_local_parquet_sources(result.selected_plan()).is_empty());
let literal_sources = native_literal_batch_sources(result.selected_plan());
assert_eq!(literal_sources.len(), 1);
assert_eq!(literal_sources[0].row_count(), 0);
assert_eq!(native_aggregate_functions(result.selected_plan()), vec![AggregateFunction::Count]);
fs::remove_dir_all(dir).unwrap();
}
fn pushed_down_scan_only_decimal_predicate_uses_reader_pushdown() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-pushed-decimal-post-scan");
let path = dir.join("input.parquet");
write_decimal_filter_parquet_file(&path);
let scan = parquet_scan_with_pushed_down_decimal_predicate(&path);
let plan = aggregate_sort_limit_plan_without_filter(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.source_diagnostics.len(), 1);
assert!(result.source_diagnostics[0].direct_device_read);
assert_eq!(result.source_diagnostics[0].predicate_disposition.as_str(), "reader_pushdown");
assert!(result.source_diagnostics[0].rejected_features.is_empty());
let source = native_local_parquet_source(result.selected_plan());
assert!(source.reader_predicate.is_some());
assert_eq!(source.pure_filter_columns, vec!["payload"]);
assert!(native_filter_exprs(result.selected_plan()).is_empty());
fs::remove_dir_all(dir).unwrap();
}
#[test]
fn source_contracts_tc_pred_cases_1() {
testing::run_named_cases(&[
(
"pushed_down_decimal_equality_predicate_uses_reader_pushdown",
pushed_down_decimal_equality_predicate_uses_reader_pushdown,
),
(
"pushed_down_integer_divide_predicate_stays_a_native_post_scan_filter",
pushed_down_integer_divide_predicate_stays_a_native_post_scan_filter,
),
(
"pushed_down_case_decimal_divide_predicate_uses_native_post_scan_filter",
pushed_down_case_decimal_divide_predicate_uses_native_post_scan_filter,
),
(
"pushed_down_parquet_predicate_with_different_filter_uses_native_post_scan_filter",
pushed_down_parquet_predicate_with_different_filter_uses_native_post_scan_filter,
),
(
"pushed_down_parquet_predicate_with_filter_is_reader_pushdown",
pushed_down_parquet_predicate_with_filter_is_reader_pushdown,
),
(
"pushed_down_decimal_in_list_predicate_uses_reader_pushdown",
pushed_down_decimal_in_list_predicate_uses_reader_pushdown,
),
(
"pushed_down_parquet_predicate_subset_of_filter_is_reader_pushdown_with_native_verification",
pushed_down_parquet_predicate_subset_of_filter_is_reader_pushdown_with_native_verification,
),
(
"pushed_down_parquet_predicate_uses_filter_input_schema_for_validation",
pushed_down_parquet_predicate_uses_filter_input_schema_for_validation,
),
(
"pushed_down_scan_only_decimal_predicate_uses_reader_pushdown",
pushed_down_scan_only_decimal_predicate_uses_reader_pushdown,
),
(
"pushed_down_parquet_predicate_without_filter_is_reader_pushdown",
pushed_down_parquet_predicate_without_filter_is_reader_pushdown,
),
(
"pushed_down_substring_prefix_predicate_is_rewritten_to_reader_pushdown_ranges",
pushed_down_substring_prefix_predicate_is_rewritten_to_reader_pushdown_ranges,
),
]);
}
fn pushed_down_decimal_equality_predicate_uses_reader_pushdown() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-pushed-decimal-equality-reader");
let path = dir.join("input.parquet");
write_decimal_filter_parquet_file(&path);
let scan = parquet_scan_with_pushed_down_decimal_equality_predicate(&path);
let scan_schema = scan.schema();
let predicate =
binary(col("payload", &scan_schema).unwrap(), Operator::Eq, lit(ScalarValue::Int32(Some(10))), &scan_schema)
.unwrap();
let plan = aggregate_sort_limit_plan_with_filter(scan, predicate, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.source_diagnostics.len(), 1);
assert!(result.source_diagnostics[0].direct_device_read);
assert_eq!(result.source_diagnostics[0].predicate_disposition.as_str(), "reader_pushdown");
assert!(result.source_diagnostics[0].rejected_features.is_empty());
let source = native_local_parquet_source(result.selected_plan());
assert!(source.reader_predicate.is_some());
assert_eq!(source.pure_filter_columns, vec!["payload"]);
assert!(native_filter_exprs(result.selected_plan()).is_empty());
fs::remove_dir_all(dir).unwrap();
}
fn pushed_down_decimal_in_list_predicate_uses_reader_pushdown() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-pushed-decimal-in-list-reader");
let path = dir.join("input.parquet");
write_decimal_filter_parquet_file(&path);
let scan = parquet_scan_with_pushed_down_decimal_in_list_predicate(&path);
let plan = aggregate_sort_limit_plan_without_filter(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.source_diagnostics.len(), 1);
assert!(result.source_diagnostics[0].direct_device_read);
assert_eq!(result.source_diagnostics[0].predicate_disposition.as_str(), "reader_pushdown");
assert!(result.source_diagnostics[0].rejected_features.is_empty());
let source = native_local_parquet_source(result.selected_plan());
assert!(source.reader_predicate.is_some());
assert_eq!(source.pure_filter_columns, vec!["payload"]);
assert!(native_filter_exprs(result.selected_plan()).is_empty());
fs::remove_dir_all(dir).unwrap();
}
fn pushed_down_case_decimal_divide_predicate_uses_native_post_scan_filter() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-pushed-case-decimal-divide");
let path = dir.join("input.parquet");
write_household_demographics_like_parquet_file(&path);
let scan = parquet_scan_with_pushed_down_case_decimal_divide_predicate(&path);
let plan = aggregate_sort_limit_plan_without_filter(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.source_diagnostics.len(), 1);
assert!(result.source_diagnostics[0].direct_device_read);
assert_eq!(result.source_diagnostics[0].predicate_disposition.as_str(), "native_post_scan_only");
assert!(result.source_diagnostics[0].rejected_features.is_empty());
let source = native_local_parquet_source(result.selected_plan());
assert!(source.reader_predicate.is_none());
assert_eq!(native_filter_exprs(result.selected_plan()).len(), 1);
fs::remove_dir_all(dir).unwrap();
}
fn pushed_down_integer_divide_predicate_stays_a_native_post_scan_filter() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-pushed-integer-divide");
let path = dir.join("input.parquet");
write_household_demographics_like_parquet_file(&path);
let scan = parquet_scan_with_pushed_down_integer_divide_predicate(&path);
let plan = aggregate_sort_limit_plan_without_filter(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.source_diagnostics.len(), 1);
assert_eq!(result.source_diagnostics[0].predicate_disposition.as_str(), "native_post_scan_only");
let source = native_local_parquet_source(result.selected_plan());
assert!(source.reader_predicate.is_none());
assert_eq!(native_filter_exprs(result.selected_plan()).len(), 1);
fs::remove_dir_all(dir).unwrap();
}
fn pushed_down_parquet_predicate_with_different_filter_uses_native_post_scan_filter() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-pushed-predicate-mismatch");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_with_pushed_down_predicate_value(&path, 20);
let plan = aggregate_sort_limit_plan(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1);
assert_eq!(result.not_selected_count(), 0);
assert!(matches!(result.report.outcomes()[0], CandidateOutcome::Selected { .. }));
assert_eq!(result.source_diagnostics.len(), 1);
assert!(result.source_diagnostics[0].direct_device_read);
assert_eq!(result.source_diagnostics[0].predicate_disposition.as_str(), "native_post_scan_only");
assert!(result.source_diagnostics[0].rejected_features.is_empty());
let source = native_local_parquet_source(result.selected_plan());
assert!(source.reader_predicate.is_none());
assert_eq!(native_filter_exprs(result.selected_plan()).len(), 1);
fs::remove_dir_all(dir).unwrap();
}
fn full_file_byte_range_coalesces_to_native_source() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-ranged-file");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_with_byte_range(&path);
let plan = aggregate_sort_limit_plan(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1);
assert_eq!(result.not_selected_count(), 0);
let source = native_local_parquet_source(result.selected_plan());
assert_eq!(source.file_count(), 1);
assert_eq!(source.files[0].file_size_bytes, Some(fs::metadata(&path).unwrap().len()));
let expected_stable_id_prefix = format!("local:{}:{}:mtime=", path.display(), fs::metadata(&path).unwrap().len());
assert!(
source.files[0].stable_id.starts_with(&expected_stable_id_prefix),
"stable id should include local path, size, and mtime: {}",
source.files[0].stable_id
);
assert!(source.diagnostics.direct_device_read);
assert_eq!(source.diagnostics.ranged_file_count, 1);
assert_eq!(source.diagnostics.byte_range_disposition.as_str(), "coalesced_full_files");
assert!(source.diagnostics.rejected_features.is_empty());
assert_eq!(result.source_diagnostics.len(), 1);
assert_eq!(result.source_diagnostics[0].ranged_file_count, 1);
assert!(result.source_diagnostics[0].direct_device_read);
assert_eq!(result.source_diagnostics[0].byte_range_disposition.as_str(), "coalesced_full_files");
assert!(result.source_diagnostics[0].rejected_features.is_empty());
fs::remove_dir_all(dir).unwrap();
}
fn partial_byte_range_maps_to_row_group_selection() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-partial-ranged-file");
let path = dir.join("input.parquet");
write_row_group_pruning_parquet_file(&path);
let row_group_offsets = row_group_range_pruning_offsets(&path);
assert!(row_group_offsets.len() >= 2, "fixture must have at least two row groups");
let scan = parquet_scan_with_explicit_byte_ranges(&path, vec![(row_group_offsets[0], row_group_offsets[1])]);
let plan = aggregate_sort_limit_plan(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
let source = native_local_parquet_source(result.selected_plan());
assert_eq!(source.file_count(), 1);
assert_eq!(source.files[0].row_groups.as_deref(), Some(&[0][..]));
assert_eq!(source.diagnostics.ranged_file_count, 1);
assert_eq!(source.diagnostics.byte_range_disposition.as_str(), "row_group_selection");
assert_eq!(source.diagnostics.selected_row_groups, Some(1));
assert!(source.diagnostics.direct_device_read);
assert!(source.diagnostics.rejected_features.is_empty());
assert_eq!(result.source_diagnostics.len(), 1);
assert_eq!(result.source_diagnostics[0].ranged_file_count, 1);
assert_eq!(result.source_diagnostics[0].byte_range_disposition.as_str(), "row_group_selection");
assert_eq!(result.source_diagnostics[0].selected_row_groups, Some(1));
assert!(result.source_diagnostics[0].rejected_features.is_empty());
fs::remove_dir_all(dir).unwrap();
}
#[test]
fn source_contracts_tc_range_cases_1() {
testing::run_named_cases(&[
(
"partial_byte_range_owns_row_groups_by_start_offset_boundaries",
partial_byte_range_owns_row_groups_by_start_offset_boundaries,
),
(
"partial_byte_range_rejects_access_plan_length_mismatch",
partial_byte_range_rejects_access_plan_length_mismatch,
),
(
"mixed_partial_byte_range_and_full_file_reports_selected_row_group_total",
mixed_partial_byte_range_and_full_file_reports_selected_row_group_total,
),
(
"partial_byte_range_with_access_plan_lowers_to_selected_row_groups",
partial_byte_range_with_access_plan_lowers_to_selected_row_groups,
),
(
"partial_byte_range_ignores_row_level_selection_outside_range",
partial_byte_range_ignores_row_level_selection_outside_range,
),
("partial_byte_range_maps_to_row_group_selection", partial_byte_range_maps_to_row_group_selection),
(
"partial_byte_range_with_access_plan_and_extra_extension_is_rejected",
partial_byte_range_with_access_plan_and_extra_extension_is_rejected,
),
(
"overlapping_byte_ranges_are_rejected_with_source_reason",
overlapping_byte_ranges_are_rejected_with_source_reason,
),
("full_file_byte_range_coalesces_to_native_source", full_file_byte_range_coalesces_to_native_source),
(
"split_full_file_byte_ranges_coalesce_without_duplicate_native_files",
split_full_file_byte_ranges_coalesce_without_duplicate_native_files,
),
]);
}
fn partial_byte_range_owns_row_groups_by_start_offset_boundaries() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-row-group-byte-range");
let path = dir.join("input.parquet");
write_row_group_pruning_parquet_file(&path);
let row_group_offsets = row_group_range_pruning_offsets(&path);
assert!(row_group_offsets.len() >= 4, "fixture must have at least four row groups");
let scan = parquet_scan_with_explicit_byte_ranges(&path, vec![(row_group_offsets[1], row_group_offsets[3])]);
let plan = aggregate_sort_limit_plan(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
let source = native_local_parquet_source(result.selected_plan());
assert_eq!(source.files[0].row_groups.as_deref(), Some(&[1, 2][..]));
assert_eq!(source.diagnostics.ranged_file_count, 1);
assert_eq!(source.diagnostics.byte_range_disposition.as_str(), "row_group_selection");
assert_eq!(source.diagnostics.selected_row_groups, Some(2));
assert!(source.diagnostics.direct_device_read);
assert!(source.diagnostics.rejected_features.is_empty());
assert_eq!(result.source_diagnostics[0].byte_range_disposition.as_str(), "row_group_selection");
assert_eq!(result.source_diagnostics[0].selected_row_groups, Some(2));
assert!(result.source_diagnostics[0].rejected_features.is_empty());
fs::remove_dir_all(dir).unwrap();
}
fn partial_byte_range_with_access_plan_lowers_to_selected_row_groups() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-ranged-access-plan");
let path = dir.join("input.parquet");
write_row_group_pruning_parquet_file(&path);
let row_group_offsets = row_group_range_pruning_offsets(&path);
assert!(row_group_offsets.len() >= 4, "fixture must have at least four row groups");
let access_plan = ParquetAccessPlan::new(vec![
RowGroupAccess::Skip,
RowGroupAccess::Scan,
RowGroupAccess::Skip,
RowGroupAccess::Scan,
]);
let scan = parquet_scan_with_explicit_byte_ranges_and_access_plan(
&path,
vec![(row_group_offsets[1], row_group_offsets[3])],
access_plan,
);
let plan = aggregate_sort_limit_plan(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
let source = native_local_parquet_source(result.selected_plan());
assert_eq!(source.file_count(), 1);
assert_eq!(source.files[0].row_groups.as_deref(), Some(&[1][..]));
assert_eq!(source.diagnostics.ranged_file_count, 1);
assert_eq!(source.diagnostics.byte_range_disposition.as_str(), "row_group_selection");
assert_eq!(source.diagnostics.selected_row_groups, Some(1));
assert!(source.diagnostics.direct_device_read);
assert!(source.diagnostics.rejected_features.is_empty());
assert_eq!(result.source_diagnostics.len(), 1);
assert_eq!(result.source_diagnostics[0].byte_range_disposition.as_str(), "row_group_selection");
assert_eq!(result.source_diagnostics[0].selected_row_groups, Some(1));
assert!(result.source_diagnostics[0].rejected_features.is_empty());
fs::remove_dir_all(dir).unwrap();
}
fn partial_byte_range_ignores_row_level_selection_outside_range() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-ranged-selection-outside");
let path = dir.join("input.parquet");
write_row_group_pruning_parquet_file(&path);
let row_group_offsets = row_group_range_pruning_offsets(&path);
assert!(row_group_offsets.len() >= 4, "fixture must have at least four row groups");
let row_selection = RowSelection::from(vec![RowSelector::select(1), RowSelector::skip(1)]);
let access_plan = ParquetAccessPlan::new(vec![
RowGroupAccess::Selection(row_selection),
RowGroupAccess::Scan,
RowGroupAccess::Skip,
RowGroupAccess::Skip,
]);
let scan = parquet_scan_with_explicit_byte_ranges_and_access_plan(
&path,
vec![(row_group_offsets[1], row_group_offsets[2])],
access_plan,
);
let plan = aggregate_sort_limit_plan(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
let source = native_local_parquet_source(result.selected_plan());
assert_eq!(source.files[0].row_groups.as_deref(), Some(&[1][..]));
assert_eq!(source.diagnostics.byte_range_disposition.as_str(), "row_group_selection");
assert_eq!(source.diagnostics.selected_row_groups, Some(1));
assert!(source.diagnostics.rejected_features.is_empty());
fs::remove_dir_all(dir).unwrap();
}
fn partial_byte_range_rejects_access_plan_length_mismatch() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-ranged-access-plan-len");
let path = dir.join("input.parquet");
write_row_group_pruning_parquet_file(&path);
let row_group_offsets = row_group_range_pruning_offsets(&path);
assert!(row_group_offsets.len() >= 4, "fixture must have at least four row groups");
let access_plan = ParquetAccessPlan::new(vec![RowGroupAccess::Scan]);
let scan = parquet_scan_with_explicit_byte_ranges_and_access_plan(
&path,
vec![(row_group_offsets[2], row_group_offsets[3])],
access_plan,
);
let plan = aggregate_sort_limit_plan(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert!(result.plan.is_none());
assert_eq!(result.not_selected_count(), 1);
assert!(matches!(
result.report.outcomes()[0],
CandidateOutcome::NotSupported { capability_reason: CapabilityReason::NativeSourceByteRange, .. }
));
assert_eq!(result.source_diagnostics[0].byte_range_disposition.as_str(), "rejected");
assert!(result.source_diagnostics[0].rejected_features.iter().any(|feature| feature.as_str() == "byte_range"));
fs::remove_dir_all(dir).unwrap();
}
fn mixed_partial_byte_range_and_full_file_reports_selected_row_group_total() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-mixed-ranged-files");
let first = dir.join("part-000.parquet");
let second = dir.join("part-001.parquet");
write_row_group_pruning_parquet_file(&first);
write_row_group_pruning_parquet_file(&second);
let row_group_offsets = row_group_range_pruning_offsets(&first);
assert!(row_group_offsets.len() >= 2, "fixture must have at least two row groups");
let scan = parquet_scan_with_named_file_byte_ranges(
&dir,
"part-000.parquet",
vec![(row_group_offsets[0], row_group_offsets[1])],
);
let plan = aggregate_sort_limit_plan(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
let source = native_local_parquet_source(result.selected_plan());
assert_eq!(source.file_count(), 2);
assert_eq!(source.diagnostics.ranged_file_count, 1);
assert_eq!(source.diagnostics.byte_range_disposition.as_str(), "row_group_selection");
assert_eq!(source.diagnostics.selected_row_groups, Some(5));
assert_eq!(result.source_diagnostics[0].selected_row_groups, Some(5));
assert!(source.diagnostics.rejected_features.is_empty());
fs::remove_dir_all(dir).unwrap();
}
fn partial_byte_range_with_access_plan_and_extra_extension_is_rejected() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-ranged-mixed-extension");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let access_plan = ParquetAccessPlan::new(vec![
RowGroupAccess::Skip,
RowGroupAccess::Scan,
RowGroupAccess::Skip,
RowGroupAccess::Scan,
]);
let scan = parquet_scan_with_partial_byte_range_access_plan_and_extra_extension(&path, access_plan);
let plan = aggregate_sort_limit_plan(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert!(result.plan.is_none());
assert_eq!(result.not_selected_count(), 1);
assert!(matches!(
result.report.outcomes()[0],
CandidateOutcome::NotSupported { capability_reason: CapabilityReason::NativeSourceByteRange, .. }
));
assert_eq!(result.source_diagnostics.len(), 1);
assert_eq!(result.source_diagnostics[0].byte_range_disposition.as_str(), "rejected");
assert!(result.source_diagnostics[0].rejected_features.iter().any(|feature| feature.as_str() == "byte_range"));
fs::remove_dir_all(dir).unwrap();
}
fn split_full_file_byte_ranges_coalesce_without_duplicate_native_files() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-split-ranged-file");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_with_split_full_byte_ranges(&path);
let plan = aggregate_sort_limit_plan(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1);
assert_eq!(result.not_selected_count(), 0);
let source = native_local_parquet_source(result.selected_plan());
assert_eq!(source.file_count(), 1);
assert_eq!(source.files[0].file_size_bytes, Some(fs::metadata(&path).unwrap().len()));
assert_eq!(source.diagnostics.file_count, 1);
assert_eq!(source.diagnostics.ranged_file_count, 2);
assert_eq!(source.diagnostics.byte_range_disposition.as_str(), "coalesced_full_files");
assert!(source.diagnostics.direct_device_read);
assert!(source.diagnostics.rejected_features.is_empty());
assert_eq!(result.source_diagnostics[0].file_count, 1);
assert_eq!(result.source_diagnostics[0].ranged_file_count, 2);
assert!(result.source_diagnostics[0].rejected_features.is_empty());
fs::remove_dir_all(dir).unwrap();
}
fn overlapping_byte_ranges_are_rejected_with_source_reason() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-overlap-ranged-file");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_with_overlapping_byte_ranges(&path);
let plan = aggregate_sort_limit_plan(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert!(result.plan.is_none());
assert_eq!(result.not_selected_count(), 1);
assert!(matches!(
result.report.outcomes()[0],
CandidateOutcome::NotSupported { capability_reason: CapabilityReason::NativeSourceByteRange, .. }
));
assert_eq!(result.source_diagnostics[0].ranged_file_count, 2);
assert_eq!(result.source_diagnostics[0].byte_range_disposition.as_str(), "rejected");
assert!(result.source_diagnostics[0].rejected_features.iter().any(|feature| feature.as_str() == "byte_range"));
fs::remove_dir_all(dir).unwrap();
}
fn assert_s3_remote_native_source(source: &LocalParquetSource, expected_uri: &str) {
assert_eq!(source.file_count(), 1);
let file = &source.files[0];
assert_eq!(file.location.read_capability(), "object_store:s3");
assert_eq!(file.resolved_source.original_location, expected_uri);
assert_eq!(file.resolved_source.reader_location(), expected_uri);
assert!(file.stable_id.starts_with("object_store:s3:fnv64:"));
assert!(!file.stable_id.contains("bucket"));
assert!(!file.stable_id.contains("table/input.parquet"));
assert_eq!(source.diagnostics.source_read_strategy, file.resolved_source.read_strategy().as_str());
assert_eq!(source.diagnostics.credential_mode, file.resolved_source.credential_mode().as_str());
assert!(!source.diagnostics.direct_device_read);
assert!(source.diagnostics.rejected_features.iter().all(|feature| feature.as_str() != "non_local_object_store"));
}
#[test]
fn s3_parquet_object_store_scan_lowers_to_remote_native_source() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-s3-store");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_with_object_store_url_and_location(
&path,
ObjectStoreUrl::parse("s3://bucket").unwrap(),
"table/input.parquet",
);
let expected_uri = "s3://bucket/table/input.parquet";
let result = lower_physical_plan_to_native(scan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.source_diagnostics.len(), 1);
assert!(result.source_diagnostics[0].rejected_features.is_empty());
assert!(!result.source_diagnostics[0].direct_device_read);
let source = native_local_parquet_source(result.selected_plan());
assert_s3_remote_native_source(source, expected_uri);
assert_eq!(result.source_diagnostics[0].source_read_strategy, source.diagnostics.source_read_strategy);
assert_eq!(result.source_diagnostics[0].credential_mode, source.diagnostics.credential_mode);
fs::remove_dir_all(dir).unwrap();
}
#[test]
fn s3_multi_file_parquet_object_store_scan_rejects_unverified_footer_schema() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-s3-store-multi-file");
let first_path = dir.join("part-0.parquet");
let second_path = dir.join("part-1.parquet");
write_numeric_parquet_file(&first_path);
write_numeric_parquet_file(&second_path);
let scan = parquet_scan_with_object_store_url(&dir, ObjectStoreUrl::parse("s3://bucket").unwrap());
let result = lower_physical_plan_to_native(scan, &ParquetReadStrategyConfig::default()).unwrap();
assert!(result.plan.is_none());
assert_eq!(result.not_selected_count(), 1);
assert!(matches!(
result.report.outcomes()[0],
CandidateOutcome::NotSupported { capability_reason: CapabilityReason::NativeSourceSchemaMapping, .. }
));
fs::remove_dir_all(dir).unwrap();
}
#[test]
fn s3_parquet_scan_reads_through_the_configured_s3_storage() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-s3-storage");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_with_object_store_url_and_location(
&path,
ObjectStoreUrl::parse("s3://bucket").unwrap(),
"table/input.parquet",
);
let config = AlgeonNativeExecutionConfig::default().with_s3_storage(
S3SourceStorageProfile::default_chain().with_endpoint("http://127.0.0.1:9").with_path_style_access(true),
);
let result = lower_physical_plan_to_native(scan, &config.source_read_strategy()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
let resolved = &native_local_parquet_source(result.selected_plan()).files[0].resolved_source;
assert_eq!(resolved.storage_profile_id(), Some(S3_STORAGE_PROFILE_ID));
assert_eq!(resolved.credential_mode(), CredentialMode::ScopedResolved);
let registered = config.source_storage_profiles();
assert_eq!(registered.iter().map(SourceStorageProfile::id).collect::<Vec<_>>(), [S3_STORAGE_PROFILE_ID]);
fs::remove_dir_all(dir).unwrap();
}
#[test]
fn non_s3_parquet_object_store_is_rejected_with_source_reason() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-remote-store");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_with_object_store_url(&path, ObjectStoreUrl::parse("test:///").unwrap());
let plan = aggregate_sort_limit_plan(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert!(result.plan.is_none());
assert_eq!(result.not_selected_count(), 1);
assert!(matches!(
result.report.outcomes()[0],
CandidateOutcome::NotSupported { capability_reason: CapabilityReason::NativeSourceNonLocalObjectStore, .. }
));
assert_eq!(result.source_diagnostics.len(), 1);
assert_eq!(result.source_diagnostics[0].file_count, 1);
assert!(
result.source_diagnostics[0]
.rejected_features
.iter()
.any(|feature| feature.as_str() == "non_local_object_store")
);
assert_eq!(result.source_diagnostics[0].partition_column_disposition.as_str(), "absent");
fs::remove_dir_all(dir).unwrap();
}
fn row_group_range_pruning_offsets(path: &std::path::Path) -> Vec<i64> {
let reader = SerializedFileReader::try_from(path).unwrap();
reader
.metadata()
.row_groups()
.iter()
.map(|row_group| {
let column = row_group.column(0);
column.dictionary_page_offset().unwrap_or_else(|| column.data_page_offset())
})
.collect()
}
#[test]
fn custom_parquet_file_reader_factory_is_rejected_with_source_reason() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-custom-reader-factory");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_with_custom_reader_factory(&path);
let plan = aggregate_sort_limit_plan(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert!(result.plan.is_none());
assert_eq!(result.not_selected_count(), 1);
assert!(matches!(
result.report.outcomes()[0],
CandidateOutcome::NotSupported {
capability_reason: CapabilityReason::NativeSourceParquetFileReaderFactory,
..
}
));
assert_eq!(result.source_diagnostics.len(), 1);
assert!(!result.source_diagnostics[0].direct_device_read);
assert!(
result.source_diagnostics[0]
.rejected_features
.iter()
.any(|feature| feature.as_str() == "parquet_file_reader_factory")
);
fs::remove_dir_all(dir).unwrap();
}
#[test]
fn reader_factory_named_like_the_datafusion_default_is_rejected_with_source_reason() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-default-lookalike-reader-factory");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_with_default_lookalike_reader_factory(&path);
let plan = aggregate_sort_limit_plan(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert!(result.plan.is_none());
assert!(matches!(
result.report.outcomes()[0],
CandidateOutcome::NotSupported {
capability_reason: CapabilityReason::NativeSourceParquetFileReaderFactory,
..
}
));
fs::remove_dir_all(dir).unwrap();
}
fn parquet_partition_columns_are_represented_in_native_source_contract() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-partition-columns");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_with_partition_value(&path, ScalarValue::Int32(Some(7)));
let plan = partition_column_group_by_plan(scan);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.source_diagnostics.len(), 1);
assert_eq!(result.source_diagnostics[0].partition_column_disposition.as_str(), "represented");
assert!(result.source_diagnostics[0].rejected_features.is_empty());
let source = native_local_parquet_source(result.selected_plan());
assert_eq!(source.partition_columns.len(), 1);
assert_eq!(source.partition_columns[0].name, "part");
assert_eq!(source.files[0].partition_values, vec![LiteralValue::Int32(7)]);
assert!(!source.read_columns.contains(&"part".to_string()));
assert!(source.projected_output_column_source_refs.contains(&SourceColumnRef::Partition(0)));
fs::remove_dir_all(dir).unwrap();
}
fn parquet_partition_typed_null_values_are_represented_in_native_source_contract() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-partition-typed-null");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_with_partition_value(&path, ScalarValue::Int32(None));
let plan = partition_column_group_by_plan(scan);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.source_diagnostics.len(), 1);
assert_eq!(result.source_diagnostics[0].partition_column_disposition.as_str(), "represented");
assert!(result.source_diagnostics[0].rejected_features.is_empty());
let source = native_local_parquet_source(result.selected_plan());
assert_eq!(source.partition_columns.len(), 1);
assert_eq!(source.partition_columns[0].data_type, DataType::Int32);
assert_eq!(source.files[0].partition_values, vec![LiteralValue::Null(DataType::Int32)]);
fs::remove_dir_all(dir).unwrap();
}
fn parquet_partition_columns_without_literal_values_are_rejected_with_source_reason() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-partition-missing-values");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_with_partition_columns(&path);
let plan = partition_column_group_by_plan(scan);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert!(result.plan.is_none());
assert_eq!(result.not_selected_count(), 1);
assert!(matches!(
result.report.outcomes()[0],
CandidateOutcome::NotSupported { capability_reason: CapabilityReason::NativeSourcePartitionValues, .. }
));
assert_eq!(result.source_diagnostics.len(), 1);
assert_eq!(result.source_diagnostics[0].partition_column_disposition.as_str(), "rejected");
fs::remove_dir_all(dir).unwrap();
}
fn unused_parquet_binary_partition_columns_do_not_block_native_lowering() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-partition-binary");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_with_typed_partition_value(
&path,
Field::new("part", DataType::Binary, false),
ScalarValue::Binary(Some(vec![1, 2, 3])),
);
let plan = aggregate_sort_limit_plan(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.source_diagnostics.len(), 1);
assert_eq!(result.source_diagnostics[0].partition_column_disposition.as_str(), "absent");
let source = native_local_parquet_source(result.selected_plan());
assert!(source.partition_columns.is_empty());
assert!(source.files[0].partition_values.is_empty());
fs::remove_dir_all(dir).unwrap();
}
fn unused_parquet_utf8_partition_columns_are_not_represented_in_native_source_contract() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-partition-utf8");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_with_typed_partition_value(
&path,
Field::new("part", DataType::Utf8, false),
ScalarValue::Utf8(Some("2026".to_string())),
);
let plan = aggregate_sort_limit_plan(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.source_diagnostics.len(), 1);
assert_eq!(result.source_diagnostics[0].partition_column_disposition.as_str(), "absent");
assert!(result.source_diagnostics[0].rejected_features.is_empty());
let source = native_local_parquet_source(result.selected_plan());
assert!(source.partition_columns.is_empty());
assert!(source.files[0].partition_values.is_empty());
fs::remove_dir_all(dir).unwrap();
}
#[test]
fn source_contracts_tc_part_cases_1() {
testing::run_named_cases(&[
(
"unused_parquet_utf8_view_partition_values_do_not_enter_native_source_contract",
unused_parquet_utf8_view_partition_values_do_not_enter_native_source_contract,
),
(
"unused_parquet_utf8_view_partition_typed_null_values_do_not_enter_native_source_contract",
unused_parquet_utf8_view_partition_typed_null_values_do_not_enter_native_source_contract,
),
(
"unused_parquet_utf8_partition_columns_are_not_represented_in_native_source_contract",
unused_parquet_utf8_partition_columns_are_not_represented_in_native_source_contract,
),
(
"parquet_partition_columns_are_represented_in_native_source_contract",
parquet_partition_columns_are_represented_in_native_source_contract,
),
(
"parquet_partition_columns_without_literal_values_are_rejected_with_source_reason",
parquet_partition_columns_without_literal_values_are_rejected_with_source_reason,
),
(
"unused_parquet_binary_partition_columns_do_not_block_native_lowering",
unused_parquet_binary_partition_columns_do_not_block_native_lowering,
),
(
"parquet_partition_typed_null_values_are_represented_in_native_source_contract",
parquet_partition_typed_null_values_are_represented_in_native_source_contract,
),
]);
}
fn unused_parquet_utf8_view_partition_values_do_not_enter_native_source_contract() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-partition-utf8-view");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_with_typed_partition_value(
&path,
Field::new("part", DataType::Utf8View, false),
ScalarValue::Utf8View(Some("2026".to_string())),
);
let plan = aggregate_sort_limit_plan(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.source_diagnostics.len(), 1);
assert_eq!(result.source_diagnostics[0].partition_column_disposition.as_str(), "absent");
assert!(result.source_diagnostics[0].rejected_features.is_empty());
let source = native_local_parquet_source(result.selected_plan());
assert!(source.partition_columns.is_empty());
assert!(source.files[0].partition_values.is_empty());
fs::remove_dir_all(dir).unwrap();
}
fn unused_parquet_utf8_view_partition_typed_null_values_do_not_enter_native_source_contract() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-partition-utf8-view-null");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_with_typed_partition_value(
&path,
Field::new("part", DataType::Utf8View, true),
ScalarValue::Utf8View(None),
);
let plan = aggregate_sort_limit_plan(scan, Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.source_diagnostics.len(), 1);
assert_eq!(result.source_diagnostics[0].partition_column_disposition.as_str(), "absent");
assert!(result.source_diagnostics[0].rejected_features.is_empty());
let source = native_local_parquet_source(result.selected_plan());
assert!(source.partition_columns.is_empty());
assert!(source.files[0].partition_values.is_empty());
fs::remove_dir_all(dir).unwrap();
}
fn unused_unsupported_binary_schema_fields_do_not_block_native_lowering() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-source-binary-schema");
let path = dir.join("input.parquet");
write_binary_schema_parquet_file(&path);
let plan = aggregate_sort_limit_plan(parquet_scan(&path), Operator::Plus);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
let source = native_local_parquet_source(result.selected_plan());
assert!(!source.read_columns.contains(&"blob".to_string()));
assert!(source.schema.field_with_name("blob").is_err());
fs::remove_dir_all(dir).unwrap();
}
fn list_utf8_projection_lowers_as_native_carry_column() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-list-carry");
let path = dir.join("input.parquet");
write_list_utf8_parquet_file(&path);
let scan = parquet_scan(&path);
let scan_schema = scan.schema();
let plan: Arc<dyn ExecutionPlan> = Arc::new(
ProjectionExec::try_new(
[
ProjectionExpr { expr: col("key", &scan_schema).unwrap(), alias: "key".to_string() },
ProjectionExpr { expr: col("tags", &scan_schema).unwrap(), alias: "tags".to_string() },
],
scan,
)
.unwrap(),
);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
let source = native_local_parquet_source(result.selected_plan());
assert_eq!(source.read_columns, vec!["key".to_string(), "tags".to_string()]);
fs::remove_dir_all(dir).unwrap();
}
fn numeric_fixed_size_list_projection_lowers_as_native_carry_column() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-fixed-size-list-carry");
let path = dir.join("input.parquet");
write_fixed_size_list_int32_parquet_file(&path);
let scan = parquet_scan(&path);
let scan_schema = scan.schema();
let plan: Arc<dyn ExecutionPlan> = Arc::new(
ProjectionExec::try_new(
[
ProjectionExpr { expr: col("key", &scan_schema).unwrap(), alias: "key".to_string() },
ProjectionExpr { expr: col("features", &scan_schema).unwrap(), alias: "features".to_string() },
],
scan,
)
.unwrap(),
);
let expected_features = scan_schema.field_with_name("features").unwrap();
let DataType::FixedSizeList(expected_child, expected_dimension) = expected_features.data_type() else {
panic!("Parquet fixture must retain FixedSizeList schema");
};
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
let source = native_local_parquet_source(result.selected_plan());
assert_eq!(source.read_columns, vec!["features".to_string(), "key".to_string()]);
let source_features = source.schema.field_with_name("features").unwrap();
let DataType::FixedSizeList(source_child, source_dimension) = source_features.data_type() else {
panic!("native source must retain FixedSizeList schema");
};
assert_eq!(source_dimension, expected_dimension);
assert_eq!(source_child.data_type(), expected_child.data_type());
assert_eq!(source_child.is_nullable(), expected_child.is_nullable());
assert_eq!(source_features.is_nullable(), expected_features.is_nullable());
fs::remove_dir_all(dir).unwrap();
}
#[test]
fn list_utf8_sort_key_is_rejected_locally_during_native_planning() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-list-sort-reject");
let path = dir.join("input.parquet");
write_list_utf8_parquet_file(&path);
let scan = parquet_scan(&path);
let scan_schema = scan.schema();
let sort_exprs =
LexOrdering::new([PhysicalSortExpr::new(col("tags", &scan_schema).unwrap(), ArrowSortOptions::default())])
.unwrap();
let plan: Arc<dyn ExecutionPlan> = Arc::new(SortExec::new(sort_exprs, scan));
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_native_rejection(&result, CapabilityReason::UnsupportedType);
fs::remove_dir_all(dir).unwrap();
}
fn unprojected_struct_and_map_schema_fields_do_not_block_native_lowering() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-unprojected-struct-map");
let path = dir.join("input.parquet");
write_struct_map_parquet_file(&path);
let scan = parquet_scan(&path);
let scan_schema = scan.schema();
let plan: Arc<dyn ExecutionPlan> = Arc::new(
ProjectionExec::try_new(
[
ProjectionExpr { expr: col("key", &scan_schema).unwrap(), alias: "key".to_string() },
ProjectionExpr { expr: col("m00", &scan_schema).unwrap(), alias: "m00".to_string() },
],
scan,
)
.unwrap(),
);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
let source = native_local_parquet_source(result.selected_plan());
assert_eq!(source.read_columns, vec!["key".to_string(), "m00".to_string()]);
assert!(source.schema.field_with_name("details").is_err());
assert!(source.schema.field_with_name("attrs").is_err());
fs::remove_dir_all(dir).unwrap();
}
#[test]
fn source_contracts_tc_schema_cases_1() {
testing::run_named_cases(&[
(
"projected_struct_schema_field_is_rejected_with_unsupported_type",
projected_struct_schema_field_is_rejected_with_unsupported_type,
),
(
"unused_unsupported_binary_schema_fields_do_not_block_native_lowering",
unused_unsupported_binary_schema_fields_do_not_block_native_lowering,
),
("list_utf8_projection_lowers_as_native_carry_column", list_utf8_projection_lowers_as_native_carry_column),
(
"numeric_fixed_size_list_projection_lowers_as_native_carry_column",
numeric_fixed_size_list_projection_lowers_as_native_carry_column,
),
(
"projected_map_schema_field_is_rejected_with_unsupported_type",
projected_map_schema_field_is_rejected_with_unsupported_type,
),
(
"unprojected_struct_and_map_schema_fields_do_not_block_native_lowering",
unprojected_struct_and_map_schema_fields_do_not_block_native_lowering,
),
]);
}
fn projected_struct_schema_field_is_rejected_with_unsupported_type() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-projected-struct-reject");
let path = dir.join("input.parquet");
write_struct_map_parquet_file(&path);
let scan = parquet_scan(&path);
let scan_schema = scan.schema();
let plan: Arc<dyn ExecutionPlan> = Arc::new(
ProjectionExec::try_new(
[ProjectionExpr { expr: col("details", &scan_schema).unwrap(), alias: "details".to_string() }],
scan,
)
.unwrap(),
);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_native_rejection(&result, CapabilityReason::UnsupportedType);
fs::remove_dir_all(dir).unwrap();
}
fn projected_map_schema_field_is_rejected_with_unsupported_type() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-projected-map-reject");
let path = dir.join("input.parquet");
write_struct_map_parquet_file(&path);
let scan = parquet_scan(&path);
let scan_schema = scan.schema();
let plan: Arc<dyn ExecutionPlan> = Arc::new(
ProjectionExec::try_new(
[ProjectionExpr { expr: col("attrs", &scan_schema).unwrap(), alias: "attrs".to_string() }],
scan,
)
.unwrap(),
);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_native_rejection(&result, CapabilityReason::UnsupportedType);
fs::remove_dir_all(dir).unwrap();
}
#[test]
fn parquet_missing_nullable_columns_are_represented_as_native_typed_nulls() {
let dir = unique_temp_dir("algeon-datafusion-native-lowering-missing-nullable-columns");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan_with_missing_nullable_column(&path);
let scan_schema = scan.schema();
let group_by = PhysicalGroupBy::new_single(vec![(
col("missing_payload", &scan_schema).unwrap(),
"missing_payload".to_string(),
)]);
let sum = AggregateExprBuilder::new(sum_udaf(), vec![col("m00", &scan_schema).unwrap()])
.schema(Arc::clone(&scan_schema))
.alias("sum_m00")
.build()
.unwrap();
let aggregate: Arc<dyn ExecutionPlan> = Arc::new(
AggregateExec::try_new(AggregateMode::Single, group_by, vec![Arc::new(sum)], vec![None], scan, scan_schema)
.unwrap(),
);
let aggregate_schema = aggregate.schema();
let sort_exprs = LexOrdering::new([PhysicalSortExpr::new(
col("sum_m00", &aggregate_schema).unwrap(),
ArrowSortOptions::new(true, false),
)])
.unwrap();
let sort: Arc<dyn ExecutionPlan> = Arc::new(SortExec::new(sort_exprs, aggregate).with_fetch(Some(1)));
let plan = Arc::new(LocalLimitExec::new(sort, 1));
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.not_selected_count(), 0, "outcomes={:?}", result.report.outcomes());
assert_eq!(result.source_diagnostics.len(), 1);
assert_eq!(result.source_diagnostics[0].schema_mapping_disposition.as_str(), "missing_nullable_columns");
assert!(result.source_diagnostics[0].rejected_features.is_empty());
let source = native_local_parquet_source(result.selected_plan());
assert_eq!(source.missing_columns.len(), 1);
assert_eq!(source.missing_columns[0].name, "missing_payload");
assert_eq!(
source.projected_output_column_source_refs,
vec![SourceColumnRef::File(2), SourceColumnRef::MissingNull(0)]
);
fs::remove_dir_all(dir).unwrap();
}
#[test]
fn projection_over_repartitioned_fetch_limited_coalesce_lowers_with_its_limit() {
use datafusion::physical_plan::{Partitioning, coalesce_partitions::CoalescePartitionsExec};
let dir = unique_temp_dir("algeon-datafusion-native-lowering-repartitioned-coalesce-fetch");
let path = dir.join("input.parquet");
write_numeric_parquet_file(&path);
let scan = parquet_scan(&path);
let scan_schema = scan.schema();
let limited: Arc<dyn ExecutionPlan> = Arc::new(CoalescePartitionsExec::new(scan).with_fetch(Some(2)));
let repartitioned: Arc<dyn ExecutionPlan> =
Arc::new(RepartitionExec::try_new(limited, Partitioning::RoundRobinBatch(4)).unwrap());
let plan: Arc<dyn ExecutionPlan> = Arc::new(
ProjectionExec::try_new(
[ProjectionExpr { expr: col("payload", &scan_schema).unwrap(), alias: "payload".to_string() }],
repartitioned,
)
.unwrap(),
);
let result = lower_physical_plan_to_native(plan, &ParquetReadStrategyConfig::default()).unwrap();
assert_eq!(result.selected_count(), 1, "outcomes={:?}", result.report.outcomes());
let native = result.selected_plan().to_string();
assert!(native.contains("Limit"), "{native}");
fs::remove_dir_all(dir).unwrap();
}