Skip to content

Commit 65426ea

Browse files
adriangbadriangb
committed
Add tests
1 parent bd57e68 commit 65426ea

File tree

7 files changed

+252
-30
lines changed

7 files changed

+252
-30
lines changed

‎datafusion/common/src/scalar/mod.rs‎

Lines changed: 1 addition & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -3704,13 +3704,7 @@ impl ScalarValue {
37043704
}
37053705

37063706
let scalar_array = self.to_array()?;
3707-
// Use cast_column for struct types to handle field reordering by name
3708-
let cast_arr = if matches!(target_type, DataType::Struct(_)) {
3709-
let target_field = Field::new("", target_type.clone(), true);
3710-
crate::nested_struct::cast_column(&scalar_array, &target_field, cast_options)?
3711-
} else {
3712-
cast_with_options(&scalar_array, target_type, cast_options)?
3713-
};
3707+
let cast_arr = cast_with_options(&scalar_array, target_type, cast_options)?;
37143708
ScalarValue::try_from_array(&cast_arr, 0)
37153709
}
37163710

‎datafusion/core/src/datasource/physical_plan/parquet.rs‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -169,8 +169,7 @@ mod tests {
169169
if self.pushdown_predicate {
170170
source = source
171171
.with_pushdown_filters(true)
172-
.with_reorder_filters(true)
173-
.with_filter_effectiveness_threshold(0.0);
172+
.with_reorder_filters(true);
174173
} else {
175174
source = source.with_pushdown_filters(false);
176175
}

‎datafusion/core/tests/parquet/filter_pushdown.rs‎

Lines changed: 103 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,9 @@
2626
//! select * from data limit 10;
2727
//! ```
2828
29+
use arrow::array::{ArrayRef, Int32Array, Int64Array, StructArray};
2930
use arrow::compute::concat_batches;
31+
use arrow::datatypes::{DataType, Field, Fields, Schema};
3032
use arrow::record_batch::RecordBatch;
3133
use datafusion::physical_plan::collect;
3234
use datafusion::physical_plan::metrics::{MetricValue, MetricsSet};
@@ -35,7 +37,9 @@ use datafusion::prelude::{
3537
};
3638
use datafusion::test_util::parquet::{ParquetScanOptions, TestParquetFile};
3739
use datafusion_expr::utils::{conjunction, disjunction, split_conjunction};
40+
use datafusion_functions::expr_fn::get_field;
3841
use std::path::Path;
42+
use std::sync::Arc;
3943

4044
use datafusion_common::test_util::parquet_test_data;
4145
use datafusion_execution::config::SessionConfig;
@@ -731,3 +735,102 @@ impl PredicateCacheTest {
731735
Ok(())
732736
}
733737
}
738+
739+
/// Number of rows for struct filter pushdown tests
740+
const STRUCT_NUM_ROWS: usize = 4096;
741+
742+
/// Creates test batches with a struct column containing an `id` field.
743+
///
744+
/// Schema: struct_col { id: Int32 }, value: Int64
745+
/// Data: id values from 0 to STRUCT_NUM_ROWS-1, value same as id
746+
fn create_struct_test_batches() -> Vec<RecordBatch> {
747+
let struct_fields = Fields::from(vec![Field::new("id", DataType::Int32, false)]);
748+
let schema = Arc::new(Schema::new(vec![
749+
Field::new("struct_col", DataType::Struct(struct_fields.clone()), false),
750+
Field::new("value", DataType::Int64, false),
751+
]));
752+
753+
// Create the data: struct_col.id = [0, 1, 2, ...], value = [0, 1, 2, ...]
754+
let ids: Vec<i32> = (0..STRUCT_NUM_ROWS as i32).collect();
755+
let id_array = Int32Array::from(ids);
756+
let struct_array = StructArray::from(vec![(
757+
Arc::new(Field::new("id", DataType::Int32, false)),
758+
Arc::new(id_array) as ArrayRef,
759+
)]);
760+
761+
let values: Vec<i64> = (0..STRUCT_NUM_ROWS as i64).collect();
762+
let value_array = Int64Array::from(values);
763+
764+
let batch =
765+
RecordBatch::try_new(schema, vec![Arc::new(struct_array), Arc::new(value_array)])
766+
.unwrap();
767+
768+
vec![batch]
769+
}
770+
771+
/// Tests that filter pushdown works for struct column field access.
772+
///
773+
/// Verifies that:
774+
/// 1. Filters on struct fields (e.g., `struct_col['id'] = 500`) are pushed down
775+
/// 2. The `pushdown_rows_pruned` metric shows rows were actually pruned
776+
/// 3. The correct number of rows are returned
777+
#[tokio::test]
778+
async fn struct_filter_pushdown() {
779+
let batches = create_struct_test_batches();
780+
781+
// Set the row group size smaller so we can test row group pruning
782+
let props = WriterProperties::builder()
783+
.set_max_row_group_size(1024)
784+
.build();
785+
786+
let tempdir = TempDir::new_in(Path::new(".")).unwrap();
787+
788+
let test_parquet_file = TestParquetFile::try_new(
789+
tempdir.path().join("struct_data.parquet"),
790+
props,
791+
batches,
792+
)
793+
.unwrap();
794+
795+
// Test 1: Selective equality filter on struct field
796+
// struct_col['id'] = 500 should match exactly 1 row
797+
let case = TestCase::new(&test_parquet_file)
798+
.with_name("struct_field_equality")
799+
.with_filter(get_field(col("struct_col"), "id").eq(lit(500)))
800+
.with_pushdown_expected(PushdownExpected::Some)
801+
.with_expected_rows(1);
802+
case.run().await;
803+
804+
// Test 2: Comparison filter on struct field
805+
// struct_col['id'] > 3000 should match 1095 rows (4095 - 3000)
806+
let case = TestCase::new(&test_parquet_file)
807+
.with_name("struct_field_comparison")
808+
.with_filter(get_field(col("struct_col"), "id").gt(lit(3000)))
809+
.with_pushdown_expected(PushdownExpected::Some)
810+
.with_expected_rows(1095);
811+
case.run().await;
812+
813+
// Test 3: Conjunction filter with struct field
814+
// struct_col['id'] >= 1000 AND struct_col['id'] < 2000 should match 1000 rows
815+
let case = TestCase::new(&test_parquet_file)
816+
.with_name("struct_field_conjunction")
817+
.with_filter(
818+
conjunction([
819+
get_field(col("struct_col"), "id").gt_eq(lit(1000)),
820+
get_field(col("struct_col"), "id").lt(lit(2000)),
821+
])
822+
.unwrap(),
823+
)
824+
.with_pushdown_expected(PushdownExpected::Some)
825+
.with_expected_rows(1000);
826+
case.run().await;
827+
828+
// Test 4: Filter that matches nothing
829+
// struct_col['id'] = 9999 should match 0 rows (max id is 4095)
830+
let case = TestCase::new(&test_parquet_file)
831+
.with_name("struct_field_no_match")
832+
.with_filter(get_field(col("struct_col"), "id").eq(lit(9999)))
833+
.with_pushdown_expected(PushdownExpected::Some)
834+
.with_expected_rows(0);
835+
case.run().await;
836+
}

‎datafusion/expr-common/src/columnar_value.rs‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,15 +20,14 @@
2020
use arrow::{
2121
array::{Array, ArrayRef, Date32Array, Date64Array, NullArray},
2222
compute::{CastOptions, kernels, max, min},
23-
datatypes::{DataType, Field},
23+
datatypes::DataType,
2424
util::pretty::pretty_format_columns,
2525
};
2626
use datafusion_common::internal_datafusion_err;
2727
use datafusion_common::{
2828
Result, ScalarValue,
2929
format::DEFAULT_CAST_OPTIONS,
3030
internal_err,
31-
nested_struct,
3231
scalar::{date_to_timestamp_multiplier, ensure_timestamp_in_bounds},
3332
};
3433
use std::fmt;

‎datafusion/sqllogictest/test_files/parquet_filter_pushdown.slt‎

Lines changed: 136 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -721,3 +721,139 @@ from t1
721721
where struct_col['field'][1]['nested'] > 1;
722722
----
723723
2
724+
725+
###
726+
# Comprehensive struct column filter pushdown tests
727+
# Tests for various struct access patterns being pushed down to parquet
728+
###
729+
730+
statement ok
731+
COPY (
732+
SELECT * FROM VALUES
733+
({id: 1, name: 'alice', score: 75.0, status: 'active', optional_field: arrow_cast(10, 'Int64')},
734+
[{item_id: 1, value: 100.0}, {item_id: 2, value: 200.0}],
735+
{outer: {inner: 1}}),
736+
({id: 2, name: 'bob', score: 45.0, status: 'inactive', optional_field: arrow_cast(NULL, 'Int64')},
737+
[{item_id: 3, value: 300.0}, {item_id: 4, value: 400.0}],
738+
{outer: {inner: 2}}),
739+
({id: 3, name: 'charlie', score: 90.0, status: 'active', optional_field: arrow_cast(30, 'Int64')},
740+
[{item_id: 5, value: 500.0}],
741+
{outer: {inner: 3}})
742+
AS t(struct_col, list_of_structs, nested_struct)
743+
) TO 'test_files/scratch/parquet_filter_pushdown/struct_tests.parquet';
744+
745+
statement ok
746+
DROP TABLE IF EXISTS struct_test;
747+
748+
statement ok
749+
CREATE EXTERNAL TABLE struct_test
750+
STORED AS PARQUET
751+
LOCATION 'test_files/scratch/parquet_filter_pushdown/struct_tests.parquet';
752+
753+
# Category 1: Simple struct field access
754+
755+
# Test equality filter on struct field - verify pushdown in EXPLAIN
756+
query TT
757+
EXPLAIN SELECT struct_col FROM struct_test WHERE struct_col['id'] = 1;
758+
----
759+
logical_plan
760+
01)Filter: get_field(struct_test.struct_col, Utf8("id")) = Int64(1)
761+
02)--TableScan: struct_test projection=[struct_col], partial_filters=[get_field(struct_test.struct_col, Utf8("id")) = Int64(1)]
762+
physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/struct_tests.parquet]]}, projection=[struct_col], file_type=parquet, predicate=get_field(struct_col@0, id) = 1
763+
764+
query ?
765+
SELECT struct_col FROM struct_test WHERE struct_col['id'] = 1;
766+
----
767+
{id: 1, name: alice, score: 75.0, status: active, optional_field: 10}
768+
769+
# Test comparison filter on struct field
770+
query ?
771+
SELECT struct_col FROM struct_test WHERE struct_col['score'] > 50.0 ORDER BY struct_col['id'];
772+
----
773+
{id: 1, name: alice, score: 75.0, status: active, optional_field: 10}
774+
{id: 3, name: charlie, score: 90.0, status: active, optional_field: 30}
775+
776+
# Test string equality filter on struct field
777+
query ?
778+
SELECT struct_col FROM struct_test WHERE struct_col['status'] = 'active' ORDER BY struct_col['id'];
779+
----
780+
{id: 1, name: alice, score: 75.0, status: active, optional_field: 10}
781+
{id: 3, name: charlie, score: 90.0, status: active, optional_field: 30}
782+
783+
# Category 2: List of structs access
784+
785+
# Test list element struct field access - verify pushdown in EXPLAIN
786+
query TT
787+
EXPLAIN SELECT list_of_structs FROM struct_test WHERE list_of_structs[1]['item_id'] = 1;
788+
----
789+
logical_plan
790+
01)Filter: get_field(array_element(struct_test.list_of_structs, Int64(1)), Utf8("item_id")) = Int64(1)
791+
02)--TableScan: struct_test projection=[list_of_structs], partial_filters=[get_field(array_element(struct_test.list_of_structs, Int64(1)), Utf8("item_id")) = Int64(1)]
792+
physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/struct_tests.parquet]]}, projection=[list_of_structs], file_type=parquet, predicate=get_field(array_element(list_of_structs@1, 1), item_id) = 1
793+
794+
query ?
795+
SELECT list_of_structs FROM struct_test WHERE list_of_structs[1]['item_id'] = 1;
796+
----
797+
[{item_id: 1, value: 100.0}, {item_id: 2, value: 200.0}]
798+
799+
# Test list element struct field with comparison
800+
query ?
801+
SELECT list_of_structs FROM struct_test WHERE list_of_structs[1]['value'] > 200.0 ORDER BY struct_col['id'];
802+
----
803+
[{item_id: 3, value: 300.0}, {item_id: 4, value: 400.0}]
804+
[{item_id: 5, value: 500.0}]
805+
806+
# Category 3: Nested struct access
807+
808+
# Test nested struct field access - verify pushdown in EXPLAIN
809+
query TT
810+
EXPLAIN SELECT nested_struct FROM struct_test WHERE nested_struct['outer']['inner'] = 2;
811+
----
812+
logical_plan
813+
01)Filter: get_field(struct_test.nested_struct, Utf8("outer"), Utf8("inner")) = Int64(2)
814+
02)--TableScan: struct_test projection=[nested_struct], partial_filters=[get_field(struct_test.nested_struct, Utf8("outer"), Utf8("inner")) = Int64(2)]
815+
physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/struct_tests.parquet]]}, projection=[nested_struct], file_type=parquet, predicate=get_field(nested_struct@2, outer, inner) = 2
816+
817+
query ?
818+
SELECT nested_struct FROM struct_test WHERE nested_struct['outer']['inner'] = 2;
819+
----
820+
{outer: {inner: 2}}
821+
822+
# Category 4: Combined predicates with AND/OR
823+
824+
# Test AND with multiple struct field predicates
825+
query ?
826+
SELECT struct_col FROM struct_test
827+
WHERE struct_col['status'] = 'active' AND struct_col['score'] > 80.0;
828+
----
829+
{id: 3, name: charlie, score: 90.0, status: active, optional_field: 30}
830+
831+
# Test OR with struct field predicates
832+
query ?
833+
SELECT struct_col FROM struct_test
834+
WHERE struct_col['id'] = 1 OR struct_col['id'] = 3
835+
ORDER BY struct_col['id'];
836+
----
837+
{id: 1, name: alice, score: 75.0, status: active, optional_field: 10}
838+
{id: 3, name: charlie, score: 90.0, status: active, optional_field: 30}
839+
840+
# Category 5: NULL handling
841+
842+
# Test IS NULL filter on struct field
843+
query ?
844+
SELECT struct_col FROM struct_test WHERE struct_col['optional_field'] IS NULL;
845+
----
846+
{id: 2, name: bob, score: 45.0, status: inactive, optional_field: NULL}
847+
848+
# Test IS NOT NULL filter on struct field
849+
query ?
850+
SELECT struct_col FROM struct_test
851+
WHERE struct_col['optional_field'] IS NOT NULL
852+
ORDER BY struct_col['id'];
853+
----
854+
{id: 1, name: alice, score: 75.0, status: active, optional_field: 10}
855+
{id: 3, name: charlie, score: 90.0, status: active, optional_field: 30}
856+
857+
# Cleanup
858+
statement ok
859+
DROP TABLE struct_test;

‎datafusion/sqllogictest/test_files/spark/hash/crc32.slt‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -81,7 +81,7 @@ SELECT crc32(arrow_cast('Spark', 'BinaryView'));
8181
----
8282
1557323817
8383

84-
query I
84+
# Upstream arrow-rs issue: https://github.com/apache/arrow-rs/issues/8841
85+
# This should succeed after we receive the fix
86+
query error Arrow error: Compute error: Internal Error: Cannot cast BinaryView to BinaryArray of expected type
8587
select crc32(arrow_cast(null, 'Dictionary(Int32, Utf8)'))
86-
----
87-
NULL

‎datafusion/sqllogictest/test_files/struct.slt‎

Lines changed: 7 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -492,12 +492,9 @@ Struct("r": Utf8, "c": Float64)
492492
statement ok
493493
drop table t;
494494

495-
statement ok
495+
query error DataFusion error: Optimizer rule 'simplify_expressions' failed[\s\S]*Arrow error: Cast error: Cannot cast string 'a' to value of Float64 type
496496
create table t as values({r: 'a', c: 1}), ({c: 2.3, r: 'b'});
497497

498-
statement ok
499-
drop table t;
500-
501498
##################################
502499
## Test Coalesce with Struct
503500
##################################
@@ -563,12 +560,10 @@ create table t(a struct(r varchar, c int), b struct(r varchar, c float)) as valu
563560
(row('purple', 1), row('green', 2.3));
564561

565562
# out of order struct literal
566-
statement ok
563+
# TODO: This query should not fail
564+
statement error DataFusion error: Optimizer rule 'simplify_expressions' failed[\s\S]*Arrow error: Cast error: Cannot cast string 'b' to value of Int32 type
567565
create table t(a struct(r varchar, c int)) as values ({r: 'a', c: 1}), ({c: 2, r: 'b'});
568566

569-
statement ok
570-
drop table t;
571-
572567
##################################
573568
## Test Array of Struct
574569
##################################
@@ -578,11 +573,9 @@ select [{r: 'a', c: 1}, {r: 'b', c: 2}];
578573
----
579574
[{r: a, c: 1}, {r: b, c: 2}]
580575

581-
# out of order struct literal in array
582-
query ?
576+
# Can't create a list of struct with different field types
577+
query error
583578
select [{r: 'a', c: 1}, {c: 2, r: 'b'}];
584-
----
585-
[{c: 1, r: a}, {c: 2, r: b}]
586579

587580
statement ok
588581
create table t(a struct(r varchar, c int), b struct(r varchar, c float)) as values (row('a', 1), row('b', 2.3));
@@ -599,11 +592,9 @@ drop table t;
599592
statement ok
600593
create table t(a struct(r varchar, c int), b struct(c float, r varchar)) as values (row('a', 1), row(2.3, 'b'));
601594

602-
# create array with different struct field order
603-
query T
595+
# create array with different struct type is not valid
596+
query error
604597
select arrow_typeof([a, b]) from t;
605-
----
606-
List(Struct("c": Float32, "r": Utf8View))
607598

608599
statement ok
609600
drop table t;

0 commit comments

Comments
 (0)