You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/04/08 04:23:39 UTC
(paimon-rust) branch main updated: feat: abstract FormatFileReader and introduce ORC & avro readers (#225)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new dd71c58 feat: abstract FormatFileReader and introduce ORC & avro readers (#225)
dd71c58 is described below
commit dd71c58d842f8238d1c51a16c20e288a13aa87b0
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Wed Apr 8 12:23:33 2026 +0800
feat: abstract FormatFileReader and introduce ORC & avro readers (#225)
---
Cargo.toml | 1 +
crates/integration_tests/tests/read_tables.rs | 289 ++++++
.../datafusion/src/physical_plan/scan.rs | 3 +-
crates/paimon/Cargo.toml | 2 +
crates/paimon/src/arrow/filtering.rs | 91 ++
crates/paimon/src/arrow/format/avro.rs | 1001 +++++++++++++++++++
crates/paimon/src/arrow/format/mod.rs | 76 ++
crates/paimon/src/arrow/format/orc.rs | 195 ++++
crates/paimon/src/arrow/format/parquet.rs | 881 +++++++++++++++++
crates/paimon/src/arrow/mod.rs | 1 +
crates/paimon/src/arrow/reader.rs | 1040 +++-----------------
crates/paimon/src/spec/types.rs | 9 +
crates/paimon/src/table/read_builder.rs | 12 +-
crates/test_utils.rs | 4 +-
dev/spark/provision.py | 79 ++
15 files changed, 2749 insertions(+), 935 deletions(-)
diff --git a/Cargo.toml b/Cargo.toml
index cfba61f..c2dd6a5 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -30,6 +30,7 @@ rust-version = "1.86.0"
[workspace.dependencies]
arrow = "57.0"
arrow-array = { version = "57.0", features = ["ffi"] }
+arrow-buffer = "57.0"
arrow-schema = "57.0"
arrow-cast = "57.0"
arrow-ord = "57.0"
diff --git a/crates/integration_tests/tests/read_tables.rs b/crates/integration_tests/tests/read_tables.rs
index de7bd65..5e2ac48 100644
--- a/crates/integration_tests/tests/read_tables.rs
+++ b/crates/integration_tests/tests/read_tables.rs
@@ -2292,3 +2292,292 @@ async fn test_read_data_evolution_table_only_row_id_with_row_ranges() {
"Row range filtered count ({total_rows}) should be <= full count ({full_count})"
);
}
+
+// ---------------------------------------------------------------------------
+// Full types integration tests (parquet + orc + avro)
+// ---------------------------------------------------------------------------
+
+#[tokio::test]
+async fn test_read_full_types_table() {
+ use arrow_array::{
+ BinaryArray, BooleanArray, Date32Array, Decimal128Array, Float32Array, Float64Array,
+ Int16Array, Int64Array, Int8Array, ListArray, MapArray, StructArray,
+ TimestampMicrosecondArray,
+ };
+
+ let (_, batches) = scan_and_read_with_fs_catalog("full_types_table", None).await;
+
+ let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
+ assert_eq!(total_rows, 3, "full_types_table should have 3 rows");
+
+ // Collect all rows sorted by id.
+ // We verify primitive types via a tuple, then complex/extra decimal types separately.
+ #[allow(clippy::type_complexity)]
+ let mut rows: Vec<(
+ i32,
+ bool,
+ i8,
+ i16,
+ i32,
+ i64,
+ f32,
+ f64,
+ i128,
+ i128,
+ i128,
+ String,
+ Vec<u8>,
+ i32,
+ i64,
+ i64,
+ Vec<i32>,
+ Vec<(String, i32)>,
+ (String, i32),
+ )> = Vec::new();
+ for batch in &batches {
+ let id = batch
+ .column_by_name("id")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let col_boolean = batch
+ .column_by_name("col_boolean")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<BooleanArray>()
+ .unwrap();
+ let col_tinyint = batch
+ .column_by_name("col_tinyint")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int8Array>()
+ .unwrap();
+ let col_smallint = batch
+ .column_by_name("col_smallint")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int16Array>()
+ .unwrap();
+ let col_int = batch
+ .column_by_name("col_int")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let col_bigint = batch
+ .column_by_name("col_bigint")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ let col_float = batch
+ .column_by_name("col_float")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Float32Array>()
+ .unwrap();
+ let col_double = batch
+ .column_by_name("col_double")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Float64Array>()
+ .unwrap();
+ let col_decimal = batch
+ .column_by_name("col_decimal")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Decimal128Array>()
+ .unwrap();
+ let col_decimal5 = batch
+ .column_by_name("col_decimal5")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Decimal128Array>()
+ .unwrap();
+ let col_decimal38 = batch
+ .column_by_name("col_decimal38")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Decimal128Array>()
+ .unwrap();
+ let col_string = batch
+ .column_by_name("col_string")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ let col_binary = batch
+ .column_by_name("col_binary")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .unwrap();
+ let col_date = batch
+ .column_by_name("col_date")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Date32Array>()
+ .unwrap();
+ let col_ts = batch
+ .column_by_name("col_timestamp")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<TimestampMicrosecondArray>()
+ .unwrap();
+ let col_ts_ltz = batch
+ .column_by_name("col_timestamp_ltz")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<TimestampMicrosecondArray>()
+ .unwrap();
+ let col_array = batch
+ .column_by_name("col_array")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<ListArray>()
+ .unwrap();
+ let col_map = batch
+ .column_by_name("col_map")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<MapArray>()
+ .unwrap();
+ let col_struct = batch
+ .column_by_name("col_struct")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+
+ for i in 0..batch.num_rows() {
+ // Extract ARRAY<INT>
+ let list_values = col_array.value(i);
+ let int_arr = list_values.as_any().downcast_ref::<Int32Array>().unwrap();
+ let arr_vals: Vec<i32> = (0..int_arr.len()).map(|j| int_arr.value(j)).collect();
+
+ // Extract MAP<STRING, INT>
+ let map_val = col_map.value(i);
+ let map_struct = map_val.as_any().downcast_ref::<StructArray>().unwrap();
+ let keys = map_struct
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ let values = map_struct
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let mut map_entries: Vec<(String, i32)> = (0..keys.len())
+ .map(|j| (keys.value(j).to_string(), values.value(j)))
+ .collect();
+ map_entries.sort_by(|a, b| a.0.cmp(&b.0));
+
+ // Extract STRUCT<name: STRING, value: INT>
+ let struct_name = col_struct
+ .column_by_name("name")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ let struct_value = col_struct
+ .column_by_name("value")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+
+ rows.push((
+ id.value(i),
+ col_boolean.value(i),
+ col_tinyint.value(i),
+ col_smallint.value(i),
+ col_int.value(i),
+ col_bigint.value(i),
+ col_float.value(i),
+ col_double.value(i),
+ col_decimal.value(i),
+ col_decimal5.value(i),
+ col_decimal38.value(i),
+ col_string.value(i).to_string(),
+ col_binary.value(i).to_vec(),
+ col_date.value(i),
+ col_ts.value(i),
+ col_ts_ltz.value(i),
+ arr_vals,
+ map_entries,
+ (struct_name.value(i).to_string(), struct_value.value(i)),
+ ));
+ }
+ }
+ rows.sort_by_key(|r| r.0);
+
+ assert_eq!(rows.len(), 3);
+
+ // id=1 (parquet)
+ let r = &rows[0];
+ assert_eq!(r.0, 1);
+ assert!(r.1); // boolean = true
+ assert_eq!(r.2, 1i8); // tinyint
+ assert_eq!(r.3, 100i16); // smallint
+ assert_eq!(r.4, 1000); // int
+ assert_eq!(r.5, 100000i64); // bigint
+ assert!((r.6 - 1.5f32).abs() < f32::EPSILON); // float
+ assert!((r.7 - 2.5f64).abs() < f64::EPSILON); // double
+ assert_eq!(r.8, 12345); // decimal(10,2) = 123.45 * 100
+ assert_eq!(r.9, 12345); // decimal(5,0) = 12345
+ assert_eq!(r.10, 12_345_678_901_234_567_890_000i128); // decimal(38,18) 12345.678901234567890 * 10^18
+ assert_eq!(r.11, "parquet-hello"); // string
+ assert_eq!(r.12, vec![0xDE, 0xAD, 0xBE, 0xEF]); // binary
+ assert_eq!(r.13, 19723); // date: 2024-01-01 days since epoch
+ assert_eq!(r.14, 1_704_103_200_123_456); // ts micros
+ assert_eq!(r.15, 1_704_103_200_123_456); // ts_ltz micros
+ assert_eq!(r.16, vec![1, 2, 3]); // array
+ assert_eq!(r.17, vec![("a".into(), 10), ("b".into(), 20)]); // map
+ assert_eq!(r.18, ("alice".into(), 100)); // struct
+
+ // id=2 (orc)
+ let r = &rows[1];
+ assert_eq!(r.0, 2);
+ assert!(!r.1); // boolean = false
+ assert_eq!(r.2, 2i8);
+ assert_eq!(r.3, 200i16);
+ assert_eq!(r.4, 2000);
+ assert_eq!(r.5, 200000i64);
+ assert!((r.6 - 3.5f32).abs() < f32::EPSILON);
+ assert!((r.7 - 4.5f64).abs() < f64::EPSILON);
+ assert_eq!(r.8, 67890); // 678.90 * 100
+ assert_eq!(r.9, 99999); // decimal(5,0)
+ assert_eq!(r.10, 99_999_999_999_999_999_999_000i128); // decimal(38,18) 99999.999999999999999 * 10^18
+ assert_eq!(r.11, "orc-world");
+ assert_eq!(r.12, vec![0xCA, 0xFE, 0xBA, 0xBE]);
+ assert_eq!(r.13, 19889); // date: 2024-06-15 days since epoch
+ assert_eq!(r.14, 1_718_454_600_456_789);
+ assert_eq!(r.15, 1_718_454_600_456_789);
+ assert_eq!(r.16, vec![4, 5]); // array
+ assert_eq!(r.17, vec![("c".into(), 30)]); // map
+ assert_eq!(r.18, ("bob".into(), 200)); // struct
+
+ // id=3 (avro)
+ let r = &rows[2];
+ assert_eq!(r.0, 3);
+ assert!(r.1); // boolean = true
+ assert_eq!(r.2, 3i8);
+ assert_eq!(r.3, 300i16);
+ assert_eq!(r.4, 3000);
+ assert_eq!(r.5, 300000i64);
+ assert!((r.6 - 5.5f32).abs() < f32::EPSILON);
+ assert!((r.7 - 6.5f64).abs() < f64::EPSILON);
+ assert_eq!(r.8, 99999); // 999.99 * 100
+ assert_eq!(r.9, 0); // decimal(5,0)
+ assert_eq!(r.10, 1); // decimal(38,18) = 0.000000000000000001 * 10^18
+ assert_eq!(r.11, "avro-test");
+ assert_eq!(r.12, vec![0x01, 0x02, 0x03, 0x04]);
+ assert_eq!(r.13, 20453); // date: 2025-12-31 days since epoch
+ assert_eq!(r.14, 1_767_225_599_999_999);
+ assert_eq!(r.15, 1_767_225_599_999_999);
+ assert_eq!(r.16, vec![6]); // array
+ assert_eq!(r.17, vec![("d".into(), 40), ("e".into(), 50)]); // map
+ assert_eq!(r.18, ("carol".into(), 300)); // struct
+}
diff --git a/crates/integrations/datafusion/src/physical_plan/scan.rs b/crates/integrations/datafusion/src/physical_plan/scan.rs
index 1a181eb..57ecfa4 100644
--- a/crates/integrations/datafusion/src/physical_plan/scan.rs
+++ b/crates/integrations/datafusion/src/physical_plan/scan.rs
@@ -313,6 +313,7 @@ mod tests {
vec![("id", vec![1, 2, 3, 4]), ("value", vec![5, 20, 30, 40])],
Some(2),
);
+ let file_size = fs::metadata(bucket_dir.join("data.parquet")).unwrap().len() as i64;
let file_io = FileIOBuilder::new("file").build().unwrap();
let table_schema = TableSchema::new(
@@ -336,7 +337,7 @@ mod tests {
.with_bucket(0)
.with_bucket_path(local_file_path(&bucket_dir))
.with_total_buckets(1)
- .with_data_files(vec![test_data_file("data.parquet", 4)])
+ .with_data_files(vec![test_data_file("data.parquet", 4, file_size)])
.with_raw_convertible(true)
.build()
.unwrap();
diff --git a/crates/paimon/Cargo.toml b/crates/paimon/Cargo.toml
index d5d75ae..4f81fdc 100644
--- a/crates/paimon/Cargo.toml
+++ b/crates/paimon/Cargo.toml
@@ -56,11 +56,13 @@ serde_avro_fast = { version = "2.0.2", features = ["snappy", "zstandard"] }
indexmap = "2.5.0"
roaring = "0.11"
arrow-array = { workspace = true }
+arrow-buffer = { workspace = true }
arrow-cast = { workspace = true }
arrow-ord = { workspace = true }
arrow-schema = { workspace = true }
futures = "0.3"
parquet = { workspace = true, features = ["async", "zstd"] }
+orc-rust = "0.7.0"
async-stream = "0.3.6"
reqwest = { version = "0.12", features = ["json"] }
# DLF authentication dependencies
diff --git a/crates/paimon/src/arrow/filtering.rs b/crates/paimon/src/arrow/filtering.rs
index da2da14..f88c59b 100644
--- a/crates/paimon/src/arrow/filtering.rs
+++ b/crates/paimon/src/arrow/filtering.rs
@@ -26,6 +26,97 @@ pub(crate) fn reader_pruning_predicates(data_predicates: Vec<Predicate>) -> Vec<
.collect()
}
+/// Remap predicates from table-level indices to file-level indices.
+/// Predicates referencing fields not present in the file are resolved based on
+/// NULL semantics: the missing column is treated as all-NULL, so `IS NULL`
+/// becomes `AlwaysTrue` and all other operators become `AlwaysFalse`.
+pub(crate) fn remap_predicates_to_file(
+ predicates: &[Predicate],
+ table_fields: &[DataField],
+ file_fields: &[DataField],
+) -> Vec<Predicate> {
+ let mapping = build_field_mapping(table_fields, file_fields);
+ predicates
+ .iter()
+ .map(|p| remap_predicate(p, &mapping))
+ .collect()
+}
+
+fn remap_predicate(predicate: &Predicate, mapping: &[Option<usize>]) -> Predicate {
+ match predicate {
+ Predicate::Leaf {
+ column,
+ index,
+ data_type,
+ op,
+ literals,
+ } => {
+ match mapping.get(*index).copied().flatten() {
+ Some(file_index) => Predicate::Leaf {
+ column: column.clone(),
+ index: file_index,
+ data_type: data_type.clone(),
+ op: *op,
+ literals: literals.clone(),
+ },
+ // Column missing from file → all values are NULL.
+ None => match op {
+ PredicateOperator::IsNull => Predicate::AlwaysTrue,
+ _ => Predicate::AlwaysFalse,
+ },
+ }
+ }
+ Predicate::And(children) => {
+ let remapped: Vec<_> = children
+ .iter()
+ .map(|c| remap_predicate(c, mapping))
+ .collect();
+ if remapped.iter().any(|p| matches!(p, Predicate::AlwaysFalse)) {
+ Predicate::AlwaysFalse
+ } else {
+ let filtered: Vec<_> = remapped
+ .into_iter()
+ .filter(|p| !matches!(p, Predicate::AlwaysTrue))
+ .collect();
+ match filtered.len() {
+ 0 => Predicate::AlwaysTrue,
+ 1 => filtered.into_iter().next().unwrap(),
+ _ => Predicate::and(filtered),
+ }
+ }
+ }
+ Predicate::Or(children) => {
+ let remapped: Vec<_> = children
+ .iter()
+ .map(|c| remap_predicate(c, mapping))
+ .collect();
+ if remapped.iter().any(|p| matches!(p, Predicate::AlwaysTrue)) {
+ Predicate::AlwaysTrue
+ } else {
+ let filtered: Vec<_> = remapped
+ .into_iter()
+ .filter(|p| !matches!(p, Predicate::AlwaysFalse))
+ .collect();
+ match filtered.len() {
+ 0 => Predicate::AlwaysFalse,
+ 1 => filtered.into_iter().next().unwrap(),
+ _ => Predicate::or(filtered),
+ }
+ }
+ }
+ Predicate::Not(inner) => {
+ let remapped = remap_predicate(inner, mapping);
+ match remapped {
+ Predicate::AlwaysTrue => Predicate::AlwaysFalse,
+ Predicate::AlwaysFalse => Predicate::AlwaysTrue,
+ other => Predicate::Not(Box::new(other)),
+ }
+ }
+ Predicate::AlwaysTrue => Predicate::AlwaysTrue,
+ Predicate::AlwaysFalse => Predicate::AlwaysFalse,
+ }
+}
+
pub(crate) fn build_field_mapping(
table_fields: &[DataField],
file_fields: &[DataField],
diff --git a/crates/paimon/src/arrow/format/avro.rs b/crates/paimon/src/arrow/format/avro.rs
new file mode 100644
index 0000000..08c9aba
--- /dev/null
+++ b/crates/paimon/src/arrow/format/avro.rs
@@ -0,0 +1,1001 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use super::{FilePredicates, FormatFileReader};
+use crate::arrow::build_target_arrow_schema;
+use crate::io::FileRead;
+use crate::spec::{DataField, DataType, MapType, RowType};
+use crate::table::{ArrowRecordBatchStream, RowRange};
+use crate::Error;
+use arrow_array::{
+ BinaryArray, BooleanArray, Date32Array, Decimal128Array, Float32Array, Float64Array,
+ Int16Array, Int32Array, Int64Array, Int8Array, ListArray, MapArray, RecordBatch, StringArray,
+ StructArray, TimestampMicrosecondArray, TimestampMillisecondArray, TimestampNanosecondArray,
+};
+use arrow_buffer::{BooleanBuffer, NullBuffer, OffsetBuffer, ScalarBuffer};
+use arrow_schema::SchemaRef;
+use async_stream::try_stream;
+use async_trait::async_trait;
+use futures::StreamExt;
+use std::collections::HashMap;
+use std::sync::Arc;
+
+// ---------------------------------------------------------------------------
+// AvroValue: a serde_json::Value replacement that handles Avro bytes
+// ---------------------------------------------------------------------------
+
+/// Lightweight value type that can represent all Avro primitives including bytes.
+/// `serde_json::Value` rejects byte arrays, so we need our own.
+#[derive(Debug, Clone, PartialEq)]
+enum AvroValue {
+ Null,
+ Bool(bool),
+ Int(i64),
+ Float(f64),
+ String(String),
+ Bytes(Vec<u8>),
+ /// Avro array / sequence.
+ Array(Vec<AvroValue>),
+ /// Union wrapper or record: `{"type": value}` produced by serde_avro_fast.
+ Object(HashMap<String, AvroValue>),
+}
+
+impl AvroValue {
+ fn as_bool(&self) -> Option<bool> {
+ match self {
+ AvroValue::Bool(b) => Some(*b),
+ _ => None,
+ }
+ }
+
+ fn as_i64(&self) -> Option<i64> {
+ match self {
+ AvroValue::Int(n) => Some(*n),
+ _ => None,
+ }
+ }
+
+ fn as_f64(&self) -> Option<f64> {
+ match self {
+ AvroValue::Float(f) => Some(*f),
+ AvroValue::Int(n) => Some(*n as f64),
+ _ => None,
+ }
+ }
+
+ fn as_str(&self) -> Option<&str> {
+ match self {
+ AvroValue::String(s) => Some(s),
+ _ => None,
+ }
+ }
+
+ fn as_bytes(&self) -> Option<&[u8]> {
+ match self {
+ AvroValue::Bytes(b) => Some(b),
+ AvroValue::String(s) => Some(s.as_bytes()),
+ _ => None,
+ }
+ }
+}
+
+impl<'de> serde::Deserialize<'de> for AvroValue {
+ fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ struct AvroValueVisitor;
+
+ impl<'de> serde::de::Visitor<'de> for AvroValueVisitor {
+ type Value = AvroValue;
+
+ fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+ f.write_str("any Avro value")
+ }
+
+ fn visit_bool<E>(self, v: bool) -> Result<AvroValue, E> {
+ Ok(AvroValue::Bool(v))
+ }
+
+ fn visit_i8<E>(self, v: i8) -> Result<AvroValue, E> {
+ Ok(AvroValue::Int(v as i64))
+ }
+
+ fn visit_i16<E>(self, v: i16) -> Result<AvroValue, E> {
+ Ok(AvroValue::Int(v as i64))
+ }
+
+ fn visit_i32<E>(self, v: i32) -> Result<AvroValue, E> {
+ Ok(AvroValue::Int(v as i64))
+ }
+
+ fn visit_i64<E>(self, v: i64) -> Result<AvroValue, E> {
+ Ok(AvroValue::Int(v))
+ }
+
+ fn visit_u64<E>(self, v: u64) -> Result<AvroValue, E> {
+ Ok(AvroValue::Int(v as i64))
+ }
+
+ fn visit_f32<E>(self, v: f32) -> Result<AvroValue, E> {
+ Ok(AvroValue::Float(v as f64))
+ }
+
+ fn visit_f64<E>(self, v: f64) -> Result<AvroValue, E> {
+ Ok(AvroValue::Float(v))
+ }
+
+ fn visit_str<E>(self, v: &str) -> Result<AvroValue, E> {
+ Ok(AvroValue::String(v.to_owned()))
+ }
+
+ fn visit_string<E>(self, v: String) -> Result<AvroValue, E> {
+ Ok(AvroValue::String(v))
+ }
+
+ fn visit_bytes<E>(self, v: &[u8]) -> Result<AvroValue, E> {
+ Ok(AvroValue::Bytes(v.to_vec()))
+ }
+
+ fn visit_byte_buf<E>(self, v: Vec<u8>) -> Result<AvroValue, E> {
+ Ok(AvroValue::Bytes(v))
+ }
+
+ fn visit_none<E>(self) -> Result<AvroValue, E> {
+ Ok(AvroValue::Null)
+ }
+
+ fn visit_unit<E>(self) -> Result<AvroValue, E> {
+ Ok(AvroValue::Null)
+ }
+
+ fn visit_map<A>(self, mut map: A) -> Result<AvroValue, A::Error>
+ where
+ A: serde::de::MapAccess<'de>,
+ {
+ let mut m = HashMap::new();
+ while let Some((k, v)) = map.next_entry::<String, AvroValue>()? {
+ m.insert(k, v);
+ }
+ Ok(AvroValue::Object(m))
+ }
+
+ fn visit_seq<A>(self, mut seq: A) -> Result<AvroValue, A::Error>
+ where
+ A: serde::de::SeqAccess<'de>,
+ {
+ let mut v = Vec::new();
+ while let Some(elem) = seq.next_element::<AvroValue>()? {
+ v.push(elem);
+ }
+ Ok(AvroValue::Array(v))
+ }
+ }
+
+ deserializer.deserialize_any(AvroValueVisitor)
+ }
+}
+
+pub(crate) struct AvroFormatReader;
+
+const DEFAULT_BATCH_SIZE: usize = 8192;
+
+#[async_trait]
+impl FormatFileReader for AvroFormatReader {
+ async fn read_batch_stream(
+ &self,
+ reader: Box<dyn FileRead>,
+ file_size: u64,
+ read_fields: &[DataField],
+ _predicates: Option<&FilePredicates>,
+ batch_size: Option<usize>,
+ row_selection: Option<Vec<RowRange>>,
+ ) -> crate::Result<ArrowRecordBatchStream> {
+ // NOTE: Avro OCF requires sequential reading, so we load the entire file into memory.
+ // This is fine for typical Paimon data files but may be problematic for very large files.
+ let file_bytes = reader.read(0..file_size).await?;
+
+ let read_fields = read_fields.to_vec();
+ let target_schema = build_target_arrow_schema(&read_fields)?;
+ let batch_size = batch_size.unwrap_or(DEFAULT_BATCH_SIZE);
+
+ // Deserialize all Avro records from the OCF file.
+ let mut reader =
+ serde_avro_fast::object_container_file_encoding::Reader::from_slice(&file_bytes)
+ .map_err(|e| Error::UnexpectedError {
+ message: format!("Failed to open Avro file: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+
+ let mut all_records: Vec<HashMap<String, AvroValue>> = Vec::new();
+ for result in reader.deserialize_borrowed::<HashMap<String, AvroValue>>() {
+ let record = result.map_err(|e| Error::UnexpectedError {
+ message: format!("Failed to deserialize Avro record: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ all_records.push(record);
+ }
+
+ // Apply row selection filtering.
+ let records: Vec<HashMap<String, AvroValue>> = match row_selection {
+ Some(ref ranges) => {
+ let total_rows = all_records.len();
+ let mask = ranges_to_mask(total_rows, ranges);
+ all_records
+ .into_iter()
+ .enumerate()
+ .filter(|(i, _)| mask[*i])
+ .map(|(_, r)| r)
+ .collect()
+ }
+ None => all_records,
+ };
+
+ Ok(try_stream! {
+ for chunk in records.chunks(batch_size) {
+ let batch = records_to_batch(chunk, &read_fields, &target_schema)?;
+ yield batch;
+ }
+ }
+ .boxed())
+ }
+}
+
+// ---------------------------------------------------------------------------
+// Row ranges → boolean mask
+// ---------------------------------------------------------------------------
+
+fn ranges_to_mask(total_rows: usize, ranges: &[RowRange]) -> Vec<bool> {
+ let mut mask = vec![false; total_rows];
+ if total_rows == 0 {
+ return mask;
+ }
+ let file_end = total_rows as i64 - 1;
+ for r in ranges {
+ let from = r.from().max(0) as usize;
+ let to = (r.to().min(file_end) as usize).min(total_rows - 1);
+ for item in mask.iter_mut().take(to + 1).skip(from) {
+ *item = true;
+ }
+ }
+ mask
+}
+
+// ---------------------------------------------------------------------------
+// Avro records → Arrow RecordBatch conversion
+// ---------------------------------------------------------------------------
+
+fn records_to_batch(
+ records: &[HashMap<String, AvroValue>],
+ fields: &[DataField],
+ schema: &SchemaRef,
+) -> crate::Result<RecordBatch> {
+ let num_rows = records.len();
+ let mut columns: Vec<Arc<dyn arrow_array::Array>> = Vec::with_capacity(fields.len());
+
+ for field in fields {
+ let col = build_column(records, field.name(), field.data_type(), num_rows)?;
+ columns.push(col);
+ }
+
+ if columns.is_empty() {
+ RecordBatch::try_new_with_options(
+ schema.clone(),
+ columns,
+ &arrow_array::RecordBatchOptions::new().with_row_count(Some(num_rows)),
+ )
+ } else {
+ RecordBatch::try_new(schema.clone(), columns)
+ }
+ .map_err(|e| Error::UnexpectedError {
+ message: format!("Failed to build Avro RecordBatch: {e}"),
+ source: Some(Box::new(e)),
+ })
+}
+
+fn build_column(
+ records: &[HashMap<String, AvroValue>],
+ name: &str,
+ data_type: &DataType,
+ num_rows: usize,
+) -> crate::Result<Arc<dyn arrow_array::Array>> {
+ Ok(match data_type {
+ DataType::Boolean(_) => {
+ let arr: BooleanArray = (0..num_rows)
+ .map(|i| get_field(&records[i], name).and_then(|v| v.as_bool()))
+ .collect();
+ Arc::new(arr)
+ }
+ DataType::TinyInt(_) => {
+ let arr: Int8Array = (0..num_rows)
+ .map(|i| {
+ get_field(&records[i], name)
+ .and_then(|v| v.as_i64())
+ .map(|v| v as i8)
+ })
+ .collect();
+ Arc::new(arr)
+ }
+ DataType::SmallInt(_) => {
+ let arr: Int16Array = (0..num_rows)
+ .map(|i| {
+ get_field(&records[i], name)
+ .and_then(|v| v.as_i64())
+ .map(|v| v as i16)
+ })
+ .collect();
+ Arc::new(arr)
+ }
+ DataType::Int(_) => {
+ let arr: Int32Array = (0..num_rows)
+ .map(|i| {
+ get_field(&records[i], name)
+ .and_then(|v| v.as_i64())
+ .map(|v| v as i32)
+ })
+ .collect();
+ Arc::new(arr)
+ }
+ DataType::BigInt(_) => {
+ let arr: Int64Array = (0..num_rows)
+ .map(|i| get_field(&records[i], name).and_then(|v| v.as_i64()))
+ .collect();
+ Arc::new(arr)
+ }
+ DataType::Float(_) => {
+ let arr: Float32Array = (0..num_rows)
+ .map(|i| {
+ get_field(&records[i], name)
+ .and_then(|v| v.as_f64())
+ .map(|v| v as f32)
+ })
+ .collect();
+ Arc::new(arr)
+ }
+ DataType::Double(_) => {
+ let arr: Float64Array = (0..num_rows)
+ .map(|i| get_field(&records[i], name).and_then(|v| v.as_f64()))
+ .collect();
+ Arc::new(arr)
+ }
+ DataType::Char(_) | DataType::VarChar(_) => {
+ let arr: StringArray = (0..num_rows)
+ .map(|i| get_field(&records[i], name).and_then(|v| v.as_str()))
+ .collect();
+ Arc::new(arr)
+ }
+ DataType::Binary(_) | DataType::VarBinary(_) => {
+ let values: Vec<Option<&[u8]>> = (0..num_rows)
+ .map(|i| get_field(&records[i], name).and_then(|v| v.as_bytes()))
+ .collect();
+ let arr: BinaryArray = values.into_iter().collect();
+ Arc::new(arr)
+ }
+ DataType::Date(_) => {
+ let arr: Date32Array = (0..num_rows)
+ .map(|i| {
+ get_field(&records[i], name)
+ .and_then(|v| v.as_i64())
+ .map(|v| v as i32)
+ })
+ .collect();
+ Arc::new(arr)
+ }
+ DataType::Decimal(d) => {
+ let precision = u8::try_from(d.precision()).map_err(|_| Error::Unsupported {
+ message: "Decimal precision exceeds u8".to_string(),
+ })?;
+ let scale = i8::try_from(d.scale() as i32).map_err(|_| Error::Unsupported {
+ message: "Decimal scale out of i8 range".to_string(),
+ })?;
+ let arr: Decimal128Array = (0..num_rows)
+ .map(|i| {
+ get_field(&records[i], name).and_then(|v| match v {
+ // Avro decimal is encoded as big-endian two's complement bytes.
+ AvroValue::Bytes(b) => Some(bytes_to_i128_be(b)),
+ AvroValue::Int(n) => Some(*n as i128),
+ // serde_avro_fast may deserialize decimal as string representation.
+ AvroValue::String(s) => parse_decimal_string(s, scale),
+ _ => None,
+ })
+ })
+ .collect::<Decimal128Array>()
+ .with_precision_and_scale(precision, scale)
+ .map_err(|e| Error::UnexpectedError {
+ message: format!("Failed to build Decimal128Array: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ Arc::new(arr)
+ }
+ DataType::Timestamp(t) => {
+ build_timestamp_column(records, name, num_rows, t.precision(), None)
+ }
+ DataType::LocalZonedTimestamp(t) => build_timestamp_column(
+ records,
+ name,
+ num_rows,
+ t.precision(),
+ Some(Arc::from("UTC")),
+ ),
+ DataType::Array(arr_type) => {
+ build_array_column(records, name, arr_type.element_type(), num_rows)?
+ }
+ DataType::Map(map_type) => build_map_column(records, name, map_type, num_rows)?,
+ DataType::Row(row_type) => build_row_column(records, name, row_type, num_rows)?,
+ other => {
+ return Err(Error::Unsupported {
+ message: format!("Avro reader does not support data type: {other:?}"),
+ });
+ }
+ })
+}
+
+fn build_timestamp_column(
+ records: &[HashMap<String, AvroValue>],
+ name: &str,
+ num_rows: usize,
+ precision: u32,
+ tz: Option<Arc<str>>,
+) -> Arc<dyn arrow_array::Array> {
+ let values: Vec<Option<i64>> = (0..num_rows)
+ .map(|i| get_field(&records[i], name).and_then(|v| v.as_i64()))
+ .collect();
+ match precision {
+ 0..=3 => Arc::new(TimestampMillisecondArray::from(values).with_timezone_opt(tz)),
+ 4..=6 => Arc::new(TimestampMicrosecondArray::from(values).with_timezone_opt(tz)),
+ _ => Arc::new(TimestampNanosecondArray::from(values).with_timezone_opt(tz)),
+ }
+}
+
+fn build_array_column(
+ records: &[HashMap<String, AvroValue>],
+ name: &str,
+ element_type: &DataType,
+ num_rows: usize,
+) -> crate::Result<Arc<dyn arrow_array::Array>> {
+ let arrow_element_type = crate::arrow::paimon_type_to_arrow(element_type)?;
+ let arrow_element_field =
+ arrow_schema::Field::new("element", arrow_element_type, element_type.is_nullable());
+
+ let mut offsets = vec![0i32];
+ let mut element_records: Vec<HashMap<String, AvroValue>> = Vec::new();
+
+ for record in records.iter().take(num_rows) {
+ match get_field(record, name) {
+ Some(AvroValue::Array(arr)) => {
+ for elem in arr {
+ let mut m = HashMap::new();
+ m.insert("element".to_string(), elem.clone());
+ element_records.push(m);
+ }
+ offsets.push(offsets.last().unwrap() + arr.len() as i32);
+ }
+ _ => {
+ offsets.push(*offsets.last().unwrap());
+ }
+ }
+ }
+
+ let element_col = build_column(
+ &element_records,
+ "element",
+ element_type,
+ element_records.len(),
+ )?;
+
+ let offsets_buf = OffsetBuffer::new(ScalarBuffer::from(offsets));
+ let nulls = NullBuffer::new(BooleanBuffer::from(
+ (0..num_rows)
+ .map(|i| get_field(&records[i], name).is_some())
+ .collect::<Vec<_>>(),
+ ));
+
+ let list_arr = ListArray::try_new(
+ Arc::new(arrow_element_field),
+ offsets_buf,
+ element_col,
+ Some(nulls),
+ )
+ .map_err(|e| Error::UnexpectedError {
+ message: format!("Failed to build ListArray: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ Ok(Arc::new(list_arr))
+}
+
+fn build_map_column(
+ records: &[HashMap<String, AvroValue>],
+ name: &str,
+ map_type: &MapType,
+ num_rows: usize,
+) -> crate::Result<Arc<dyn arrow_array::Array>> {
+ let arrow_key_type = crate::arrow::paimon_type_to_arrow(map_type.key_type())?;
+ let arrow_value_type = crate::arrow::paimon_type_to_arrow(map_type.value_type())?;
+
+ let mut offsets = vec![0i32];
+ let mut key_records: Vec<HashMap<String, AvroValue>> = Vec::new();
+ let mut value_records: Vec<HashMap<String, AvroValue>> = Vec::new();
+
+ for record in records.iter().take(num_rows) {
+ match get_field_raw(record, name) {
+ Some(AvroValue::Object(map)) => {
+ for (k, v) in map {
+ let mut km = HashMap::new();
+ km.insert("key".to_string(), AvroValue::String(k.clone()));
+ key_records.push(km);
+ let mut vm = HashMap::new();
+ vm.insert("value".to_string(), v.clone());
+ value_records.push(vm);
+ }
+ offsets.push(offsets.last().unwrap() + map.len() as i32);
+ }
+ _ => {
+ offsets.push(*offsets.last().unwrap());
+ }
+ }
+ }
+
+ let key_col = build_column(&key_records, "key", map_type.key_type(), key_records.len())?;
+ let value_col = build_column(
+ &value_records,
+ "value",
+ map_type.value_type(),
+ value_records.len(),
+ )?;
+
+ let struct_arr = StructArray::try_new(
+ vec![
+ Arc::new(arrow_schema::Field::new("key", arrow_key_type, false)),
+ Arc::new(arrow_schema::Field::new(
+ "value",
+ arrow_value_type.clone(),
+ map_type.value_type().is_nullable(),
+ )),
+ ]
+ .into(),
+ vec![key_col, value_col],
+ None,
+ )
+ .map_err(|e| Error::UnexpectedError {
+ message: format!("Failed to build map StructArray: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+
+ let entries_field = arrow_schema::Field::new(
+ "entries",
+ arrow_schema::DataType::Struct(struct_arr.fields().clone()),
+ false,
+ );
+
+ let offsets_buf = OffsetBuffer::new(ScalarBuffer::from(offsets));
+ let nulls = NullBuffer::new(BooleanBuffer::from(
+ (0..num_rows)
+ .map(|i| get_field_raw(&records[i], name).is_some())
+ .collect::<Vec<_>>(),
+ ));
+
+ let map_arr = MapArray::try_new(
+ Arc::new(entries_field),
+ offsets_buf,
+ struct_arr,
+ Some(nulls),
+ false,
+ )
+ .map_err(|e| Error::UnexpectedError {
+ message: format!("Failed to build MapArray: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ Ok(Arc::new(map_arr))
+}
+
+fn build_row_column(
+ records: &[HashMap<String, AvroValue>],
+ name: &str,
+ row_type: &RowType,
+ num_rows: usize,
+) -> crate::Result<Arc<dyn arrow_array::Array>> {
+ let sub_records: Vec<HashMap<String, AvroValue>> = (0..num_rows)
+ .map(|i| match get_field_raw(&records[i], name) {
+ Some(AvroValue::Object(obj)) => obj.clone(),
+ _ => HashMap::new(),
+ })
+ .collect();
+
+ let mut child_columns: Vec<Arc<dyn arrow_array::Array>> = Vec::new();
+ let mut arrow_fields: Vec<Arc<arrow_schema::Field>> = Vec::new();
+
+ for field in row_type.fields() {
+ let col = build_column(&sub_records, field.name(), field.data_type(), num_rows)?;
+ let arrow_type = crate::arrow::paimon_type_to_arrow(field.data_type())?;
+ arrow_fields.push(Arc::new(arrow_schema::Field::new(
+ field.name(),
+ arrow_type,
+ field.data_type().is_nullable(),
+ )));
+ child_columns.push(col);
+ }
+
+ let nulls = NullBuffer::new(BooleanBuffer::from(
+ (0..num_rows)
+ .map(|i| get_field_raw(&records[i], name).is_some())
+ .collect::<Vec<_>>(),
+ ));
+
+ let struct_arr = StructArray::try_new(arrow_fields.into(), child_columns, Some(nulls))
+ .map_err(|e| Error::UnexpectedError {
+ message: format!("Failed to build StructArray: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ Ok(Arc::new(struct_arr))
+}
+
+/// Parse a decimal string (e.g. "999.99") into unscaled i128 with the given scale.
+/// For example, "999.99" with scale=2 → 99999; "0.000000000000000001" with scale=18 → 1.
+fn parse_decimal_string(s: &str, scale: i8) -> Option<i128> {
+ let negative = s.starts_with('-');
+ let s = s.strip_prefix('-').unwrap_or(s);
+ let (integer_part, frac_part) = match s.find('.') {
+ Some(pos) => (&s[..pos], &s[pos + 1..]),
+ None => (s, ""),
+ };
+ let frac_len = frac_part.len() as i8;
+ let combined = format!("{}{}", integer_part, frac_part);
+ let unscaled: i128 = combined.parse().ok()?;
+ // Adjust if the fractional digits differ from the target scale.
+ let result = if frac_len < scale {
+ unscaled * 10i128.pow((scale - frac_len) as u32)
+ } else if frac_len > scale {
+ unscaled / 10i128.pow((frac_len - scale) as u32)
+ } else {
+ unscaled
+ };
+ Some(if negative { -result } else { result })
+}
+
+/// Decode big-endian two's complement bytes into i128 (Avro decimal encoding).
+fn bytes_to_i128_be(bytes: &[u8]) -> i128 {
+ if bytes.is_empty() {
+ return 0;
+ }
+ // Sign-extend: if the high bit is set, fill with 0xFF, otherwise 0x00.
+ let sign_byte = if bytes[0] & 0x80 != 0 { 0xFF } else { 0x00 };
+ let mut buf = [sign_byte; 16];
+ let start = 16 - bytes.len();
+ buf[start..].copy_from_slice(bytes);
+ i128::from_be_bytes(buf)
+}
+
+/// Look up a field in an Avro record, unwrapping union encoding.
+fn get_field<'a>(record: &'a HashMap<String, AvroValue>, name: &str) -> Option<&'a AvroValue> {
+ record.get(name).and_then(unwrap_avro_union)
+}
+
+/// Look up a field without union unwrapping — for Map/Row types whose Object
+/// shape overlaps with union wrappers.
+fn get_field_raw<'a>(record: &'a HashMap<String, AvroValue>, name: &str) -> Option<&'a AvroValue> {
+ record.get(name).and_then(|v| match v {
+ AvroValue::Null => None,
+ other => Some(other),
+ })
+}
+
+/// Unwrap Avro union encoding: `{"type": value}` → `value`, or `"null"` → `None`.
+fn unwrap_avro_union(v: &AvroValue) -> Option<&AvroValue> {
+ match v {
+ AvroValue::Null => None,
+ AvroValue::Object(map) if map.len() == 1 => {
+ let (key, inner) = map.iter().next().unwrap();
+ if key == "null" {
+ None
+ } else {
+ Some(inner)
+ }
+ }
+ other => Some(other),
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::spec::{
+ BigIntType, BooleanType, DataField, DataType, DecimalType, DoubleType, FloatType, IntType,
+ SmallIntType, TinyIntType, VarBinaryType, VarCharType,
+ };
+ use arrow_array::Array;
+
+ // Helper to build AvroValue variants concisely in tests.
+ fn av_int(v: i64) -> AvroValue {
+ AvroValue::Int(v)
+ }
+ fn av_float(v: f64) -> AvroValue {
+ AvroValue::Float(v)
+ }
+ fn av_bool(v: bool) -> AvroValue {
+ AvroValue::Bool(v)
+ }
+ fn av_str(v: &str) -> AvroValue {
+ AvroValue::String(v.to_string())
+ }
+ fn av_bytes(v: &[u8]) -> AvroValue {
+ AvroValue::Bytes(v.to_vec())
+ }
+ fn av_null() -> AvroValue {
+ AvroValue::Null
+ }
+ fn av_union(key: &str, val: AvroValue) -> AvroValue {
+ AvroValue::Object(HashMap::from([(key.to_string(), val)]))
+ }
+
+ // -----------------------------------------------------------------------
+ // unwrap_avro_union
+ // -----------------------------------------------------------------------
+
+ #[test]
+ fn test_unwrap_avro_union_null() {
+ assert!(unwrap_avro_union(&av_null()).is_none());
+ }
+
+ #[test]
+ fn test_unwrap_avro_union_plain_value() {
+ let v = av_int(42);
+ assert_eq!(unwrap_avro_union(&v), Some(&av_int(42)));
+ }
+
+ #[test]
+ fn test_unwrap_avro_union_wrapped_value() {
+ let v = av_union("int", av_int(42));
+ assert_eq!(unwrap_avro_union(&v), Some(&av_int(42)));
+ }
+
+ #[test]
+ fn test_unwrap_avro_union_null_key() {
+ let v = av_union("null", av_null());
+ assert!(unwrap_avro_union(&v).is_none());
+ }
+
+ // -----------------------------------------------------------------------
+ // get_field
+ // -----------------------------------------------------------------------
+
+ #[test]
+ fn test_get_field_present() {
+ let mut record = HashMap::new();
+ record.insert("name".to_string(), av_str("alice"));
+ assert_eq!(get_field(&record, "name"), Some(&av_str("alice")));
+ }
+
+ #[test]
+ fn test_get_field_missing() {
+ let record: HashMap<String, AvroValue> = HashMap::new();
+ assert!(get_field(&record, "name").is_none());
+ }
+
+ #[test]
+ fn test_get_field_union_wrapped() {
+ let mut record = HashMap::new();
+ record.insert("age".to_string(), av_union("int", av_int(30)));
+ assert_eq!(get_field(&record, "age"), Some(&av_int(30)));
+ }
+
+ // -----------------------------------------------------------------------
+ // ranges_to_mask
+ // -----------------------------------------------------------------------
+
+ #[test]
+ fn test_ranges_to_mask_single_range() {
+ let ranges = vec![RowRange::new(1, 3)];
+ let mask = ranges_to_mask(5, &ranges);
+ assert_eq!(mask, vec![false, true, true, true, false]);
+ }
+
+ #[test]
+ fn test_ranges_to_mask_full_range() {
+ let ranges = vec![RowRange::new(0, 4)];
+ let mask = ranges_to_mask(5, &ranges);
+ assert_eq!(mask, vec![true, true, true, true, true]);
+ }
+
+ #[test]
+ fn test_ranges_to_mask_multiple_ranges() {
+ let ranges = vec![RowRange::new(0, 0), RowRange::new(3, 4)];
+ let mask = ranges_to_mask(5, &ranges);
+ assert_eq!(mask, vec![true, false, false, true, true]);
+ }
+
+ // -----------------------------------------------------------------------
+ // build_column + records_to_batch
+ // -----------------------------------------------------------------------
+
+ fn make_records(rows: Vec<Vec<(&str, AvroValue)>>) -> Vec<HashMap<String, AvroValue>> {
+ rows.into_iter()
+ .map(|fields| {
+ fields
+ .into_iter()
+ .map(|(k, v)| (k.to_string(), v))
+ .collect()
+ })
+ .collect()
+ }
+
+ #[test]
+ fn test_build_column_int() {
+ let records = make_records(vec![
+ vec![("x", av_int(1))],
+ vec![("x", av_int(2))],
+ vec![("x", av_int(3))],
+ ]);
+ let col = build_column(&records, "x", &DataType::Int(IntType::new()), 3).unwrap();
+ let arr = col.as_any().downcast_ref::<Int32Array>().unwrap();
+ assert_eq!(arr.values().as_ref(), &[1, 2, 3]);
+ }
+
+ #[test]
+ fn test_build_column_bigint() {
+ let records = make_records(vec![vec![("v", av_int(100))], vec![("v", av_int(200))]]);
+ let col = build_column(&records, "v", &DataType::BigInt(BigIntType::new()), 2).unwrap();
+ let arr = col.as_any().downcast_ref::<Int64Array>().unwrap();
+ assert_eq!(arr.values().as_ref(), &[100, 200]);
+ }
+
+ #[test]
+ fn test_build_column_boolean() {
+ let records = make_records(vec![
+ vec![("b", av_bool(true))],
+ vec![("b", av_bool(false))],
+ ]);
+ let col = build_column(&records, "b", &DataType::Boolean(BooleanType::new()), 2).unwrap();
+ let arr = col.as_any().downcast_ref::<BooleanArray>().unwrap();
+ assert!(arr.value(0));
+ assert!(!arr.value(1));
+ }
+
+ #[test]
+ fn test_build_column_string() {
+ let records = make_records(vec![
+ vec![("s", av_str("hello"))],
+ vec![("s", av_str("world"))],
+ ]);
+ let col = build_column(
+ &records,
+ "s",
+ &DataType::VarChar(VarCharType::new(100).unwrap()),
+ 2,
+ )
+ .unwrap();
+ let arr = col.as_any().downcast_ref::<StringArray>().unwrap();
+ assert_eq!(arr.value(0), "hello");
+ assert_eq!(arr.value(1), "world");
+ }
+
+ #[test]
+ fn test_build_column_float_double() {
+ let records = make_records(vec![vec![("f", av_float(1.5)), ("d", av_float(2.5))]]);
+ let fcol = build_column(&records, "f", &DataType::Float(FloatType::new()), 1).unwrap();
+ let dcol = build_column(&records, "d", &DataType::Double(DoubleType::new()), 1).unwrap();
+ let farr = fcol.as_any().downcast_ref::<Float32Array>().unwrap();
+ let darr = dcol.as_any().downcast_ref::<Float64Array>().unwrap();
+ assert!((farr.value(0) - 1.5_f32).abs() < f32::EPSILON);
+ assert!((darr.value(0) - 2.5_f64).abs() < f64::EPSILON);
+ }
+
+ #[test]
+ fn test_build_column_with_nulls() {
+ let records = make_records(vec![
+ vec![("x", av_int(10))],
+ vec![("x", av_null())],
+ vec![("x", av_int(30))],
+ ]);
+ let col = build_column(&records, "x", &DataType::Int(IntType::new()), 3).unwrap();
+ let arr = col.as_any().downcast_ref::<Int32Array>().unwrap();
+ assert!(!arr.is_null(0));
+ assert!(arr.is_null(1));
+ assert!(!arr.is_null(2));
+ assert_eq!(arr.value(0), 10);
+ assert_eq!(arr.value(2), 30);
+ }
+
+ #[test]
+ fn test_build_column_tinyint_smallint() {
+ let records = make_records(vec![vec![("t", av_int(7)), ("s", av_int(300))]]);
+ let tcol = build_column(&records, "t", &DataType::TinyInt(TinyIntType::new()), 1).unwrap();
+ let scol =
+ build_column(&records, "s", &DataType::SmallInt(SmallIntType::new()), 1).unwrap();
+ let tarr = tcol.as_any().downcast_ref::<Int8Array>().unwrap();
+ let sarr = scol.as_any().downcast_ref::<Int16Array>().unwrap();
+ assert_eq!(tarr.value(0), 7);
+ assert_eq!(sarr.value(0), 300);
+ }
+
+ #[test]
+ fn test_build_column_binary() {
+ let records = make_records(vec![
+ vec![("b", av_bytes(&[0xDE, 0xAD]))],
+ vec![("b", av_bytes(&[0xBE, 0xEF]))],
+ ]);
+ let col = build_column(
+ &records,
+ "b",
+ &DataType::VarBinary(VarBinaryType::new(100).unwrap()),
+ 2,
+ )
+ .unwrap();
+ let arr = col.as_any().downcast_ref::<BinaryArray>().unwrap();
+ assert_eq!(arr.value(0), &[0xDE, 0xAD]);
+ assert_eq!(arr.value(1), &[0xBE, 0xEF]);
+ }
+
+ #[test]
+ fn test_build_column_decimal_from_bytes() {
+ // 12345 in big-endian two's complement = [0x30, 0x39]
+ let records = make_records(vec![
+ vec![("d", av_bytes(&[0x30, 0x39]))],
+ vec![("d", av_bytes(&[0x01, 0xA4]))], // 420
+ ]);
+ let col = build_column(
+ &records,
+ "d",
+ &DataType::Decimal(DecimalType::new(10, 2).unwrap()),
+ 2,
+ )
+ .unwrap();
+ let arr = col.as_any().downcast_ref::<Decimal128Array>().unwrap();
+ assert_eq!(arr.value(0), 12345); // 123.45 * 100
+ assert_eq!(arr.value(1), 420); // 4.20 * 100
+ }
+
+ #[test]
+ fn test_bytes_to_i128_be() {
+ assert_eq!(bytes_to_i128_be(&[0x30, 0x39]), 12345);
+ assert_eq!(bytes_to_i128_be(&[0xFF]), -1);
+ assert_eq!(bytes_to_i128_be(&[]), 0);
+ assert_eq!(bytes_to_i128_be(&[0x00, 0x01]), 1);
+ }
+
+ #[test]
+ fn test_records_to_batch_basic() {
+ let fields = vec![
+ DataField::new(0, "id".to_string(), DataType::Int(IntType::new())),
+ DataField::new(
+ 1,
+ "name".to_string(),
+ DataType::VarChar(VarCharType::new(50).unwrap()),
+ ),
+ ];
+ let schema = crate::arrow::build_target_arrow_schema(&fields).unwrap();
+ let records = make_records(vec![
+ vec![("id", av_int(1)), ("name", av_str("alice"))],
+ vec![("id", av_int(2)), ("name", av_str("bob"))],
+ ]);
+ let batch = records_to_batch(&records, &fields, &schema).unwrap();
+ assert_eq!(batch.num_rows(), 2);
+ assert_eq!(batch.num_columns(), 2);
+ }
+
+ #[test]
+ fn test_records_to_batch_empty() {
+ let fields = vec![DataField::new(
+ 0,
+ "id".to_string(),
+ DataType::Int(IntType::new()),
+ )];
+ let schema = crate::arrow::build_target_arrow_schema(&fields).unwrap();
+ let records: Vec<HashMap<String, AvroValue>> = vec![];
+ let batch = records_to_batch(&records, &fields, &schema).unwrap();
+ assert_eq!(batch.num_rows(), 0);
+ }
+}
diff --git a/crates/paimon/src/arrow/format/mod.rs b/crates/paimon/src/arrow/format/mod.rs
new file mode 100644
index 0000000..63fd4e0
--- /dev/null
+++ b/crates/paimon/src/arrow/format/mod.rs
@@ -0,0 +1,76 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod avro;
+mod orc;
+mod parquet;
+
+use crate::io::FileRead;
+use crate::spec::{DataField, Predicate};
+use crate::table::{ArrowRecordBatchStream, RowRange};
+use crate::Error;
+use async_trait::async_trait;
+
+/// Predicates with the file-level field context needed for pushdown.
+/// Only used by formats that support predicate pushdown (e.g. Parquet).
+pub(crate) struct FilePredicates {
+ /// Predicates with indices already remapped to file-level fields.
+ pub predicates: Vec<Predicate>,
+ /// File-level fields (full file schema), used for stats access and row filtering.
+ pub file_fields: Vec<DataField>,
+}
+
+/// Format-agnostic file reader that produces Arrow RecordBatch streams.
+///
+/// Each implementation (Parquet, ORC, ...) handles:
+/// - Column projection
+/// - Predicate pushdown (row-group/stripe pruning + row-level filtering)
+/// - Row range selection
+#[async_trait]
+pub(crate) trait FormatFileReader: Send + Sync {
+ /// Read a single data file, returning a stream of RecordBatches
+ /// containing only the projected columns (using names from the file's schema).
+ ///
+ /// `row_selection` is a pre-merged list of 0-based inclusive row ranges
+ /// (DV + row_ranges already combined by the caller).
+ async fn read_batch_stream(
+ &self,
+ reader: Box<dyn FileRead>,
+ file_size: u64,
+ read_fields: &[DataField],
+ predicates: Option<&FilePredicates>,
+ batch_size: Option<usize>,
+ row_selection: Option<Vec<RowRange>>,
+ ) -> crate::Result<ArrowRecordBatchStream>;
+}
+
+/// Create a format reader based on the file extension.
+pub(crate) fn create_format_reader(path: &str) -> crate::Result<Box<dyn FormatFileReader>> {
+ if path.to_ascii_lowercase().ends_with(".parquet") {
+ Ok(Box::new(parquet::ParquetFormatReader))
+ } else if path.to_ascii_lowercase().ends_with(".orc") {
+ Ok(Box::new(orc::OrcFormatReader))
+ } else if path.to_ascii_lowercase().ends_with(".avro") {
+ Ok(Box::new(avro::AvroFormatReader))
+ } else {
+ Err(Error::Unsupported {
+ message: format!(
+ "unsupported file format: expected .parquet, .orc, or .avro, got: {path}"
+ ),
+ })
+ }
+}
diff --git a/crates/paimon/src/arrow/format/orc.rs b/crates/paimon/src/arrow/format/orc.rs
new file mode 100644
index 0000000..8806fec
--- /dev/null
+++ b/crates/paimon/src/arrow/format/orc.rs
@@ -0,0 +1,195 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use super::{FilePredicates, FormatFileReader};
+use crate::io::FileRead;
+use crate::spec::DataField;
+use crate::table::{ArrowRecordBatchStream, RowRange};
+use crate::Error;
+use async_trait::async_trait;
+use bytes::Bytes;
+use futures::{future::BoxFuture, StreamExt};
+use orc_rust::projection::ProjectionMask;
+use orc_rust::reader::AsyncChunkReader;
+use orc_rust::ArrowReaderBuilder;
+
+pub(crate) struct OrcFormatReader;
+
+#[async_trait]
+impl FormatFileReader for OrcFormatReader {
+ async fn read_batch_stream(
+ &self,
+ reader: Box<dyn FileRead>,
+ file_size: u64,
+ read_fields: &[DataField],
+ // TODO: support predicate pushdown for ORC (stripe pruning + row-level filtering)
+ _predicates: Option<&FilePredicates>,
+ batch_size: Option<usize>,
+ row_selection: Option<Vec<RowRange>>,
+ ) -> crate::Result<ArrowRecordBatchStream> {
+ let orc_reader = OrcFileReader::new(file_size, reader);
+
+ let builder = ArrowReaderBuilder::try_new_async(orc_reader)
+ .await
+ .map_err(|e| Error::UnexpectedError {
+ message: format!("Failed to open ORC file: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+
+ let projected_names: Vec<&str> = read_fields.iter().map(|f| f.name()).collect();
+ let projection =
+ ProjectionMask::named_roots(builder.file_metadata().root_data_type(), &projected_names);
+
+ let mut builder = builder.with_projection(projection);
+
+ if let Some(size) = batch_size {
+ builder = builder.with_batch_size(size);
+ }
+
+ if let Some(ref ranges) = row_selection {
+ let total_rows: usize = builder
+ .file_metadata()
+ .stripe_metadatas()
+ .iter()
+ .map(|s| s.number_of_rows() as usize)
+ .sum();
+ let selection = build_range_row_selection(total_rows, ranges);
+ builder = builder.with_row_selection(selection);
+ }
+
+ let stream = builder.build_async();
+ Ok(stream
+ .map(|r| {
+ r.map_err(|e| Error::UnexpectedError {
+ message: format!("ORC read error: {e}"),
+ source: Some(Box::new(e)),
+ })
+ })
+ .boxed())
+ }
+}
+
+// ---------------------------------------------------------------------------
+// Row ranges → orc_rust::RowSelection
+// ---------------------------------------------------------------------------
+
+fn build_range_row_selection(
+ total_rows: usize,
+ row_ranges: &[RowRange],
+) -> orc_rust::row_selection::RowSelection {
+ if total_rows == 0 {
+ return orc_rust::row_selection::RowSelection::default();
+ }
+
+ let file_end = total_rows as i64 - 1;
+ let mut local_ranges: Vec<(usize, usize)> = row_ranges
+ .iter()
+ .filter_map(|r| {
+ if r.to() < 0 || r.from() > file_end {
+ return None;
+ }
+ let local_start = r.from().max(0) as usize;
+ let local_end = (r.to().min(file_end) + 1) as usize;
+ Some((local_start, local_end))
+ })
+ .collect();
+ local_ranges.sort_by_key(|&(s, _)| s);
+
+ orc_rust::row_selection::RowSelection::from_consecutive_ranges(
+ local_ranges.into_iter().map(|(s, e)| s..e),
+ total_rows,
+ )
+}
+
+// ---------------------------------------------------------------------------
+// OrcFileReader — adapts paimon FileRead to orc-rust AsyncChunkReader
+// ---------------------------------------------------------------------------
+
+struct OrcFileReader {
+ file_size: u64,
+ r: Box<dyn FileRead>,
+}
+
+impl OrcFileReader {
+ fn new(file_size: u64, r: Box<dyn FileRead>) -> Self {
+ Self { file_size, r }
+ }
+}
+
+impl AsyncChunkReader for OrcFileReader {
+ fn len(&mut self) -> BoxFuture<'_, std::io::Result<u64>> {
+ Box::pin(std::future::ready(Ok(self.file_size)))
+ }
+
+ fn get_bytes(
+ &mut self,
+ offset_from_start: u64,
+ length: u64,
+ ) -> BoxFuture<'_, std::io::Result<Bytes>> {
+ Box::pin(async move {
+ self.r
+ .read(offset_from_start..offset_from_start + length)
+ .await
+ .map_err(|e| std::io::Error::other(e.to_string()))
+ })
+ }
+}
+
+// ---------------------------------------------------------------------------
+// Tests
+// ---------------------------------------------------------------------------
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use orc_rust::row_selection::RowSelector;
+
+ #[test]
+ fn test_build_range_row_selection_single_range() {
+ let ranges = vec![RowRange::new(2, 4)];
+ let sel = build_range_row_selection(6, &ranges);
+ // rows 0,1 skip; 2,3,4 select; 5 skip
+ let expected: orc_rust::row_selection::RowSelection = vec![
+ RowSelector::skip(2),
+ RowSelector::select(3),
+ RowSelector::skip(1),
+ ]
+ .into();
+ assert_eq!(sel, expected);
+ }
+
+ #[test]
+ fn test_build_range_row_selection_with_offset() {
+ let ranges = vec![RowRange::new(1, 3)];
+ let sel = build_range_row_selection(5, &ranges);
+ let expected: orc_rust::row_selection::RowSelection = vec![
+ RowSelector::skip(1),
+ RowSelector::select(3),
+ RowSelector::skip(1),
+ ]
+ .into();
+ assert_eq!(sel, expected);
+ }
+
+ #[test]
+ fn test_build_range_row_selection_out_of_file() {
+ let ranges = vec![RowRange::new(10, 20)];
+ let sel = build_range_row_selection(5, &ranges);
+ let expected: orc_rust::row_selection::RowSelection = vec![RowSelector::skip(5)].into();
+ assert_eq!(sel, expected);
+ }
+}
diff --git a/crates/paimon/src/arrow/format/parquet.rs b/crates/paimon/src/arrow/format/parquet.rs
new file mode 100644
index 0000000..915bd56
--- /dev/null
+++ b/crates/paimon/src/arrow/format/parquet.rs
@@ -0,0 +1,881 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use super::{FilePredicates, FormatFileReader};
+use crate::arrow::filtering::{predicates_may_match_with_schema, StatsAccessor};
+use crate::io::FileRead;
+use crate::spec::{DataField, DataType, Datum, Predicate, PredicateOperator};
+use crate::table::{ArrowRecordBatchStream, RowRange};
+use crate::Error;
+use arrow_array::{
+ Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Decimal128Array, Float32Array,
+ Float64Array, Int16Array, Int32Array, Int64Array, Int8Array, RecordBatch, Scalar, StringArray,
+};
+use arrow_ord::cmp::{
+ eq as arrow_eq, gt as arrow_gt, gt_eq as arrow_gt_eq, lt as arrow_lt, lt_eq as arrow_lt_eq,
+ neq as arrow_neq,
+};
+use arrow_schema::ArrowError;
+use async_trait::async_trait;
+use bytes::Bytes;
+use futures::future::BoxFuture;
+use futures::{StreamExt, TryFutureExt};
+use parquet::arrow::arrow_reader::{
+ ArrowPredicate, ArrowPredicateFn, ArrowReaderOptions, RowFilter, RowSelection, RowSelector,
+};
+use parquet::arrow::async_reader::{AsyncFileReader, MetadataFetch};
+use parquet::arrow::{ParquetRecordBatchStreamBuilder, ProjectionMask};
+use parquet::file::metadata::ParquetMetaDataReader;
+use parquet::file::metadata::{ParquetMetaData, RowGroupMetaData};
+use parquet::file::statistics::Statistics as ParquetStatistics;
+use std::collections::HashMap;
+use std::ops::Range;
+use std::sync::Arc;
+
+pub(crate) struct ParquetFormatReader;
+
+#[async_trait]
+impl FormatFileReader for ParquetFormatReader {
+ async fn read_batch_stream(
+ &self,
+ reader: Box<dyn FileRead>,
+ file_size: u64,
+ read_fields: &[DataField],
+ predicates: Option<&FilePredicates>,
+ batch_size: Option<usize>,
+ row_selection: Option<Vec<RowRange>>,
+ ) -> crate::Result<ArrowRecordBatchStream> {
+ let arrow_file_reader = ArrowFileReader::new(file_size, reader);
+
+ let mut batch_stream_builder =
+ ParquetRecordBatchStreamBuilder::new(arrow_file_reader).await?;
+
+ let parquet_schema = batch_stream_builder.parquet_schema().clone();
+ let root_schema = parquet_schema.root_schema();
+ let root_indices: Vec<usize> = read_fields
+ .iter()
+ .filter_map(|f| {
+ root_schema
+ .get_fields()
+ .iter()
+ .position(|pf| pf.name() == f.name())
+ })
+ .collect();
+
+ let mask = ProjectionMask::roots(&parquet_schema, root_indices);
+ batch_stream_builder = batch_stream_builder.with_projection(mask);
+
+ let empty_predicates = Vec::new();
+ let (preds, file_fields): (&[Predicate], &[DataField]) = match predicates {
+ Some(fp) => (&fp.predicates, &fp.file_fields),
+ None => (&empty_predicates, &[]),
+ };
+
+ let parquet_row_filter = build_parquet_row_filter(&parquet_schema, preds, file_fields)?;
+ if let Some(f) = parquet_row_filter {
+ batch_stream_builder = batch_stream_builder.with_row_filter(f);
+ }
+
+ let predicate_row_selection = build_predicate_row_selection(
+ batch_stream_builder.metadata().row_groups(),
+ preds,
+ file_fields,
+ )?;
+ let mut combined_selection = predicate_row_selection;
+
+ if let Some(ref ranges) = row_selection {
+ let range_selection =
+ build_row_ranges_selection(batch_stream_builder.metadata().row_groups(), ranges);
+ combined_selection =
+ intersect_optional_row_selections(combined_selection, Some(range_selection));
+ }
+ if let Some(sel) = combined_selection {
+ batch_stream_builder = batch_stream_builder.with_row_selection(sel);
+ }
+ if let Some(size) = batch_size {
+ batch_stream_builder = batch_stream_builder.with_batch_size(size);
+ }
+
+ let batch_stream = batch_stream_builder.build()?;
+ Ok(batch_stream.map(|r| r.map_err(Error::from)).boxed())
+ }
+}
+
+// ---------------------------------------------------------------------------
+// Parquet row-filter helpers
+// ---------------------------------------------------------------------------
+
+fn build_parquet_row_filter(
+ parquet_schema: &parquet::schema::types::SchemaDescriptor,
+ predicates: &[Predicate],
+ file_fields: &[DataField],
+) -> crate::Result<Option<RowFilter>> {
+ if predicates.is_empty() {
+ return Ok(None);
+ }
+
+ let mut filters: Vec<Box<dyn ArrowPredicate>> = Vec::new();
+
+ for predicate in predicates {
+ if let Some(filter) = build_parquet_arrow_predicate(parquet_schema, predicate, file_fields)?
+ {
+ filters.push(filter);
+ }
+ }
+
+ if filters.is_empty() {
+ Ok(None)
+ } else {
+ Ok(Some(RowFilter::new(filters)))
+ }
+}
+
+fn build_parquet_arrow_predicate(
+ parquet_schema: &parquet::schema::types::SchemaDescriptor,
+ predicate: &Predicate,
+ file_fields: &[DataField],
+) -> crate::Result<Option<Box<dyn ArrowPredicate>>> {
+ let Predicate::Leaf {
+ index,
+ data_type: _,
+ op,
+ literals,
+ ..
+ } = predicate
+ else {
+ return Ok(None);
+ };
+ if !predicate_supported_for_parquet_row_filter(*op) {
+ return Ok(None);
+ }
+
+ let Some(file_field) = file_fields.get(*index) else {
+ return Ok(None);
+ };
+ let Some(root_index) = parquet_root_index(parquet_schema, file_field.name()) else {
+ return Ok(None);
+ };
+ if !parquet_row_filter_literals_supported(*op, literals, file_field.data_type())? {
+ return Ok(None);
+ }
+
+ let projection = ProjectionMask::roots(parquet_schema, [root_index]);
+ let op = *op;
+ let data_type = file_field.data_type().clone();
+ let literals = literals.to_vec();
+ Ok(Some(Box::new(ArrowPredicateFn::new(
+ projection,
+ move |batch: RecordBatch| {
+ let Some(column) = batch.columns().first() else {
+ return Ok(BooleanArray::new_null(batch.num_rows()));
+ };
+ evaluate_exact_leaf_predicate(column, &data_type, op, &literals)
+ },
+ ))))
+}
+
+fn predicate_supported_for_parquet_row_filter(op: PredicateOperator) -> bool {
+ matches!(
+ op,
+ PredicateOperator::IsNull
+ | PredicateOperator::IsNotNull
+ | PredicateOperator::Eq
+ | PredicateOperator::NotEq
+ | PredicateOperator::Lt
+ | PredicateOperator::LtEq
+ | PredicateOperator::Gt
+ | PredicateOperator::GtEq
+ | PredicateOperator::In
+ | PredicateOperator::NotIn
+ )
+}
+
+fn parquet_row_filter_literals_supported(
+ op: PredicateOperator,
+ literals: &[Datum],
+ file_data_type: &DataType,
+) -> crate::Result<bool> {
+ match op {
+ PredicateOperator::IsNull | PredicateOperator::IsNotNull => Ok(true),
+ PredicateOperator::Eq
+ | PredicateOperator::NotEq
+ | PredicateOperator::Lt
+ | PredicateOperator::LtEq
+ | PredicateOperator::Gt
+ | PredicateOperator::GtEq => {
+ let Some(literal) = literals.first() else {
+ return Ok(false);
+ };
+ Ok(literal_scalar_for_parquet_filter(literal, file_data_type)?.is_some())
+ }
+ PredicateOperator::In | PredicateOperator::NotIn => {
+ for literal in literals {
+ if literal_scalar_for_parquet_filter(literal, file_data_type)?.is_none() {
+ return Ok(false);
+ }
+ }
+ Ok(true)
+ }
+ }
+}
+
+fn parquet_root_index(
+ parquet_schema: &parquet::schema::types::SchemaDescriptor,
+ root_name: &str,
+) -> Option<usize> {
+ parquet_schema
+ .root_schema()
+ .get_fields()
+ .iter()
+ .position(|field| field.name() == root_name)
+}
+
+// ---------------------------------------------------------------------------
+// Predicate evaluation helpers
+// ---------------------------------------------------------------------------
+
+fn evaluate_exact_leaf_predicate(
+ array: &ArrayRef,
+ data_type: &DataType,
+ op: PredicateOperator,
+ literals: &[Datum],
+) -> Result<BooleanArray, ArrowError> {
+ match op {
+ PredicateOperator::IsNull => Ok(boolean_mask_from_predicate(array.len(), |row_index| {
+ array.is_null(row_index)
+ })),
+ PredicateOperator::IsNotNull => Ok(boolean_mask_from_predicate(array.len(), |row_index| {
+ array.is_valid(row_index)
+ })),
+ PredicateOperator::In | PredicateOperator::NotIn => {
+ evaluate_set_membership_predicate(array, data_type, op, literals)
+ }
+ PredicateOperator::Eq
+ | PredicateOperator::NotEq
+ | PredicateOperator::Lt
+ | PredicateOperator::LtEq
+ | PredicateOperator::Gt
+ | PredicateOperator::GtEq => {
+ let Some(literal) = literals.first() else {
+ return Ok(BooleanArray::from(vec![true; array.len()]));
+ };
+ let Some(scalar) = literal_scalar_for_parquet_filter(literal, data_type)
+ .map_err(|e| ArrowError::ComputeError(e.to_string()))?
+ else {
+ return Ok(BooleanArray::from(vec![true; array.len()]));
+ };
+ let result = evaluate_column_predicate(array, &scalar, op)?;
+ Ok(sanitize_filter_mask(result))
+ }
+ }
+}
+
+fn evaluate_set_membership_predicate(
+ array: &ArrayRef,
+ data_type: &DataType,
+ op: PredicateOperator,
+ literals: &[Datum],
+) -> Result<BooleanArray, ArrowError> {
+ if literals.is_empty() {
+ return Ok(match op {
+ PredicateOperator::In => BooleanArray::from(vec![false; array.len()]),
+ PredicateOperator::NotIn => {
+ boolean_mask_from_predicate(array.len(), |row_index| array.is_valid(row_index))
+ }
+ _ => unreachable!(),
+ });
+ }
+
+ let mut combined = match op {
+ PredicateOperator::In => BooleanArray::from(vec![false; array.len()]),
+ PredicateOperator::NotIn => {
+ boolean_mask_from_predicate(array.len(), |row_index| array.is_valid(row_index))
+ }
+ _ => unreachable!(),
+ };
+
+ for literal in literals {
+ let Some(scalar) = literal_scalar_for_parquet_filter(literal, data_type)
+ .map_err(|e| ArrowError::ComputeError(e.to_string()))?
+ else {
+ return Ok(BooleanArray::from(vec![true; array.len()]));
+ };
+ let comparison_op = match op {
+ PredicateOperator::In => PredicateOperator::Eq,
+ PredicateOperator::NotIn => PredicateOperator::NotEq,
+ _ => unreachable!(),
+ };
+ let mask = sanitize_filter_mask(evaluate_column_predicate(array, &scalar, comparison_op)?);
+ combined = combine_filter_masks(&combined, &mask, matches!(op, PredicateOperator::In));
+ }
+
+ Ok(combined)
+}
+
+fn evaluate_column_predicate(
+ column: &ArrayRef,
+ scalar: &Scalar<ArrayRef>,
+ op: PredicateOperator,
+) -> Result<BooleanArray, ArrowError> {
+ match op {
+ PredicateOperator::Eq => arrow_eq(column, scalar),
+ PredicateOperator::NotEq => arrow_neq(column, scalar),
+ PredicateOperator::Lt => arrow_lt(column, scalar),
+ PredicateOperator::LtEq => arrow_lt_eq(column, scalar),
+ PredicateOperator::Gt => arrow_gt(column, scalar),
+ PredicateOperator::GtEq => arrow_gt_eq(column, scalar),
+ PredicateOperator::IsNull
+ | PredicateOperator::IsNotNull
+ | PredicateOperator::In
+ | PredicateOperator::NotIn => Ok(BooleanArray::new_null(column.len())),
+ }
+}
+
+fn sanitize_filter_mask(mask: BooleanArray) -> BooleanArray {
+ if mask.null_count() == 0 {
+ return mask;
+ }
+
+ boolean_mask_from_predicate(mask.len(), |row_index| {
+ mask.is_valid(row_index) && mask.value(row_index)
+ })
+}
+
+fn combine_filter_masks(left: &BooleanArray, right: &BooleanArray, use_or: bool) -> BooleanArray {
+ debug_assert_eq!(left.len(), right.len());
+ boolean_mask_from_predicate(left.len(), |row_index| {
+ if use_or {
+ left.value(row_index) || right.value(row_index)
+ } else {
+ left.value(row_index) && right.value(row_index)
+ }
+ })
+}
+
+fn boolean_mask_from_predicate(
+ len: usize,
+ mut predicate: impl FnMut(usize) -> bool,
+) -> BooleanArray {
+ BooleanArray::from((0..len).map(&mut predicate).collect::<Vec<_>>())
+}
+
+// ---------------------------------------------------------------------------
+// Row-group statistics pruning
+// ---------------------------------------------------------------------------
+
+struct ParquetRowGroupStats<'a> {
+ row_group: &'a RowGroupMetaData,
+ column_indices: &'a [Option<usize>],
+}
+
+impl StatsAccessor for ParquetRowGroupStats<'_> {
+ fn row_count(&self) -> i64 {
+ self.row_group.num_rows()
+ }
+
+ fn null_count(&self, index: usize) -> Option<i64> {
+ let _ = index;
+ None
+ }
+
+ fn min_value(&self, index: usize, data_type: &DataType) -> Option<Datum> {
+ let column_index = self.column_indices.get(index).copied().flatten()?;
+ parquet_stats_to_datum(
+ self.row_group.column(column_index).statistics()?,
+ data_type,
+ true,
+ )
+ }
+
+ fn max_value(&self, index: usize, data_type: &DataType) -> Option<Datum> {
+ let column_index = self.column_indices.get(index).copied().flatten()?;
+ parquet_stats_to_datum(
+ self.row_group.column(column_index).statistics()?,
+ data_type,
+ false,
+ )
+ }
+}
+
+fn build_predicate_row_selection(
+ row_groups: &[RowGroupMetaData],
+ predicates: &[Predicate],
+ file_fields: &[DataField],
+) -> crate::Result<Option<RowSelection>> {
+ if predicates.is_empty() || row_groups.is_empty() {
+ return Ok(None);
+ }
+
+ // Predicates have already been remapped to file-level indices by the caller
+ // (remap_predicates_to_file in reader.rs), so we use an identity mapping here.
+ let identity_mapping: Vec<Option<usize>> = (0..file_fields.len()).map(Some).collect();
+ let column_indices = build_row_group_column_indices(row_groups[0].columns(), file_fields);
+ let mut selectors = Vec::with_capacity(row_groups.len());
+ let mut all_selected = true;
+
+ for row_group in row_groups {
+ let stats = ParquetRowGroupStats {
+ row_group,
+ column_indices: &column_indices,
+ };
+ let may_match =
+ predicates_may_match_with_schema(predicates, &stats, &identity_mapping, file_fields);
+ if !may_match {
+ all_selected = false;
+ }
+ selectors.push(if may_match {
+ RowSelector::select(row_group.num_rows() as usize)
+ } else {
+ RowSelector::skip(row_group.num_rows() as usize)
+ });
+ }
+
+ if all_selected {
+ Ok(None)
+ } else {
+ Ok(Some(selectors.into()))
+ }
+}
+
+fn build_row_group_column_indices(
+ columns: &[parquet::file::metadata::ColumnChunkMetaData],
+ file_fields: &[DataField],
+) -> Vec<Option<usize>> {
+ let mut by_root_name: HashMap<&str, Option<usize>> = HashMap::new();
+ for (column_index, column) in columns.iter().enumerate() {
+ let Some(root_name) = column.column_path().parts().first() else {
+ continue;
+ };
+ let entry = by_root_name
+ .entry(root_name.as_str())
+ .or_insert(Some(column_index));
+ if entry.is_some() && *entry != Some(column_index) {
+ *entry = None;
+ }
+ }
+
+ file_fields
+ .iter()
+ .map(|field| by_root_name.get(field.name()).copied().flatten())
+ .collect()
+}
+
+// ---------------------------------------------------------------------------
+// Parquet statistics → Datum conversion
+// ---------------------------------------------------------------------------
+
+fn parquet_stats_to_datum(
+ stats: &ParquetStatistics,
+ data_type: &DataType,
+ is_min: bool,
+) -> Option<Datum> {
+ let exact = if is_min {
+ stats.min_is_exact()
+ } else {
+ stats.max_is_exact()
+ };
+ if !exact {
+ return None;
+ }
+
+ match (stats, data_type) {
+ (ParquetStatistics::Boolean(stats), DataType::Boolean(_)) => {
+ exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
+ .copied()
+ .map(Datum::Bool)
+ }
+ (ParquetStatistics::Int32(stats), DataType::TinyInt(_)) => {
+ exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
+ .and_then(|value| i8::try_from(*value).ok())
+ .map(Datum::TinyInt)
+ }
+ (ParquetStatistics::Int32(stats), DataType::SmallInt(_)) => {
+ exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
+ .and_then(|value| i16::try_from(*value).ok())
+ .map(Datum::SmallInt)
+ }
+ (ParquetStatistics::Int32(stats), DataType::Int(_)) => {
+ exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
+ .copied()
+ .map(Datum::Int)
+ }
+ (ParquetStatistics::Int32(stats), DataType::Date(_)) => {
+ exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
+ .copied()
+ .map(Datum::Date)
+ }
+ (ParquetStatistics::Int32(stats), DataType::Time(_)) => {
+ exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
+ .copied()
+ .map(Datum::Time)
+ }
+ (ParquetStatistics::Int64(stats), DataType::BigInt(_)) => {
+ exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
+ .copied()
+ .map(Datum::Long)
+ }
+ (ParquetStatistics::Int64(stats), DataType::Timestamp(ts)) if ts.precision() <= 3 => {
+ exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
+ .copied()
+ .map(|millis| Datum::Timestamp { millis, nanos: 0 })
+ }
+ (ParquetStatistics::Int64(stats), DataType::LocalZonedTimestamp(ts))
+ if ts.precision() <= 3 =>
+ {
+ exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
+ .copied()
+ .map(|millis| Datum::LocalZonedTimestamp { millis, nanos: 0 })
+ }
+ (ParquetStatistics::Float(stats), DataType::Float(_)) => {
+ exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
+ .copied()
+ .map(Datum::Float)
+ }
+ (ParquetStatistics::Double(stats), DataType::Double(_)) => {
+ exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
+ .copied()
+ .map(Datum::Double)
+ }
+ (ParquetStatistics::ByteArray(stats), DataType::Char(_))
+ | (ParquetStatistics::ByteArray(stats), DataType::VarChar(_)) => {
+ exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
+ .and_then(|value| std::str::from_utf8(value.data()).ok())
+ .map(|value| Datum::String(value.to_string()))
+ }
+ (ParquetStatistics::ByteArray(stats), DataType::Binary(_))
+ | (ParquetStatistics::ByteArray(stats), DataType::VarBinary(_)) => {
+ exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
+ .map(|value| Datum::Bytes(value.data().to_vec()))
+ }
+ (ParquetStatistics::FixedLenByteArray(stats), DataType::Binary(_))
+ | (ParquetStatistics::FixedLenByteArray(stats), DataType::VarBinary(_)) => {
+ exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
+ .map(|value| Datum::Bytes(value.data().to_vec()))
+ }
+ _ => None,
+ }
+}
+
+fn exact_parquet_value<'a, T>(
+ is_min: bool,
+ min: Option<&'a T>,
+ max: Option<&'a T>,
+) -> Option<&'a T> {
+ if is_min {
+ min
+ } else {
+ max
+ }
+}
+
+// ---------------------------------------------------------------------------
+// Literal → Arrow scalar conversion
+// ---------------------------------------------------------------------------
+
+fn literal_scalar_for_parquet_filter(
+ literal: &Datum,
+ file_data_type: &DataType,
+) -> crate::Result<Option<Scalar<ArrayRef>>> {
+ let array: ArrayRef = match file_data_type {
+ DataType::Boolean(_) => match literal {
+ Datum::Bool(value) => Arc::new(BooleanArray::new_scalar(*value).into_inner()),
+ _ => return Ok(None),
+ },
+ DataType::TinyInt(_) => {
+ match integer_literal(literal).and_then(|value| i8::try_from(value).ok()) {
+ Some(value) => Arc::new(Int8Array::new_scalar(value).into_inner()),
+ None => return Ok(None),
+ }
+ }
+ DataType::SmallInt(_) => {
+ match integer_literal(literal).and_then(|value| i16::try_from(value).ok()) {
+ Some(value) => Arc::new(Int16Array::new_scalar(value).into_inner()),
+ None => return Ok(None),
+ }
+ }
+ DataType::Int(_) => {
+ match integer_literal(literal).and_then(|value| i32::try_from(value).ok()) {
+ Some(value) => Arc::new(Int32Array::new_scalar(value).into_inner()),
+ None => return Ok(None),
+ }
+ }
+ DataType::BigInt(_) => {
+ match integer_literal(literal).and_then(|value| i64::try_from(value).ok()) {
+ Some(value) => Arc::new(Int64Array::new_scalar(value).into_inner()),
+ None => return Ok(None),
+ }
+ }
+ DataType::Float(_) => match float32_literal(literal) {
+ Some(value) => Arc::new(Float32Array::new_scalar(value).into_inner()),
+ None => return Ok(None),
+ },
+ DataType::Double(_) => match float64_literal(literal) {
+ Some(value) => Arc::new(Float64Array::new_scalar(value).into_inner()),
+ None => return Ok(None),
+ },
+ DataType::Char(_) | DataType::VarChar(_) => match literal {
+ Datum::String(value) => Arc::new(StringArray::new_scalar(value.as_str()).into_inner()),
+ _ => return Ok(None),
+ },
+ DataType::Binary(_) | DataType::VarBinary(_) => match literal {
+ Datum::Bytes(value) => Arc::new(BinaryArray::new_scalar(value.as_slice()).into_inner()),
+ _ => return Ok(None),
+ },
+ DataType::Date(_) => match literal {
+ Datum::Date(value) => Arc::new(Date32Array::new_scalar(*value).into_inner()),
+ _ => return Ok(None),
+ },
+ DataType::Decimal(decimal) => match literal {
+ Datum::Decimal {
+ unscaled,
+ precision,
+ scale,
+ } if *precision <= decimal.precision() && *scale == decimal.scale() => {
+ let precision =
+ u8::try_from(decimal.precision()).map_err(|_| Error::Unsupported {
+ message: "Decimal precision exceeds Arrow decimal128 range".to_string(),
+ })?;
+ let scale =
+ i8::try_from(decimal.scale() as i32).map_err(|_| Error::Unsupported {
+ message: "Decimal scale exceeds Arrow decimal128 range".to_string(),
+ })?;
+ Arc::new(
+ Decimal128Array::new_scalar(*unscaled)
+ .into_inner()
+ .with_precision_and_scale(precision, scale)
+ .map_err(|e| Error::UnexpectedError {
+ message: format!(
+ "Failed to build decimal scalar for parquet row filter: {e}"
+ ),
+ source: Some(Box::new(e)),
+ })?,
+ )
+ }
+ _ => return Ok(None),
+ },
+ DataType::Time(_)
+ | DataType::Timestamp(_)
+ | DataType::LocalZonedTimestamp(_)
+ | DataType::Array(_)
+ | DataType::Map(_)
+ | DataType::Multiset(_)
+ | DataType::Row(_) => return Ok(None),
+ };
+
+ Ok(Some(Scalar::new(array)))
+}
+
+fn integer_literal(literal: &Datum) -> Option<i128> {
+ match literal {
+ Datum::TinyInt(value) => Some(i128::from(*value)),
+ Datum::SmallInt(value) => Some(i128::from(*value)),
+ Datum::Int(value) => Some(i128::from(*value)),
+ Datum::Long(value) => Some(i128::from(*value)),
+ _ => None,
+ }
+}
+
+fn float32_literal(literal: &Datum) -> Option<f32> {
+ match literal {
+ Datum::Float(value) => Some(*value),
+ Datum::Double(value) => {
+ let casted = *value as f32;
+ ((casted as f64) == *value).then_some(casted)
+ }
+ _ => None,
+ }
+}
+
+fn float64_literal(literal: &Datum) -> Option<f64> {
+ match literal {
+ Datum::Float(value) => Some(f64::from(*value)),
+ Datum::Double(value) => Some(*value),
+ _ => None,
+ }
+}
+
+// ---------------------------------------------------------------------------
+// Row selection helpers (DV, row ranges)
+// ---------------------------------------------------------------------------
+
+fn intersect_optional_row_selections(
+ left: Option<RowSelection>,
+ right: Option<RowSelection>,
+) -> Option<RowSelection> {
+ match (left, right) {
+ (Some(left), Some(right)) => Some(left.intersection(&right)),
+ (Some(selection), None) | (None, Some(selection)) => Some(selection),
+ (None, None) => None,
+ }
+}
+
+/// Build a Parquet [RowSelection] from inclusive `[from, to]` file-local row ranges (0-based).
+fn build_row_ranges_selection(
+ row_group_metadata_list: &[RowGroupMetaData],
+ row_ranges: &[RowRange],
+) -> RowSelection {
+ let total_rows: i64 = row_group_metadata_list.iter().map(|rg| rg.num_rows()).sum();
+ if total_rows == 0 {
+ return vec![].into();
+ }
+
+ let file_end = total_rows - 1;
+ let mut local_ranges: Vec<(usize, usize)> = row_ranges
+ .iter()
+ .filter_map(|r| {
+ if r.to() < 0 || r.from() > file_end {
+ return None;
+ }
+ let local_start = r.from().max(0) as usize;
+ let local_end = (r.to().min(file_end) + 1) as usize;
+ Some((local_start, local_end))
+ })
+ .collect();
+ local_ranges.sort_by_key(|&(s, _)| s);
+
+ let mut selectors: Vec<RowSelector> = Vec::new();
+ let mut cursor: usize = 0;
+ for (start, end) in &local_ranges {
+ if *start > cursor {
+ selectors.push(RowSelector::skip(*start - cursor));
+ }
+ let select_start = (*start).max(cursor);
+ if *end > select_start {
+ selectors.push(RowSelector::select(*end - select_start));
+ }
+ cursor = cursor.max(*end);
+ }
+ let total = total_rows as usize;
+ if cursor < total {
+ selectors.push(RowSelector::skip(total - cursor));
+ }
+ selectors.into()
+}
+
+// ---------------------------------------------------------------------------
+// ArrowFileReader — async Parquet IO adapter
+// ---------------------------------------------------------------------------
+
+/// ArrowFileReader is a wrapper around a FileRead that impls parquets AsyncFileReader.
+///
+/// # TODO
+///
+/// [ParquetObjectReader](https://docs.rs/parquet/latest/src/parquet/arrow/async_reader/store.rs.html#64)
+/// contains the following hints to speed up metadata loading, similar to iceberg, we can consider adding them to this struct:
+///
+/// - `metadata_size_hint`: Provide a hint as to the size of the parquet file's footer.
+/// - `preload_column_index`: Load the Column Index as part of [`Self::get_metadata`].
+/// - `preload_offset_index`: Load the Offset Index as part of [`Self::get_metadata`].
+struct ArrowFileReader {
+ file_size: u64,
+ r: Box<dyn FileRead>,
+}
+
+impl ArrowFileReader {
+ fn new(file_size: u64, r: Box<dyn FileRead>) -> Self {
+ Self { file_size, r }
+ }
+
+ fn read_bytes(&mut self, range: Range<u64>) -> BoxFuture<'_, parquet::errors::Result<Bytes>> {
+ Box::pin(self.r.read(range.start..range.end).map_err(|err| {
+ let err_msg = format!("{err}");
+ parquet::errors::ParquetError::External(err_msg.into())
+ }))
+ }
+}
+
+impl MetadataFetch for ArrowFileReader {
+ fn fetch(&mut self, range: Range<u64>) -> BoxFuture<'_, parquet::errors::Result<Bytes>> {
+ self.read_bytes(range)
+ }
+}
+
+impl AsyncFileReader for ArrowFileReader {
+ fn get_bytes(&mut self, range: Range<u64>) -> BoxFuture<'_, parquet::errors::Result<Bytes>> {
+ self.read_bytes(range)
+ }
+
+ fn get_metadata(
+ &mut self,
+ options: Option<&ArrowReaderOptions>,
+ ) -> BoxFuture<'_, parquet::errors::Result<Arc<ParquetMetaData>>> {
+ let metadata_opts = options.map(|o| o.metadata_options().clone());
+ Box::pin(async move {
+ let file_size = self.file_size;
+ let metadata = ParquetMetaDataReader::new()
+ .with_metadata_options(metadata_opts)
+ .load_and_finish(self, file_size)
+ .await?;
+ Ok(Arc::new(metadata))
+ })
+ }
+}
+
+// ---------------------------------------------------------------------------
+// Tests
+// ---------------------------------------------------------------------------
+
+#[cfg(test)]
+mod tests {
+ use super::build_parquet_row_filter;
+ use crate::spec::{DataField, DataType, Datum, IntType, PredicateBuilder};
+ use parquet::schema::{parser::parse_message_type, types::SchemaDescriptor};
+ use std::sync::Arc;
+
+ fn test_fields() -> Vec<DataField> {
+ vec![
+ DataField::new(0, "id".to_string(), DataType::Int(IntType::new())),
+ DataField::new(1, "score".to_string(), DataType::Int(IntType::new())),
+ ]
+ }
+
+ fn test_parquet_schema() -> SchemaDescriptor {
+ SchemaDescriptor::new(Arc::new(
+ parse_message_type(
+ "
+ message test_schema {
+ OPTIONAL INT32 id;
+ OPTIONAL INT32 score;
+ }
+ ",
+ )
+ .expect("test schema should parse"),
+ ))
+ }
+
+ #[test]
+ fn test_build_parquet_row_filter_supports_null_and_membership_predicates() {
+ let fields = test_fields();
+ let builder = PredicateBuilder::new(&fields);
+ let predicates = vec![
+ builder
+ .is_null("id")
+ .expect("is null predicate should build"),
+ builder
+ .is_in("score", vec![Datum::Int(7)])
+ .expect("in predicate should build"),
+ builder
+ .is_not_in("score", vec![Datum::Int(9)])
+ .expect("not in predicate should build"),
+ ];
+
+ let row_filter = build_parquet_row_filter(&test_parquet_schema(), &predicates, &fields)
+ .expect("parquet row filter should build");
+
+ assert!(row_filter.is_some());
+ }
+}
diff --git a/crates/paimon/src/arrow/mod.rs b/crates/paimon/src/arrow/mod.rs
index f524d59..f48b9da 100644
--- a/crates/paimon/src/arrow/mod.rs
+++ b/crates/paimon/src/arrow/mod.rs
@@ -16,6 +16,7 @@
// under the License.
pub(crate) mod filtering;
+pub(crate) mod format;
mod reader;
pub(crate) mod schema_evolution;
diff --git a/crates/paimon/src/arrow/reader.rs b/crates/paimon/src/arrow/reader.rs
index 4524ed4..3f19bd8 100644
--- a/crates/paimon/src/arrow/reader.rs
+++ b/crates/paimon/src/arrow/reader.rs
@@ -16,46 +16,22 @@
// under the License.
use crate::arrow::build_target_arrow_schema;
-use crate::arrow::filtering::{
- build_field_mapping, predicates_may_match_with_schema, StatsAccessor,
-};
+use crate::arrow::format::create_format_reader;
use crate::arrow::schema_evolution::{create_index_mapping, NULL_FIELD_INDEX};
use crate::deletion_vector::{DeletionVector, DeletionVectorFactory};
-use crate::io::{FileIO, FileRead, FileStatus};
-use crate::spec::{
- DataField, DataFileMeta, DataType, Datum, Predicate, PredicateOperator, ROW_ID_FIELD_NAME,
-};
+use crate::io::FileIO;
+use crate::spec::{DataField, DataFileMeta, Predicate, ROW_ID_FIELD_NAME};
use crate::table::schema_manager::SchemaManager;
use crate::table::ArrowRecordBatchStream;
use crate::table::RowRange;
use crate::{DataSplit, Error};
-use arrow_array::{
- Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Decimal128Array, Float32Array,
- Float64Array, Int16Array, Int32Array, Int64Array, Int8Array, RecordBatch, Scalar, StringArray,
-};
+use arrow_array::{Array, Int64Array, RecordBatch};
use arrow_cast::cast;
-use arrow_ord::cmp::{
- eq as arrow_eq, gt as arrow_gt, gt_eq as arrow_gt_eq, lt as arrow_lt, lt_eq as arrow_lt_eq,
- neq as arrow_neq,
-};
-use arrow_schema::ArrowError;
use async_stream::try_stream;
-use bytes::Bytes;
-use futures::future::BoxFuture;
-use futures::{StreamExt, TryFutureExt};
-use parquet::arrow::arrow_reader::{
- ArrowPredicate, ArrowPredicateFn, ArrowReaderOptions, RowFilter, RowSelection, RowSelector,
-};
-use parquet::arrow::async_reader::{AsyncFileReader, MetadataFetch};
-use parquet::arrow::{ParquetRecordBatchStreamBuilder, ProjectionMask};
-use parquet::file::metadata::ParquetMetaDataReader;
-use parquet::file::metadata::{ParquetMetaData, RowGroupMetaData};
-use parquet::file::statistics::Statistics as ParquetStatistics;
+use futures::StreamExt;
use std::collections::HashMap;
-use std::ops::Range;
use std::sync::Arc;
-use tokio::try_join;
/// Builder to create ArrowReader
pub struct ArrowReaderBuilder {
@@ -404,14 +380,14 @@ fn read_single_file_stream(
let target_schema = build_target_arrow_schema(&read_type)?;
let file_fields = data_fields.clone().unwrap_or_else(|| table_fields.clone());
- // Compute index mapping and determine which columns to read from the parquet file.
+ // Compute index mapping and determine which columns to read from the file.
// If data_fields is provided, use field-ID-based mapping; otherwise use read_type names directly.
- let (parquet_read_fields, index_mapping) = if let Some(ref df) = data_fields {
+ let (projected_read_fields, index_mapping) = if let Some(ref df) = data_fields {
let mapping = create_index_mapping(&read_type, df);
match mapping {
Some(ref idx_map) => {
// Only read data fields that are referenced by the index mapping.
- // Dedup by data field index to avoid duplicate parquet column projections.
+ // Dedup by data field index to avoid duplicate column projections.
let mut seen = std::collections::HashSet::new();
let fields_to_read: Vec<DataField> = idx_map
.iter()
@@ -430,89 +406,47 @@ fn read_single_file_stream(
(read_type.clone(), None)
};
- let parquet_column_names: Vec<String> = parquet_read_fields
- .iter()
- .map(|f| f.name().to_string())
- .collect();
-
- Ok(try_stream! {
- let path_to_read = split.data_file_path(&file_meta);
- if !path_to_read.to_ascii_lowercase().ends_with(".parquet") {
- Err(Error::Unsupported {
- message: format!(
- "unsupported file format: only .parquet is supported, got: {path_to_read}"
- ),
- })?
- }
-
- let parquet_file = file_io.new_input(&path_to_read)?;
- let (parquet_metadata, parquet_reader) =
- try_join!(parquet_file.metadata(), parquet_file.reader())?;
- let arrow_file_reader = ArrowFileReader::new(parquet_metadata, parquet_reader);
-
- let mut batch_stream_builder = ParquetRecordBatchStreamBuilder::new(arrow_file_reader).await?;
-
- // Project columns by root-level index to correctly handle complex types
- // (ARRAY, MAP, STRUCT). `ProjectionMask::columns` matches leaf column names
- // which doesn't work for nested types; `ProjectionMask::roots` uses top-level
- // field indices instead.
- let parquet_schema = batch_stream_builder.parquet_schema().clone();
- let root_schema = parquet_schema.root_schema();
- let root_indices: Vec<usize> = parquet_column_names
- .iter()
- .filter_map(|name| {
- root_schema
- .get_fields()
- .iter()
- .position(|f| f.name() == name)
- })
- .collect();
-
- let mask = ProjectionMask::roots(&parquet_schema, root_indices);
- batch_stream_builder = batch_stream_builder.with_projection(mask);
- let row_filter =
- build_parquet_row_filter(&parquet_schema, &predicates, &table_fields, &file_fields)?;
- if let Some(row_filter) = row_filter {
- batch_stream_builder = batch_stream_builder.with_row_filter(row_filter);
- }
-
- let predicate_row_selection = build_predicate_row_selection(
- batch_stream_builder.metadata().row_groups(),
+ // Remap predicates from table-level to file-level indices.
+ let file_predicates = {
+ let remapped = crate::arrow::filtering::remap_predicates_to_file(
&predicates,
&table_fields,
&file_fields,
- )?;
- let mut row_selection = predicate_row_selection;
-
- if let Some(ref dv) = dv {
- if !dv.is_empty() {
- let delete_row_selection =
- build_deletes_row_selection(batch_stream_builder.metadata().row_groups(), dv)?;
- row_selection = intersect_optional_row_selections(
- row_selection,
- Some(delete_row_selection),
- );
- }
- }
- if let Some(ref ranges) = row_ranges {
- let range_selection = build_row_ranges_selection(
- batch_stream_builder.metadata().row_groups(),
- ranges,
- file_meta.first_row_id.unwrap_or(0),
- );
- row_selection = intersect_optional_row_selections(
- row_selection,
- Some(range_selection),
- );
- }
- if let Some(row_selection) = row_selection {
- batch_stream_builder = batch_stream_builder.with_row_selection(row_selection);
- }
- if let Some(size) = batch_size {
- batch_stream_builder = batch_stream_builder.with_batch_size(size);
+ );
+ if remapped.is_empty() {
+ None
+ } else {
+ Some(crate::arrow::format::FilePredicates {
+ predicates: remapped,
+ file_fields: file_fields.clone(),
+ })
}
+ };
+
+ Ok(try_stream! {
+ let path_to_read = split.data_file_path(&file_meta);
+ let format_reader = create_format_reader(&path_to_read)?;
+ let input_file = file_io.new_input(&path_to_read)?;
+ let file_reader = input_file.reader().await?;
+ let local_ranges = row_ranges.as_ref().map(|ranges| {
+ to_local_row_ranges(ranges, file_meta.first_row_id.unwrap_or(0), file_meta.row_count)
+ });
+
+ let row_selection = merge_row_selection(
+ file_meta.row_count,
+ dv.as_deref(),
+ local_ranges.as_deref(),
+ );
+
+ let mut batch_stream = format_reader.read_batch_stream(
+ Box::new(file_reader),
+ file_meta.file_size as u64,
+ &projected_read_fields,
+ file_predicates.as_ref(),
+ batch_size,
+ row_selection,
+ ).await?;
- let mut batch_stream = batch_stream_builder.build()?;
while let Some(batch) = batch_stream.next().await {
let batch = batch?;
let num_rows = batch.num_rows();
@@ -857,621 +791,94 @@ fn merge_files_by_columns(
.boxed())
}
-fn intersect_optional_row_selections(
- left: Option<RowSelection>,
- right: Option<RowSelection>,
-) -> Option<RowSelection> {
- match (left, right) {
- (Some(left), Some(right)) => Some(left.intersection(&right)),
- (Some(selection), None) | (None, Some(selection)) => Some(selection),
- (None, None) => None,
- }
-}
-
-fn sanitize_filter_mask(mask: BooleanArray) -> BooleanArray {
- if mask.null_count() == 0 {
- return mask;
- }
-
- boolean_mask_from_predicate(mask.len(), |row_index| {
- mask.is_valid(row_index) && mask.value(row_index)
- })
-}
-
-fn evaluate_exact_leaf_predicate(
- array: &ArrayRef,
- data_type: &DataType,
- op: PredicateOperator,
- literals: &[Datum],
-) -> Result<BooleanArray, ArrowError> {
- match op {
- PredicateOperator::IsNull => Ok(boolean_mask_from_predicate(array.len(), |row_index| {
- array.is_null(row_index)
- })),
- PredicateOperator::IsNotNull => Ok(boolean_mask_from_predicate(array.len(), |row_index| {
- array.is_valid(row_index)
- })),
- PredicateOperator::In | PredicateOperator::NotIn => {
- evaluate_set_membership_predicate(array, data_type, op, literals)
- }
- PredicateOperator::Eq
- | PredicateOperator::NotEq
- | PredicateOperator::Lt
- | PredicateOperator::LtEq
- | PredicateOperator::Gt
- | PredicateOperator::GtEq => {
- let Some(literal) = literals.first() else {
- return Ok(BooleanArray::from(vec![true; array.len()]));
- };
- let Some(scalar) = literal_scalar_for_parquet_filter(literal, data_type)
- .map_err(|e| ArrowError::ComputeError(e.to_string()))?
- else {
- return Ok(BooleanArray::from(vec![true; array.len()]));
- };
- let result = evaluate_column_predicate(array, &scalar, op)?;
- Ok(sanitize_filter_mask(result))
- }
- }
-}
-
-fn evaluate_set_membership_predicate(
- array: &ArrayRef,
- data_type: &DataType,
- op: PredicateOperator,
- literals: &[Datum],
-) -> Result<BooleanArray, ArrowError> {
- if literals.is_empty() {
- return Ok(match op {
- PredicateOperator::In => BooleanArray::from(vec![false; array.len()]),
- PredicateOperator::NotIn => {
- boolean_mask_from_predicate(array.len(), |row_index| array.is_valid(row_index))
+/// Convert absolute RowRanges to file-local 0-based ranges.
+fn to_local_row_ranges(
+ row_ranges: &[RowRange],
+ first_row_id: i64,
+ row_count: i64,
+) -> Vec<RowRange> {
+ let file_end = first_row_id + row_count - 1;
+ row_ranges
+ .iter()
+ .filter_map(|r| {
+ if r.to() < first_row_id || r.from() > file_end {
+ return None;
}
- PredicateOperator::IsNull
- | PredicateOperator::IsNotNull
- | PredicateOperator::Eq
- | PredicateOperator::NotEq
- | PredicateOperator::Lt
- | PredicateOperator::LtEq
- | PredicateOperator::Gt
- | PredicateOperator::GtEq => unreachable!(),
- });
- }
-
- let mut combined = match op {
- PredicateOperator::In => BooleanArray::from(vec![false; array.len()]),
- PredicateOperator::NotIn => {
- boolean_mask_from_predicate(array.len(), |row_index| array.is_valid(row_index))
- }
- PredicateOperator::IsNull
- | PredicateOperator::IsNotNull
- | PredicateOperator::Eq
- | PredicateOperator::NotEq
- | PredicateOperator::Lt
- | PredicateOperator::LtEq
- | PredicateOperator::Gt
- | PredicateOperator::GtEq => unreachable!(),
- };
-
- for literal in literals {
- let Some(scalar) = literal_scalar_for_parquet_filter(literal, data_type)
- .map_err(|e| ArrowError::ComputeError(e.to_string()))?
- else {
- return Ok(BooleanArray::from(vec![true; array.len()]));
- };
- let comparison_op = match op {
- PredicateOperator::In => PredicateOperator::Eq,
- PredicateOperator::NotIn => PredicateOperator::NotEq,
- PredicateOperator::IsNull
- | PredicateOperator::IsNotNull
- | PredicateOperator::Eq
- | PredicateOperator::NotEq
- | PredicateOperator::Lt
- | PredicateOperator::LtEq
- | PredicateOperator::Gt
- | PredicateOperator::GtEq => unreachable!(),
- };
- let mask = sanitize_filter_mask(evaluate_column_predicate(array, &scalar, comparison_op)?);
- combined = combine_filter_masks(&combined, &mask, matches!(op, PredicateOperator::In));
- }
-
- Ok(combined)
-}
-
-fn combine_filter_masks(left: &BooleanArray, right: &BooleanArray, use_or: bool) -> BooleanArray {
- debug_assert_eq!(left.len(), right.len());
- boolean_mask_from_predicate(left.len(), |row_index| {
- if use_or {
- left.value(row_index) || right.value(row_index)
- } else {
- left.value(row_index) && right.value(row_index)
- }
- })
-}
-
-fn boolean_mask_from_predicate(
- len: usize,
- mut predicate: impl FnMut(usize) -> bool,
-) -> BooleanArray {
- BooleanArray::from((0..len).map(&mut predicate).collect::<Vec<_>>())
-}
-
-struct ParquetRowGroupStats<'a> {
- row_group: &'a RowGroupMetaData,
- column_indices: &'a [Option<usize>],
+ let local_from = (r.from() - first_row_id).max(0);
+ let local_to = (r.to() - first_row_id).min(row_count - 1);
+ Some(RowRange::new(local_from, local_to))
+ })
+ .collect()
}
-impl StatsAccessor for ParquetRowGroupStats<'_> {
- fn row_count(&self) -> i64 {
- self.row_group.num_rows()
+/// Merge DV and row_ranges into a unified list of 0-based inclusive RowRanges.
+/// Returns `None` if no filtering is needed (no DV and no ranges).
+///
+/// Complexity: O(D + R) where D = number of deleted rows, R = number of ranges.
+fn merge_row_selection(
+ row_count: i64,
+ dv: Option<&DeletionVector>,
+ row_ranges: Option<&[RowRange]>,
+) -> Option<Vec<RowRange>> {
+ let has_dv = dv.is_some_and(|d| !d.is_empty());
+ let has_ranges = row_ranges.is_some();
+ if !has_dv && !has_ranges {
+ return None;
}
- fn null_count(&self, index: usize) -> Option<i64> {
- let _ = index;
- // parquet::Statistics::null_count_opt() may return Some(0) even when
- // the null-count statistic is absent, so treating it as authoritative
- // would make IS NULL / IS NOT NULL pruning unsafe. Fail open here.
- None
+ // Fast path: no DV, just return row_ranges as-is.
+ if !has_dv {
+ return row_ranges.map(|r| r.to_vec());
}
- fn min_value(&self, index: usize, data_type: &DataType) -> Option<Datum> {
- let column_index = self.column_indices.get(index).copied().flatten()?;
- parquet_stats_to_datum(
- self.row_group.column(column_index).statistics()?,
- data_type,
- true,
- )
- }
+ // Build non-deleted ranges from DV (sorted iterator).
+ let dv_ranges = dv_to_non_deleted_ranges(dv.unwrap(), row_count);
- fn max_value(&self, index: usize, data_type: &DataType) -> Option<Datum> {
- let column_index = self.column_indices.get(index).copied().flatten()?;
- parquet_stats_to_datum(
- self.row_group.column(column_index).statistics()?,
- data_type,
- false,
- )
+ match row_ranges {
+ Some(ranges) => Some(intersect_sorted_ranges(&dv_ranges, ranges)),
+ None => Some(dv_ranges),
}
}
-fn build_predicate_row_selection(
- row_groups: &[RowGroupMetaData],
- predicates: &[Predicate],
- table_fields: &[DataField],
- file_fields: &[DataField],
-) -> crate::Result<Option<RowSelection>> {
- if predicates.is_empty() || row_groups.is_empty() {
- return Ok(None);
- }
-
- let field_mapping = build_field_mapping(table_fields, file_fields);
- let column_indices = build_row_group_column_indices(row_groups[0].columns(), file_fields);
- let mut selectors = Vec::with_capacity(row_groups.len());
- let mut all_selected = true;
-
- for row_group in row_groups {
- let stats = ParquetRowGroupStats {
- row_group,
- column_indices: &column_indices,
- };
- let may_match =
- predicates_may_match_with_schema(predicates, &stats, &field_mapping, file_fields);
- if !may_match {
- all_selected = false;
+/// Convert a DeletionVector into sorted non-deleted inclusive RowRanges.
+/// The DV iterator yields sorted deleted positions.
+fn dv_to_non_deleted_ranges(dv: &DeletionVector, row_count: i64) -> Vec<RowRange> {
+ let mut result = Vec::new();
+ let mut cursor: i64 = 0;
+ for deleted in dv.iter() {
+ let del = deleted as i64;
+ if del >= row_count {
+ break;
}
- selectors.push(if may_match {
- RowSelector::select(row_group.num_rows() as usize)
- } else {
- RowSelector::skip(row_group.num_rows() as usize)
- });
- }
-
- if all_selected {
- Ok(None)
- } else {
- Ok(Some(selectors.into()))
- }
-}
-
-fn build_parquet_row_filter(
- parquet_schema: &parquet::schema::types::SchemaDescriptor,
- predicates: &[Predicate],
- table_fields: &[DataField],
- file_fields: &[DataField],
-) -> crate::Result<Option<RowFilter>> {
- // Keep decode-time filtering intentionally narrow for the submit-ready
- // Parquet path. Unsupported predicate shapes or types remain residual
- // filters above the reader.
- if predicates.is_empty() {
- return Ok(None);
- }
-
- let field_mapping = build_field_mapping(table_fields, file_fields);
- let mut filters: Vec<Box<dyn ArrowPredicate>> = Vec::new();
-
- for predicate in predicates {
- if let Some(filter) =
- build_parquet_arrow_predicate(parquet_schema, predicate, &field_mapping, file_fields)?
- {
- filters.push(filter);
+ if del > cursor {
+ result.push(RowRange::new(cursor, del - 1));
}
+ cursor = del + 1;
}
-
- if filters.is_empty() {
- Ok(None)
- } else {
- Ok(Some(RowFilter::new(filters)))
- }
-}
-
-fn build_parquet_arrow_predicate(
- parquet_schema: &parquet::schema::types::SchemaDescriptor,
- predicate: &Predicate,
- field_mapping: &[Option<usize>],
- file_fields: &[DataField],
-) -> crate::Result<Option<Box<dyn ArrowPredicate>>> {
- let Predicate::Leaf {
- index,
- data_type: _,
- op,
- literals,
- ..
- } = predicate
- else {
- return Ok(None);
- };
- if !predicate_supported_for_parquet_row_filter(*op) {
- return Ok(None);
- }
-
- let Some(file_index) = field_mapping.get(*index).copied().flatten() else {
- return Ok(None);
- };
- let Some(file_field) = file_fields.get(file_index) else {
- return Ok(None);
- };
- let Some(root_index) = parquet_root_index(parquet_schema, file_field.name()) else {
- return Ok(None);
- };
- if !parquet_row_filter_literals_supported(*op, literals, file_field.data_type())? {
- return Ok(None);
+ if cursor < row_count {
+ result.push(RowRange::new(cursor, row_count - 1));
}
-
- let projection = ProjectionMask::roots(parquet_schema, [root_index]);
- let op = *op;
- let data_type = file_field.data_type().clone();
- let literals = literals.to_vec();
- Ok(Some(Box::new(ArrowPredicateFn::new(
- projection,
- move |batch: RecordBatch| {
- let Some(column) = batch.columns().first() else {
- return Ok(BooleanArray::new_null(batch.num_rows()));
- };
- evaluate_exact_leaf_predicate(column, &data_type, op, &literals)
- },
- ))))
-}
-
-fn predicate_supported_for_parquet_row_filter(op: PredicateOperator) -> bool {
- matches!(
- op,
- PredicateOperator::IsNull
- | PredicateOperator::IsNotNull
- | PredicateOperator::Eq
- | PredicateOperator::NotEq
- | PredicateOperator::Lt
- | PredicateOperator::LtEq
- | PredicateOperator::Gt
- | PredicateOperator::GtEq
- | PredicateOperator::In
- | PredicateOperator::NotIn
- )
+ result
}
-fn parquet_row_filter_literals_supported(
- op: PredicateOperator,
- literals: &[Datum],
- file_data_type: &DataType,
-) -> crate::Result<bool> {
- match op {
- PredicateOperator::IsNull | PredicateOperator::IsNotNull => Ok(true),
- PredicateOperator::Eq
- | PredicateOperator::NotEq
- | PredicateOperator::Lt
- | PredicateOperator::LtEq
- | PredicateOperator::Gt
- | PredicateOperator::GtEq => {
- let Some(literal) = literals.first() else {
- return Ok(false);
- };
- Ok(literal_scalar_for_parquet_filter(literal, file_data_type)?.is_some())
+/// Intersect two sorted lists of inclusive RowRanges using a merge-style scan.
+fn intersect_sorted_ranges(a: &[RowRange], b: &[RowRange]) -> Vec<RowRange> {
+ let mut result = Vec::new();
+ let (mut i, mut j) = (0, 0);
+ while i < a.len() && j < b.len() {
+ let from = a[i].from().max(b[j].from());
+ let to = a[i].to().min(b[j].to());
+ if from <= to {
+ result.push(RowRange::new(from, to));
}
- PredicateOperator::In | PredicateOperator::NotIn => {
- for literal in literals {
- if literal_scalar_for_parquet_filter(literal, file_data_type)?.is_none() {
- return Ok(false);
- }
- }
- Ok(true)
- }
- }
-}
-
-fn parquet_root_index(
- parquet_schema: &parquet::schema::types::SchemaDescriptor,
- root_name: &str,
-) -> Option<usize> {
- parquet_schema
- .root_schema()
- .get_fields()
- .iter()
- .position(|field| field.name() == root_name)
-}
-
-fn evaluate_column_predicate(
- column: &ArrayRef,
- scalar: &Scalar<ArrayRef>,
- op: PredicateOperator,
-) -> Result<BooleanArray, ArrowError> {
- match op {
- PredicateOperator::Eq => arrow_eq(column, scalar),
- PredicateOperator::NotEq => arrow_neq(column, scalar),
- PredicateOperator::Lt => arrow_lt(column, scalar),
- PredicateOperator::LtEq => arrow_lt_eq(column, scalar),
- PredicateOperator::Gt => arrow_gt(column, scalar),
- PredicateOperator::GtEq => arrow_gt_eq(column, scalar),
- PredicateOperator::IsNull
- | PredicateOperator::IsNotNull
- | PredicateOperator::In
- | PredicateOperator::NotIn => Ok(BooleanArray::new_null(column.len())),
- }
-}
-
-fn literal_scalar_for_parquet_filter(
- literal: &Datum,
- file_data_type: &DataType,
-) -> crate::Result<Option<Scalar<ArrayRef>>> {
- let array: ArrayRef = match file_data_type {
- DataType::Boolean(_) => match literal {
- Datum::Bool(value) => Arc::new(BooleanArray::new_scalar(*value).into_inner()),
- _ => return Ok(None),
- },
- DataType::TinyInt(_) => {
- match integer_literal(literal).and_then(|value| i8::try_from(value).ok()) {
- Some(value) => Arc::new(Int8Array::new_scalar(value).into_inner()),
- None => return Ok(None),
- }
- }
- DataType::SmallInt(_) => {
- match integer_literal(literal).and_then(|value| i16::try_from(value).ok()) {
- Some(value) => Arc::new(Int16Array::new_scalar(value).into_inner()),
- None => return Ok(None),
- }
- }
- DataType::Int(_) => {
- match integer_literal(literal).and_then(|value| i32::try_from(value).ok()) {
- Some(value) => Arc::new(Int32Array::new_scalar(value).into_inner()),
- None => return Ok(None),
- }
- }
- DataType::BigInt(_) => {
- match integer_literal(literal).and_then(|value| i64::try_from(value).ok()) {
- Some(value) => Arc::new(Int64Array::new_scalar(value).into_inner()),
- None => return Ok(None),
- }
- }
- DataType::Float(_) => match float32_literal(literal) {
- Some(value) => Arc::new(Float32Array::new_scalar(value).into_inner()),
- None => return Ok(None),
- },
- DataType::Double(_) => match float64_literal(literal) {
- Some(value) => Arc::new(Float64Array::new_scalar(value).into_inner()),
- None => return Ok(None),
- },
- DataType::Char(_) | DataType::VarChar(_) => match literal {
- Datum::String(value) => Arc::new(StringArray::new_scalar(value.as_str()).into_inner()),
- _ => return Ok(None),
- },
- DataType::Binary(_) | DataType::VarBinary(_) => match literal {
- Datum::Bytes(value) => Arc::new(BinaryArray::new_scalar(value.as_slice()).into_inner()),
- _ => return Ok(None),
- },
- DataType::Date(_) => match literal {
- Datum::Date(value) => Arc::new(Date32Array::new_scalar(*value).into_inner()),
- _ => return Ok(None),
- },
- DataType::Decimal(decimal) => match literal {
- Datum::Decimal {
- unscaled,
- precision,
- scale,
- } if *precision <= decimal.precision() && *scale == decimal.scale() => {
- let precision =
- u8::try_from(decimal.precision()).map_err(|_| Error::Unsupported {
- message: "Decimal precision exceeds Arrow decimal128 range".to_string(),
- })?;
- let scale =
- i8::try_from(decimal.scale() as i32).map_err(|_| Error::Unsupported {
- message: "Decimal scale exceeds Arrow decimal128 range".to_string(),
- })?;
- Arc::new(
- Decimal128Array::new_scalar(*unscaled)
- .into_inner()
- .with_precision_and_scale(precision, scale)
- .map_err(|e| Error::UnexpectedError {
- message: format!(
- "Failed to build decimal scalar for parquet row filter: {e}"
- ),
- source: Some(Box::new(e)),
- })?,
- )
- }
- _ => return Ok(None),
- },
- DataType::Time(_)
- | DataType::Timestamp(_)
- | DataType::LocalZonedTimestamp(_)
- | DataType::Array(_)
- | DataType::Map(_)
- | DataType::Multiset(_)
- | DataType::Row(_) => return Ok(None),
- };
-
- Ok(Some(Scalar::new(array)))
-}
-
-fn integer_literal(literal: &Datum) -> Option<i128> {
- match literal {
- Datum::TinyInt(value) => Some(i128::from(*value)),
- Datum::SmallInt(value) => Some(i128::from(*value)),
- Datum::Int(value) => Some(i128::from(*value)),
- Datum::Long(value) => Some(i128::from(*value)),
- _ => None,
- }
-}
-
-fn float32_literal(literal: &Datum) -> Option<f32> {
- match literal {
- Datum::Float(value) => Some(*value),
- Datum::Double(value) => {
- let casted = *value as f32;
- ((casted as f64) == *value).then_some(casted)
- }
- _ => None,
- }
-}
-
-fn float64_literal(literal: &Datum) -> Option<f64> {
- match literal {
- Datum::Float(value) => Some(f64::from(*value)),
- Datum::Double(value) => Some(*value),
- _ => None,
- }
-}
-
-fn build_row_group_column_indices(
- columns: &[parquet::file::metadata::ColumnChunkMetaData],
- file_fields: &[DataField],
-) -> Vec<Option<usize>> {
- let mut by_root_name: HashMap<&str, Option<usize>> = HashMap::new();
- for (column_index, column) in columns.iter().enumerate() {
- let Some(root_name) = column.column_path().parts().first() else {
- continue;
- };
- let entry = by_root_name
- .entry(root_name.as_str())
- .or_insert(Some(column_index));
- if entry.is_some() && *entry != Some(column_index) {
- *entry = None;
- }
- }
-
- file_fields
- .iter()
- .map(|field| by_root_name.get(field.name()).copied().flatten())
- .collect()
-}
-
-fn parquet_stats_to_datum(
- stats: &ParquetStatistics,
- data_type: &DataType,
- is_min: bool,
-) -> Option<Datum> {
- let exact = if is_min {
- stats.min_is_exact()
- } else {
- stats.max_is_exact()
- };
- if !exact {
- return None;
- }
-
- match (stats, data_type) {
- (ParquetStatistics::Boolean(stats), DataType::Boolean(_)) => {
- exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
- .copied()
- .map(Datum::Bool)
- }
- (ParquetStatistics::Int32(stats), DataType::TinyInt(_)) => {
- exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
- .and_then(|value| i8::try_from(*value).ok())
- .map(Datum::TinyInt)
- }
- (ParquetStatistics::Int32(stats), DataType::SmallInt(_)) => {
- exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
- .and_then(|value| i16::try_from(*value).ok())
- .map(Datum::SmallInt)
- }
- (ParquetStatistics::Int32(stats), DataType::Int(_)) => {
- exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
- .copied()
- .map(Datum::Int)
- }
- (ParquetStatistics::Int32(stats), DataType::Date(_)) => {
- exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
- .copied()
- .map(Datum::Date)
- }
- (ParquetStatistics::Int32(stats), DataType::Time(_)) => {
- exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
- .copied()
- .map(Datum::Time)
- }
- (ParquetStatistics::Int64(stats), DataType::BigInt(_)) => {
- exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
- .copied()
- .map(Datum::Long)
- }
- (ParquetStatistics::Int64(stats), DataType::Timestamp(ts)) if ts.precision() <= 3 => {
- exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
- .copied()
- .map(|millis| Datum::Timestamp { millis, nanos: 0 })
- }
- (ParquetStatistics::Int64(stats), DataType::LocalZonedTimestamp(ts))
- if ts.precision() <= 3 =>
- {
- exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
- .copied()
- .map(|millis| Datum::LocalZonedTimestamp { millis, nanos: 0 })
- }
- (ParquetStatistics::Float(stats), DataType::Float(_)) => {
- exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
- .copied()
- .map(Datum::Float)
- }
- (ParquetStatistics::Double(stats), DataType::Double(_)) => {
- exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
- .copied()
- .map(Datum::Double)
- }
- (ParquetStatistics::ByteArray(stats), DataType::Char(_))
- | (ParquetStatistics::ByteArray(stats), DataType::VarChar(_)) => {
- exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
- .and_then(|value| std::str::from_utf8(value.data()).ok())
- .map(|value| Datum::String(value.to_string()))
- }
- (ParquetStatistics::ByteArray(stats), DataType::Binary(_))
- | (ParquetStatistics::ByteArray(stats), DataType::VarBinary(_)) => {
- exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
- .map(|value| Datum::Bytes(value.data().to_vec()))
- }
- (ParquetStatistics::FixedLenByteArray(stats), DataType::Binary(_))
- | (ParquetStatistics::FixedLenByteArray(stats), DataType::VarBinary(_)) => {
- exact_parquet_value(is_min, stats.min_opt(), stats.max_opt())
- .map(|value| Datum::Bytes(value.data().to_vec()))
+ // Advance the range that ends first.
+ if a[i].to() < b[j].to() {
+ i += 1;
+ } else {
+ j += 1;
}
- _ => None,
- }
-}
-
-fn exact_parquet_value<'a, T>(
- is_min: bool,
- min: Option<&'a T>,
- max: Option<&'a T>,
-) -> Option<&'a T> {
- if is_min {
- min
- } else {
- max
}
+ result
}
/// Expand row_ranges into a flat sequence of selected row IDs for a file.
@@ -1536,226 +943,3 @@ fn append_null_row_id_column(
let array: Arc<dyn arrow_array::Array> = Arc::new(Int64Array::new_null(batch.num_rows()));
insert_column_at(batch, array, insert_index, output_schema)
}
-
-/// Build a Parquet [RowSelection] from inclusive `[from, to]` row ranges.
-fn build_row_ranges_selection(
- row_group_metadata_list: &[RowGroupMetaData],
- row_ranges: &[RowRange],
- file_first_row_id: i64,
-) -> RowSelection {
- let total_rows: i64 = row_group_metadata_list.iter().map(|rg| rg.num_rows()).sum();
- if total_rows == 0 {
- return vec![].into();
- }
-
- let file_end_row_id = file_first_row_id.saturating_add(total_rows - 1);
- let mut local_ranges: Vec<(usize, usize)> = row_ranges
- .iter()
- .filter_map(|r| {
- if r.to() < file_first_row_id || r.from() > file_end_row_id {
- return None;
- }
- let local_start = (r.from() - file_first_row_id).max(0) as usize;
- let clamped_to = r.to().min(file_end_row_id);
- let local_end = (clamped_to + 1 - file_first_row_id) as usize;
- Some((local_start, local_end))
- })
- .collect();
- local_ranges.sort_by_key(|&(s, _)| s);
-
- let mut selectors: Vec<RowSelector> = Vec::new();
- let mut cursor: usize = 0;
- for (start, end) in &local_ranges {
- if *start > cursor {
- selectors.push(RowSelector::skip(*start - cursor));
- }
- let select_start = (*start).max(cursor);
- if *end > select_start {
- selectors.push(RowSelector::select(*end - select_start));
- }
- cursor = cursor.max(*end);
- }
- let total = total_rows as usize;
- if cursor < total {
- selectors.push(RowSelector::skip(total - cursor));
- }
- selectors.into()
-}
-
-/// Builds a Parquet [RowSelection] from deletion vector.
-/// Only rows not in the deletion vector are selected; deleted rows are skipped at read time.
-/// todo: Uses [DeletionVectorIterator] with [advance_to](DeletionVectorIterator::advance_to) when skipping row groups similar to iceberg-rust
-fn build_deletes_row_selection(
- row_group_metadata_list: &[RowGroupMetaData],
- deletion_vector: &DeletionVector,
-) -> crate::Result<RowSelection> {
- let mut delete_iter = deletion_vector.iter();
-
- let mut results: Vec<RowSelector> = Vec::new();
- let mut current_row_group_base_idx: u64 = 0;
- let mut next_deleted_row_idx_opt = delete_iter.next();
-
- for row_group_metadata in row_group_metadata_list {
- let row_group_num_rows = row_group_metadata.num_rows() as u64;
- let next_row_group_base_idx = current_row_group_base_idx + row_group_num_rows;
-
- let mut next_deleted_row_idx = match next_deleted_row_idx_opt {
- Some(next_deleted_row_idx) => {
- if next_deleted_row_idx >= next_row_group_base_idx {
- results.push(RowSelector::select(row_group_num_rows as usize));
- current_row_group_base_idx += row_group_num_rows;
- continue;
- }
- next_deleted_row_idx
- }
- None => {
- results.push(RowSelector::select(row_group_num_rows as usize));
- current_row_group_base_idx += row_group_num_rows;
- continue;
- }
- };
-
- let mut current_idx = current_row_group_base_idx;
- 'chunks: while next_deleted_row_idx < next_row_group_base_idx {
- if current_idx < next_deleted_row_idx {
- let run_length = next_deleted_row_idx - current_idx;
- results.push(RowSelector::select(run_length as usize));
- current_idx += run_length;
- }
- let mut run_length = 0u64;
- while next_deleted_row_idx == current_idx
- && next_deleted_row_idx < next_row_group_base_idx
- {
- run_length += 1;
- current_idx += 1;
- next_deleted_row_idx_opt = delete_iter.next();
- next_deleted_row_idx = match next_deleted_row_idx_opt {
- Some(v) => v,
- None => {
- results.push(RowSelector::skip(run_length as usize));
- break 'chunks;
- }
- };
- }
- if run_length > 0 {
- results.push(RowSelector::skip(run_length as usize));
- }
- }
- if current_idx < next_row_group_base_idx {
- results.push(RowSelector::select(
- (next_row_group_base_idx - current_idx) as usize,
- ));
- }
- current_row_group_base_idx += row_group_num_rows;
- }
-
- Ok(results.into())
-}
-
-/// ArrowFileReader is a wrapper around a FileRead that impls parquets AsyncFileReader.
-///
-/// # TODO
-///
-/// [ParquetObjectReader](https://docs.rs/parquet/latest/src/parquet/arrow/async_reader/store.rs.html#64)
-/// contains the following hints to speed up metadata loading, similar to iceberg, we can consider adding them to this struct:
-///
-/// - `metadata_size_hint`: Provide a hint as to the size of the parquet file's footer.
-/// - `preload_column_index`: Load the Column Index as part of [`Self::get_metadata`].
-/// - `preload_offset_index`: Load the Offset Index as part of [`Self::get_metadata`].
-struct ArrowFileReader<R: FileRead> {
- meta: FileStatus,
- r: R,
-}
-
-impl<R: FileRead> ArrowFileReader<R> {
- /// Create a new ArrowFileReader
- fn new(meta: FileStatus, r: R) -> Self {
- Self { meta, r }
- }
-
- fn read_bytes(&mut self, range: Range<u64>) -> BoxFuture<'_, parquet::errors::Result<Bytes>> {
- Box::pin(self.r.read(range.start..range.end).map_err(|err| {
- let err_msg = format!("{err}");
- parquet::errors::ParquetError::External(err_msg.into())
- }))
- }
-}
-
-impl<R: FileRead> MetadataFetch for ArrowFileReader<R> {
- fn fetch(&mut self, range: Range<u64>) -> BoxFuture<'_, parquet::errors::Result<Bytes>> {
- self.read_bytes(range)
- }
-}
-
-impl<R: FileRead> AsyncFileReader for ArrowFileReader<R> {
- fn get_bytes(&mut self, range: Range<u64>) -> BoxFuture<'_, parquet::errors::Result<Bytes>> {
- self.read_bytes(range)
- }
-
- fn get_metadata(
- &mut self,
- options: Option<&ArrowReaderOptions>,
- ) -> BoxFuture<'_, parquet::errors::Result<Arc<ParquetMetaData>>> {
- let metadata_opts = options.map(|o| o.metadata_options().clone());
- Box::pin(async move {
- let file_size = self.meta.size;
- let metadata = ParquetMetaDataReader::new()
- .with_metadata_options(metadata_opts)
- .load_and_finish(self, file_size)
- .await?;
- Ok(Arc::new(metadata))
- })
- }
-}
-
-#[cfg(test)]
-mod tests {
- use super::build_parquet_row_filter;
- use crate::spec::{DataField, DataType, Datum, IntType, PredicateBuilder};
- use parquet::schema::{parser::parse_message_type, types::SchemaDescriptor};
- use std::sync::Arc;
-
- fn test_fields() -> Vec<DataField> {
- vec![
- DataField::new(0, "id".to_string(), DataType::Int(IntType::new())),
- DataField::new(1, "score".to_string(), DataType::Int(IntType::new())),
- ]
- }
-
- fn test_parquet_schema() -> SchemaDescriptor {
- SchemaDescriptor::new(Arc::new(
- parse_message_type(
- "
- message test_schema {
- OPTIONAL INT32 id;
- OPTIONAL INT32 score;
- }
- ",
- )
- .expect("test schema should parse"),
- ))
- }
-
- #[test]
- fn test_build_parquet_row_filter_supports_null_and_membership_predicates() {
- let fields = test_fields();
- let builder = PredicateBuilder::new(&fields);
- let predicates = vec![
- builder
- .is_null("id")
- .expect("is null predicate should build"),
- builder
- .is_in("score", vec![Datum::Int(7)])
- .expect("in predicate should build"),
- builder
- .is_not_in("score", vec![Datum::Int(9)])
- .expect("not in predicate should build"),
- ];
-
- let row_filter =
- build_parquet_row_filter(&test_parquet_schema(), &predicates, &fields, &fields)
- .expect("parquet row filter should build");
-
- assert!(row_filter.is_some());
- }
-}
diff --git a/crates/paimon/src/spec/types.rs b/crates/paimon/src/spec/types.rs
index 1bb4566..d86c068 100644
--- a/crates/paimon/src/spec/types.rs
+++ b/crates/paimon/src/spec/types.rs
@@ -1160,6 +1160,15 @@ impl FromStr for VarBinaryType {
type Err = Error;
fn from_str(s: &str) -> Result<Self, Self::Err> {
+ // "BYTES" is an alias for VARBINARY(MAX_LENGTH) in Paimon Java.
+ if s == "BYTES" || s == "BYTES NOT NULL" {
+ let nullable = !s.contains("NOT NULL");
+ return Ok(VarBinaryType {
+ nullable,
+ length: Self::MAX_LENGTH,
+ });
+ }
+
if !s.starts_with(serde_utils::VARBINARY::NAME) {
return DataTypeInvalidSnafu {
message: "Invalid VARBINARY type. Expected string to start with 'VARBINARY'.",
diff --git a/crates/paimon/src/table/read_builder.rs b/crates/paimon/src/table/read_builder.rs
index c8c3d86..1ead1fc 100644
--- a/crates/paimon/src/table/read_builder.rs
+++ b/crates/paimon/src/table/read_builder.rs
@@ -375,6 +375,7 @@ mod tests {
vec![("id", vec![1, 2, 3, 4]), ("value", vec![1, 2, 20, 30])],
Some(2),
);
+ let file_size = fs::metadata(&parquet_path).unwrap().len() as i64;
let file_io = FileIOBuilder::new("file").build().unwrap();
let table_schema = TableSchema::new(
@@ -398,7 +399,7 @@ mod tests {
.with_bucket(0)
.with_bucket_path(local_file_path(&bucket_dir))
.with_total_buckets(1)
- .with_data_files(vec![test_data_file("data.parquet", 4)])
+ .with_data_files(vec![test_data_file("data.parquet", 4, file_size)])
.with_raw_convertible(true)
.build()
.unwrap();
@@ -433,6 +434,7 @@ mod tests {
vec![("id", vec![1, 2, 3, 4]), ("value", vec![1, 2, 20, 30])],
Some(2),
);
+ let file_size = fs::metadata(&parquet_path).unwrap().len() as i64;
let file_io = FileIOBuilder::new("file").build().unwrap();
let table_schema = TableSchema::new(
@@ -456,7 +458,7 @@ mod tests {
.with_bucket(0)
.with_bucket_path(local_file_path(&bucket_dir))
.with_total_buckets(1)
- .with_data_files(vec![test_data_file("data.parquet", 4)])
+ .with_data_files(vec![test_data_file("data.parquet", 4, file_size)])
.with_raw_convertible(true)
.build()
.unwrap();
@@ -489,6 +491,7 @@ mod tests {
vec![("id", vec![1, 2, 3, 4]), ("value", vec![5, 20, 30, 40])],
Some(2),
);
+ let file_size = fs::metadata(&parquet_path).unwrap().len() as i64;
let file_io = FileIOBuilder::new("file").build().unwrap();
let table_schema = TableSchema::new(
@@ -512,7 +515,7 @@ mod tests {
.with_bucket(0)
.with_bucket_path(local_file_path(&bucket_dir))
.with_total_buckets(1)
- .with_data_files(vec![test_data_file("data.parquet", 4)])
+ .with_data_files(vec![test_data_file("data.parquet", 4, file_size)])
.with_raw_convertible(true)
.build()
.unwrap();
@@ -546,6 +549,7 @@ mod tests {
vec![("id", vec![1, 2, 3, 4]), ("value", vec![1, 2, 20, 30])],
Some(2),
);
+ let file_size = fs::metadata(bucket_dir.join("data.parquet")).unwrap().len() as i64;
let file_io = FileIOBuilder::new("file").build().unwrap();
let table_schema = TableSchema::new(
@@ -571,7 +575,7 @@ mod tests {
.with_bucket(0)
.with_bucket_path(local_file_path(&bucket_dir))
.with_total_buckets(1)
- .with_data_files(vec![test_data_file("data.parquet", 4)])
+ .with_data_files(vec![test_data_file("data.parquet", 4, file_size)])
.with_raw_convertible(true)
.build()
.unwrap();
diff --git a/crates/test_utils.rs b/crates/test_utils.rs
index 48f6fa2..cc88192 100644
--- a/crates/test_utils.rs
+++ b/crates/test_utils.rs
@@ -63,13 +63,13 @@ pub(crate) fn local_file_path(path: &Path) -> String {
}
}
-pub(crate) fn test_data_file<T>(file_name: &str, row_count: i64) -> T
+pub(crate) fn test_data_file<T>(file_name: &str, row_count: i64, file_size: i64) -> T
where
T: DeserializeOwned,
{
serde_json::from_value(serde_json::json!({
"_FILE_NAME": file_name,
- "_FILE_SIZE": 0,
+ "_FILE_SIZE": file_size,
"_ROW_COUNT": row_count,
"_MIN_KEY": [],
"_MAX_KEY": [],
diff --git a/dev/spark/provision.py b/dev/spark/provision.py
index d782d46..82483e2 100644
--- a/dev/spark/provision.py
+++ b/dev/spark/provision.py
@@ -710,6 +710,85 @@ def main():
"""
)
+ # ===== Full types table: parquet, orc, avro =====
+ # Note: Spark 3.x does not support parameterized timestamp precision (e.g. TIMESTAMP(3)),
+ # so all timestamps here use the default precision 6 (microseconds).
+ spark.sql(
+ """
+ CREATE TABLE IF NOT EXISTS full_types_table (
+ id INT,
+ col_boolean BOOLEAN,
+ col_tinyint TINYINT,
+ col_smallint SMALLINT,
+ col_int INT,
+ col_bigint BIGINT,
+ col_float FLOAT,
+ col_double DOUBLE,
+ col_decimal DECIMAL(10, 2),
+ col_decimal5 DECIMAL(5),
+ col_decimal38 DECIMAL(38, 18),
+ col_string STRING,
+ col_binary BINARY,
+ col_date DATE,
+ col_timestamp TIMESTAMP_NTZ,
+ col_timestamp_ltz TIMESTAMP,
+ col_array ARRAY<INT>,
+ col_map MAP<STRING, INT>,
+ col_struct STRUCT<name: STRING, value: INT>
+ ) USING paimon
+ TBLPROPERTIES (
+ 'file.format' = 'parquet'
+ )
+ """
+ )
+ # Parquet row
+ spark.sql(
+ """
+ INSERT INTO full_types_table VALUES
+ (1, true, CAST(1 AS TINYINT), CAST(100 AS SMALLINT), 1000, 100000,
+ CAST(1.5 AS FLOAT), 2.5, CAST(123.45 AS DECIMAL(10,2)),
+ CAST(12345 AS DECIMAL(5)), CAST(12345.678901234567890 AS DECIMAL(38,18)),
+ 'parquet-hello', X'DEADBEEF',
+ DATE '2024-01-01',
+ TIMESTAMP_NTZ '2024-01-01 10:00:00.123456',
+ TIMESTAMP '2024-01-01 10:00:00.123456',
+ array(1, 2, 3), map('a', 10, 'b', 20),
+ named_struct('name', 'alice', 'value', 100))
+ """
+ )
+ # Switch to ORC
+ spark.sql("ALTER TABLE full_types_table SET TBLPROPERTIES ('file.format' = 'orc')")
+ spark.sql(
+ """
+ INSERT INTO full_types_table VALUES
+ (2, false, CAST(2 AS TINYINT), CAST(200 AS SMALLINT), 2000, 200000,
+ CAST(3.5 AS FLOAT), 4.5, CAST(678.90 AS DECIMAL(10,2)),
+ CAST(99999 AS DECIMAL(5)), CAST(99999.999999999999999 AS DECIMAL(38,18)),
+ 'orc-world', X'CAFEBABE',
+ DATE '2024-06-15',
+ TIMESTAMP_NTZ '2024-06-15 12:30:00.456789',
+ TIMESTAMP '2024-06-15 12:30:00.456789',
+ array(4, 5), map('c', 30),
+ named_struct('name', 'bob', 'value', 200))
+ """
+ )
+ # Switch to Avro
+ spark.sql("ALTER TABLE full_types_table SET TBLPROPERTIES ('file.format' = 'avro')")
+ spark.sql(
+ """
+ INSERT INTO full_types_table VALUES
+ (3, true, CAST(3 AS TINYINT), CAST(300 AS SMALLINT), 3000, 300000,
+ CAST(5.5 AS FLOAT), 6.5, CAST(999.99 AS DECIMAL(10,2)),
+ CAST(0 AS DECIMAL(5)), CAST(0.000000000000000001 AS DECIMAL(38,18)),
+ 'avro-test', X'01020304',
+ DATE '2025-12-31',
+ TIMESTAMP_NTZ '2025-12-31 23:59:59.999999',
+ TIMESTAMP '2025-12-31 23:59:59.999999',
+ array(6), map('d', 40, 'e', 50),
+ named_struct('name', 'carol', 'value', 300))
+ """
+ )
+
if __name__ == "__main__":
main()