Skip to content

Commit feca4dc

Browse files
authored
feat: add conservative ORC predicate pushdown (#388)
1 parent 4a73e4a commit feca4dc

4 files changed

Lines changed: 738 additions & 13 deletions

File tree

crates/integration_tests/tests/read_tables.rs

Lines changed: 167 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -96,8 +96,19 @@ async fn scan_and_read_with_fs_catalog(
9696
async fn scan_and_read_with_filter(
9797
table: &paimon::Table,
9898
filter: Predicate,
99+
) -> (Plan, Vec<RecordBatch>) {
100+
scan_and_read_with_projection_and_filter(table, None, filter).await
101+
}
102+
103+
async fn scan_and_read_with_projection_and_filter(
104+
table: &paimon::Table,
105+
projection: Option<&[&str]>,
106+
filter: Predicate,
99107
) -> (Plan, Vec<RecordBatch>) {
100108
let mut read_builder = table.new_read_builder();
109+
if let Some(cols) = projection {
110+
read_builder.with_projection(cols);
111+
}
101112
read_builder.with_filter(filter);
102113
let scan = read_builder.new_scan();
103114
let plan = scan.plan().await.expect("Failed to plan scan");
@@ -2935,6 +2946,162 @@ async fn test_read_full_types_table() {
29352946
assert_eq!(r.18, ("carol".into(), 300)); // struct
29362947
}
29372948

2949+
#[tokio::test]
2950+
async fn test_read_orc_with_filter_only_column_projection() {
2951+
use paimon::spec::{Datum, PredicateBuilder};
2952+
2953+
let catalog = create_file_system_catalog();
2954+
let table = get_table_from_catalog(&catalog, "full_types_table").await;
2955+
let pb = PredicateBuilder::new(table.schema().fields());
2956+
let filter = pb
2957+
.equal("id", Datum::Int(2))
2958+
.expect("Failed to build id predicate");
2959+
2960+
let (_, batches) =
2961+
scan_and_read_with_projection_and_filter(&table, Some(&["col_string"]), filter).await;
2962+
2963+
let mut values = Vec::new();
2964+
for batch in &batches {
2965+
assert_eq!(batch.num_columns(), 1);
2966+
assert_eq!(batch.schema().field(0).name(), "col_string");
2967+
let col_string = batch
2968+
.column_by_name("col_string")
2969+
.and_then(|c| c.as_any().downcast_ref::<StringArray>())
2970+
.expect("Expected StringArray for col_string");
2971+
values.extend((0..batch.num_rows()).map(|row| col_string.value(row).to_string()));
2972+
}
2973+
2974+
assert_eq!(values, vec!["orc-world"]);
2975+
}
2976+
2977+
async fn assert_full_types_orc_filter_matches(
2978+
filter: Predicate,
2979+
projected_column: &str,
2980+
expected_string_values: &[&str],
2981+
) {
2982+
let catalog = create_file_system_catalog();
2983+
let table = get_table_from_catalog(&catalog, "full_types_table").await;
2984+
2985+
let (_, batches) =
2986+
scan_and_read_with_projection_and_filter(&table, Some(&[projected_column]), filter).await;
2987+
2988+
let mut values = Vec::new();
2989+
for batch in &batches {
2990+
assert_eq!(batch.num_columns(), 1);
2991+
assert_eq!(batch.schema().field(0).name(), projected_column);
2992+
let column = batch
2993+
.column_by_name(projected_column)
2994+
.and_then(|c| c.as_any().downcast_ref::<StringArray>())
2995+
.expect("Expected StringArray for projected column");
2996+
values.extend((0..batch.num_rows()).map(|row| column.value(row).to_string()));
2997+
}
2998+
2999+
assert_eq!(values, expected_string_values);
3000+
}
3001+
3002+
#[tokio::test]
3003+
async fn test_read_orc_with_supported_predicate_pushdown_types() {
3004+
use paimon::spec::{Datum, PredicateBuilder};
3005+
3006+
let catalog = create_file_system_catalog();
3007+
let table = get_table_from_catalog(&catalog, "full_types_table").await;
3008+
let pb = PredicateBuilder::new(table.schema().fields());
3009+
3010+
let cases = vec![
3011+
(
3012+
"col_boolean_eq",
3013+
pb.equal("col_boolean", Datum::Bool(false))
3014+
.expect("build boolean predicate"),
3015+
vec!["orc-world"],
3016+
),
3017+
(
3018+
"col_tinyint_eq",
3019+
pb.equal("col_tinyint", Datum::TinyInt(2))
3020+
.expect("build tinyint predicate"),
3021+
vec!["orc-world"],
3022+
),
3023+
(
3024+
"col_smallint_eq",
3025+
pb.equal("col_smallint", Datum::SmallInt(200))
3026+
.expect("build smallint predicate"),
3027+
vec!["orc-world"],
3028+
),
3029+
(
3030+
"col_int_eq",
3031+
pb.equal("col_int", Datum::Int(2000))
3032+
.expect("build int predicate"),
3033+
vec!["orc-world"],
3034+
),
3035+
(
3036+
"col_bigint_eq",
3037+
pb.equal("col_bigint", Datum::Long(200000))
3038+
.expect("build bigint predicate"),
3039+
vec!["orc-world"],
3040+
),
3041+
(
3042+
"col_string_gte",
3043+
pb.greater_or_equal("col_string", Datum::String("orc-world".to_string()))
3044+
.expect("build string lower-bound predicate"),
3045+
vec!["parquet-hello", "orc-world"],
3046+
),
3047+
(
3048+
"col_string_lte",
3049+
pb.less_or_equal("col_string", Datum::String("orc-world".to_string()))
3050+
.expect("build string upper-bound predicate"),
3051+
vec!["orc-world", "avro-test"],
3052+
),
3053+
];
3054+
3055+
for (case_name, filter, expected_string_values) in cases {
3056+
let (_, batches) =
3057+
scan_and_read_with_projection_and_filter(&table, Some(&["col_string"]), filter).await;
3058+
3059+
let mut values = Vec::new();
3060+
for batch in &batches {
3061+
assert_eq!(batch.num_columns(), 1);
3062+
assert_eq!(batch.schema().field(0).name(), "col_string");
3063+
let col_string = batch
3064+
.column_by_name("col_string")
3065+
.and_then(|c| c.as_any().downcast_ref::<StringArray>())
3066+
.expect("Expected StringArray for col_string");
3067+
values.extend((0..batch.num_rows()).map(|row| col_string.value(row).to_string()));
3068+
}
3069+
3070+
assert_eq!(values, expected_string_values, "case {case_name}");
3071+
}
3072+
}
3073+
3074+
#[tokio::test]
3075+
async fn test_read_orc_with_unsupported_date_predicate_remains_residual() {
3076+
use paimon::spec::{Datum, PredicateBuilder};
3077+
3078+
let catalog = create_file_system_catalog();
3079+
let table = get_table_from_catalog(&catalog, "full_types_table").await;
3080+
let pb = PredicateBuilder::new(table.schema().fields());
3081+
let filter = pb
3082+
.greater_or_equal("col_date", Datum::Date(19889))
3083+
.expect("build date predicate");
3084+
3085+
assert_full_types_orc_filter_matches(filter, "col_string", &["orc-world", "avro-test"]).await;
3086+
}
3087+
3088+
#[tokio::test]
3089+
async fn test_read_orc_predicate_pushdown_remains_conservative() {
3090+
use paimon::spec::{Datum, PredicateBuilder};
3091+
3092+
let catalog = create_file_system_catalog();
3093+
let table = get_table_from_catalog(&catalog, "full_types_table").await;
3094+
let pb = PredicateBuilder::new(table.schema().fields());
3095+
let filter = pb.equal("id", Datum::Int(2)).expect("build id predicate");
3096+
3097+
assert!(
3098+
!table.new_read_builder().is_exact_filter_pushdown(&filter),
3099+
"ORC reader pruning must not make data predicates exact at the table boundary"
3100+
);
3101+
3102+
assert_full_types_orc_filter_matches(filter, "col_string", &["orc-world"]).await;
3103+
}
3104+
29383105
#[tokio::test]
29393106
async fn test_read_full_types_boundary_table() {
29403107
use arrow_array::{

crates/paimon/src/arrow/format/mod.rs

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,8 @@ pub(crate) struct FilePredicates {
4343
///
4444
/// Each implementation (Parquet, ORC, ...) handles:
4545
/// - Column projection
46-
/// - Predicate pushdown (row-group/stripe pruning + row-level filtering)
46+
/// - Predicate pushdown where supported (row-group/stripe pruning and, for
47+
/// some formats, row-level filtering)
4748
/// - Row range selection
4849
#[async_trait]
4950
pub(crate) trait FormatFileReader: Send + Sync {

0 commit comments

Comments
 (0)