feat: add conservative ORC predicate pushdown

crates/integration_tests/tests/read_tables.rs

@@ -96,8 +96,19 @@ async fn scan_and_read_with_fs_catalog(
 async fn scan_and_read_with_filter(
     table: &paimon::Table,
     filter: Predicate,
+) -> (Plan, Vec<RecordBatch>) {
+    scan_and_read_with_projection_and_filter(table, None, filter).await
+}
+
+async fn scan_and_read_with_projection_and_filter(
+    table: &paimon::Table,
+    projection: Option<&[&str]>,
+    filter: Predicate,
 ) -> (Plan, Vec<RecordBatch>) {
     let mut read_builder = table.new_read_builder();
+    if let Some(cols) = projection {
+        read_builder.with_projection(cols);
+    }
     read_builder.with_filter(filter);
     let scan = read_builder.new_scan();
     let plan = scan.plan().await.expect("Failed to plan scan");
@@ -2935,6 +2946,162 @@ async fn test_read_full_types_table() {
     assert_eq!(r.18, ("carol".into(), 300)); // struct
 }
 
+#[tokio::test]
+async fn test_read_orc_with_filter_only_column_projection() {
+    use paimon::spec::{Datum, PredicateBuilder};
+
+    let catalog = create_file_system_catalog();
+    let table = get_table_from_catalog(&catalog, "full_types_table").await;
+    let pb = PredicateBuilder::new(table.schema().fields());
+    let filter = pb
+        .equal("id", Datum::Int(2))
+        .expect("Failed to build id predicate");
+
+    let (_, batches) =
+        scan_and_read_with_projection_and_filter(&table, Some(&["col_string"]), filter).await;
+
+    let mut values = Vec::new();
+    for batch in &batches {
+        assert_eq!(batch.num_columns(), 1);
+        assert_eq!(batch.schema().field(0).name(), "col_string");
+        let col_string = batch
+            .column_by_name("col_string")
+            .and_then(|c| c.as_any().downcast_ref::<StringArray>())
+            .expect("Expected StringArray for col_string");
+        values.extend((0..batch.num_rows()).map(|row| col_string.value(row).to_string()));
+    }
+
+    assert_eq!(values, vec!["orc-world"]);
+}
+
+async fn assert_full_types_orc_filter_matches(
+    filter: Predicate,
+    projected_column: &str,
+    expected_string_values: &[&str],
+) {
+    let catalog = create_file_system_catalog();
+    let table = get_table_from_catalog(&catalog, "full_types_table").await;
+
+    let (_, batches) =
+        scan_and_read_with_projection_and_filter(&table, Some(&[projected_column]), filter).await;
+
+    let mut values = Vec::new();
+    for batch in &batches {
+        assert_eq!(batch.num_columns(), 1);
+        assert_eq!(batch.schema().field(0).name(), projected_column);
+        let column = batch
+            .column_by_name(projected_column)
+            .and_then(|c| c.as_any().downcast_ref::<StringArray>())
+            .expect("Expected StringArray for projected column");
+        values.extend((0..batch.num_rows()).map(|row| column.value(row).to_string()));
+    }
+
+    assert_eq!(values, expected_string_values);
+}
+
+#[tokio::test]
+async fn test_read_orc_with_supported_predicate_pushdown_types() {
+    use paimon::spec::{Datum, PredicateBuilder};
+
+    let catalog = create_file_system_catalog();
+    let table = get_table_from_catalog(&catalog, "full_types_table").await;
+    let pb = PredicateBuilder::new(table.schema().fields());
+
+    let cases = vec![
+        (
+            "col_boolean_eq",
+            pb.equal("col_boolean", Datum::Bool(false))
+                .expect("build boolean predicate"),
+            vec!["orc-world"],
+        ),
+        (
+            "col_tinyint_eq",
+            pb.equal("col_tinyint", Datum::TinyInt(2))
+                .expect("build tinyint predicate"),
+            vec!["orc-world"],
+        ),
+        (
+            "col_smallint_eq",
+            pb.equal("col_smallint", Datum::SmallInt(200))
+                .expect("build smallint predicate"),
+            vec!["orc-world"],
+        ),
+        (
+            "col_int_eq",
+            pb.equal("col_int", Datum::Int(2000))
+                .expect("build int predicate"),
+            vec!["orc-world"],
+        ),
+        (
+            "col_bigint_eq",
+            pb.equal("col_bigint", Datum::Long(200000))
+                .expect("build bigint predicate"),
+            vec!["orc-world"],
+        ),
+        (
+            "col_string_gte",
+            pb.greater_or_equal("col_string", Datum::String("orc-world".to_string()))
+                .expect("build string lower-bound predicate"),
+            vec!["parquet-hello", "orc-world"],
+        ),
+        (
+            "col_string_lte",
+            pb.less_or_equal("col_string", Datum::String("orc-world".to_string()))
+                .expect("build string upper-bound predicate"),
+            vec!["orc-world", "avro-test"],
+        ),
+    ];
+
+    for (case_name, filter, expected_string_values) in cases {
+        let (_, batches) =
+            scan_and_read_with_projection_and_filter(&table, Some(&["col_string"]), filter).await;
+
+        let mut values = Vec::new();
+        for batch in &batches {
+            assert_eq!(batch.num_columns(), 1);
+            assert_eq!(batch.schema().field(0).name(), "col_string");
+            let col_string = batch
+                .column_by_name("col_string")
+                .and_then(|c| c.as_any().downcast_ref::<StringArray>())
+                .expect("Expected StringArray for col_string");
+            values.extend((0..batch.num_rows()).map(|row| col_string.value(row).to_string()));
+        }
+
+        assert_eq!(values, expected_string_values, "case {case_name}");
+    }
+}
+
+#[tokio::test]
+async fn test_read_orc_with_unsupported_date_predicate_remains_residual() {
+    use paimon::spec::{Datum, PredicateBuilder};
+
+    let catalog = create_file_system_catalog();
+    let table = get_table_from_catalog(&catalog, "full_types_table").await;
+    let pb = PredicateBuilder::new(table.schema().fields());
+    let filter = pb
+        .greater_or_equal("col_date", Datum::Date(19889))
+        .expect("build date predicate");
+
+    assert_full_types_orc_filter_matches(filter, "col_string", &["orc-world", "avro-test"]).await;
+}
+
+#[tokio::test]
+async fn test_read_orc_predicate_pushdown_remains_conservative() {
+    use paimon::spec::{Datum, PredicateBuilder};
+
+    let catalog = create_file_system_catalog();
+    let table = get_table_from_catalog(&catalog, "full_types_table").await;
+    let pb = PredicateBuilder::new(table.schema().fields());
+    let filter = pb.equal("id", Datum::Int(2)).expect("build id predicate");
+
+    assert!(
+        !table.new_read_builder().is_exact_filter_pushdown(&filter),
+        "ORC reader pruning must not make data predicates exact at the table boundary"
+    );
+
+    assert_full_types_orc_filter_matches(filter, "col_string", &["orc-world"]).await;
+}
+
 #[tokio::test]
 async fn test_read_full_types_boundary_table() {
     use arrow_array::{

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

@@ -43,7 +43,8 @@ pub(crate) struct FilePredicates {
 ///
 /// Each implementation (Parquet, ORC, ...) handles:
 /// - Column projection
-/// - Predicate pushdown (row-group/stripe pruning + row-level filtering)
+/// - Predicate pushdown where supported (row-group/stripe pruning and, for
+///   some formats, row-level filtering)
 /// - Row range selection
 #[async_trait]
 pub(crate) trait FormatFileReader: Send + Sync {