You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/08/07 11:14:02 UTC
(datafusion) branch main updated: chore(deps): Update to arrow/parquet 59.2.0 (#24030)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new c8f89cf408 chore(deps): Update to arrow/parquet 59.2.0 (#24030)
c8f89cf408 is described below
commit c8f89cf408a2ecac51eafa423dc256da56652fa5
Author: Andrew Lamb <an...@nerdnetworks.org>
AuthorDate: Fri Aug 7 07:01:50 2026 -0400
chore(deps): Update to arrow/parquet 59.2.0 (#24030)
## Which issue does this PR close?
- Related to https://github.com/apache/arrow-rs/issues/9879
## Rationale for this change
Upgrade to latest arrow / parquet library
## What changes are included in this PR?
1. Update version pins
2. Update for API deprecation (I will comment inline)
## Are these changes tested?
yes, by CI
## Are there any user-facing changes?
Not yet
---
Cargo.lock | 107 +++++++++++----------
Cargo.toml | 18 ++--
benchmarks/src/cancellation.rs | 5 +-
.../examples/data_io/parquet_advanced_index.rs | 50 +++++++---
datafusion/datasource-parquet/src/bloom_filter.rs | 7 +-
datafusion/datasource-parquet/src/reader.rs | 52 ++++------
datafusion/functions-aggregate/src/array_agg.rs | 2 +-
.../proto/tests/cases/roundtrip_logical_plan.rs | 4 +-
docs/source/library-user-guide/upgrading/55.0.0.md | 51 ++++++++++
9 files changed, 175 insertions(+), 121 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 93627f290f..3ddb32f60f 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -164,9 +164,9 @@ checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50"
[[package]]
name = "arrow"
-version = "59.1.0"
+version = "59.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b952ca5a8046ad741b60f142d6eca4aeebcad615694202bc64c5341f23e32c5b"
+checksum = "61d285d16bce7d0be61912f7928342b673067b6b7d7ef6cc179258ba7de1fecf"
dependencies = [
"arrow-arith",
"arrow-array",
@@ -187,9 +187,9 @@ dependencies = [
[[package]]
name = "arrow-arith"
-version = "59.1.0"
+version = "59.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "64a13b8d3008c4e9063c597a08f46446fe3fd5789277127672d6c0bdbb43b1ff"
+checksum = "757ef1836251e88222542a7da2623bc1c9cb9e20afefa6db2c41e79991cd91d4"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -201,9 +201,9 @@ dependencies = [
[[package]]
name = "arrow-array"
-version = "59.1.0"
+version = "59.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9486151b2f0785bafc6fa04fc5c99fcb4495455662e58787ea32eaaed33c4192"
+checksum = "bc9a4a4b2b5ecd0e04df03471661cb61f28bed3c7fd50994715129b01b2edb97"
dependencies = [
"ahash",
"arrow-buffer",
@@ -213,6 +213,7 @@ dependencies = [
"chrono-tz",
"half",
"hashbrown 0.17.1",
+ "libc",
"num-complex",
"num-integer",
"num-traits",
@@ -220,9 +221,9 @@ dependencies = [
[[package]]
name = "arrow-avro"
-version = "59.1.0"
+version = "59.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2e4f9b23a0d7b613acb59fa20bdbe0f80ffdae6411498378340b3915e45f5b84"
+checksum = "9fb45cd6bd2b25c0965793b83200eaca82214273a8030fbbc2d783e4c7c65a61"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -244,21 +245,21 @@ dependencies = [
[[package]]
name = "arrow-buffer"
-version = "59.1.0"
+version = "59.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c4776577a87794bfdf0b4e90e2ea12454fa7738ea2823c4be5b9d1851da7b434"
+checksum = "c12b576ef18c1deb80925a248b25ad84f419198d791b8e293fc6aaa60441fe90"
dependencies = [
"bytes",
"half",
- "num-bigint",
+ "num-bigint 0.5.1",
"num-traits",
]
[[package]]
name = "arrow-cast"
-version = "59.1.0"
+version = "59.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a9ad451ce4f98710828a455b96991b8f031deb2e67f5fcad6773f017e4a69c3a"
+checksum = "68338a9096a5dc9bc11927c58c43a8526d96bf6abd2012ef6c0c9f505991cc79"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -267,7 +268,7 @@ dependencies = [
"arrow-schema",
"arrow-select",
"atoi",
- "base64 0.22.1",
+ "base64 0.23.0",
"chrono",
"comfy-table",
"half",
@@ -278,9 +279,9 @@ dependencies = [
[[package]]
name = "arrow-csv"
-version = "59.1.0"
+version = "59.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8aa7bf96d6141a7bcca2eed57c7c9767d2a2175281857b8a7b68308992864784"
+checksum = "25011b52b346407d497ef0030e12b45e4f2d0cc279efc09c4f3d09106db30e36"
dependencies = [
"arrow-array",
"arrow-cast",
@@ -293,9 +294,9 @@ dependencies = [
[[package]]
name = "arrow-data"
-version = "59.1.0"
+version = "59.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b38fe43e2e8704360f1464e6e8cc4fc381ef02cc4fb0192afa8df1aaa0115c66"
+checksum = "723fe4aeed7604e00b9883a465af4ff0a0e6c44c03e41a68c3d1cbc403e0e44d"
dependencies = [
"arrow-buffer",
"arrow-schema",
@@ -306,9 +307,9 @@ dependencies = [
[[package]]
name = "arrow-flight"
-version = "59.1.0"
+version = "59.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "42115e09dbb694b5955da998912121451c6910b338228cb80a5701370dba43ff"
+checksum = "2bebfacc9d71f0728f6774164e4d4254b5e504d2b46812d0512d8290ec119a64"
dependencies = [
"arrow-arith",
"arrow-array",
@@ -321,11 +322,10 @@ dependencies = [
"arrow-schema",
"arrow-select",
"arrow-string",
- "base64 0.22.1",
+ "base64 0.23.0",
"bytes",
"futures",
"once_cell",
- "paste",
"prost",
"prost-types",
"tonic",
@@ -334,9 +334,9 @@ dependencies = [
[[package]]
name = "arrow-ipc"
-version = "59.1.0"
+version = "59.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "29dac499fcbc6ba74ee0324057821d381929a48526a3966bd9dffb44aa06d98c"
+checksum = "149437b14371f5b9ec60f5ddc751483ae99d7a7072653c0075e5e469156eea7b"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -350,9 +350,9 @@ dependencies = [
[[package]]
name = "arrow-json"
-version = "59.1.0"
+version = "59.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0fe05e916ddc50f4c7a363cd69c0ef5894fcee063517e9a0b8582f0c56746af6"
+checksum = "f18b9123ccfec418a663f821c9a034af339711678c11ffe00d3ec07da5ff9f7e"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -375,9 +375,9 @@ dependencies = [
[[package]]
name = "arrow-ord"
-version = "59.1.0"
+version = "59.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0e13dbdc2a9c053c10c7baa6e30faee04a180aa7ce88e471835850ce37abd20b"
+checksum = "e6c08dff0686cf23ca4f562803f191ccbeb726dbae6309cd4b4aaf65e0f2c979"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -388,9 +388,9 @@ dependencies = [
[[package]]
name = "arrow-row"
-version = "59.1.0"
+version = "59.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4d5a1f8c733d15260b305683472ee8ad89c62cbd706703ca873b90d051b41592"
+checksum = "bbec439386df71ad570e6758a946111322b9e9dc8db83b5527321f0b4c9119c2"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -401,9 +401,9 @@ dependencies = [
[[package]]
name = "arrow-schema"
-version = "59.1.0"
+version = "59.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d9e4969dc350d571766247143ab36a5187d095d3d3690970408bc630d47c69e5"
+checksum = "e6fed2ca0d1eade57e811cbe73b98ad50cc08a1183e13b2d2aa43a7df593f40e"
dependencies = [
"bitflags",
"serde",
@@ -413,9 +413,9 @@ dependencies = [
[[package]]
name = "arrow-select"
-version = "59.1.0"
+version = "59.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "402770dba90865359d98d1ef92ef16e23d75c0cca9c2c880c8a05468b7743bf9"
+checksum = "466b19cf75130b891dc1b23a84b343c714c62c64c9c62e365c76aa0ff90a53fb"
dependencies = [
"ahash",
"arrow-array",
@@ -427,9 +427,9 @@ dependencies = [
[[package]]
name = "arrow-string"
-version = "59.1.0"
+version = "59.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a2b0afbb8b9016700938291123df30838b89decc3213dba00852021988b170d3"
+checksum = "c838a25bb3691e919e0f617616ac51a4ff8517a952e29ca133cf0c22b2ce65b1"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -1013,7 +1013,7 @@ checksum = "4d6867f1565b3aad85681f1015055b087fcfd840d6aeee6eee7f2da317603695"
dependencies = [
"autocfg",
"libm",
- "num-bigint",
+ "num-bigint 0.4.6",
"num-integer",
"num-traits",
]
@@ -4067,9 +4067,9 @@ checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154"
[[package]]
name = "lz4_flex"
-version = "0.13.0"
+version = "0.14.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "db9a0d582c2874f68138a16ce1867e0ffde6c0bb0a0df85e1f36d04146db488a"
+checksum = "ecbdfe44b1bd960b68170b417450a628c43f7cf56bb3c5317e61cb230ee7f226"
dependencies = [
"twox-hash",
]
@@ -4212,7 +4212,7 @@ version = "0.4.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "35bd024e8b2ff75562e5f34e7f4905839deb4b22955ef5e73d2fea1b9813cb23"
dependencies = [
- "num-bigint",
+ "num-bigint 0.4.6",
"num-complex",
"num-integer",
"num-iter",
@@ -4230,6 +4230,16 @@ dependencies = [
"num-traits",
]
+[[package]]
+name = "num-bigint"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "93e7820bc0a80a0238e650327316f929ba18d5be054b647490a3a6a339f3e7c0"
+dependencies = [
+ "num-integer",
+ "num-traits",
+]
+
[[package]]
name = "num-complex"
version = "0.4.6"
@@ -4271,7 +4281,7 @@ version = "0.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f83d14da390562dca69fc84082e73e548e1ad308d24accdedd2720017cb37824"
dependencies = [
- "num-bigint",
+ "num-bigint 0.4.6",
"num-integer",
"num-traits",
]
@@ -4477,9 +4487,9 @@ dependencies = [
[[package]]
name = "parquet"
-version = "59.1.0"
+version = "59.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5302d4da74d6596a1f11f9928767995b53bca657cbeea1e4e8c5074f8a1157dd"
+checksum = "7065842956a20c2a536924ce8e4d9955f7422451511b9eb7500d7bfe5077e59c"
dependencies = [
"ahash",
"arrow-array",
@@ -4488,7 +4498,7 @@ dependencies = [
"arrow-ipc",
"arrow-schema",
"arrow-select",
- "base64 0.22.1",
+ "base64 0.23.0",
"brotli",
"bytes",
"chrono",
@@ -4497,11 +4507,10 @@ dependencies = [
"half",
"hashbrown 0.17.1",
"lz4_flex",
- "num-bigint",
+ "num-bigint 0.5.1",
"num-integer",
"num-traits",
"object_store",
- "paste",
"ring",
"seq-macro",
"simdutf8",
@@ -4536,12 +4545,6 @@ dependencies = [
"syn 2.0.119",
]
-[[package]]
-name = "paste"
-version = "1.0.15"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "57c0d7b74b563b49d38dae00a0c37d4d6de9b432382b2892f0574ddcae73fd0a"
-
[[package]]
name = "pbjson"
version = "0.8.0"
diff --git a/Cargo.toml b/Cargo.toml
index 03b90480fe..a0482f85f2 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -89,30 +89,30 @@ version = "54.1.0"
#
# See for more details: https://github.com/rust-lang/cargo/issues/11329
apache-avro = { version = "0.21", default-features = false }
-arrow = { version = "59.1.0", features = [
+arrow = { version = "59.2.0", features = [
"prettyprint",
"chrono-tz",
] }
-arrow-avro = { version = "59.1.0", default-features = false, features = [
+arrow-avro = { version = "59.2.0", default-features = false, features = [
"deflate",
"snappy",
"zstd",
"bzip2",
"xz",
] }
-arrow-buffer = { version = "59.1.0", default-features = false }
-arrow-data = { version = "59.1.0", default-features = false }
-arrow-flight = { version = "59.1.0", features = [
+arrow-buffer = { version = "59.2.0", default-features = false }
+arrow-data = { version = "59.2.0", default-features = false }
+arrow-flight = { version = "59.2.0", features = [
"flight-sql-experimental",
] }
# Both codecs are required here to make sure that code paths like
# file-spilling have access to all compression codecs.
-arrow-ipc = { version = "59.1.0", default-features = false, features = [
+arrow-ipc = { version = "59.2.0", default-features = false, features = [
"lz4",
"zstd",
] }
-arrow-ord = { version = "59.1.0", default-features = false }
-arrow-schema = { version = "59.1.0", default-features = false }
+arrow-ord = { version = "59.2.0", default-features = false }
+arrow-schema = { version = "59.2.0", default-features = false }
async-trait = "0.1.89"
bigdecimal = "0.4.8"
bytes = "1.11"
@@ -178,7 +178,7 @@ memchr = "2.8.1"
num-traits = { version = "0.2" }
object_store = { version = "0.13.2", default-features = false }
parking_lot = "0.12"
-parquet = { version = "59.1.0", default-features = false, features = [
+parquet = { version = "59.2.0", default-features = false, features = [
"arrow",
"async",
"object_store",
diff --git a/benchmarks/src/cancellation.rs b/benchmarks/src/cancellation.rs
index 1048fa0989..5f7fdcc43d 100644
--- a/benchmarks/src/cancellation.rs
+++ b/benchmarks/src/cancellation.rs
@@ -37,8 +37,9 @@ use datafusion::prelude::*;
use datafusion_common::instant::Instant;
use futures::TryStreamExt;
use object_store::ObjectStore;
+use object_store::buffered::BufWriter;
use parquet::arrow::AsyncArrowWriter;
-use parquet::arrow::async_writer::ParquetObjectWriter;
+use rand::Rng;
use rand::distr::Alphanumeric;
use rand::prelude::*;
use tokio::runtime::Runtime;
@@ -301,7 +302,7 @@ async fn generate_data(
});
let to_write = RecordBatch::try_from_iter(data).unwrap();
let path = object_store::path::Path::from(format!("{file_num}.parquet").as_str());
- let object_store_writer = ParquetObjectWriter::new(Arc::clone(&store) as _, path);
+ let object_store_writer = BufWriter::new(Arc::clone(&store) as _, path);
let mut writer =
AsyncArrowWriter::try_new(object_store_writer, to_write.schema(), None)?;
diff --git a/datafusion-examples/examples/data_io/parquet_advanced_index.rs b/datafusion-examples/examples/data_io/parquet_advanced_index.rs
index 563b536915..b6440eb3e2 100644
--- a/datafusion-examples/examples/data_io/parquet_advanced_index.rs
+++ b/datafusion-examples/examples/data_io/parquet_advanced_index.rs
@@ -41,7 +41,8 @@ use datafusion::parquet::arrow::ArrowWriter;
use datafusion::parquet::arrow::arrow_reader::{
ArrowReaderOptions, ParquetRecordBatchReaderBuilder, RowSelection, RowSelector,
};
-use datafusion::parquet::arrow::async_reader::{AsyncFileReader, ParquetObjectReader};
+use datafusion::parquet::arrow::async_reader::AsyncFileReader;
+use datafusion::parquet::errors::ParquetError;
use datafusion::parquet::file::metadata::{PageIndexPolicy, ParquetMetaData};
use datafusion::parquet::file::properties::{EnabledStatistics, WriterProperties};
use datafusion::parquet::schema::types::ColumnPath;
@@ -59,7 +60,7 @@ use bytes::Bytes;
use datafusion::datasource::memory::DataSourceExec;
use futures::FutureExt;
use futures::future::BoxFuture;
-use object_store::ObjectStore;
+use object_store::{ObjectStore, ObjectStoreExt};
use tempfile::TempDir;
use url::Url;
@@ -554,12 +555,19 @@ impl ParquetFileReaderFactory for CachedParquetFileReaderFactory {
&self,
_partition_index: usize,
partitioned_file: PartitionedFile,
- metadata_size_hint: Option<usize>,
+ _metadata_size_hint: Option<usize>,
_metrics: &ExecutionPlanMetricsSet,
) -> Result<Box<dyn AsyncFileReader + Send>> {
// for this example we ignore the partition index and metrics
// but in a real system you would likely use them to report details on
// the performance of the reader.
+ //
+ // We also ignore the metadata size hint as this reader always serves
+ // metadata from the pre-populated `self.metadata` cache, so it never
+ // performs the footer fetch the hint is meant to optimize. A real
+ // implementation would likely pass the hint to
+ // `ParquetMetaDataReader::with_prefetch_hint` to reduce the number of
+ // IO requests needed to load the footer.
let filename = partitioned_file
.object_meta
.location
@@ -570,13 +578,7 @@ impl ParquetFileReaderFactory for CachedParquetFileReaderFactory {
.to_string();
let object_store = Arc::clone(&self.object_store);
- let mut inner =
- ParquetObjectReader::new(object_store, partitioned_file.object_meta.location)
- .with_file_size(partitioned_file.object_meta.size);
-
- if let Some(hint) = metadata_size_hint {
- inner = inner.with_footer_size_hint(hint)
- };
+ let location = partitioned_file.object_meta.location;
let metadata = self
.metadata
@@ -585,16 +587,18 @@ impl ParquetFileReaderFactory for CachedParquetFileReaderFactory {
Ok(Box::new(ParquetReaderWithCache {
filename,
metadata: Arc::clone(metadata),
- inner,
+ object_store,
+ location,
}))
}
}
-/// wrapper around a ParquetObjectReader that caches metadata
+/// An [`AsyncFileReader`] that reads from an [`ObjectStore`] and caches metadata
struct ParquetReaderWithCache {
filename: String,
metadata: Arc<ParquetMetaData>,
- inner: ParquetObjectReader,
+ object_store: Arc<dyn ObjectStore>,
+ location: object_store::path::Path,
}
impl AsyncFileReader for ParquetReaderWithCache {
@@ -603,7 +607,15 @@ impl AsyncFileReader for ParquetReaderWithCache {
range: Range<u64>,
) -> BoxFuture<'_, datafusion::parquet::errors::Result<Bytes>> {
println!("get_bytes: {} Reading range {:?}", self.filename, range);
- self.inner.get_bytes(range)
+ let object_store = Arc::clone(&self.object_store);
+ let location = self.location.clone();
+ async move {
+ object_store
+ .get_range(&location, range)
+ .await
+ .map_err(|e| ParquetError::External(Box::new(e)))
+ }
+ .boxed()
}
fn get_byte_ranges(
@@ -614,7 +626,15 @@ impl AsyncFileReader for ParquetReaderWithCache {
"get_byte_ranges: {} Reading ranges {:?}",
self.filename, ranges
);
- self.inner.get_byte_ranges(ranges)
+ let object_store = Arc::clone(&self.object_store);
+ let location = self.location.clone();
+ async move {
+ object_store
+ .get_ranges(&location, &ranges)
+ .await
+ .map_err(|e| ParquetError::External(Box::new(e)))
+ }
+ .boxed()
}
fn get_metadata(
diff --git a/datafusion/datasource-parquet/src/bloom_filter.rs b/datafusion/datasource-parquet/src/bloom_filter.rs
index 4cfe8bf1f8..8e55c12ba0 100644
--- a/datafusion/datasource-parquet/src/bloom_filter.rs
+++ b/datafusion/datasource-parquet/src/bloom_filter.rs
@@ -253,7 +253,6 @@ mod tests {
use object_store::{ObjectStore, ObjectStoreExt};
use parquet::arrow::ArrowWriter;
use parquet::arrow::ParquetRecordBatchStreamBuilder;
- use parquet::arrow::async_reader::ParquetObjectReader;
use parquet::file::properties::{EnabledStatistics, WriterProperties};
fn build_test_pruning_predicate(
@@ -651,14 +650,10 @@ mod tests {
let file_metrics =
ParquetFileMetrics::new(0, object_meta.location.as_ref(), &metrics);
let store: Arc<dyn ObjectStore> = Arc::new(in_memory);
- let inner =
- ParquetObjectReader::new(Arc::clone(&store), object_meta.location.clone())
- .with_file_size(object_meta.size);
-
let partitioned_file = PartitionedFile::new_from_meta(object_meta);
let reader =
- ParquetFileReader::new(file_metrics.clone(), store, inner, partitioned_file);
+ ParquetFileReader::new(file_metrics.clone(), store, partitioned_file);
let mut builder = ParquetRecordBatchStreamBuilder::new(reader).await.unwrap();
let access_plan = ParquetAccessPlan::new_all(builder.metadata().num_row_groups());
diff --git a/datafusion/datasource-parquet/src/reader.rs b/datafusion/datasource-parquet/src/reader.rs
index ee2d3a17d5..71b0020f32 100644
--- a/datafusion/datasource-parquet/src/reader.rs
+++ b/datafusion/datasource-parquet/src/reader.rs
@@ -27,10 +27,12 @@ use datafusion_execution::cache::cache_manager::FileMetadata;
use datafusion_execution::cache::cache_manager::FileMetadataCache;
use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet;
use futures::FutureExt;
+use futures::TryFutureExt;
use futures::future::BoxFuture;
-use object_store::ObjectStore;
+use object_store::{ObjectStore, ObjectStoreExt};
use parquet::arrow::arrow_reader::ArrowReaderOptions;
-use parquet::arrow::async_reader::{AsyncFileReader, ParquetObjectReader};
+use parquet::arrow::async_reader::AsyncFileReader;
+use parquet::errors::ParquetError;
use parquet::file::metadata::ParquetMetaData;
use std::any::Any;
use std::fmt::Debug;
@@ -99,21 +101,10 @@ impl ParquetFileReaderFactory for DefaultParquetFileReaderFactory {
partitioned_file.object_meta.location.as_ref(),
metrics,
);
- let store = Arc::clone(&self.store);
- let mut inner = ParquetObjectReader::new(
- store,
- partitioned_file.object_meta.location.clone(),
- )
- .with_file_size(partitioned_file.object_meta.size);
-
- if let Some(hint) = metadata_size_hint {
- inner = inner.with_footer_size_hint(hint)
- };
let reader = ParquetFileReader::new(
file_metrics,
Arc::clone(&self.store),
- inner,
partitioned_file,
)
.with_metadata_hint(metadata_size_hint);
@@ -158,22 +149,10 @@ impl ParquetFileReaderFactory for CachedParquetFileReaderFactory {
partitioned_file.object_meta.location.as_ref(),
metrics,
);
- let store = Arc::clone(&self.store);
-
- let mut inner = ParquetObjectReader::new(
- store,
- partitioned_file.object_meta.location.clone(),
- )
- .with_file_size(partitioned_file.object_meta.size);
-
- if let Some(hint) = metadata_size_hint {
- inner = inner.with_footer_size_hint(hint)
- };
let reader = ParquetFileReader::new(
file_metrics,
Arc::clone(&self.store),
- inner,
partitioned_file,
)
.with_metadata_hint(metadata_size_hint)
@@ -185,9 +164,8 @@ impl ParquetFileReaderFactory for CachedParquetFileReaderFactory {
/// Implements [`AsyncFileReader`] for a parquet file in object storage.
///
-/// This implementation uses the [`ParquetObjectReader`] to read data from the
-/// object store on demand, as required, tracking the number of bytes read via
-/// [`ParquetFileMetrics`].
+/// This implementation reads data directly from the underlying [`ObjectStore`]
+/// on demand, as required, tracking the number of bytes read.
///
/// When configured via [`Self::with_metadata_cache`], [`Self::get_metadata`]
/// reads footer and page metadata from the cache when available and populates
@@ -201,7 +179,6 @@ impl ParquetFileReaderFactory for CachedParquetFileReaderFactory {
pub struct ParquetFileReader {
file_metrics: ParquetFileMetrics,
store: Arc<dyn ObjectStore>,
- inner: ParquetObjectReader,
partitioned_file: PartitionedFile,
metadata_cache: Option<Arc<FileMetadataCache>>,
metadata_size_hint: Option<usize>,
@@ -219,13 +196,11 @@ impl ParquetFileReader {
pub(crate) fn new(
file_metrics: ParquetFileMetrics,
store: Arc<dyn ObjectStore>,
- inner: ParquetObjectReader,
partitioned_file: PartitionedFile,
) -> Self {
Self {
file_metrics,
store,
- inner,
partitioned_file,
metadata_cache: None,
metadata_size_hint: None,
@@ -267,7 +242,10 @@ impl AsyncFileReader for ParquetFileReader {
) -> BoxFuture<'_, parquet::errors::Result<Bytes>> {
let bytes_scanned = range.end - range.start;
self.file_metrics.bytes_scanned.add(bytes_scanned as usize);
- self.inner.get_bytes(range)
+ self.store
+ .get_range(&self.partitioned_file.object_meta.location, range)
+ .map_err(|e| ParquetError::External(Box::new(e)))
+ .boxed()
}
fn get_byte_ranges(
@@ -279,7 +257,13 @@ impl AsyncFileReader for ParquetFileReader {
{
let total: u64 = ranges.iter().map(|r| r.end - r.start).sum();
self.file_metrics.bytes_scanned.add(total as usize);
- self.inner.get_byte_ranges(ranges)
+ async move {
+ self.store
+ .get_ranges(&self.partitioned_file.object_meta.location, &ranges)
+ .await
+ .map_err(|e| ParquetError::External(Box::new(e)))
+ }
+ .boxed()
}
fn get_metadata<'a>(
@@ -308,7 +292,7 @@ impl AsyncFileReader for ParquetFileReader {
.fetch_metadata()
.await
.map_err(|e| {
- parquet::errors::ParquetError::General(format!(
+ ParquetError::General(format!(
"Failed to fetch metadata for file {}: {e}",
object_meta.location,
))
diff --git a/datafusion/functions-aggregate/src/array_agg.rs b/datafusion/functions-aggregate/src/array_agg.rs
index cfacd77196..eaf7f9addb 100644
--- a/datafusion/functions-aggregate/src/array_agg.rs
+++ b/datafusion/functions-aggregate/src/array_agg.rs
@@ -1741,7 +1741,7 @@ mod tests {
acc2.update_batch(&[data(["b", "c", "a"])])?;
acc1 = merge(acc1, acc2)?;
- assert_eq!(acc1.size(), 282);
+ assert_eq!(acc1.size(), 290);
Ok(())
}
diff --git a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs
index 1418998b43..b32ed268c0 100644
--- a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs
+++ b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs
@@ -2216,7 +2216,7 @@ fn round_trip_scalar_values_and_data_types() {
Arc::new(Field::new(
"entries",
DataType::Struct(Fields::from(vec![
- Field::new("key", DataType::Int32, true),
+ Field::new("key", DataType::Int32, false),
Field::new("value", DataType::Utf8, false),
])),
false,
@@ -2228,7 +2228,7 @@ fn round_trip_scalar_values_and_data_types() {
Arc::new(Field::new(
"entries",
DataType::Struct(Fields::from(vec![
- Field::new("key", DataType::Int32, true),
+ Field::new("key", DataType::Int32, false),
Field::new("value", DataType::Utf8, true),
])),
false,
diff --git a/docs/source/library-user-guide/upgrading/55.0.0.md b/docs/source/library-user-guide/upgrading/55.0.0.md
index 1997b8e1c1..884d8fab13 100644
--- a/docs/source/library-user-guide/upgrading/55.0.0.md
+++ b/docs/source/library-user-guide/upgrading/55.0.0.md
@@ -1110,3 +1110,54 @@ SELECT array_distance(
Previously, this query returned `2.0`, the distance between `[1, 2]` and
`[1, 4]`. It now returns a planning error stating that `array_distance` only
supports one-dimensional arrays.
+
+### `ParquetObjectReader` / `ParquetObjectWriter` deprecated upstream
+
+The [`parquet` crate] deprecated [`ParquetObjectReader`]
+and [`ParquetObjectWriter`] in favor of implementing
+[`AsyncFileReader`] directly (see the example on the [`AsyncFileReader`] trait and
+[`parquet/examples/object_store.rs`] in `arrow-rs`) or passing an
+[`BufWriter`] straight to [`AsyncArrowWriter`].
+
+**Who is affected:**
+
+- Custom [`ParquetFileReaderFactory`] implementations that construct a
+ [`ParquetObjectReader`] directly and now see a deprecation warning after
+ upgrading the `parquet` dependency.
+
+**Migration guide:**
+
+If your [`AsyncFileReader`] implementation exists mainly to read from an
+[`ObjectStore`] and track metrics, consider using DataFusion's
+[`ParquetFileReader`] instead of wrapping a [`ParquetObjectReader`]:
+
+```rust,ignore
+// Before
+let inner = ParquetObjectReader::new(store, location).with_file_size(size);
+Ok(Box::new(MyReader { inner, file_metrics, partitioned_file }))
+
+// After
+Ok(Box::new(ParquetFileReader {
+ file_metrics,
+ store,
+ metadata_size_hint,
+ partitioned_file,
+}))
+```
+
+If you need custom behavior (I/O coalescing, byte caching, a dedicated I/O
+runtime), implement `AsyncFileReader` directly against your `ObjectStore`,
+following the pattern in [`parquet/examples/object_store.rs`]
+
+See [PR #24030](https://github.com/apache/datafusion/pull/24030) for details.
+
+[`parquet` crate]: https://crates.io/crates/parquet
+[`parquetobjectreader`]: https://docs.rs/parquet/59.1.0/parquet/arrow/async_reader/struct.ParquetObjectReader.html
+[`parquetobjectwriter`]: https://docs.rs/parquet/59.1.0/parquet/arrow/async_writer/struct.ParquetObjectWriter.html
+[`parquetfilereader`]: https://docs.rs/datafusion/latest/datafusion/datasource/physical_plan/parquet/struct.ParquetFileReader.html
+[`parquetfilereaderfactory`]: https://docs.rs/datafusion/latest/datafusion/datasource/physical_plan/parquet/trait.ParquetFileReaderFactory.html
+[`asyncfilereader`]: https://docs.rs/parquet/59.1.0/parquet/arrow/async_reader/trait.AsyncFileReader.html
+[`objectstore`]: https://docs.rs/object_store/latest/object_store/trait.ObjectStore.html
+[`bufwriter`]: https://docs.rs/tokio/latest/tokio/io/struct.BufWriter.html
+[`asyncarrowwriter`]: https://docs.rs/parquet/59.1.0/parquet/arrow/async_writer/struct.AsyncArrowWriter.html
+[`parquet/examples/object_store.rs`]: https://github.com/apache/arrow-rs/blob/main/parquet/examples/object_store.rs
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org