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/09/18 21:19:50 UTC
(datafusion) branch gh-readonly-queue/main/pr-25335-efd36e7b66a143cb88181cea0dc50405b793af6c created (now b631f2c7d9)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-25335-efd36e7b66a143cb88181cea0dc50405b793af6c
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at b631f2c7d9 chore: Update to arrow/parquet `60.0.0` and `object_store` `0.14.2` (#25335)
This branch includes the following new commits:
new b631f2c7d9 chore: Update to arrow/parquet `60.0.0` and `object_store` `0.14.2` (#25335)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: chore: Update to arrow/parquet `60.0.0` and `object_store` `0.14.2` (#25335)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-25335-efd36e7b66a143cb88181cea0dc50405b793af6c
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit b631f2c7d92d0a38637a8d8ae980e2474b891f34
Author: Andrew Lamb <an...@nerdnetworks.org>
AuthorDate: Fri Sep 18 21:19:28 2026 +0000
chore: Update to arrow/parquet `60.0.0` and `object_store` `0.14.2` (#25335)
## Which issue does this PR close?
- Related to apache/arrow-rs#9880
- replaces / closes #24366
- closes https://github.com/apache/datafusion/pull/23693
## Rationale for this change
Update to the latest arrow/parquet release, `60.0.0`. See the upstream
[CHANGELOG](https://github.com/apache/arrow-rs/blob/60.0.0/CHANGELOG.md)
for details.
## What changes are included in this PR?
1. Update to arrow/parquet `60.0.0` and `object_store` `0.14.2`
2. Adjust code for upstream API differences (see inline comments)
3. Update test expectations for upstream behavior changes
## What is the testing strategy for this PR?
Existing CI.
## Are there any user-facing changes?
New dependency versions.
Also, following the upstream change, a few DataFusion APIs now use
arrow's new
[`Metadata`](https://docs.rs/arrow/latest/arrow/datatypes/struct.Metadata.html)
type (thanks @emilk !) instead of `HashMap<String, String>`
---------
Co-authored-by: Claude Fable 5 <no...@anthropic.com>
Co-authored-by: Copilot Autofix powered by AI <17...@users.noreply.github.com>
---
Cargo.lock | 301 ++++++++++++++-------
Cargo.toml | 24 +-
datafusion-cli/src/main.rs | 20 +-
datafusion/catalog-listing/src/helpers.rs | 4 +-
datafusion/catalog-listing/src/table.rs | 4 +-
datafusion/common/src/dfschema.rs | 17 +-
datafusion/common/src/hash_utils.rs | 4 +-
datafusion/common/src/heap_size.rs | 22 +-
datafusion/common/src/metadata.rs | 94 ++++---
datafusion/common/src/nested_struct.rs | 12 +-
datafusion/common/src/param_value.rs | 2 +-
datafusion/common/src/scalar/mod.rs | 2 +-
datafusion/common/src/utils/memory.rs | 2 +-
datafusion/common/src/utils/mod.rs | 10 +-
datafusion/core/src/datasource/file_format/csv.rs | 1 +
.../core/src/datasource/file_format/parquet.rs | 44 ++-
.../core/tests/parquet/string_in_list_pruning.rs | 10 +-
datafusion/core/tests/sql/path_partition.rs | 2 +
.../tests/user_defined/user_defined_aggregates.rs | 14 +-
.../user_defined/user_defined_scalar_functions.rs | 8 +-
.../user_defined/user_defined_window_functions.rs | 8 +-
datafusion/datasource-arrow/src/file_format.rs | 6 +-
datafusion/datasource-arrow/src/source.rs | 7 +-
datafusion/datasource-parquet/src/file_format.rs | 4 +-
datafusion/datasource-parquet/src/metadata.rs | 26 +-
datafusion/datasource-parquet/src/opener/mod.rs | 15 +-
datafusion/datasource-parquet/src/page_filter.rs | 87 +++---
datafusion/datasource-parquet/src/reader.rs | 6 +-
.../src/statistics_order_tests.rs | 35 ++-
datafusion/execution/src/cache/mod.rs | 6 +-
datafusion/expr/src/expr.rs | 55 ++--
datafusion/expr/src/logical_plan/builder.rs | 8 +-
datafusion/expr/src/logical_plan/plan.rs | 7 +-
datafusion/ffi/src/record_batch_stream.rs | 4 +-
datafusion/ffi/src/udaf/mod.rs | 6 +-
.../src/aggregate/groups_accumulator/nulls.rs | 4 +-
.../functions-aggregate/benches/array_agg.rs | 6 +-
datafusion/functions-aggregate/src/array_agg.rs | 4 +-
datafusion/functions/src/binaries.rs | 2 +-
datafusion/functions/src/core/arrow_field.rs | 17 +-
datafusion/functions/src/core/arrow_metadata.rs | 16 +-
datafusion/functions/src/core/with_metadata.rs | 11 +-
datafusion/functions/src/string/common.rs | 4 +-
datafusion/functions/src/string/split_part.rs | 2 +-
datafusion/functions/src/strings.rs | 4 +-
datafusion/functions/src/unicode/common.rs | 2 +-
datafusion/functions/src/unicode/initcap.rs | 18 +-
datafusion/functions/src/unicode/substr.rs | 2 +-
datafusion/functions/src/unicode/substrindex.rs | 4 +-
datafusion/optimizer/src/optimizer.rs | 5 +-
.../optimizer/src/propagate_empty_relation.rs | 4 +-
datafusion/optimizer/src/push_down_filter.rs | 4 +-
.../src/simplify_expressions/udf_preimage.rs | 6 +-
.../physical-expr-common/src/binary_view_map.rs | 5 +-
datafusion/physical-expr-common/src/utils.rs | 2 +-
datafusion/physical-expr/benches/case_when.rs | 9 +-
datafusion/physical-expr/src/expressions/cast.rs | 5 +-
.../physical-expr/src/expressions/try_cast.rs | 7 +-
datafusion/physical-expr/src/planner.rs | 26 +-
datafusion/physical-expr/src/scalar_subquery.rs | 2 +-
.../group_values/multi_group_by/bytes_view.rs | 6 +-
datafusion/physical-plan/src/joins/utils.rs | 8 +-
.../physical-plan/src/spill/spill_manager.rs | 33 +--
datafusion/physical-plan/src/union.rs | 4 +-
datafusion/proto-common/src/from_proto/mod.rs | 8 +-
datafusion/proto-common/src/to_proto/mod.rs | 8 +-
datafusion/proto/src/logical_plan/to_proto.rs | 6 +-
.../proto/tests/cases/roundtrip_logical_plan.rs | 8 +-
datafusion/proto/tests/cases/serialize.rs | 6 +-
datafusion/sql/src/statement.rs | 4 +-
datafusion/sql/tests/cases/params.rs | 12 +-
datafusion/sql/tests/common/mod.rs | 21 +-
datafusion/sqllogictest/src/test_context.rs | 5 +-
.../test_files/datetime/arith_interval_double.slt | 12 +-
.../sqllogictest/test_files/explain_analyze.slt | 2 +-
.../test_files/push_down_filter_parquet.slt | 44 +--
datafusion/sqllogictest/test_files/scalar.slt | 2 +-
docs/source/library-user-guide/upgrading/56.0.0.md | 40 +++
78 files changed, 725 insertions(+), 552 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 061df543c5..221bbf2583 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -33,15 +33,15 @@ dependencies = [
[[package]]
name = "alloc-no-stdlib"
-version = "2.0.4"
+version = "3.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cc7bb162ec39d46ab1ca8c77bf72e890535becd1751bb45f64c597edb4c8c6b3"
+checksum = "c2fb6cfd47bf496ff64095c20eaba0c201404ee38714d4142fcfa1dc334fcc7a"
[[package]]
name = "alloc-stdlib"
-version = "0.2.2"
+version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "94fb8275041c72129eb51b7d0322c29b8387a0386127718b096429201a5d6ece"
+checksum = "0b5c1865780388bfa186411ab5f247819487fc4864c6e9c3106611fa347586e1"
dependencies = [
"alloc-no-stdlib",
]
@@ -158,9 +158,9 @@ checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50"
[[package]]
name = "arrow"
-version = "59.3.0"
+version = "60.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7c14b3d39f306bc28fd639d59f06e17a0f377d0021e1b7e9054e4d6fedc98774"
+checksum = "234ce58e76c04e8162c2e4bc3fe53fa7c2418ef5fba7bb2838b8531b7d4fbf14"
dependencies = [
"arrow-arith",
"arrow-array",
@@ -176,14 +176,14 @@ dependencies = [
"arrow-select",
"arrow-string",
"half",
- "rand 0.9.4",
+ "rand 0.10.1",
]
[[package]]
name = "arrow-arith"
-version = "59.3.0"
+version = "60.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ce2961626677665b2195eb59242af4c7befe7b8737ca2050295389362380104e"
+checksum = "2303ebedb1071f4f72d71171e383859a9a285c94d99e4892685a96fca9890720"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -195,9 +195,9 @@ dependencies = [
[[package]]
name = "arrow-array"
-version = "59.3.0"
+version = "60.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1e5f6adeffdf587d7a31db5d2266189624b526730cd3627f9ff9fedae97ad584"
+checksum = "51dff6e4b9f158864a0aeb6a131ad53858ecbae7a2fc8989307b9b3c6b2f122e"
dependencies = [
"ahash",
"arrow-buffer",
@@ -215,9 +215,9 @@ dependencies = [
[[package]]
name = "arrow-avro"
-version = "59.3.0"
+version = "60.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9145b685d586482102e4fe3d40349fe00086c0b219d51cae9daaead4698849ea"
+checksum = "edad137a3609e2c6a4af0b5ff1c5c049b7af532649d235c7e22e9a9cad933b35"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -228,20 +228,20 @@ dependencies = [
"flate2",
"indexmap 2.14.2",
"liblzma",
- "rand 0.9.4",
+ "rand 0.10.1",
"serde",
"serde_json",
"snap",
"strum_macros",
"uuid",
- "zstd 0.13.3",
+ "zstd",
]
[[package]]
name = "arrow-buffer"
-version = "59.3.0"
+version = "60.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "097d193003ce7995d5d087089069ec2a6e0187faf5a6f8c9f38af2645d987182"
+checksum = "c7235b863533e9db3ab86905b4521251de11226275b436ff3cb2d0010558d1c5"
dependencies = [
"bytes",
"half",
@@ -251,9 +251,9 @@ dependencies = [
[[package]]
name = "arrow-cast"
-version = "59.3.0"
+version = "60.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "635c9c635668ad26adf76cce8fb276c4be7cf06e63bd516de7da514f9680ee53"
+checksum = "d52cf840f0c34aeeb7e8b05aba72d66878e6c001827f423331f0b260209e7b7e"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -271,11 +271,22 @@ dependencies = [
"ryu",
]
+[[package]]
+name = "arrow-cmp"
+version = "60.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "73a0e9adb28d95739fabcdae637261a2c14e618caf58ba1b88857120f9ee2687"
+dependencies = [
+ "arrow-array",
+ "arrow-buffer",
+ "arrow-schema",
+]
+
[[package]]
name = "arrow-csv"
-version = "59.3.0"
+version = "60.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4c2ebf8d631e79b02c16cf5ae860561272c26024ec88fce389a56aaddd558e86"
+checksum = "2ad83176cc677f996096d694a3cccd7310069faeee4b3df9e2dabf92fdcf8b4b"
dependencies = [
"arrow-array",
"arrow-cast",
@@ -288,9 +299,9 @@ dependencies = [
[[package]]
name = "arrow-data"
-version = "59.3.0"
+version = "60.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9ba2f832eaeca24b8f26143dba750e42ee4ab51cf7d65e701ca9607cfda9f358"
+checksum = "5aa06d5e22786c4966ffc4d2f171d0cb65d95f6bdea108a14e8c335e9faa437f"
dependencies = [
"arrow-buffer",
"arrow-schema",
@@ -301,9 +312,9 @@ dependencies = [
[[package]]
name = "arrow-flight"
-version = "59.3.0"
+version = "60.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "be0e6d452fff35cb4a3ef1a719d03fdfe653e590e8012b57d3057a316e295402"
+checksum = "fbc77d8656483656cbf966e04b62f193f74111ff7d9cc15f58a549314a3b061e"
dependencies = [
"arrow-arith",
"arrow-array",
@@ -328,9 +339,9 @@ dependencies = [
[[package]]
name = "arrow-ipc"
-version = "59.3.0"
+version = "60.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dcc41681ea80f521df14c36725b74d4c60702c47f0793af2be469c04527e2599"
+checksum = "86c523472d22e31f984df7a44fc36e6323d3c9b05ffe53d267ee4c3aa7c128d2"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -339,14 +350,14 @@ dependencies = [
"arrow-select",
"flatbuffers",
"lz4_flex",
- "zstd 0.13.3",
+ "zstd",
]
[[package]]
name = "arrow-json"
-version = "59.3.0"
+version = "60.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a2f57d7a81969f24ccf80809587b76c09897e6f829d2d65a5976bfb3218851f1"
+checksum = "342fc6600a93a48038d853431803c941436b7b51f756763e76cad9784b7e3718"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -369,12 +380,13 @@ dependencies = [
[[package]]
name = "arrow-ord"
-version = "59.3.0"
+version = "60.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2c900759f3bd8354fd4196bc4403eee846894dc2adf66b4225472006a0bf18c5"
+checksum = "5a7ac2c6a9cdc782fa9548d3763461eaec8c93347e297fddbaa44807e70a3698"
dependencies = [
"arrow-array",
"arrow-buffer",
+ "arrow-cmp",
"arrow-data",
"arrow-schema",
"arrow-select",
@@ -382,9 +394,9 @@ dependencies = [
[[package]]
name = "arrow-row"
-version = "59.3.0"
+version = "60.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f4c6425032e28266e3fc4ff680805e57e670d6ea92473043f3e65b7ed6ac79f2"
+checksum = "758b87e9648d80001e0916c083334ed93074c677a81a96144d5c84bd2d8eab2a"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -395,9 +407,9 @@ dependencies = [
[[package]]
name = "arrow-schema"
-version = "59.3.0"
+version = "60.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "10fab8d4563491417ba801fab29d205104d20d4bdf37bda6cd1cf425cff598cd"
+checksum = "f85a9a58cd03560526b0066487574aacb9008e413504fa0ee23d61141e96c338"
dependencies = [
"bitflags",
"serde",
@@ -407,13 +419,14 @@ dependencies = [
[[package]]
name = "arrow-select"
-version = "59.3.0"
+version = "60.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fc58569193c2525915f3cc6310edba3792f1200f65d6e9ed330aa33e691493b8"
+checksum = "ba3482b7d03bf4a3bf0b195f08a7f2b071595ccd54897e35b975ac63d3104977"
dependencies = [
"ahash",
"arrow-array",
"arrow-buffer",
+ "arrow-cmp",
"arrow-data",
"arrow-schema",
"num-traits",
@@ -421,9 +434,9 @@ dependencies = [
[[package]]
name = "arrow-string"
-version = "59.3.0"
+version = "60.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2e0813f3c35c1cfea65e14c20a953440f7783c088b7ad2d0db162ccdeefcec14"
+checksum = "f300b4400b1e7cd58f7cbbd5dd2a0072cc59e73bfaee2a152b3ac660e46d9dd1"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -516,9 +529,9 @@ dependencies = [
[[package]]
name = "atoi"
-version = "2.0.0"
+version = "3.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f28d99ec8bfea296261ca1af174f24225171fea9664ba9003cbebee704810528"
+checksum = "e7a8bbe9949e43a1edaa043038c68703b04774156afdfb62ba2cef5bf93d67be"
dependencies = [
"num-traits",
]
@@ -1138,9 +1151,9 @@ dependencies = [
[[package]]
name = "brotli"
-version = "8.0.2"
+version = "9.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4bd8b9603c7aa97359dbd97ecf258968c95f3adddd6db2f7e7a5bef101c84560"
+checksum = "f8b851b75c23ca7873623d612fe49bd1989aeb03d08fb9432187eb253d3d4c6b"
dependencies = [
"alloc-no-stdlib",
"alloc-stdlib",
@@ -1149,9 +1162,9 @@ dependencies = [
[[package]]
name = "brotli-decompressor"
-version = "5.0.0"
+version = "6.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "874bb8112abecc98cbd6d81ea4fa7e94fb9449648c93cc89aa40c81c24d7de03"
+checksum = "46087801e40b785410c352bee60e9b6226312709cf52d5fff92776cdf5cf6490"
dependencies = [
"alloc-no-stdlib",
"alloc-stdlib",
@@ -1375,11 +1388,21 @@ version = "1.0.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570"
+[[package]]
+name = "combine"
+version = "4.6.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ba5a308b75df32fe02788e748662718f03fde005016435c444eea572398219fd"
+dependencies = [
+ "bytes",
+ "memchr",
+]
+
[[package]]
name = "comfy-table"
-version = "7.2.2"
+version = "8.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "958c5d6ecf1f214b4c2bbbbf6ab9523a864bd136dcf71a7e8904799acfe1ad47"
+checksum = "136c8c4c3823846e8ba6d4bda011b4e5d5827b8bb0ac26c31f9ab31abe9f54f2"
dependencies = [
"unicode-segmentation",
"unicode-width 0.2.2",
@@ -1396,8 +1419,8 @@ dependencies = [
"flate2",
"liblzma",
"memchr",
- "zstd 0.14.0",
- "zstd-safe 8.0.0",
+ "zstd",
+ "zstd-safe",
]
[[package]]
@@ -1509,6 +1532,16 @@ version = "2.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "217698eaf96b4a3f0bc4f3662aaa55bdf913cd54d7204591faa790070c6d0853"
+[[package]]
+name = "crc-fast"
+version = "1.10.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e75b2483e97a5a7da73ac68a05b629f9c53cff58d8ed1c77866079e18b00dba5"
+dependencies = [
+ "digest 0.10.7",
+ "spin",
+]
+
[[package]]
name = "crc32fast"
version = "1.5.1"
@@ -1771,7 +1804,7 @@ dependencies = [
"unicode-width 0.2.2",
"url",
"uuid",
- "zstd 0.14.0",
+ "zstd",
]
[[package]]
@@ -1959,7 +1992,7 @@ dependencies = [
"tokio",
"tokio-util",
"url",
- "zstd 0.14.0",
+ "zstd",
]
[[package]]
@@ -3346,8 +3379,6 @@ dependencies = [
"cfg-if",
"crunchy",
"num-traits",
- "rand 0.9.4",
- "rand_distr",
"zerocopy",
]
@@ -3892,6 +3923,55 @@ dependencies = [
"syn 2.0.119",
]
+[[package]]
+name = "jni"
+version = "0.22.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5efd9a482cf3a427f00d6b35f14332adc7902ce91efb778580e180ff90fa3498"
+dependencies = [
+ "cfg-if",
+ "combine",
+ "jni-macros",
+ "jni-sys",
+ "log",
+ "simd_cesu8",
+ "thiserror",
+ "walkdir",
+ "windows-link",
+]
+
+[[package]]
+name = "jni-macros"
+version = "0.22.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a00109accc170f0bdb141fed3e393c565b6f5e072365c3bd58f5b062591560a3"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "rustc_version",
+ "simd_cesu8",
+ "syn 2.0.119",
+]
+
+[[package]]
+name = "jni-sys"
+version = "0.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c6377a88cb3910bee9b0fa88d4f42e1d2da8e79915598f65fb0c7ee14c878af2"
+dependencies = [
+ "jni-sys-macros",
+]
+
+[[package]]
+name = "jni-sys-macros"
+version = "0.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "38c0b942f458fe50cdac086d2f946512305e5631e720728f2a61aabcd47a6264"
+dependencies = [
+ "quote",
+ "syn 2.0.119",
+]
+
[[package]]
name = "jobserver"
version = "0.1.34"
@@ -4440,14 +4520,16 @@ dependencies = [
[[package]]
name = "object_store"
-version = "0.13.2"
+version = "0.14.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "622acbc9100d3c10e2ee15804b0caa40e55c933d5aa53814cd520805b7958a49"
+checksum = "f1796bc93603f78c5760a69f2d58badc9618d22adade0a95385bb2adbae4eb94"
dependencies = [
"async-trait",
- "base64 0.22.1",
+ "aws-lc-rs",
+ "base64 0.23.1",
"bytes",
"chrono",
+ "crc-fast",
"form_urlencoded",
"futures-channel",
"futures-core",
@@ -4456,14 +4538,14 @@ dependencies = [
"http-body-util",
"humantime",
"hyper",
- "itertools 0.14.0",
- "md-5 0.10.6",
+ "itertools 0.15.0",
+ "md-5 0.11.0",
+ "nix",
"parking_lot",
"percent-encoding",
"quick-xml",
"rand 0.10.1",
"reqwest",
- "ring",
"rustls-pki-types",
"serde",
"serde_json",
@@ -4475,6 +4557,7 @@ dependencies = [
"walkdir",
"wasm-bindgen-futures",
"web-time",
+ "windows-sys 0.61.2",
]
[[package]]
@@ -4554,9 +4637,9 @@ dependencies = [
[[package]]
name = "parquet"
-version = "59.3.0"
+version = "60.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ff322f54b1a0f9288e614ed1f2d329b380af5476420db19f46ffb865e1163d73"
+checksum = "8af83d2940bc0510f9aef86d865f56fdc6095f87ab115ac885a80b7c5226d3ba"
dependencies = [
"ahash",
"arrow-array",
@@ -4584,7 +4667,7 @@ dependencies = [
"snap",
"tokio",
"twox-hash",
- "zstd 0.13.3",
+ "zstd",
]
[[package]]
@@ -4990,9 +5073,9 @@ dependencies = [
[[package]]
name = "quick-xml"
-version = "0.39.2"
+version = "0.41.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "958f21e8e7ceb5a1aa7fa87fab28e7c75976e0bfe7e23ff069e0a260f894067d"
+checksum = "e660451e55124f798a69a5af3f49ccfbefbd41910eefd25caf2393e1f3473ec1"
dependencies = [
"memchr",
"serde",
@@ -5024,6 +5107,7 @@ version = "0.11.15"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4fcb935c5bec503c2f0e306bdd3e58bb9029dcb14fa8d9ac76e3a5256ac0763e"
dependencies = [
+ "aws-lc-rs",
"bytes",
"getrandom 0.3.4",
"lru-slab",
@@ -5312,11 +5396,11 @@ checksum = "ba39f3699c378cd8970968dcbff9c43159ea4cfbd88d43c00b22f2ef10a435d2"
[[package]]
name = "reqwest"
-version = "0.12.28"
+version = "0.13.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147"
+checksum = "16a1cfa75cc186dd73d5818e510e042e40927bccc9c236b061cea97e1eb08029"
dependencies = [
- "base64 0.22.1",
+ "base64 0.23.1",
"bytes",
"futures-core",
"futures-util",
@@ -5333,11 +5417,8 @@ dependencies = [
"pin-project-lite",
"quinn",
"rustls",
- "rustls-native-certs",
"rustls-pki-types",
- "serde",
- "serde_json",
- "serde_urlencoded",
+ "rustls-platform-verifier",
"sync_wrapper",
"tokio",
"tokio-rustls",
@@ -5472,6 +5553,33 @@ dependencies = [
"zeroize",
]
+[[package]]
+name = "rustls-platform-verifier"
+version = "0.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "26d1e2536ce4f35f4846aa13bff16bd0ff40157cdb14cc056c7b14ba41233ba0"
+dependencies = [
+ "core-foundation",
+ "core-foundation-sys",
+ "jni",
+ "log",
+ "once_cell",
+ "rustls",
+ "rustls-native-certs",
+ "rustls-platform-verifier-android",
+ "rustls-webpki",
+ "security-framework",
+ "security-framework-sys",
+ "webpki-root-certs",
+ "windows-sys 0.61.2",
+]
+
+[[package]]
+name = "rustls-platform-verifier-android"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f"
+
[[package]]
name = "rustls-webpki"
version = "0.103.15"
@@ -5848,6 +5956,16 @@ version = "0.3.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "703d5c7ef118737c72f1af64ad2f6f8c5e1921f818cdcb97b8fe6fc69bf66214"
+[[package]]
+name = "simd_cesu8"
+version = "1.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "11031e251abf8611c80f460e19dbdeb54a66db918e49c65a7065b46ac7aec520"
+dependencies = [
+ "rustc_version",
+ "simdutf8",
+]
+
[[package]]
name = "simdutf8"
version = "0.1.5"
@@ -5912,6 +6030,12 @@ dependencies = [
"windows-sys 0.61.2",
]
+[[package]]
+name = "spin"
+version = "0.10.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "023a211cb3138dbc438680b32560ad89f699977624c9f8dbb95a47d5b4c07dd3"
+
[[package]]
name = "sqllogictest"
version = "0.29.1"
@@ -7056,9 +7180,9 @@ dependencies = [
[[package]]
name = "wasm-streams"
-version = "0.4.2"
+version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "15053d8d85c7eccdbefef60f06769760a563c7f0a9d6902a13d35c7800b0ad65"
+checksum = "9d1ec4f6517c9e11ae630e200b2b65d193279042e28edd4a2cda233e46670bbb"
dependencies = [
"futures-util",
"js-sys",
@@ -7099,6 +7223,15 @@ dependencies = [
"wasm-bindgen",
]
+[[package]]
+name = "webpki-root-certs"
+version = "1.0.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b96554aa2acc8ccdb7e1c9a58a7a68dd5d13bccc69cd124cb09406db612a1c9b"
+dependencies = [
+ "rustls-pki-types",
+]
+
[[package]]
name = "whoami"
version = "2.1.1"
@@ -7581,31 +7714,13 @@ version = "1.0.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa"
-[[package]]
-name = "zstd"
-version = "0.13.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e91ee311a569c327171651566e07972200e76fcfe2242a4fa446149a3881c08a"
-dependencies = [
- "zstd-safe 7.2.4",
-]
-
[[package]]
name = "zstd"
version = "0.14.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bf06bd8162af0734b344780deb55b42a2429ae430870d13fcc12f238e880fe6e"
dependencies = [
- "zstd-safe 8.0.0",
-]
-
-[[package]]
-name = "zstd-safe"
-version = "7.2.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8f49c4d5f0abb602a93fb8736af2a4f4dd9512e36f7f570d66e65ff867ed3b9d"
-dependencies = [
- "zstd-sys",
+ "zstd-safe",
]
[[package]]
diff --git a/Cargo.toml b/Cargo.toml
index 34ff33de37..622c947d7d 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -90,31 +90,31 @@ version = "55.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.2.0", features = [
+arrow = { version = "60.0.0", features = [
"prettyprint",
"chrono-tz",
] }
-arrow-avro = { version = "59.2.0", default-features = false, features = [
+arrow-avro = { version = "60.0.0", default-features = false, features = [
"deflate",
"snappy",
"zstd",
"bzip2",
"xz",
] }
-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",
+arrow-buffer = { version = "60.0.0", default-features = false }
+arrow-data = { version = "60.0.0", default-features = false }
+arrow-flight = { version = "60.0.0", features = [
+ "flight-sql",
] }
# Both codecs are required here to make sure that code paths like
# file-spilling have access to all compression codecs.
-arrow-ipc = { version = "59.2.0", default-features = false, features = [
+arrow-ipc = { version = "60.0.0", default-features = false, features = [
"lz4",
"zstd",
] }
-arrow-ord = { version = "59.2.0", default-features = false }
-arrow-schema = { version = "59.2.0", default-features = false }
-arrow-select = { version = "59.2.0", default-features = false }
+arrow-ord = { version = "60.0.0", default-features = false }
+arrow-schema = { version = "60.0.0", default-features = false }
+arrow-select = { version = "60.0.0", default-features = false }
async-trait = "0.1.89"
bigdecimal = "0.4.8"
bytes = "1.11"
@@ -179,9 +179,9 @@ log = "^0.4"
loom = "0.7"
memchr = "2.8.1"
num-traits = { version = "0.2" }
-object_store = { version = "0.13.2", default-features = false }
+object_store = { version = "0.14.2", default-features = false }
parking_lot = "0.12"
-parquet = { version = "59.2.0", default-features = false, features = [
+parquet = { version = "60.0.0", default-features = false, features = [
"arrow",
"async",
"object_store",
diff --git a/datafusion-cli/src/main.rs b/datafusion-cli/src/main.rs
index f82206a5bd..2f84a9aaf4 100644
--- a/datafusion-cli/src/main.rs
+++ b/datafusion-cli/src/main.rs
@@ -662,9 +662,9 @@ mod tests {
+-----------------------------------+-----------------+---------------------+------+------------------+
| filename | file_size_bytes | metadata_size_bytes | hits | extra |
+-----------------------------------+-----------------+---------------------+------+------------------+
- | alltypes_plain.parquet | 1851 | 8794 | 1 | page_index=false |
- | alltypes_tiny_pages.parquet | 454233 | 268970 | 2 | page_index=true |
- | lz4_raw_compressed_larger.parquet | 380836 | 1331 | 1 | page_index=false |
+ | alltypes_plain.parquet | 1851 | 8938 | 1 | page_index=false |
+ | alltypes_tiny_pages.parquet | 454233 | 268826 | 2 | page_index=true |
+ | lz4_raw_compressed_larger.parquet | 380836 | 1315 | 1 | page_index=false |
+-----------------------------------+-----------------+---------------------+------+------------------+
");
@@ -693,9 +693,9 @@ mod tests {
+-----------------------------------+-----------------+---------------------+------+------------------+
| filename | file_size_bytes | metadata_size_bytes | hits | extra |
+-----------------------------------+-----------------+---------------------+------+------------------+
- | alltypes_plain.parquet | 1851 | 8794 | 4 | page_index=false |
- | alltypes_tiny_pages.parquet | 454233 | 268970 | 2 | page_index=true |
- | lz4_raw_compressed_larger.parquet | 380836 | 1331 | 2 | page_index=false |
+ | alltypes_plain.parquet | 1851 | 8938 | 4 | page_index=false |
+ | alltypes_tiny_pages.parquet | 454233 | 268826 | 2 | page_index=true |
+ | lz4_raw_compressed_larger.parquet | 380836 | 1315 | 2 | page_index=false |
+-----------------------------------+-----------------+---------------------+------+------------------+
");
@@ -862,14 +862,14 @@ mod tests {
])?
.sort(vec![col("filename").sort(true, false)])?;
let rbs = df.collect().await?;
- assert_snapshot!(batches_to_string(&rbs),@r"
+ assert_snapshot!(batches_to_string(&rbs),@r#"
+---------------------+-----------+-----------------+------+------+
| metadata_size_bytes | filename | file_size_bytes | etag | hits |
+---------------------+-----------+-----------------+------+------+
- | 212 | 0.parquet | 3642 | 0 | 2 |
- | 212 | 1.parquet | 3642 | 1 | 2 |
+ | 216 | 0.parquet | 3620 | "0" | 2 |
+ | 216 | 1.parquet | 3620 | "1" | 2 |
+---------------------+-----------+-----------------+------+------+
- ");
+ "#);
Ok(())
}
diff --git a/datafusion/catalog-listing/src/helpers.rs b/datafusion/catalog-listing/src/helpers.rs
index dc090378a8..cf0d170809 100644
--- a/datafusion/catalog-listing/src/helpers.rs
+++ b/datafusion/catalog-listing/src/helpers.rs
@@ -30,7 +30,7 @@ use datafusion_expr::{BinaryExpr, Operator, lit, utils};
use arrow::{
array::AsArray,
- datatypes::{DataType, Field},
+ datatypes::{DataType, Field, Metadata},
record_batch::RecordBatch,
};
use datafusion_expr::execution_props::ExecutionProps;
@@ -423,7 +423,7 @@ pub async fn pruned_partition_list<'a>(
.iter()
.map(|(n, d)| Field::new(n, d.clone(), true))
.collect(),
- Default::default(),
+ Metadata::new(),
)?;
Ok(objects
diff --git a/datafusion/catalog-listing/src/table.rs b/datafusion/catalog-listing/src/table.rs
index 6c294fe077..b04c267d1e 100644
--- a/datafusion/catalog-listing/src/table.rs
+++ b/datafusion/catalog-listing/src/table.rs
@@ -20,7 +20,7 @@ use crate::helpers::{
expr_applicable_for_cols, filter_partitioned_file, pruned_partition_list,
};
use crate::{ListingOptions, ListingTableConfig};
-use arrow::datatypes::{Field, Schema, SchemaBuilder, SchemaRef};
+use arrow::datatypes::{Field, Metadata, Schema, SchemaBuilder, SchemaRef};
use async_trait::async_trait;
use datafusion_catalog::{ScanArgs, ScanResult, Session, TableProvider};
use datafusion_common::stats::{Precision, is_known_empty};
@@ -1044,7 +1044,7 @@ impl ListingTable {
.iter()
.map(|(name, data_type)| Field::new(name, data_type.clone(), true))
.collect(),
- Default::default(),
+ Metadata::new(),
)?;
file_groups
diff --git a/datafusion/common/src/dfschema.rs b/datafusion/common/src/dfschema.rs
index a0e2f0590a..258d74ecb3 100644
--- a/datafusion/common/src/dfschema.rs
+++ b/datafusion/common/src/dfschema.rs
@@ -18,7 +18,7 @@
//! DFSchema is an extended schema struct that DataFusion uses to provide support for
//! fields with optional relation names.
-use std::collections::{BTreeSet, HashMap, HashSet};
+use std::collections::{BTreeSet, HashSet};
use std::fmt::Write as _;
use std::fmt::{Display, Formatter};
use std::hash::Hash;
@@ -32,7 +32,7 @@ use crate::{
use arrow::compute::can_cast_types;
use arrow::datatypes::{
- DataType, Field, FieldRef, Fields, Schema, SchemaBuilder, SchemaRef,
+ DataType, Field, FieldRef, Fields, Metadata, Schema, SchemaBuilder, SchemaRef,
};
/// A reference-counted reference to a [DFSchema].
@@ -154,7 +154,7 @@ impl DFSchema {
/// Create a `DFSchema` from an Arrow schema where all the fields have a given qualifier
pub fn new_with_metadata(
qualified_fields: Vec<(Option<TableReference>, Arc<Field>)>,
- metadata: HashMap<String, String>,
+ metadata: impl Into<Metadata>,
) -> Result<Self> {
let (qualifiers, fields): (Vec<Option<TableReference>>, Vec<Arc<Field>>) =
qualified_fields.into_iter().unzip();
@@ -173,7 +173,7 @@ impl DFSchema {
/// Create a new `DFSchema` from a list of Arrow [Field]s
pub fn from_unqualified_fields(
fields: Fields,
- metadata: HashMap<String, String>,
+ metadata: impl Into<Metadata>,
) -> Result<Self> {
let field_count = fields.len();
let schema = Arc::new(Schema::new_with_metadata(fields, metadata));
@@ -865,7 +865,7 @@ impl DFSchema {
}
/// Get metadata of this schema
- pub fn metadata(&self) -> &HashMap<String, String> {
+ pub fn metadata(&self) -> &Metadata {
&self.inner.metadata
}
@@ -1192,7 +1192,7 @@ impl ToDFSchema for Vec<Field> {
let field_count = self.len();
let schema = Schema {
fields: self.into(),
- metadata: HashMap::new(),
+ metadata: Metadata::new(),
};
let dfschema = DFSchema {
inner: schema.into(),
@@ -1234,7 +1234,7 @@ pub trait ExprSchema: std::fmt::Debug {
}
/// Returns the column's optional metadata.
- fn metadata(&self, col: &Column) -> Result<&HashMap<String, String>> {
+ fn metadata(&self, col: &Column) -> Result<&Metadata> {
Ok(self.field_from_column(col)?.metadata())
}
@@ -1258,7 +1258,7 @@ impl<P: AsRef<DFSchema> + std::fmt::Debug> ExprSchema for P {
self.as_ref().data_type(col)
}
- fn metadata(&self, col: &Column) -> Result<&HashMap<String, String>> {
+ fn metadata(&self, col: &Column) -> Result<&Metadata> {
ExprSchema::metadata(self.as_ref(), col)
}
@@ -1391,6 +1391,7 @@ pub fn qualified_name(qualifier: Option<&TableReference>, name: &str) -> String
#[cfg(test)]
mod tests {
use crate::assert_contains;
+ use std::collections::HashMap;
use super::*;
diff --git a/datafusion/common/src/hash_utils.rs b/datafusion/common/src/hash_utils.rs
index a716ca7548..1e6d96ecfd 100644
--- a/datafusion/common/src/hash_utils.rs
+++ b/datafusion/common/src/hash_utils.rs
@@ -689,8 +689,8 @@ fn hash_map_array(
let offsets = array.offsets();
// Create hashes for each entry in each row
- let first_offset = offsets.first().copied().unwrap_or_default() as usize;
- let last_offset = offsets.last().copied().unwrap_or_default() as usize;
+ let first_offset = offsets.first() as usize;
+ let last_offset = offsets.last() as usize;
let entries_len = last_offset - first_offset;
// Only hash the entries that are actually referenced
diff --git a/datafusion/common/src/heap_size.rs b/datafusion/common/src/heap_size.rs
index 3dbe096953..14649ca161 100644
--- a/datafusion/common/src/heap_size.rs
+++ b/datafusion/common/src/heap_size.rs
@@ -47,12 +47,12 @@ use arrow::array::{
};
use arrow::datatypes::{
DataType, Field, Fields, IntervalDayTime, IntervalMonthDayNano, IntervalUnit,
- TimeUnit, UnionFields, UnionMode, i256,
+ Metadata, TimeUnit, UnionFields, UnionMode, i256,
};
use chrono::{DateTime, Utc};
use half::f16;
use hashbrown::HashSet;
-use std::collections::HashMap;
+use std::collections::{BTreeMap, HashMap};
use std::fmt::Debug;
use std::sync::Arc;
@@ -396,6 +396,24 @@ impl DFHeapSize for UnionFields {
}
}
+impl<K: DFHeapSize, V: DFHeapSize> DFHeapSize for BTreeMap<K, V> {
+ fn heap_size(&self, ctx: &mut DFHeapSizeCtx) -> usize {
+ // BTreeMap does not provide a way to get its heap size, so this
+ // approximates it as the entries' sizes, ignoring node overhead.
+ self.iter()
+ .map(|(k, v)| size_of::<(K, V)>() + k.heap_size(ctx) + v.heap_size(ctx))
+ .sum()
+ }
+}
+
+impl DFHeapSize for Metadata {
+ fn heap_size(&self, ctx: &mut DFHeapSizeCtx) -> usize {
+ // `as_arc` returns `None` when the metadata is empty; the `Arc` impl
+ // dedupes instances that share the same allocation.
+ self.as_arc().map_or(0, |map| map.heap_size(ctx))
+ }
+}
+
impl DFHeapSize for Field {
fn heap_size(&self, ctx: &mut DFHeapSizeCtx) -> usize {
self.name().heap_size(ctx)
diff --git a/datafusion/common/src/metadata.rs b/datafusion/common/src/metadata.rs
index d6d8fb7b0e..34242e9b52 100644
--- a/datafusion/common/src/metadata.rs
+++ b/datafusion/common/src/metadata.rs
@@ -17,7 +17,7 @@
use std::{collections::BTreeMap, sync::Arc};
-use arrow::datatypes::{DataType, Field, FieldRef};
+use arrow::datatypes::{DataType, Field, FieldRef, Metadata};
use hashbrown::HashMap;
use crate::{DataFusionError, ScalarValue, error::_plan_err};
@@ -84,14 +84,8 @@ impl From<ScalarValue> for ScalarAndMetadata {
/// Returns a planning error with suitably formatted type representations if
/// actual and expected do not compare to equal.
pub fn check_metadata_with_storage_equal(
- actual: (
- &DataType,
- Option<&std::collections::HashMap<String, String>>,
- ),
- expected: (
- &DataType,
- Option<&std::collections::HashMap<String, String>>,
- ),
+ actual: (&DataType, Option<&Metadata>),
+ expected: (&DataType, Option<&Metadata>),
what: &str,
context: &str,
) -> Result<(), DataFusionError> {
@@ -131,7 +125,7 @@ pub fn check_metadata_with_storage_equal(
/// renderings.
pub fn format_type_and_metadata(
data_type: &DataType,
- metadata: Option<&std::collections::HashMap<String, String>>,
+ metadata: Option<&Metadata>,
) -> String {
match metadata {
Some(metadata) if !metadata.is_empty() => {
@@ -179,10 +173,7 @@ pub fn format_type_and_metadata(
pub struct FieldMetadata {
/// The inner metadata of a literal expression, which is a map of string
/// keys to string values.
- ///
- /// Note this is not a `HashMap` because `HashMap` does not provide
- /// implementations for traits like `Debug` and `Hash`.
- inner: Arc<BTreeMap<String, String>>,
+ inner: Metadata,
}
impl Default for FieldMetadata {
@@ -195,7 +186,7 @@ impl FieldMetadata {
/// Create a new empty metadata instance.
pub fn new_empty() -> Self {
Self {
- inner: Arc::new(BTreeMap::new()),
+ inner: Metadata::new(),
}
}
@@ -261,31 +252,28 @@ impl FieldMetadata {
}
/// Create a new metadata instance from a `Field`'s metadata.
+ ///
+ /// This is cheap, as [`Metadata`] is cheap to clone.
pub fn new_from_field(field: &Field) -> Self {
- let inner = field
- .metadata()
- .iter()
- .map(|(k, v)| (k.to_string(), v.to_string()))
- .collect();
Self {
- inner: Arc::new(inner),
+ inner: field.metadata().clone(),
}
}
/// Create a new metadata instance from a map of string keys to string values.
pub fn new(inner: BTreeMap<String, String>) -> Self {
Self {
- inner: Arc::new(inner),
+ inner: Metadata::from(inner),
}
}
- /// Get the inner metadata as a reference to a `BTreeMap`.
- pub fn inner(&self) -> &BTreeMap<String, String> {
+ /// Get the inner [`Metadata`]
+ pub fn inner(&self) -> &Metadata {
&self.inner
}
- /// Return the inner metadata
- pub fn into_inner(self) -> Arc<BTreeMap<String, String>> {
+ /// Return the inner [`Metadata`]
+ pub fn into_inner(self) -> Metadata {
self.inner
}
@@ -294,8 +282,7 @@ impl FieldMetadata {
if other.is_empty() {
return;
}
- let other = Arc::unwrap_or_clone(other.into_inner());
- Arc::make_mut(&mut self.inner).extend(other);
+ self.inner.extend(other.inner);
}
/// Returns true if the metadata is empty.
@@ -310,10 +297,14 @@ impl FieldMetadata {
/// Convert this `FieldMetadata` into a `HashMap<String, String>`
pub fn to_hashmap(&self) -> std::collections::HashMap<String, String> {
- self.inner
- .iter()
- .map(|(k, v)| (k.to_string(), v.to_string()))
- .collect()
+ (&self.inner).into()
+ }
+
+ /// Convert this `FieldMetadata` into an arrow [`Metadata`]
+ ///
+ /// This is cheap, as [`Metadata`] is cheap to clone.
+ pub fn to_metadata(&self) -> Metadata {
+ self.inner.clone()
}
/// Updates the metadata on the Field with this metadata, if it is not empty.
@@ -322,7 +313,7 @@ impl FieldMetadata {
return field;
}
- field.with_metadata(self.to_hashmap())
+ field.with_metadata(self.inner.clone())
}
/// Updates the metadata on the FieldRef with this metadata, if it is not empty.
@@ -331,11 +322,29 @@ impl FieldMetadata {
return field_ref;
}
- Arc::make_mut(&mut field_ref).set_metadata(self.to_hashmap());
+ Arc::make_mut(&mut field_ref).set_metadata(self.inner.clone());
field_ref
}
}
+impl From<&FieldMetadata> for Metadata {
+ fn from(value: &FieldMetadata) -> Self {
+ value.to_metadata()
+ }
+}
+
+impl From<Metadata> for FieldMetadata {
+ fn from(value: Metadata) -> Self {
+ Self { inner: value }
+ }
+}
+
+impl From<&Metadata> for FieldMetadata {
+ fn from(value: &Metadata) -> Self {
+ Self::from(value.clone())
+ }
+}
+
impl From<&Field> for FieldMetadata {
fn from(field: &Field) -> Self {
Self::new_from_field(field)
@@ -350,7 +359,9 @@ impl From<BTreeMap<String, String>> for FieldMetadata {
impl From<std::collections::HashMap<String, String>> for FieldMetadata {
fn from(map: std::collections::HashMap<String, String>) -> Self {
- Self::new(map.into_iter().collect())
+ Self {
+ inner: Metadata::from(map),
+ }
}
}
@@ -368,17 +379,16 @@ impl From<&std::collections::HashMap<String, String>> for FieldMetadata {
/// From hashbrown map
impl From<HashMap<String, String>> for FieldMetadata {
fn from(map: HashMap<String, String>) -> Self {
- let inner = map.into_iter().collect();
- Self::new(inner)
+ Self {
+ inner: map.into_iter().collect(),
+ }
}
}
impl From<&HashMap<String, String>> for FieldMetadata {
fn from(map: &HashMap<String, String>) -> Self {
- let inner = map
- .into_iter()
- .map(|(k, v)| (k.to_string(), v.to_string()))
- .collect();
- Self::new(inner)
+ Self {
+ inner: map.iter().map(|(k, v)| (k.clone(), v.clone())).collect(),
+ }
}
}
diff --git a/datafusion/common/src/nested_struct.rs b/datafusion/common/src/nested_struct.rs
index 2f046b92c3..e79c21dfb7 100644
--- a/datafusion/common/src/nested_struct.rs
+++ b/datafusion/common/src/nested_struct.rs
@@ -1821,8 +1821,8 @@ mod tests {
Arc::new(non_null_field(
"entries",
struct_type(vec![
- non_null_field("keys", DataType::Utf8),
- field("values", DataType::Int32),
+ non_null_field("key", DataType::Utf8),
+ field("value", DataType::Int32),
]),
)),
false,
@@ -1846,8 +1846,8 @@ mod tests {
Arc::new(non_null_field(
"entries",
struct_type(vec![
- non_null_field("keys", DataType::Utf8),
- field("values", DataType::Int32),
+ non_null_field("key", DataType::Utf8),
+ field("value", DataType::Int32),
]),
)),
false,
@@ -1874,8 +1874,8 @@ mod tests {
assert!(map.is_null(1));
let map0 = map.value(0);
let entries = map0.as_any().downcast_ref::<StructArray>().unwrap();
- let keys = get_column_as!(entries, "keys", StringArray);
- let vals = get_column_as!(entries, "values", Int32Array);
+ let keys = get_column_as!(entries, "key", StringArray);
+ let vals = get_column_as!(entries, "value", Int32Array);
assert_eq!(keys.value(0), "a");
assert_eq!(vals.value(0), 1);
}
diff --git a/datafusion/common/src/param_value.rs b/datafusion/common/src/param_value.rs
index 0fac6b529e..1979daa7e3 100644
--- a/datafusion/common/src/param_value.rs
+++ b/datafusion/common/src/param_value.rs
@@ -64,7 +64,7 @@ impl ParamValues {
check_metadata_with_storage_equal(
(
&lit.value.data_type(),
- lit.metadata.as_ref().map(|m| m.to_hashmap()).as_ref(),
+ lit.metadata.as_ref().map(|m| m.to_metadata()).as_ref(),
),
(param_type.data_type(), Some(param_type.metadata())),
"parameter",
diff --git a/datafusion/common/src/scalar/mod.rs b/datafusion/common/src/scalar/mod.rs
index cf7047e598..34c9f9e03d 100644
--- a/datafusion/common/src/scalar/mod.rs
+++ b/datafusion/common/src/scalar/mod.rs
@@ -11109,7 +11109,7 @@ mod tests {
Box::new(ScalarValue::Float32(None)),
);
let err = scalar.eq_array(&run_array, 0).unwrap_err();
- let expected = "Internal error: could not cast array of type RunEndEncoded(\"run_ends\": non-null Int16, \"values\": Float32) to arrow_array::array::run_array::RunArray<arrow_array::types::Int32Type>";
+ let expected = "Internal error: could not cast array of type RunEndEncoded(non-null Int16, Float32) to arrow_array::array::run_array::RunArray<arrow_array::types::Int32Type>";
assert!(err.to_string().starts_with(expected));
}
diff --git a/datafusion/common/src/utils/memory.rs b/datafusion/common/src/utils/memory.rs
index 119d2e73ee..bb006295f8 100644
--- a/datafusion/common/src/utils/memory.rs
+++ b/datafusion/common/src/utils/memory.rs
@@ -330,7 +330,7 @@ impl RecordBatchMemoryCounter {
array: &GenericByteViewArray<T>,
) {
self.count_buffer_memory_size(array.views().inner());
- for buffer in array.data_buffers() {
+ for buffer in array.data_buffers().iter() {
self.count_buffer_memory_size(buffer);
}
}
diff --git a/datafusion/common/src/utils/mod.rs b/datafusion/common/src/utils/mod.rs
index e047db39a5..fba76f949f 100644
--- a/datafusion/common/src/utils/mod.rs
+++ b/datafusion/common/src/utils/mod.rs
@@ -1246,13 +1246,11 @@ fn sliced_list_values<O: OffsetSizeTrait>(list: &GenericListArray<O>) -> ArrayRe
let values = list.values();
let offsets = list.offsets();
- if let (Some(first), Some(last)) = (offsets.first(), offsets.last()) {
- let first = first.as_usize();
- let last = last.as_usize();
+ let first = offsets.first().as_usize();
+ let last = offsets.last().as_usize();
- if first != 0 || last != values.len() {
- return values.slice(first, last - first);
- }
+ if first != 0 || last != values.len() {
+ return values.slice(first, last - first);
}
Arc::clone(values)
diff --git a/datafusion/core/src/datasource/file_format/csv.rs b/datafusion/core/src/datasource/file_format/csv.rs
index 2fb64fd648..8442b00900 100644
--- a/datafusion/core/src/datasource/file_format/csv.rs
+++ b/datafusion/core/src/datasource/file_format/csv.rs
@@ -136,6 +136,7 @@ mod tests {
},
range: Default::default(),
attributes: Attributes::default(),
+ extensions: Default::default(),
})
}
diff --git a/datafusion/core/src/datasource/file_format/parquet.rs b/datafusion/core/src/datasource/file_format/parquet.rs
index bfcfb74848..c5685e1d2d 100644
--- a/datafusion/core/src/datasource/file_format/parquet.rs
+++ b/datafusion/core/src/datasource/file_format/parquet.rs
@@ -165,10 +165,8 @@ mod tests {
};
use parquet::arrow::ParquetRecordBatchStreamBuilder;
use parquet::arrow::arrow_reader::ArrowReaderOptions;
- use parquet::file::metadata::{
- KeyValue, PageIndexPolicy, ParquetColumnIndex, ParquetMetaData,
- ParquetOffsetIndex,
- };
+ use parquet::file::metadata::page_index::PageIndexProvider;
+ use parquet::file::metadata::{KeyValue, PageIndexPolicy, ParquetMetaData};
use parquet::file::page_index::column_index::ColumnIndexMetaData;
use tokio::fs::File;
@@ -1116,7 +1114,7 @@ mod tests {
.await?
.metadata()
.clone();
- check_page_index_validation(builder.column_index(), builder.offset_index());
+ check_page_index_validation(builder.page_index());
let path = format!("{testdata}/alltypes_tiny_pages_plain.parquet");
let file = File::open(path).await?;
@@ -1125,34 +1123,28 @@ mod tests {
.await?
.metadata()
.clone();
- check_page_index_validation(builder.column_index(), builder.offset_index());
+ check_page_index_validation(builder.page_index());
Ok(())
}
- fn check_page_index_validation(
- page_index: Option<&ParquetColumnIndex>,
- offset_index: Option<&ParquetOffsetIndex>,
- ) {
- assert!(page_index.is_some());
- assert!(offset_index.is_some());
-
+ fn check_page_index_validation(page_index: Option<&Arc<dyn PageIndexProvider>>) {
let page_index = page_index.unwrap();
- let offset_index = offset_index.unwrap();
-
- // there is only one row group in one file.
- assert_eq!(page_index.len(), 1);
- assert_eq!(offset_index.len(), 1);
- let page_index = page_index.first().unwrap();
- let offset_index = offset_index.first().unwrap();
-
- // 13 col in one row group
- assert_eq!(page_index.len(), 13);
- assert_eq!(offset_index.len(), 13);
+ assert!(page_index.is_complete());
+
+ // there is only one row group in one file, with 13 columns.
+ // All columns have an offset index; all except column 10 also have a
+ // column index.
+ for col in 0..13 {
+ assert_eq!(page_index.column_index(0, col).is_some(), col != 10);
+ assert!(page_index.offset_index(0, col).is_some());
+ }
+ assert!(page_index.column_index(1, 0).is_none());
+ assert!(page_index.offset_index(1, 0).is_none());
// test result in int_col
- let int_col_index = page_index.get(4).unwrap();
- let int_col_offset = offset_index.get(4).unwrap().page_locations();
+ let int_col_index = page_index.column_index(0, 4).unwrap();
+ let int_col_offset = page_index.offset_index(0, 4).unwrap().page_locations();
// 325 pages in int_col
assert_eq!(int_col_offset.len(), 325);
diff --git a/datafusion/core/tests/parquet/string_in_list_pruning.rs b/datafusion/core/tests/parquet/string_in_list_pruning.rs
index f264685571..1f9d6e31cb 100644
--- a/datafusion/core/tests/parquet/string_in_list_pruning.rs
+++ b/datafusion/core/tests/parquet/string_in_list_pruning.rs
@@ -59,10 +59,14 @@ fn assert_file_layout(
rows_per_group: usize,
) {
assert_eq!(metadata.num_row_groups(), total_rows / rows_per_group);
- let offsets = metadata.offset_index().unwrap();
- for row_group in offsets {
+ let page_index = metadata.page_index().unwrap();
+ for row_group in 0..metadata.num_row_groups() {
assert_eq!(
- row_group[0].page_locations().len(),
+ page_index
+ .offset_index(row_group, 0)
+ .unwrap()
+ .page_locations()
+ .len(),
rows_per_group / ROWS_PER_UNIT
);
}
diff --git a/datafusion/core/tests/sql/path_partition.rs b/datafusion/core/tests/sql/path_partition.rs
index f8701ccc56..a427481c43 100644
--- a/datafusion/core/tests/sql/path_partition.rs
+++ b/datafusion/core/tests/sql/path_partition.rs
@@ -718,6 +718,7 @@ impl ObjectStore for MirroringObjectStore {
payload,
meta,
attributes: Attributes::default(),
+ extensions: Default::default(),
})
}
@@ -786,6 +787,7 @@ impl ObjectStore for MirroringObjectStore {
Ok(ListResult {
common_prefixes: common_prefixes.into_iter().collect(),
objects,
+ extensions: Default::default(),
})
}
diff --git a/datafusion/core/tests/user_defined/user_defined_aggregates.rs b/datafusion/core/tests/user_defined/user_defined_aggregates.rs
index d035fa25e1..0f153ebbd3 100644
--- a/datafusion/core/tests/user_defined/user_defined_aggregates.rs
+++ b/datafusion/core/tests/user_defined/user_defined_aggregates.rs
@@ -30,7 +30,7 @@ use arrow::array::{
Array, AsArray, Int32Array, PrimitiveArray, StringArray, StructArray, UInt64Array,
record_batch, types::UInt64Type,
};
-use arrow::datatypes::{Fields, Schema};
+use arrow::datatypes::{Fields, Metadata, Schema};
use arrow_schema::FieldRef;
use datafusion::common::test_util::batches_to_string;
use datafusion::dataframe::DataFrame;
@@ -1018,10 +1018,8 @@ async fn test_metadata_based_aggregate() -> Result<()> {
let data_array = Arc::new(UInt64Array::from(vec![0, 5, 10, 15, 20])) as ArrayRef;
let schema = Arc::new(Schema::new(vec![
Field::new("no_metadata", DataType::UInt64, true),
- Field::new("with_metadata", DataType::UInt64, true).with_metadata(
- std::iter::once(("modify_values".to_string(), "double_output".to_string()))
- .collect(),
- ),
+ Field::new("with_metadata", DataType::UInt64, true)
+ .with_metadata(Metadata::new().with("modify_values", "double_output")),
]));
let batch = RecordBatch::try_new(
@@ -1091,10 +1089,8 @@ async fn test_metadata_based_aggregate_as_window() -> Result<()> {
let data_array = Arc::new(UInt64Array::from(vec![0, 5, 10, 15, 20])) as ArrayRef;
let schema = Arc::new(Schema::new(vec![
Field::new("no_metadata", DataType::UInt64, true),
- Field::new("with_metadata", DataType::UInt64, true).with_metadata(
- std::iter::once(("modify_values".to_string(), "double_output".to_string()))
- .collect(),
- ),
+ Field::new("with_metadata", DataType::UInt64, true)
+ .with_metadata(Metadata::new().with("modify_values", "double_output")),
]));
let batch = RecordBatch::try_new(
diff --git a/datafusion/core/tests/user_defined/user_defined_scalar_functions.rs b/datafusion/core/tests/user_defined/user_defined_scalar_functions.rs
index 36d094cd03..ef2aac0743 100644
--- a/datafusion/core/tests/user_defined/user_defined_scalar_functions.rs
+++ b/datafusion/core/tests/user_defined/user_defined_scalar_functions.rs
@@ -25,7 +25,7 @@ use arrow::array::{
};
use arrow::array::{Int8Array, UInt64Array, as_string_array, create_array, record_batch};
use arrow::compute::kernels::numeric::add;
-use arrow::datatypes::{DataType, Field, Schema};
+use arrow::datatypes::{DataType, Field, Metadata, Schema};
use arrow_schema::extension::{Bool8, CanonicalExtensionType, ExtensionType};
use arrow_schema::{ArrowError, FieldRef, SchemaRef};
use datafusion::common::test_util::batches_to_string;
@@ -1647,10 +1647,8 @@ async fn test_metadata_based_udf() -> Result<()> {
let data_array = Arc::new(UInt64Array::from(vec![0, 5, 10, 15, 20])) as ArrayRef;
let schema = Arc::new(Schema::new(vec![
Field::new("no_metadata", DataType::UInt64, true),
- Field::new("with_metadata", DataType::UInt64, true).with_metadata(
- std::iter::once(("modify_values".to_string(), "double_output".to_string()))
- .collect(),
- ),
+ Field::new("with_metadata", DataType::UInt64, true)
+ .with_metadata(Metadata::new().with("modify_values", "double_output")),
]));
let batch = RecordBatch::try_new(
schema,
diff --git a/datafusion/core/tests/user_defined/user_defined_window_functions.rs b/datafusion/core/tests/user_defined/user_defined_window_functions.rs
index fc14b9a1c6..d79f296d07 100644
--- a/datafusion/core/tests/user_defined/user_defined_window_functions.rs
+++ b/datafusion/core/tests/user_defined/user_defined_window_functions.rs
@@ -22,7 +22,7 @@ use arrow::array::{
Array, ArrayRef, AsArray, Int64Array, RecordBatch, StringArray, UInt64Array,
record_batch,
};
-use arrow::datatypes::{DataType, Field, Schema};
+use arrow::datatypes::{DataType, Field, Metadata, Schema};
use arrow_schema::FieldRef;
use datafusion::common::test_util::batches_to_string;
use datafusion::common::{Result, ScalarValue};
@@ -868,10 +868,8 @@ async fn test_metadata_based_window_fn() -> Result<()> {
let data_array = Arc::new(UInt64Array::from(vec![0, 5, 10, 15, 20])) as ArrayRef;
let schema = Arc::new(Schema::new(vec![
Field::new("no_metadata", DataType::UInt64, true),
- Field::new("with_metadata", DataType::UInt64, true).with_metadata(
- std::iter::once(("modify_values".to_string(), "double_output".to_string()))
- .collect(),
- ),
+ Field::new("with_metadata", DataType::UInt64, true)
+ .with_metadata(Metadata::new().with("modify_values", "double_output")),
]));
let batch = RecordBatch::try_new(
diff --git a/datafusion/datasource-arrow/src/file_format.rs b/datafusion/datasource-arrow/src/file_format.rs
index 2bee57ef17..1da354b863 100644
--- a/datafusion/datasource-arrow/src/file_format.rs
+++ b/datafusion/datasource-arrow/src/file_format.rs
@@ -26,7 +26,7 @@ use std::sync::Arc;
use arrow::datatypes::{Schema, SchemaRef};
use arrow::error::ArrowError;
-use arrow::ipc::convert::fb_to_schema;
+use arrow::ipc::convert::try_fb_to_schema;
use arrow::ipc::reader::{FileReader, StreamReader};
use arrow::ipc::writer::IpcWriteOptions;
use arrow::ipc::{CompressionType, root_as_message};
@@ -481,7 +481,9 @@ async fn infer_stream_schema(
let fb_schema = message.header_as_schema().ok_or_else(|| {
ArrowError::IpcError("Unable to read IPC message schema".to_string())
})?;
- let schema = fb_to_schema(fb_schema);
+ let schema = try_fb_to_schema(fb_schema).map_err(|err| {
+ ArrowError::IpcError(format!("Unable to convert IPC schema: {err:?}"))
+ })?;
Ok(Arc::new(schema))
}
diff --git a/datafusion/datasource-arrow/src/source.rs b/datafusion/datasource-arrow/src/source.rs
index 7e8189a78d..30af5a8bee 100644
--- a/datafusion/datasource-arrow/src/source.rs
+++ b/datafusion/datasource-arrow/src/source.rs
@@ -179,7 +179,12 @@ impl FileOpener for ArrowFileOpener {
})?;
// build decoder according to footer & projection
let schema =
- arrow_ipc::convert::fb_to_schema(footer.schema().unwrap());
+ arrow_ipc::convert::try_fb_to_schema(footer.schema().unwrap())
+ .map_err(|err| {
+ exec_datafusion_err!(
+ "Unable to convert IPC schema: {err:?}"
+ )
+ })?;
let mut decoder = FileDecoder::new(schema.into(), footer.version());
if let Some(projection) = projection {
decoder = decoder.with_projection(projection);
diff --git a/datafusion/datasource-parquet/src/file_format.rs b/datafusion/datasource-parquet/src/file_format.rs
index 18f2b5a650..8bb700b68c 100644
--- a/datafusion/datasource-parquet/src/file_format.rs
+++ b/datafusion/datasource-parquet/src/file_format.rs
@@ -33,7 +33,7 @@ pub use crate::schema_coercion::{
pub use crate::sink::ParquetSink;
-use arrow::datatypes::{Fields, Schema, SchemaRef};
+use arrow::datatypes::{Fields, Metadata, Schema, SchemaRef};
use datafusion_datasource::TableSchema;
use datafusion_datasource::file_compression_type::FileCompressionType;
use datafusion_datasource::file_sink_config::FileSinkConfig;
@@ -267,7 +267,7 @@ fn clear_metadata(
.fields()
.iter()
.map(|field| {
- field.as_ref().clone().with_metadata(Default::default()) // clear meta
+ field.as_ref().clone().with_metadata(Metadata::new()) // clear meta
})
.collect::<Fields>();
Schema::new(fields)
diff --git a/datafusion/datasource-parquet/src/metadata.rs b/datafusion/datasource-parquet/src/metadata.rs
index 4781a309f5..51dc98e4f2 100644
--- a/datafusion/datasource-parquet/src/metadata.rs
+++ b/datafusion/datasource-parquet/src/metadata.rs
@@ -80,7 +80,13 @@ pub(crate) fn has_untrusted_min_max_order(
parquet_column_index: usize,
) -> bool {
let column = parquet_schema.column(parquet_column_index);
- if column.sort_order() == SortOrder::UNDEFINED {
+ // As of arrow 60, INT96 columns report `SortOrder::INT96_TIMESTAMP`
+ // rather than `UNDEFINED`; keep treating their min/max as untrusted.
+ // until <https://github.com/apache/datafusion/issues/25484>
+ if matches!(
+ column.sort_order(),
+ SortOrder::UNDEFINED | SortOrder::INT96_TIMESTAMP
+ ) {
return true;
}
requires_unsigned_byte_array_order(&column)
@@ -313,12 +319,13 @@ impl<'a> DFParquetMetadata<'a> {
}
/// Check whether `metadata` already has both the column index and the
- /// offset index populated (see [`ParquetMetaData::column_index`] and
- /// [`ParquetMetaData::offset_index`]).
+ /// offset index populated (see [`ParquetMetaData::page_index`]).
///
/// Used to decide whether page index I/O can be skipped.
fn metadata_has_page_index(metadata: &ParquetMetaData) -> bool {
- metadata.column_index().is_some() && metadata.offset_index().is_some()
+ metadata
+ .page_index()
+ .is_some_and(|page_index| page_index.is_complete())
}
/// Store `metadata` in the configured [`FileMetadataCache`], keyed by
@@ -405,7 +412,10 @@ impl<'a> DFParquetMetadata<'a> {
object_meta: &ObjectMeta,
metadata: Arc<ParquetMetaData>,
) -> Result<Arc<ParquetMetaData>> {
- if metadata.column_index().is_some() && metadata.offset_index().is_some() {
+ if metadata
+ .page_index()
+ .is_some_and(|page_index| page_index.is_complete())
+ {
return Ok(metadata);
}
let metadata =
@@ -1049,8 +1059,10 @@ impl FileMetadata for CachedParquetMetaData {
}
fn extra_info(&self) -> HashMap<String, String> {
- let page_index =
- self.0.column_index().is_some() && self.0.offset_index().is_some();
+ let page_index = self
+ .0
+ .page_index()
+ .is_some_and(|page_index| page_index.is_complete());
HashMap::from([("page_index".to_owned(), page_index.to_string())])
}
}
diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs
index 0bd39f956d..ff67fec44e 100644
--- a/datafusion/datasource-parquet/src/opener/mod.rs
+++ b/datafusion/datasource-parquet/src/opener/mod.rs
@@ -2004,8 +2004,12 @@ async fn load_page_index<T: AsyncFileReader>(
options: ArrowReaderOptions,
) -> Result<ArrowReaderMetadata> {
let parquet_metadata = reader_metadata.metadata();
- let missing_column_index = parquet_metadata.column_index().is_none();
- let missing_offset_index = parquet_metadata.offset_index().is_none();
+ let missing_column_index = !parquet_metadata
+ .page_index()
+ .is_some_and(|page_index| page_index.has_column_indexes());
+ let missing_offset_index = !parquet_metadata
+ .page_index()
+ .is_some_and(|page_index| page_index.has_offset_indexes());
// You may ask yourself: why are we even checking if the page index is already loaded here?
// Didn't we explicitly *not* load it above?
// Well it's possible that a custom implementation of `AsyncFileReader` gives you
@@ -3932,8 +3936,11 @@ mod test {
.fetch_metadata()
.await
.unwrap();
- assert!(original.column_index().is_some());
- assert!(original.offset_index().is_some());
+ assert!(
+ original
+ .page_index()
+ .is_some_and(|page_index| page_index.is_complete())
+ );
let mut opener = ParquetMorselizerBuilder::new()
.with_store(Arc::clone(&store))
.with_schema(Arc::clone(&schema))
diff --git a/datafusion/datasource-parquet/src/page_filter.rs b/datafusion/datasource-parquet/src/page_filter.rs
index f6a267a582..7190a60bda 100644
--- a/datafusion/datasource-parquet/src/page_filter.rs
+++ b/datafusion/datasource-parquet/src/page_filter.rs
@@ -37,7 +37,7 @@ use datafusion_pruning::{MAX_IN_LIST_SIZE, PruningPredicate, PruningPredicateBui
use log::{debug, trace};
use parquet::arrow::arrow_reader::statistics::StatisticsConverter;
-use parquet::file::metadata::{ParquetColumnIndex, ParquetOffsetIndex};
+use parquet::file::metadata::page_index::PageIndexProvider;
use parquet::file::page_index::offset_index::PageLocation;
use parquet::schema::types::SchemaDescriptor;
use parquet::{
@@ -47,8 +47,8 @@ use parquet::{
/// Filters a [`ParquetAccessPlan`] based on the [Parquet PageIndex], if present
///
-/// It does so by evaluating statistics from the [`ParquetColumnIndex`] and
-/// [`ParquetOffsetIndex`] and converting them to [`RowSelection`].
+/// It does so by evaluating statistics from the [`PageIndexProvider`] (column
+/// and offset indexes) and converting them to [`RowSelection`].
///
/// [Parquet PageIndex]: https://github.com/apache/parquet-format/blob/master/PageIndex.md
///
@@ -243,13 +243,18 @@ impl PagePruningAccessPlanFilter {
return PagePruningResult::new(access_plan, 0, 0);
}
- if parquet_metadata.offset_index().is_none()
- || parquet_metadata.column_index().is_none()
+ if !parquet_metadata
+ .page_index()
+ .is_some_and(|page_index| page_index.is_complete())
{
debug!(
"Can not prune pages due to lack of indexes. Have offset: {}, column index: {}",
- parquet_metadata.offset_index().is_some(),
- parquet_metadata.column_index().is_some()
+ parquet_metadata
+ .page_index()
+ .is_some_and(|page_index| page_index.has_offset_indexes()),
+ parquet_metadata
+ .page_index()
+ .is_some_and(|page_index| page_index.has_column_indexes())
);
return PagePruningResult::new(access_plan, 0, 0);
}
@@ -308,12 +313,10 @@ impl PagePruningAccessPlanFilter {
// The selection for this particular row group
let mut overall_selection = None;
- let total_pages_in_group =
- parquet_metadata.offset_index().map_or(0, |offset_index| {
- offset_index[row_group_index]
- .first()
- .map_or(0, |column| column.page_locations.len())
- });
+ let total_pages_in_group = parquet_metadata
+ .page_index()
+ .and_then(|page_index| page_index.offset_index(row_group_index, 0))
+ .map_or(0, |offset_index| offset_index.page_locations().len());
// stores the indexes of the matched pages
let mut matched_pages_in_group: HashSet<usize> =
HashSet::from_iter(0..total_pages_in_group);
@@ -598,11 +601,10 @@ fn fully_matched_page_count(
row_group_index: usize,
parquet_metadata: &ParquetMetaData,
) -> usize {
- parquet_metadata.offset_index().map_or(0, |offset_index| {
- offset_index[row_group_index]
- .first()
- .map_or(0, |column| column.page_locations.len())
- })
+ parquet_metadata
+ .page_index()
+ .and_then(|page_index| page_index.offset_index(row_group_index, 0))
+ .map_or(0, |offset_index| offset_index.page_locations().len())
}
fn prune_pages_for_predicate(
@@ -708,14 +710,13 @@ fn prune_pages_in_one_row_group(
Some((RowSelection::from(vec), values))
}
-/// Implement [`PruningStatistics`] for one column's PageIndex (column_index + offset_index)
+/// Implement [`PruningStatistics`] for one column's [`PageIndexProvider`]
#[derive(Debug)]
struct PagesPruningStatistics<'a> {
row_group_index: usize,
row_group_metadatas: &'a [RowGroupMetaData],
converter: StatisticsConverter<'a>,
- column_index: &'a ParquetColumnIndex,
- offset_index: &'a ParquetOffsetIndex,
+ page_index: &'a dyn PageIndexProvider,
page_offsets: &'a Vec<PageLocation>,
trusted_min_max: bool,
}
@@ -739,16 +740,14 @@ impl<'a> PagesPruningStatistics<'a> {
return None;
};
- let column_index = parquet_metadata.column_index()?;
- let offset_index = parquet_metadata.offset_index()?;
+ let page_index: &dyn PageIndexProvider = parquet_metadata.page_index()?.as_ref();
+ if !page_index.is_complete() {
+ return None;
+ }
let row_group_metadatas = parquet_metadata.row_groups();
- let Some(row_group_page_offsets) = offset_index.get(row_group_index) else {
- trace!("No page offsets for row group {row_group_index}, skipping");
- return None;
- };
let Some(offset_index_metadata) =
- row_group_page_offsets.get(parquet_column_index)
+ page_index.offset_index(row_group_index, parquet_column_index)
else {
trace!(
"No page offsets for column {:?} in row group {row_group_index}, skipping",
@@ -768,8 +767,7 @@ impl<'a> PagesPruningStatistics<'a> {
row_group_index,
row_group_metadatas,
converter,
- column_index,
- offset_index,
+ page_index,
page_offsets,
trusted_min_max,
})
@@ -801,11 +799,10 @@ impl PruningStatistics for PagesPruningStatistics<'_> {
if !self.trusted_min_max {
return None;
}
- match self.converter.data_page_mins(
- self.column_index,
- self.offset_index,
- [&self.row_group_index],
- ) {
+ match self
+ .converter
+ .data_page_mins(self.page_index, [&self.row_group_index])
+ {
Ok(min_values) => Some(min_values),
Err(e) => {
debug!("Error evaluating data page min values {e}");
@@ -818,11 +815,10 @@ impl PruningStatistics for PagesPruningStatistics<'_> {
if !self.trusted_min_max {
return None;
}
- match self.converter.data_page_maxes(
- self.column_index,
- self.offset_index,
- [&self.row_group_index],
- ) {
+ match self
+ .converter
+ .data_page_maxes(self.page_index, [&self.row_group_index])
+ {
Ok(min_values) => Some(min_values),
Err(e) => {
debug!("Error evaluating data page max values {e}");
@@ -836,11 +832,10 @@ impl PruningStatistics for PagesPruningStatistics<'_> {
}
fn null_counts(&self, _column: &datafusion_common::Column) -> Option<ArrayRef> {
- match self.converter.data_page_null_counts(
- self.column_index,
- self.offset_index,
- [&self.row_group_index],
- ) {
+ match self
+ .converter
+ .data_page_null_counts(self.page_index, [&self.row_group_index])
+ {
Ok(null_counts) => Some(Arc::new(null_counts)),
Err(e) => {
debug!("Error evaluating data page null counts {e}");
@@ -851,7 +846,7 @@ impl PruningStatistics for PagesPruningStatistics<'_> {
fn row_counts(&self) -> Option<ArrayRef> {
match self.converter.data_page_row_counts(
- self.offset_index,
+ self.page_index,
self.row_group_metadatas,
[&self.row_group_index],
) {
diff --git a/datafusion/datasource-parquet/src/reader.rs b/datafusion/datasource-parquet/src/reader.rs
index 6224c21473..06ab40fdd2 100644
--- a/datafusion/datasource-parquet/src/reader.rs
+++ b/datafusion/datasource-parquet/src/reader.rs
@@ -359,8 +359,10 @@ impl FileMetadata for CachedParquetMetaData {
}
fn extra_info(&self) -> HashMap<String, String> {
- let page_index =
- self.0.column_index().is_some() && self.0.offset_index().is_some();
+ let page_index = self
+ .0
+ .page_index()
+ .is_some_and(|page_index| page_index.is_complete());
HashMap::from([("page_index".to_owned(), page_index.to_string())])
}
}
diff --git a/datafusion/datasource-parquet/src/statistics_order_tests.rs b/datafusion/datasource-parquet/src/statistics_order_tests.rs
index 2156a92dcd..8b8e2c9831 100644
--- a/datafusion/datasource-parquet/src/statistics_order_tests.rs
+++ b/datafusion/datasource-parquet/src/statistics_order_tests.rs
@@ -35,6 +35,7 @@ use parquet::arrow::ArrowWriter;
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
use parquet::basic::{ColumnOrder, LogicalType, SortOrder, Type as PhysicalType};
use parquet::data_type::{ByteArray, FixedLenByteArray};
+use parquet::file::metadata::page_index::{PageIndex, PageIndexBuilder};
use parquet::file::metadata::{
ColumnChunkMetaData, ColumnIndexBuilder, FileMetaData, OffsetIndexBuilder,
PageIndexPolicy, ParquetMetaData, ParquetMetaDataReader, ParquetMetaDataWriter,
@@ -136,16 +137,25 @@ impl TestFile {
.build()
.unwrap();
- let mut column_index = metadata.column_index().unwrap().clone();
+ // Give the first row group's string column the same untrusted-looking
+ // bounds in its page index when the column order is missing or unknown.
+ let mut page_index = metadata
+ .page_index()
+ .unwrap()
+ .as_any()
+ .downcast_ref::<PageIndex>()
+ .unwrap()
+ .clone()
+ .into_builder();
if matches!(order, StatisticsOrder::Missing | StatisticsOrder::Unknown) {
let mut index = ColumnIndexBuilder::new(PhysicalType::BYTE_ARRAY);
- index.append(false, "aé".as_bytes().to_vec(), b"b".to_vec(), 0);
- column_index[0][0] = index.build().unwrap();
+ index.append(false, "aé".as_bytes().to_vec(), b"b".to_vec(), 0, None);
+ page_index.put_column_index(index.build().unwrap(), 0, 0);
}
let metadata = metadata
.into_builder()
.set_row_groups(row_groups)
- .set_column_index(Some(column_index))
+ .set_page_index(Some(Arc::new(page_index.build())))
.build();
// Keep the real data pages, and serialize the replacement statistics
@@ -177,8 +187,10 @@ impl TestFile {
bytes[new_end..new_end + 4].copy_from_slice(&metadata_len.to_le_bytes());
} else {
// Change the first union member from field 1 (TYPEORDER) to
- // an unrecognized field 2. The numeric column stays known.
- bytes[start + 2] = 0x2c;
+ // an unrecognized field 4 (fields 2 and 3 are
+ // IEEE754TotalOrder and INT96TimestampOrder as of arrow 60).
+ // The numeric column stays known.
+ bytes[start + 2] = 0x4c;
}
}
@@ -543,16 +555,21 @@ fn single_column_metadata(
.build()
.unwrap(),
)));
+ // Attach a single-page column and offset index mirroring `statistics`.
let mut column_index = ColumnIndexBuilder::new(physical_type);
column_index.append(
false,
statistics.min_bytes_opt().unwrap().to_vec(),
statistics.max_bytes_opt().unwrap().to_vec(),
0,
+ None,
);
let mut offset_index = OffsetIndexBuilder::new();
offset_index.append_row_count(3);
offset_index.append_offset_and_size(0, 1);
+ let mut page_index = PageIndexBuilder::new(1, 1);
+ page_index.put_column_index(column_index.build().unwrap(), 0, 0);
+ page_index.put_offset_index(offset_index.build(), 0, 0);
let column = ColumnChunkMetaData::builder(schema.column(0))
.set_num_values(3)
.set_statistics(statistics)
@@ -568,8 +585,7 @@ fn single_column_metadata(
vec![group],
)
.into_builder()
- .set_column_index(Some(vec![vec![column_index.build().unwrap()]]))
- .set_offset_index(Some(vec![vec![offset_index.build()]]))
+ .set_page_index(Some(Arc::new(page_index.build())))
.build()
}
@@ -737,7 +753,7 @@ fn undefined_int96_order_is_never_trusted() {
.build()
.unwrap(),
));
- assert_eq!(schema.column(0).sort_order(), SortOrder::UNDEFINED);
+ assert_eq!(schema.column(0).sort_order(), SortOrder::INT96_TIMESTAMP);
for order in [
None,
@@ -746,6 +762,7 @@ fn undefined_int96_order_is_never_trusted() {
Some(ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::UNDEFINED)),
Some(ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::SIGNED)),
Some(ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::UNSIGNED)),
+ Some(ColumnOrder::INT96_TIMESTAMP_ORDER),
] {
assert!(
has_untrusted_min_max_order(
diff --git a/datafusion/execution/src/cache/mod.rs b/datafusion/execution/src/cache/mod.rs
index f47a3f3ca4..2eba533e04 100644
--- a/datafusion/execution/src/cache/mod.rs
+++ b/datafusion/execution/src/cache/mod.rs
@@ -228,7 +228,7 @@ impl DFHeapSize for SchemaFingerprint {
#[cfg(test)]
mod schema_fingerprint_tests {
use super::*;
- use datafusion_common::arrow::datatypes::Field;
+ use datafusion_common::arrow::datatypes::{Field, Metadata};
fn fp(fields: Vec<Field>) -> SchemaFingerprint {
SchemaFingerprint::from_schema(&Schema::new(fields))
@@ -263,13 +263,13 @@ mod schema_fingerprint_tests {
let field_md = SchemaFingerprint::from_schema(&Schema::new(vec![
Field::new("id", DataType::Int64, false)
- .with_metadata([("note".to_string(), "x".to_string())].into()),
+ .with_metadata(Metadata::new().with("note", "x")),
]));
assert_eq!(plain, field_md, "field metadata must be ignored");
let schema_md = SchemaFingerprint::from_schema(
&Schema::new(vec![Field::new("id", DataType::Int64, false)])
- .with_metadata([("k".to_string(), "v".to_string())].into()),
+ .with_metadata(Metadata::new().with("k", "v")),
);
assert_eq!(plain, schema_md, "schema metadata must be ignored");
}
diff --git a/datafusion/expr/src/expr.rs b/datafusion/expr/src/expr.rs
index 3428395b54..e8e805c2b9 100644
--- a/datafusion/expr/src/expr.rs
+++ b/datafusion/expr/src/expr.rs
@@ -32,7 +32,7 @@ use crate::type_coercion::functions::value_fields_with_higher_order_udf;
use crate::{AggregateUDF, LambdaParametersProgress, ValueOrLambda, Volatility};
use crate::{ExprSchemable, Operator, Signature, WindowFrame, WindowUDF};
-use arrow::datatypes::{DataType, Field, FieldRef};
+use arrow::datatypes::{DataType, Field, FieldRef, Metadata};
use datafusion_common::cse::{HashNode, NormalizeEq, Normalizeable};
use datafusion_common::datatype::DataTypeExt;
use datafusion_common::metadata::format_type_and_metadata;
@@ -622,7 +622,7 @@ impl<'a> TreeNodeContainer<'a, Self> for Expr {
/// See the [default_column_values.rs] example implementation.
///
/// [default_column_values.rs]: https://github.com/apache/datafusion/blob/main/datafusion-examples/examples/custom_data_source/default_column_values.rs
-pub type SchemaFieldMetadata = std::collections::HashMap<String, String>;
+pub type SchemaFieldMetadata = Metadata;
/// Intersects multiple metadata instances for UNION operations.
///
@@ -3886,7 +3886,7 @@ mod test {
let subquery_schema = Arc::new(
DFSchema::from_unqualified_fields(
vec![subquery_field].into(),
- Default::default(),
+ Metadata::new(),
)
.unwrap(),
);
@@ -3934,7 +3934,7 @@ mod test {
let subquery_schema = Arc::new(
DFSchema::from_unqualified_fields(
vec![subquery_field].into(),
- Default::default(),
+ Metadata::new(),
)
.unwrap(),
);
@@ -3988,7 +3988,7 @@ mod test {
let subquery_schema = Arc::new(
DFSchema::from_unqualified_fields(
vec![subquery_field].into(),
- Default::default(),
+ Metadata::new(),
)
.unwrap(),
);
@@ -4041,7 +4041,7 @@ mod test {
let subquery_schema = Arc::new(
DFSchema::from_unqualified_fields(
vec![subquery_field].into(),
- Default::default(),
+ Metadata::new(),
)
.unwrap(),
);
@@ -4142,9 +4142,8 @@ mod test {
fn infer_placeholder_with_metadata() {
// name == $1, where name is a non-nullable string
let schema = Arc::new(Schema::new(vec![
- Field::new("name", DataType::Utf8, false).with_metadata(
- [("some_key".to_string(), "some_value".to_string())].into(),
- ),
+ Field::new("name", DataType::Utf8, false)
+ .with_metadata(Metadata::new().with("some_key", "some_value")),
]));
let df_schema = DFSchema::try_from(schema).unwrap();
@@ -4604,53 +4603,53 @@ mod test {
mod intersect_metadata_tests {
use super::super::intersect_metadata_for_union;
- use std::collections::HashMap;
+ use arrow::datatypes::Metadata;
#[test]
fn all_branches_same_metadata() {
- let m1 = HashMap::from([("key".into(), "val".into())]);
- let m2 = HashMap::from([("key".into(), "val".into())]);
+ let m1 = Metadata::new().with("key", "val");
+ let m2 = Metadata::new().with("key", "val");
let result = intersect_metadata_for_union([&m1, &m2]);
- assert_eq!(result, HashMap::from([("key".into(), "val".into())]));
+ assert_eq!(result, Metadata::new().with("key", "val"));
}
#[test]
fn conflicting_metadata_dropped() {
- let m1 = HashMap::from([("key".into(), "a".into())]);
- let m2 = HashMap::from([("key".into(), "b".into())]);
+ let m1 = Metadata::new().with("key", "a");
+ let m2 = Metadata::new().with("key", "b");
let result = intersect_metadata_for_union([&m1, &m2]);
assert!(result.is_empty());
}
#[test]
fn empty_metadata_branch_skipped() {
- let m1 = HashMap::from([("key".into(), "val".into())]);
- let m2 = HashMap::new(); // e.g. NULL literal
+ let m1 = Metadata::new().with("key", "val");
+ let m2 = Metadata::new(); // e.g. NULL literal
let result = intersect_metadata_for_union([&m1, &m2]);
- assert_eq!(result, HashMap::from([("key".into(), "val".into())]));
+ assert_eq!(result, Metadata::new().with("key", "val"));
}
#[test]
fn empty_metadata_first_branch_skipped() {
- let m1 = HashMap::new();
- let m2 = HashMap::from([("key".into(), "val".into())]);
+ let m1 = Metadata::new();
+ let m2 = Metadata::new().with("key", "val");
let result = intersect_metadata_for_union([&m1, &m2]);
- assert_eq!(result, HashMap::from([("key".into(), "val".into())]));
+ assert_eq!(result, Metadata::new().with("key", "val"));
}
#[test]
fn all_branches_empty_metadata() {
- let m1: HashMap<String, String> = HashMap::new();
- let m2: HashMap<String, String> = HashMap::new();
+ let m1 = Metadata::new();
+ let m2 = Metadata::new();
let result = intersect_metadata_for_union([&m1, &m2]);
assert!(result.is_empty());
}
#[test]
fn mixed_empty_and_conflicting() {
- let m1 = HashMap::from([("key".into(), "a".into())]);
- let m2 = HashMap::new();
- let m3 = HashMap::from([("key".into(), "b".into())]);
+ let m1 = Metadata::new().with("key", "a");
+ let m2 = Metadata::new();
+ let m3 = Metadata::new().with("key", "b");
let result = intersect_metadata_for_union([&m1, &m2, &m3]);
// m2 is skipped; m1 and m3 conflict → dropped
assert!(result.is_empty());
@@ -4658,9 +4657,7 @@ mod test {
#[test]
fn no_inputs() {
- let result = intersect_metadata_for_union(std::iter::empty::<
- &HashMap<String, String>,
- >());
+ let result = intersect_metadata_for_union(std::iter::empty::<&Metadata>());
assert!(result.is_empty());
}
}
diff --git a/datafusion/expr/src/logical_plan/builder.rs b/datafusion/expr/src/logical_plan/builder.rs
index 36aa67bbe7..a44505b15a 100644
--- a/datafusion/expr/src/logical_plan/builder.rs
+++ b/datafusion/expr/src/logical_plan/builder.rs
@@ -1882,12 +1882,8 @@ pub fn build_join_schema(
_ => (right, left),
};
- let metadata = schema1
- .metadata()
- .clone()
- .into_iter()
- .chain(schema2.metadata().clone())
- .collect();
+ let mut metadata = schema1.metadata().clone();
+ metadata.extend(schema2.metadata().clone());
let dfschema = DFSchema::new_with_metadata(qualified_fields, metadata)?;
dfschema.with_functional_dependencies(func_dependencies)
diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs
index de6d1667c7..96e7feebf6 100644
--- a/datafusion/expr/src/logical_plan/plan.rs
+++ b/datafusion/expr/src/logical_plan/plan.rs
@@ -55,7 +55,7 @@ use crate::{
use crate::statistics::StatisticsRequest;
use arrow::compute::SortOptions;
-use arrow::datatypes::{DataType, Field, FieldRef, Schema, SchemaRef};
+use arrow::datatypes::{DataType, Field, FieldRef, Metadata, Schema, SchemaRef};
use datafusion_common::cse::{NormalizeEq, Normalizeable};
use datafusion_common::format::{ExplainAnalyzeCategories, ExplainFormat, MetricType};
use datafusion_common::metadata::check_metadata_with_storage_equal;
@@ -3460,8 +3460,7 @@ impl Union {
inputs: &[Arc<LogicalPlan>],
loose_types: bool,
) -> Result<DFSchemaRef> {
- type FieldData<'a> =
- (&'a DataType, bool, Vec<&'a HashMap<String, String>>, usize);
+ type FieldData<'a> = (&'a DataType, bool, Vec<&'a Metadata>, usize);
let mut cols: Vec<(&str, FieldData)> = Vec::new();
for input in inputs.iter() {
for field in input.schema().fields() {
@@ -6067,7 +6066,7 @@ mod tests {
let schema_with_metadata = || {
DFSchema::from_unqualified_fields(
vec![Field::new("count", DataType::Int64, false)].into(),
- [("key".to_string(), "value".to_string())].into(),
+ Metadata::new().with("key", "value"),
)
.unwrap()
};
diff --git a/datafusion/ffi/src/record_batch_stream.rs b/datafusion/ffi/src/record_batch_stream.rs
index 2fd0cd0ad8..c208b7d9af 100644
--- a/datafusion/ffi/src/record_batch_stream.rs
+++ b/datafusion/ffi/src/record_batch_stream.rs
@@ -220,7 +220,7 @@ mod tests {
use std::sync::Arc;
use arrow::array::record_batch;
- use arrow::datatypes::{DataType, Field, Schema};
+ use arrow::datatypes::{DataType, Field, Metadata, Schema};
use datafusion::error::Result;
use datafusion::execution::SendableRecordBatchStream;
use datafusion::test_util::bounded_stream;
@@ -275,7 +275,7 @@ mod tests {
.schema()
.as_ref()
.clone()
- .with_metadata([("some_key".to_owned(), "some_value".to_owned())].into())
+ .with_metadata(Metadata::new().with("some_key", "some_value"))
.into();
let rb = rb.with_schema(schema)?;
diff --git a/datafusion/ffi/src/udaf/mod.rs b/datafusion/ffi/src/udaf/mod.rs
index 7bddb927a4..bb0f3567f3 100644
--- a/datafusion/ffi/src/udaf/mod.rs
+++ b/datafusion/ffi/src/udaf/mod.rs
@@ -657,9 +657,8 @@ impl From<AggregateOrderSensitivity> for FFI_AggregateOrderSensitivity {
#[cfg(test)]
mod tests {
- use std::collections::HashMap;
- use arrow::datatypes::Schema;
+ use arrow::datatypes::{Metadata, Schema};
use datafusion::common::create_array;
use datafusion::functions_aggregate::sum::Sum;
use datafusion::physical_expr::PhysicalSortExpr;
@@ -776,8 +775,7 @@ mod tests {
let foreign_udaf: Arc<dyn AggregateUDFImpl> = (&local_udaf).into();
let foreign_udaf = AggregateUDF::new_from_shared_impl(foreign_udaf);
- let metadata: HashMap<String, String> =
- std::iter::once(("a_key".to_string(), "a_value".to_string())).collect();
+ let metadata = Metadata::new().with("a_key", "a_value");
let input_field = Arc::new(
Field::new("a", DataType::Float64, false).with_metadata(metadata.clone()),
);
diff --git a/datafusion/functions-aggregate-common/src/aggregate/groups_accumulator/nulls.rs b/datafusion/functions-aggregate-common/src/aggregate/groups_accumulator/nulls.rs
index d524afe43a..e98ad4fae6 100644
--- a/datafusion/functions-aggregate-common/src/aggregate/groups_accumulator/nulls.rs
+++ b/datafusion/functions-aggregate-common/src/aggregate/groups_accumulator/nulls.rs
@@ -160,7 +160,7 @@ pub fn set_nulls_dyn(input: &dyn Array, nulls: Option<NullBuffer>) -> Result<Arr
unsafe {
Arc::new(StringViewArray::new_unchecked(
input.views().clone(),
- input.data_buffers().to_vec(),
+ Arc::clone(input.data_buffers()),
nulls,
))
}
@@ -197,7 +197,7 @@ pub fn set_nulls_dyn(input: &dyn Array, nulls: Option<NullBuffer>) -> Result<Arr
unsafe {
Arc::new(BinaryViewArray::new_unchecked(
input.views().clone(),
- input.data_buffers().to_vec(),
+ Arc::clone(input.data_buffers()),
nulls,
))
}
diff --git a/datafusion/functions-aggregate/benches/array_agg.rs b/datafusion/functions-aggregate/benches/array_agg.rs
index d7e5a51107..ad6d502eb5 100644
--- a/datafusion/functions-aggregate/benches/array_agg.rs
+++ b/datafusion/functions-aggregate/benches/array_agg.rs
@@ -33,7 +33,6 @@ use arrow::buffer::OffsetBuffer;
use arrow::util::bench_util::create_primitive_array;
use rand::Rng;
use rand::SeedableRng;
-use rand::distr::{Distribution, StandardUniform};
use rand::prelude::StdRng;
/// Returns fixed seedable RNG
@@ -65,8 +64,7 @@ pub fn create_list_array<T>(
zero_length_lists_probability: f32,
) -> ListArray
where
- T: ArrowPrimitiveType,
- StandardUniform: Distribution<T::Native>,
+ T: ArrowPrimitiveType<Native = i64>,
{
let mut nulls_builder = NullBufferBuilder::new(size);
let mut rng = StdRng::seed_from_u64(42);
@@ -89,7 +87,7 @@ where
length
}));
- let length = *offsets.last().unwrap() as usize;
+ let length = offsets.last() as usize;
let values = create_primitive_array::<T>(length, 0.0);
diff --git a/datafusion/functions-aggregate/src/array_agg.rs b/datafusion/functions-aggregate/src/array_agg.rs
index 00c637ac3c..cc00e8f45b 100644
--- a/datafusion/functions-aggregate/src/array_agg.rs
+++ b/datafusion/functions-aggregate/src/array_agg.rs
@@ -2095,7 +2095,7 @@ mod tests {
acc2.update_batch(&[string_list_data([vec!["e", "f", "g"]])])?;
acc1 = merge(acc1, acc2)?;
- assert_eq!(acc1.size(), 2290);
+ assert_eq!(acc1.size(), 2210);
Ok(())
}
@@ -2115,7 +2115,7 @@ mod tests {
]);
acc.update_batch(&[Arc::clone(&input), input])?;
- assert_eq!(acc.size(), 2295);
+ assert_eq!(acc.size(), 2135);
Ok(())
}
diff --git a/datafusion/functions/src/binaries.rs b/datafusion/functions/src/binaries.rs
index 861b7574ce..b335ba10ef 100644
--- a/datafusion/functions/src/binaries.rs
+++ b/datafusion/functions/src/binaries.rs
@@ -248,7 +248,7 @@ impl ConcatBuilder for ConcatBinaryViewBuilder {
let array = unsafe {
BinaryViewArray::new_unchecked(
ScalarBuffer::from(self.views),
- buffers,
+ buffers.into(),
null_buffer,
)
};
diff --git a/datafusion/functions/src/core/arrow_field.rs b/datafusion/functions/src/core/arrow_field.rs
index 50da43a166..46b8e9872f 100644
--- a/datafusion/functions/src/core/arrow_field.rs
+++ b/datafusion/functions/src/core/arrow_field.rs
@@ -16,7 +16,8 @@
// under the License.
use arrow::array::{
- Array, BooleanArray, MapBuilder, StringArray, StringBuilder, StructArray,
+ Array, BooleanArray, MapBuilder, MapFieldNames, StringArray, StringBuilder,
+ StructArray,
};
use arrow::datatypes::{DataType, Field, Fields};
use datafusion_common::{Result, ScalarValue, utils::take_function_args};
@@ -125,8 +126,18 @@ impl ScalarUDFImpl for ArrowFieldFunc {
// Build the metadata map array (same pattern as arrow_metadata.rs)
let metadata = field.metadata();
- let mut map_builder =
- MapBuilder::new(None, StringBuilder::new(), StringBuilder::new());
+ // Match the field names declared in `return_type` (the arrow-rs
+ // default changed to `key`/`value` in arrow 60)
+ let map_field_names = MapFieldNames {
+ entry: "entries".to_string(),
+ key: "keys".to_string(),
+ value: "values".to_string(),
+ };
+ let mut map_builder = MapBuilder::new(
+ Some(map_field_names),
+ StringBuilder::new(),
+ StringBuilder::new(),
+ );
let mut entries: Vec<_> = metadata.iter().collect();
entries.sort_by_key(|(k, _)| *k);
diff --git a/datafusion/functions/src/core/arrow_metadata.rs b/datafusion/functions/src/core/arrow_metadata.rs
index acaaa6eff5..37a2a52c19 100644
--- a/datafusion/functions/src/core/arrow_metadata.rs
+++ b/datafusion/functions/src/core/arrow_metadata.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-use arrow::array::{MapBuilder, StringBuilder};
+use arrow::array::{MapBuilder, MapFieldNames, StringBuilder};
use arrow::datatypes::{DataType, Field, Fields};
use datafusion_common::types::logical_string;
use datafusion_common::{Result, ScalarValue, exec_err, internal_err};
@@ -129,8 +129,18 @@ impl ScalarUDFImpl for ArrowMetadataFunc {
let value = metadata.get(key).cloned();
Ok(ColumnarValue::Scalar(ScalarValue::Utf8(value)))
} else if args.args.len() == 1 {
- let mut map_builder =
- MapBuilder::new(None, StringBuilder::new(), StringBuilder::new());
+ // Match the field names declared in `return_type` (the arrow-rs
+ // default changed to `key`/`value` in arrow 60)
+ let map_field_names = MapFieldNames {
+ entry: "entries".to_string(),
+ key: "keys".to_string(),
+ value: "values".to_string(),
+ };
+ let mut map_builder = MapBuilder::new(
+ Some(map_field_names),
+ StringBuilder::new(),
+ StringBuilder::new(),
+ );
let mut entries: Vec<_> = metadata.iter().collect();
entries.sort_by_key(|(k, _)| *k);
diff --git a/datafusion/functions/src/core/with_metadata.rs b/datafusion/functions/src/core/with_metadata.rs
index a86fb88f38..d5f0b65b81 100644
--- a/datafusion/functions/src/core/with_metadata.rs
+++ b/datafusion/functions/src/core/with_metadata.rs
@@ -157,7 +157,7 @@ impl ScalarUDFImpl for WithMetadataFunc {
#[cfg(test)]
mod tests {
use super::*;
- use arrow::datatypes::Field;
+ use arrow::datatypes::{Field, Metadata};
use datafusion_common::ScalarValue;
use std::sync::Arc;
@@ -197,14 +197,7 @@ mod tests {
fn merges_existing_metadata_and_overwrites_on_collision() {
let udf = WithMetadataFunc::new();
let mut existing = Field::new("x", DataType::Float64, false);
- existing.set_metadata(
- [
- ("keep".to_string(), "yes".to_string()),
- ("unit".to_string(), "old".to_string()),
- ]
- .into_iter()
- .collect(),
- );
+ existing.set_metadata(Metadata::new().with("keep", "yes").with("unit", "old"));
let input: FieldRef = Arc::new(existing);
let k = str_lit("unit");
let v = str_lit("new");
diff --git a/datafusion/functions/src/string/common.rs b/datafusion/functions/src/string/common.rs
index e37e7ad362..c13f724648 100644
--- a/datafusion/functions/src/string/common.rs
+++ b/datafusion/functions/src/string/common.rs
@@ -228,7 +228,7 @@ fn string_view_trim<Tr: Trimmer>(args: &[ArrayRef]) -> Result<ArrayRef> {
unsafe {
let array = StringViewArray::new_unchecked(
views_buf,
- string_view_array.data_buffers().to_vec(),
+ Arc::clone(string_view_array.data_buffers()),
nulls_buf,
);
Ok(Arc::new(array) as ArrayRef)
@@ -691,7 +691,7 @@ fn case_conversion_utf8view_ascii_inner<F: Fn(&u8) -> u8>(
unsafe {
StringViewArray::new_unchecked(
ScalarBuffer::from(new_views),
- completed,
+ completed.into(),
array.nulls().cloned(),
)
}
diff --git a/datafusion/functions/src/string/split_part.rs b/datafusion/functions/src/string/split_part.rs
index 9b73a1af88..401f248a24 100644
--- a/datafusion/functions/src/string/split_part.rs
+++ b/datafusion/functions/src/string/split_part.rs
@@ -497,7 +497,7 @@ fn split_part_scalar_view(
unsafe {
Ok(Arc::new(StringViewArray::new_unchecked(
views_buf,
- string_view_array.data_buffers().to_vec(),
+ Arc::clone(string_view_array.data_buffers()),
nulls,
)) as ArrayRef)
}
diff --git a/datafusion/functions/src/strings.rs b/datafusion/functions/src/strings.rs
index c788c6fb1f..4e21f6c74f 100644
--- a/datafusion/functions/src/strings.rs
+++ b/datafusion/functions/src/strings.rs
@@ -299,7 +299,7 @@ impl ConcatBuilder for ConcatStringViewBuilder {
let array = unsafe {
StringViewArray::new_unchecked(
ScalarBuffer::from(self.views),
- buffers,
+ buffers.into(),
null_buffer,
)
};
@@ -948,7 +948,7 @@ impl StringViewArrayBuilder {
let array = unsafe {
StringViewArray::new_unchecked(
ScalarBuffer::from(self.views),
- self.completed,
+ self.completed.into(),
null_buffer,
)
};
diff --git a/datafusion/functions/src/unicode/common.rs b/datafusion/functions/src/unicode/common.rs
index 5dc8f334da..75c88e0902 100644
--- a/datafusion/functions/src/unicode/common.rs
+++ b/datafusion/functions/src/unicode/common.rs
@@ -226,7 +226,7 @@ fn general_left_right_view<F: LeftRightSlicer>(
// - Each view is produced by `make_view` with correct bytes and offset
// - Out-of-line views reuse the original buffer index and adjusted offset
unsafe {
- let array = StringViewArray::new_unchecked(views, data_buffers, new_nulls);
+ let array = StringViewArray::new_unchecked(views, data_buffers.into(), new_nulls);
Ok(Arc::new(array) as ArrayRef)
}
}
diff --git a/datafusion/functions/src/unicode/initcap.rs b/datafusion/functions/src/unicode/initcap.rs
index 0332ab5d44..d18aaa945e 100644
--- a/datafusion/functions/src/unicode/initcap.rs
+++ b/datafusion/functions/src/unicode/initcap.rs
@@ -204,8 +204,8 @@ fn initcap_ascii_array<T: OffsetSizeTrait>(
) -> ArrayRef {
let offsets = string_array.offsets();
let src = string_array.value_data();
- let first_offset = offsets.first().unwrap().as_usize();
- let last_offset = offsets.last().unwrap().as_usize();
+ let first_offset = offsets.first().as_usize();
+ let last_offset = offsets.last().as_usize();
// For sliced arrays, only convert the visible bytes, not the entire input
// buffer.
@@ -490,11 +490,8 @@ mod tests {
assert_eq!(result.value(1), "Baz Qux");
// The output values buffer should be compact
- assert_eq!(*result.offsets().first().unwrap(), 0);
- assert_eq!(
- result.value_data().len(),
- *result.offsets().last().unwrap() as usize
- );
+ assert_eq!(result.offsets().first(), 0);
+ assert_eq!(result.value_data().len(), result.offsets().last() as usize);
Ok(())
}
@@ -519,11 +516,8 @@ mod tests {
assert_eq!(result.value(1), "Baz Qux");
// The output values buffer should be compact
- assert_eq!(*result.offsets().first().unwrap(), 0);
- assert_eq!(
- result.value_data().len(),
- *result.offsets().last().unwrap() as usize
- );
+ assert_eq!(result.offsets().first(), 0);
+ assert_eq!(result.value_data().len(), result.offsets().last() as usize);
Ok(())
}
}
diff --git a/datafusion/functions/src/unicode/substr.rs b/datafusion/functions/src/unicode/substr.rs
index 0cae215224..53f6080088 100644
--- a/datafusion/functions/src/unicode/substr.rs
+++ b/datafusion/functions/src/unicode/substr.rs
@@ -311,7 +311,7 @@ fn string_view_substr(
unsafe {
let array = StringViewArray::new_unchecked(
views_buf,
- string_view_array.data_buffers().to_vec(),
+ Arc::clone(string_view_array.data_buffers()),
nulls,
);
Ok(Arc::new(array) as ArrayRef)
diff --git a/datafusion/functions/src/unicode/substrindex.rs b/datafusion/functions/src/unicode/substrindex.rs
index f9f0bafa04..51cd615b92 100644
--- a/datafusion/functions/src/unicode/substrindex.rs
+++ b/datafusion/functions/src/unicode/substrindex.rs
@@ -335,7 +335,7 @@ fn substr_index_view(
unsafe {
Ok(Arc::new(StringViewArray::new_unchecked(
ScalarBuffer::from(views_buf),
- data_buffers,
+ data_buffers.into(),
nulls,
)) as ArrayRef)
}
@@ -468,7 +468,7 @@ fn substr_index_scalar_view(
unsafe {
Ok(Arc::new(StringViewArray::new_unchecked(
ScalarBuffer::from(views_buf),
- data_buffers,
+ data_buffers.into(),
string_array.nulls().cloned(),
)) as ArrayRef)
}
diff --git a/datafusion/optimizer/src/optimizer.rs b/datafusion/optimizer/src/optimizer.rs
index 49c59014fc..b6e93a3b16 100644
--- a/datafusion/optimizer/src/optimizer.rs
+++ b/datafusion/optimizer/src/optimizer.rs
@@ -790,6 +790,8 @@ fn assert_valid_optimization(
mod tests {
use std::sync::{Arc, Mutex};
+ use arrow::datatypes::Metadata;
+
use datafusion_common::tree_node::Transformed;
use datafusion_common::{
Column, DFSchema, DFSchemaRef, DataFusionError, Result, assert_contains, plan_err,
@@ -969,8 +971,7 @@ mod tests {
.iter()
.enumerate()
.map(|(i, (qualifier, field))| {
- let metadata =
- std::iter::once(("key".into(), format!("value {i}"))).collect();
+ let metadata = Metadata::new().with("key", format!("value {i}"));
let new_arrow_field = field.as_ref().clone().with_metadata(metadata);
(qualifier.cloned(), Arc::new(new_arrow_field))
diff --git a/datafusion/optimizer/src/propagate_empty_relation.rs b/datafusion/optimizer/src/propagate_empty_relation.rs
index 18ddc361a0..49e358a854 100644
--- a/datafusion/optimizer/src/propagate_empty_relation.rs
+++ b/datafusion/optimizer/src/propagate_empty_relation.rs
@@ -348,7 +348,7 @@ fn has_empty_grouping_set(group_expr: &[Expr]) -> bool {
#[cfg(test)]
mod tests {
- use arrow::datatypes::{DataType, Field, Schema};
+ use arrow::datatypes::{DataType, Field, Metadata, Schema};
use datafusion_common::{Column, DFSchema};
use datafusion_expr::logical_plan::table_scan;
@@ -800,7 +800,7 @@ mod tests {
produce_one_row: false,
schema: Arc::new(DFSchema::from_unqualified_fields(
fields.into(),
- Default::default(),
+ Metadata::new(),
)?),
});
diff --git a/datafusion/optimizer/src/push_down_filter.rs b/datafusion/optimizer/src/push_down_filter.rs
index b34827a87b..f534289c2b 100644
--- a/datafusion/optimizer/src/push_down_filter.rs
+++ b/datafusion/optimizer/src/push_down_filter.rs
@@ -1463,7 +1463,7 @@ mod tests {
use std::cmp::Ordering;
use std::fmt::{Debug, Formatter};
- use arrow::datatypes::{Field, Schema, SchemaRef};
+ use arrow::datatypes::{Field, Metadata, Schema, SchemaRef};
use async_trait::async_trait;
use datafusion_common::{DFSchemaRef, DataFusionError, ScalarValue};
@@ -4357,7 +4357,7 @@ mod tests {
let schema = Arc::new(
DFSchema::new_with_metadata(
vec![(None, Field::new("a", DataType::Int64, false).into())],
- Default::default(),
+ Metadata::new(),
)
.unwrap(),
);
diff --git a/datafusion/optimizer/src/simplify_expressions/udf_preimage.rs b/datafusion/optimizer/src/simplify_expressions/udf_preimage.rs
index d888a54d56..c71d17d303 100644
--- a/datafusion/optimizer/src/simplify_expressions/udf_preimage.rs
+++ b/datafusion/optimizer/src/simplify_expressions/udf_preimage.rs
@@ -71,7 +71,7 @@ mod test {
use std::sync::Arc;
- use arrow::datatypes::{DataType, Field};
+ use arrow::datatypes::{DataType, Field, Metadata};
use datafusion_common::{DFSchema, DFSchemaRef, Result, ScalarValue};
use datafusion_expr::{
ColumnarValue, Expr, Operator, ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl,
@@ -201,7 +201,7 @@ mod test {
Arc::new(
DFSchema::from_unqualified_fields(
vec![Field::new("x", DataType::Int32, true)].into(),
- Default::default(),
+ Metadata::new(),
)
.unwrap(),
)
@@ -215,7 +215,7 @@ mod test {
Field::new("y", DataType::Int32, false),
]
.into(),
- Default::default(),
+ Metadata::new(),
)
.unwrap(),
)
diff --git a/datafusion/physical-expr-common/src/binary_view_map.rs b/datafusion/physical-expr-common/src/binary_view_map.rs
index 0483ed5044..725a1e3c5b 100644
--- a/datafusion/physical-expr-common/src/binary_view_map.rs
+++ b/datafusion/physical-expr-common/src/binary_view_map.rs
@@ -456,8 +456,9 @@ where
};
let views = ScalarBuffer::from(self.views);
- let array =
- unsafe { BinaryViewArray::new_unchecked(views, self.completed, null_buffer) };
+ let array = unsafe {
+ BinaryViewArray::new_unchecked(views, self.completed.into(), null_buffer)
+ };
match self.output_type {
OutputType::BinaryView => Arc::new(array),
diff --git a/datafusion/physical-expr-common/src/utils.rs b/datafusion/physical-expr-common/src/utils.rs
index 5dadcdcabb..b67e391dd4 100644
--- a/datafusion/physical-expr-common/src/utils.rs
+++ b/datafusion/physical-expr-common/src/utils.rs
@@ -305,7 +305,7 @@ fn scatter_byte_view<T: ByteViewType>(
unsafe {
GenericByteViewArray::new_unchecked(
new_views,
- truthy.data_buffers().to_vec(),
+ Arc::clone(truthy.data_buffers()),
nulls,
)
}
diff --git a/datafusion/physical-expr/benches/case_when.rs b/datafusion/physical-expr/benches/case_when.rs
index b6c45f002d..e94cb9560b 100644
--- a/datafusion/physical-expr/benches/case_when.rs
+++ b/datafusion/physical-expr/benches/case_when.rs
@@ -18,7 +18,6 @@
use arrow::array::{Array, ArrayRef, Int32Array, Int32Builder, StringArray};
use arrow::datatypes::{ArrowNativeTypeOp, Field, Schema};
use arrow::record_batch::RecordBatch;
-use arrow::util::test_util::seedable_rng;
use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
use datafusion_common::ScalarValue;
use datafusion_expr::Operator;
@@ -28,7 +27,13 @@ use itertools::Itertools;
use rand::distr::Alphanumeric;
use rand::distr::uniform::SampleUniform;
use rand::rngs::StdRng;
-use rand::{Rng, RngCore};
+use rand::{Rng, RngCore, SeedableRng};
+
+/// Returns a fixed-seed RNG using this crate's `rand` version (arrow's
+/// `test_util::seedable_rng` returns its own `rand` version's `StdRng`)
+fn seedable_rng() -> StdRng {
+ StdRng::seed_from_u64(42)
+}
use std::fmt::{Display, Formatter};
use std::hint::black_box;
use std::ops::Range;
diff --git a/datafusion/physical-expr/src/expressions/cast.rs b/datafusion/physical-expr/src/expressions/cast.rs
index 845b9608d9..8e49b6ff48 100644
--- a/datafusion/physical-expr/src/expressions/cast.rs
+++ b/datafusion/physical-expr/src/expressions/cast.rs
@@ -15,7 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-use std::collections::HashMap;
use std::fmt;
use std::hash::Hash;
use std::sync::Arc;
@@ -23,7 +22,7 @@ use std::sync::Arc;
use crate::physical_expr::PhysicalExpr;
use arrow::compute::{CastOptions, can_cast_types};
-use arrow::datatypes::{DataType, DataType::*, Field, FieldRef, Schema};
+use arrow::datatypes::{DataType, DataType::*, Field, FieldRef, Metadata, Schema};
use arrow::record_batch::RecordBatch;
use arrow_schema::extension::{EXTENSION_TYPE_METADATA_KEY, EXTENSION_TYPE_NAME_KEY};
use datafusion_common::datatype::DataTypeExt;
@@ -168,7 +167,7 @@ impl CastExpr {
}
/// Explicit metadata for the output field, or `None` to pass through source metadata.
- pub fn target_metadata(&self) -> Option<&HashMap<String, String>> {
+ pub fn target_metadata(&self) -> Option<&Metadata> {
self.explicit_target.then(|| self.target_field.metadata())
}
diff --git a/datafusion/physical-expr/src/expressions/try_cast.rs b/datafusion/physical-expr/src/expressions/try_cast.rs
index c054026724..7e445a275d 100644
--- a/datafusion/physical-expr/src/expressions/try_cast.rs
+++ b/datafusion/physical-expr/src/expressions/try_cast.rs
@@ -15,7 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-use std::collections::HashMap;
use std::fmt;
use std::hash::Hash;
use std::sync::Arc;
@@ -23,7 +22,7 @@ use std::sync::Arc;
use crate::PhysicalExpr;
use arrow::compute;
use arrow::compute::CastOptions;
-use arrow::datatypes::{DataType, Field, FieldRef, Schema};
+use arrow::datatypes::{DataType, Field, FieldRef, Metadata, Schema};
use arrow::record_batch::RecordBatch;
use arrow_schema::extension::{EXTENSION_TYPE_METADATA_KEY, EXTENSION_TYPE_NAME_KEY};
use compute::can_cast_types;
@@ -124,7 +123,7 @@ impl TryCastExpr {
}
/// Explicit metadata for the output field, or `None` to pass through source metadata.
- pub fn target_metadata(&self) -> Option<&HashMap<String, String>> {
+ pub fn target_metadata(&self) -> Option<&Metadata> {
self.explicit_target.then(|| self.target_field.metadata())
}
@@ -343,6 +342,8 @@ pub fn try_cast_with_target_field(
#[cfg(test)]
mod tests {
+ use std::collections::HashMap;
+
use super::*;
use crate::expressions::col;
use arrow::array::{
diff --git a/datafusion/physical-expr/src/planner.rs b/datafusion/physical-expr/src/planner.rs
index 9522b0910c..bf6e8cc6da 100644
--- a/datafusion/physical-expr/src/planner.rs
+++ b/datafusion/physical-expr/src/planner.rs
@@ -742,7 +742,7 @@ pub fn logical2physical(expr: &Expr, schema: &Schema) -> Arc<dyn PhysicalExpr> {
#[cfg(test)]
mod tests {
use arrow::array::{ArrayRef, BooleanArray, RecordBatch, StringArray};
- use arrow::datatypes::{DataType, Field};
+ use arrow::datatypes::{DataType, Field, Metadata};
use arrow_schema::extension::{EXTENSION_TYPE_METADATA_KEY, EXTENSION_TYPE_NAME_KEY};
use datafusion_common::HashMap;
use datafusion_expr::physical_planning_context::{
@@ -884,15 +884,10 @@ mod tests {
// With exact target metadata semantics, all target metadata should propagate.
let target_field = Arc::new(
Field::new("cast_target", DataType::Int64, true).with_metadata(
- [
- (
- EXTENSION_TYPE_NAME_KEY.to_string(),
- "arrow.json".to_string(),
- ),
- (EXTENSION_TYPE_METADATA_KEY.to_string(), "{}".to_string()),
- ("custom_target_meta".to_string(), "custom_value".to_string()),
- ]
- .into(),
+ Metadata::new()
+ .with(EXTENSION_TYPE_NAME_KEY, "arrow.json")
+ .with(EXTENSION_TYPE_METADATA_KEY, "{}")
+ .with("custom_target_meta", "custom_value"),
),
);
let cast_expr = Expr::Cast(Cast::new_from_field(
@@ -962,14 +957,9 @@ mod tests {
// With exact target metadata semantics, all target metadata should propagate.
let target_field = Arc::new(
Field::new("same_type_cast", DataType::Int32, true).with_metadata(
- [
- (
- EXTENSION_TYPE_NAME_KEY.to_string(),
- "arrow.opaque".to_string(),
- ),
- ("custom_meta".to_string(), "custom_value".to_string()),
- ]
- .into(),
+ Metadata::new()
+ .with(EXTENSION_TYPE_NAME_KEY, "arrow.opaque")
+ .with("custom_meta", "custom_value"),
),
);
diff --git a/datafusion/physical-expr/src/scalar_subquery.rs b/datafusion/physical-expr/src/scalar_subquery.rs
index 927df8c2ee..d04c9f19ef 100644
--- a/datafusion/physical-expr/src/scalar_subquery.rs
+++ b/datafusion/physical-expr/src/scalar_subquery.rs
@@ -189,7 +189,7 @@ impl PhysicalExpr for ScalarSubqueryExpr {
"ScalarSubqueryExpr",
"index",
)?,
- metadata: field.metadata().clone(),
+ metadata: field.metadata().into(),
},
)),
}))
diff --git a/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/bytes_view.rs b/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/bytes_view.rs
index 8153b67d6c..24eb944763 100644
--- a/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/bytes_view.rs
+++ b/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/bytes_view.rs
@@ -402,7 +402,7 @@ impl<B: ByteViewType> ByteViewGroupValueBuilder<B> {
unsafe {
Arc::new(GenericByteViewArray::<B>::new_unchecked(
views,
- completed,
+ completed.into(),
null_buffer,
))
}
@@ -457,7 +457,7 @@ impl<B: ByteViewType> ByteViewGroupValueBuilder<B> {
unsafe {
return Arc::new(GenericByteViewArray::<B>::new_unchecked(
views,
- Vec::new(),
+ Vec::new().into(),
null_buffer,
));
}
@@ -508,7 +508,7 @@ impl<B: ByteViewType> ByteViewGroupValueBuilder<B> {
unsafe {
Arc::new(GenericByteViewArray::<B>::new_unchecked(
views,
- buffers,
+ buffers.into(),
null_buffer,
))
}
diff --git a/datafusion/physical-plan/src/joins/utils.rs b/datafusion/physical-plan/src/joins/utils.rs
index 8cba438f55..30b2972312 100644
--- a/datafusion/physical-plan/src/joins/utils.rs
+++ b/datafusion/physical-plan/src/joins/utils.rs
@@ -340,12 +340,8 @@ pub fn build_join_schema(
_ => (right, left),
};
- let metadata = schema1
- .metadata()
- .clone()
- .into_iter()
- .chain(schema2.metadata().clone())
- .collect();
+ let mut metadata = schema1.metadata().clone();
+ metadata.extend(schema2.metadata().clone());
(fields.finish().with_metadata(metadata), column_indices)
}
diff --git a/datafusion/physical-plan/src/spill/spill_manager.rs b/datafusion/physical-plan/src/spill/spill_manager.rs
index 1db4f53982..389fafb985 100644
--- a/datafusion/physical-plan/src/spill/spill_manager.rs
+++ b/datafusion/physical-plan/src/spill/spill_manager.rs
@@ -20,8 +20,7 @@
use super::{SpillReaderStream, in_progress_spill_file::InProgressSpillFile};
use crate::coop::cooperative;
use crate::{common::spawn_buffered, metrics::SpillMetrics};
-use arrow::array::{BinaryViewArray, GenericByteViewArray, StringViewArray};
-use arrow::datatypes::{ByteViewType, SchemaRef};
+use arrow::datatypes::SchemaRef;
use arrow::record_batch::RecordBatch;
use datafusion_common::{DataFusionError, Result, config::SpillCompression};
use datafusion_execution::SendableRecordBatchStream;
@@ -232,35 +231,14 @@ impl GetSlicedSize for RecordBatch {
let mut total = 0;
for array in self.columns() {
let data = array.to_data();
+ // Since https://github.com/apache/arrow-rs/issues/8230 this also
+ // accounts for the variadic data buffers retained by view arrays
total += data.get_slice_memory_size()?;
-
- // While StringViewArray holds large data buffer for non inlined string, the Arrow layout (BufferSpec)
- // does not include any data buffers. Currently, ArrayData::get_slice_memory_size()
- // under-counts memory size by accounting only views buffer although data buffer is cloned during slice()
- //
- // Therefore, we manually add the sum of the lengths used by all non inlined views
- // on top of the sliced size for views buffer. This matches the intended semantics of
- // "bytes needed if we materialized exactly this slice into fresh buffers".
- // This is a workaround until https://github.com/apache/arrow-rs/issues/8230
- if let Some(sv) = array.as_any().downcast_ref::<StringViewArray>() {
- total += byte_view_data_buffer_size(sv);
- }
- if let Some(bv) = array.as_any().downcast_ref::<BinaryViewArray>() {
- total += byte_view_data_buffer_size(bv);
- }
}
Ok(total)
}
}
-fn byte_view_data_buffer_size<T: ByteViewType>(array: &GenericByteViewArray<T>) -> usize {
- array
- .data_buffers()
- .iter()
- .map(|buffer| buffer.capacity())
- .sum()
-}
-
#[cfg(test)]
mod tests {
use super::SpillManager;
@@ -526,10 +504,11 @@ mod tests {
half_batch.get_sliced_size().unwrap()
< get_record_batch_memory_size(&half_batch)
);
+ // `get_slice_memory_size` accounts for the retained
+ // variadic data buffers as well, so it matches `get_sliced_size`
let data = arrow::array::Array::to_data(&half_batch.column(0));
let views_sliced_size = data.get_slice_memory_size()?;
- // The sliced size should be larger than sliced views buffer size
- assert!(views_sliced_size < half_batch.get_sliced_size().unwrap());
+ assert_eq!(views_sliced_size, half_batch.get_sliced_size().unwrap());
Ok(())
}
diff --git a/datafusion/physical-plan/src/union.rs b/datafusion/physical-plan/src/union.rs
index da496b99fc..1f5985fe3f 100644
--- a/datafusion/physical-plan/src/union.rs
+++ b/datafusion/physical-plan/src/union.rs
@@ -46,7 +46,7 @@ use crate::statistics::{ChildStats, StatisticsArgs};
use crate::stream::ObservedStream;
use crate::{ChildrenPropertiesMode, ReplaceChildrenOptions, validate_child_count};
-use arrow::datatypes::{Field, Schema, SchemaRef};
+use arrow::datatypes::{Field, Metadata, Schema, SchemaRef};
use arrow::record_batch::RecordBatch;
use datafusion_common::config::ConfigOptions;
use datafusion_common::stats::NdvFallback;
@@ -1044,7 +1044,7 @@ fn union_schema(inputs: &[Arc<dyn ExecutionPlan>]) -> Result<SchemaRef> {
})
.collect::<Vec<_>>();
- let all_metadata_merged = inputs
+ let all_metadata_merged: Metadata = inputs
.iter()
.flat_map(|i| i.schema().metadata().clone().into_iter())
.collect();
diff --git a/datafusion/proto-common/src/from_proto/mod.rs b/datafusion/proto-common/src/from_proto/mod.rs
index 8ddb1cf967..62b16664f1 100644
--- a/datafusion/proto-common/src/from_proto/mod.rs
+++ b/datafusion/proto-common/src/from_proto/mod.rs
@@ -29,7 +29,7 @@ use arrow::datatypes::{
TimeUnit, UnionFields, UnionMode, i256,
};
use arrow::ipc::{
- convert::fb_to_schema,
+ convert::try_fb_to_schema,
reader::{read_dictionary, read_record_batch},
root_as_message,
writer::{DictionaryTracker, IpcDataGenerator, IpcWriteOptions},
@@ -463,7 +463,11 @@ impl TryFrom<&protobuf::ScalarValue> for ScalarValue {
.to_string(),
)
})?;
- fb_to_schema(ipc_schema)
+ try_fb_to_schema(ipc_schema).map_err(|e| {
+ Error::General(format!(
+ "Error converting IPC schema while deserializing nested ScalarValue: {e}"
+ ))
+ })?
};
let message = root_as_message(ipc_message.as_slice()).map_err(|e| {
diff --git a/datafusion/proto-common/src/to_proto/mod.rs b/datafusion/proto-common/src/to_proto/mod.rs
index 2d47a84709..f9a881e76d 100644
--- a/datafusion/proto-common/src/to_proto/mod.rs
+++ b/datafusion/proto-common/src/to_proto/mod.rs
@@ -100,7 +100,7 @@ impl TryFrom<&Field> for protobuf::Field {
arrow_type: Some(Box::new(arrow_type)),
nullable: field.is_nullable(),
children: Vec::new(),
- metadata: field.metadata().clone(),
+ metadata: field.metadata().into(),
})
}
}
@@ -266,7 +266,7 @@ impl TryFrom<&Schema> for protobuf::Schema {
fn try_from(schema: &Schema) -> Result<Self, Self::Error> {
Ok(Self {
columns: convert_arc_fields_to_proto_fields(schema.fields())?,
- metadata: schema.metadata.clone(),
+ metadata: schema.metadata().into(),
})
}
}
@@ -277,7 +277,7 @@ impl TryFrom<SchemaRef> for protobuf::Schema {
fn try_from(schema: SchemaRef) -> Result<Self, Self::Error> {
Ok(Self {
columns: convert_arc_fields_to_proto_fields(schema.fields())?,
- metadata: schema.metadata.clone(),
+ metadata: schema.metadata().into(),
})
}
}
@@ -299,7 +299,7 @@ impl TryFrom<&DFSchema> for protobuf::DfSchema {
.collect::<Result<Vec<_>, Error>>()?;
Ok(Self {
columns,
- metadata: s.metadata().clone(),
+ metadata: s.metadata().into(),
})
}
}
diff --git a/datafusion/proto/src/logical_plan/to_proto.rs b/datafusion/proto/src/logical_plan/to_proto.rs
index 1dc4250065..f72b07871b 100644
--- a/datafusion/proto/src/logical_plan/to_proto.rs
+++ b/datafusion/proto/src/logical_plan/to_proto.rs
@@ -401,7 +401,7 @@ pub fn serialize_expr(
let expr = Box::new(protobuf::CastNode {
expr: Some(Box::new(serialize_expr(expr.as_ref(), codec)?)),
arrow_type: Some(field.data_type().try_into()?),
- metadata: field.metadata().clone(),
+ metadata: field.metadata().into(),
nullable: Some(field.is_nullable()),
});
protobuf::LogicalExprNode {
@@ -412,7 +412,7 @@ pub fn serialize_expr(
let expr = Box::new(protobuf::TryCastNode {
expr: Some(Box::new(serialize_expr(expr.as_ref(), codec)?)),
arrow_type: Some(field.data_type().try_into()?),
- metadata: field.metadata().clone(),
+ metadata: field.metadata().into(),
nullable: Some(field.is_nullable()),
});
protobuf::LogicalExprNode {
@@ -505,7 +505,7 @@ pub fn serialize_expr(
nullable: field.as_ref().map(|f| f.is_nullable()),
metadata: field
.as_ref()
- .map(|f| f.metadata().clone())
+ .map(|f| f.metadata().into())
.unwrap_or_default(),
})),
},
diff --git a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs
index 932c418811..eb6207cf42 100644
--- a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs
+++ b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs
@@ -21,8 +21,8 @@ use arrow::array::{
};
use arrow::datatypes::{
DECIMAL256_MAX_PRECISION, DataType, Field, FieldRef, Fields, Int32Type,
- IntervalDayTimeType, IntervalMonthDayNanoType, IntervalUnit, Schema, SchemaRef,
- TimeUnit, UnionFields, UnionMode,
+ IntervalDayTimeType, IntervalMonthDayNanoType, IntervalUnit, Metadata, Schema,
+ SchemaRef, TimeUnit, UnionFields, UnionMode,
};
use arrow::util::pretty::pretty_format_batches;
use datafusion::datasource::file_format::json::{JsonFormat, JsonFormatFactory};
@@ -1713,9 +1713,7 @@ async fn roundtrip_logical_plan_prepared_statement_with_metadata() -> Result<()>
"".to_string(),
vec![
Field::new("", DataType::Int32, true)
- .with_metadata(
- [("some_key".to_string(), "some_value".to_string())].into(),
- )
+ .with_metadata(Metadata::new().with("some_key", "some_value"))
.into(),
],
)
diff --git a/datafusion/proto/tests/cases/serialize.rs b/datafusion/proto/tests/cases/serialize.rs
index 40548fb1cf..c71ca27b3e 100644
--- a/datafusion/proto/tests/cases/serialize.rs
+++ b/datafusion/proto/tests/cases/serialize.rs
@@ -18,7 +18,7 @@
use std::sync::Arc;
use arrow::array::ArrayRef;
-use arrow::datatypes::{DataType, Field};
+use arrow::datatypes::{DataType, Field, Metadata};
use datafusion::execution::FunctionRegistry;
use datafusion::prelude::SessionContext;
@@ -203,9 +203,7 @@ fn roundtrip_placeholder_with_metadata() {
"placeholder_id".to_string(),
Some(
Field::new("", DataType::Utf8, false)
- .with_metadata(
- [("some_key".to_string(), "some_value".to_string())].into(),
- )
+ .with_metadata(Metadata::new().with("some_key", "some_value"))
.into(),
),
));
diff --git a/datafusion/sql/src/statement.rs b/datafusion/sql/src/statement.rs
index 75ebcfea8f..cdf58327e1 100644
--- a/datafusion/sql/src/statement.rs
+++ b/datafusion/sql/src/statement.rs
@@ -29,7 +29,7 @@ use crate::planner::{
};
use crate::utils::normalize_ident;
-use arrow::datatypes::{Field, FieldRef, Fields};
+use arrow::datatypes::{Field, FieldRef, Fields, Metadata};
use datafusion_common::error::_plan_err;
use datafusion_common::format::ExplainStatementOptions;
use datafusion_common::parsers::CompressionTypeVariant;
@@ -2867,7 +2867,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
let mut planner_context =
PlannerContext::new().with_prepare_param_data_types(prepare_param_data_types);
planner_context.set_table_schema(Some(DFSchemaRef::new(
- DFSchema::from_unqualified_fields(fields.clone(), Default::default())?,
+ DFSchema::from_unqualified_fields(fields.clone(), Metadata::new())?,
)));
let source = self.query_to_plan(*source, &mut planner_context)?;
if fields.len() != source.schema().fields().len() {
diff --git a/datafusion/sql/tests/cases/params.rs b/datafusion/sql/tests/cases/params.rs
index d7ad61b2bc..3d51974071 100644
--- a/datafusion/sql/tests/cases/params.rs
+++ b/datafusion/sql/tests/cases/params.rs
@@ -16,7 +16,7 @@
// under the License.
use crate::logical_plan;
-use arrow::datatypes::{DataType, Field, FieldRef};
+use arrow::datatypes::{DataType, Field, FieldRef, Metadata};
use datafusion_common::{
ParamValues, ScalarValue, assert_contains,
metadata::{ScalarAndMetadata, format_type_and_metadata},
@@ -732,9 +732,8 @@ fn test_prepare_statement_to_plan_one_param() {
fn test_update_infer_with_metadata() {
// Here the uuid field is inferred as nullable because it appears in the filter
// (and not in the update values, where its nullability would be inferred)
- let uuid_field = Field::new("", DataType::FixedSizeBinary(16), true).with_metadata(
- [("ARROW:extension:name".to_string(), "arrow.uuid".to_string())].into(),
- );
+ let uuid_field = Field::new("", DataType::FixedSizeBinary(16), true)
+ .with_metadata(Metadata::new().with("ARROW:extension:name", "arrow.uuid"));
let uuid_bytes = vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16];
let expected_types = vec![
(
@@ -801,9 +800,8 @@ fn test_update_infer_with_metadata() {
#[test]
fn test_insert_infer_with_metadata() {
- let uuid_field = Field::new("", DataType::FixedSizeBinary(16), false).with_metadata(
- [("ARROW:extension:name".to_string(), "arrow.uuid".to_string())].into(),
- );
+ let uuid_field = Field::new("", DataType::FixedSizeBinary(16), false)
+ .with_metadata(Metadata::new().with("ARROW:extension:name", "arrow.uuid"));
let uuid_bytes = vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16];
let expected_types = vec![
("$1", Some(uuid_field.clone().with_name("id").into())),
diff --git a/datafusion/sql/tests/common/mod.rs b/datafusion/sql/tests/common/mod.rs
index 66447afd73..2196d5eecd 100644
--- a/datafusion/sql/tests/common/mod.rs
+++ b/datafusion/sql/tests/common/mod.rs
@@ -168,8 +168,7 @@ impl ContextProvider for MockContextProvider {
])),
"person_with_uuid_extension" => Ok(Schema::new(vec![
Field::new("id", DataType::FixedSizeBinary(16), false).with_metadata(
- [("ARROW:extension:name".to_string(), "arrow.uuid".to_string())]
- .into(),
+ Metadata::new().with("ARROW:extension:name", "arrow.uuid"),
),
Field::new("first_name", DataType::Utf8, false),
Field::new("last_name", DataType::Utf8, false),
@@ -181,11 +180,7 @@ impl ContextProvider for MockContextProvider {
])),
"string_with_extension" => Ok(Schema::new(vec![
Field::new("value", DataType::Utf8, false).with_metadata(
- [(
- "ARROW:extension:name".to_string(),
- "example.string".to_string(),
- )]
- .into(),
+ Metadata::new().with("ARROW:extension:name", "example.string"),
),
])),
"orders" => Ok(Schema::new(vec![
@@ -232,15 +227,12 @@ impl ContextProvider for MockContextProvider {
Field::new(
"left",
DataType::List(Arc::new(
- Field::new_list_field(DataType::Int64, true).with_metadata(
- [("PARQUET:field_id".to_string(), "2".to_string())].into(),
- ),
+ Field::new_list_field(DataType::Int64, true)
+ .with_metadata(Metadata::new().with("PARQUET:field_id", "2")),
)),
false,
)
- .with_metadata(
- [("PARQUET:field_id".to_string(), "1".to_string())].into(),
- ),
+ .with_metadata(Metadata::new().with("PARQUET:field_id", "1")),
])),
"lineitem" => Ok(Schema::new(vec![
Field::new("l_orderkey", DataType::UInt32, false),
@@ -410,8 +402,7 @@ impl TypePlanner for CustomTypePlanner {
match sql_type {
sqlparser::ast::DataType::Uuid => Ok(Some(Arc::new(
Field::new("", DataType::FixedSizeBinary(16), true).with_metadata(
- [("ARROW:extension:name".to_string(), "arrow.uuid".to_string())]
- .into(),
+ Metadata::new().with("ARROW:extension:name", "arrow.uuid"),
),
))),
sqlparser::ast::DataType::Datetime(precision) => {
diff --git a/datafusion/sqllogictest/src/test_context.rs b/datafusion/sqllogictest/src/test_context.rs
index 76dcb322a9..97d9d66867 100644
--- a/datafusion/sqllogictest/src/test_context.rs
+++ b/datafusion/sqllogictest/src/test_context.rs
@@ -29,7 +29,7 @@ use arrow::array::{
};
use arrow::buffer::ScalarBuffer;
use arrow::datatypes::{
- DataType, Field, FieldRef, Fields, Schema, SchemaRef, TimeUnit, UInt32Type,
+ DataType, Field, FieldRef, Fields, Metadata, Schema, SchemaRef, TimeUnit, UInt32Type,
UnionFields,
};
use arrow::record_batch::RecordBatch;
@@ -85,8 +85,7 @@ impl TypePlanner for SqlLogicTestTypePlanner {
match sql_type {
ast::DataType::Uuid => Ok(Some(Arc::new(
Field::new("", DataType::FixedSizeBinary(16), true).with_metadata(
- [("ARROW:extension:name".to_string(), "arrow.uuid".to_string())]
- .into(),
+ Metadata::new().with("ARROW:extension:name", "arrow.uuid"),
),
))),
_ => Ok(None),
diff --git a/datafusion/sqllogictest/test_files/datetime/arith_interval_double.slt b/datafusion/sqllogictest/test_files/datetime/arith_interval_double.slt
index cb357f0cdb..4bfa7235ff 100644
--- a/datafusion/sqllogictest/test_files/datetime/arith_interval_double.slt
+++ b/datafusion/sqllogictest/test_files/datetime/arith_interval_double.slt
@@ -21,7 +21,7 @@
# interval '1 day' * 21 → 21 days
# interval '1 hour' * 3.5 → 03:30:00
-# these currently do not work - https://github.com/apache/arrow-rs/issues/9030
+# Some of these currently do not work - https://github.com/apache/arrow-rs/issues/9030
query error Invalid interval arithmetic operation: Interval\(MonthDayNano\) \* Interval\(MonthDayNano\)
SELECT interval '1 second' * 900
@@ -35,7 +35,7 @@ query error Invalid interval arithmetic operation: Interval\(MonthDayNano\) \* I
SELECT interval '1 day' * 21
-query error Invalid interval arithmetic operation: Interval\(MonthDayNano\) \* Interval\(MonthDayNano\)
+query error DataFusion error: Arrow error: Invalid argument error: Invalid interval arithmetic operation: Interval\(MonthDayNano\) \* Float64
SELECT interval '1 hour' * 3.5
@@ -50,9 +50,13 @@ SELECT arrow_typeof(interval '1 second' * 900)
# Divide an interval by a scalar
# interval '1 hour' / 1.5 → 00:40:00
-query error Invalid interval arithmetic operation: Interval\(MonthDayNano\) / Interval\(MonthDayNano\)
+query ?
SELECT interval '1 hour' / 1.5
+----
+40 mins
-query error Invalid interval arithmetic operation: Interval\(MonthDayNano\) / Interval\(MonthDayNano\)
+query T
SELECT arrow_typeof(interval '1 hour' / 1.5)
+----
+Interval(MonthDayNano)
diff --git a/datafusion/sqllogictest/test_files/explain_analyze.slt b/datafusion/sqllogictest/test_files/explain_analyze.slt
index 81b274d6c1..511ba88ed3 100644
--- a/datafusion/sqllogictest/test_files/explain_analyze.slt
+++ b/datafusion/sqllogictest/test_files/explain_analyze.slt
@@ -159,7 +159,7 @@ SELECT a+1, pow(a,2)
FROM generate_series(1, 100) as t1(a);
----
Plan with Metrics
-01)ProjectionExec: expr=[a@0 + 1 as t1.a + Int64(1), power(CAST(a@0 AS Float64), 2) as pow(t1.a,Int64(2))], metrics=[output_rows=100, elapsed_compute=<slt:ignore>, output_bytes=1632.0 B, output_batches=1, expr_0_eval_time=<slt:ignore>, expr_1_eval_time=<slt:ignore>]
+01)ProjectionExec: expr=[a@0 + 1 as t1.a + Int64(1), power(CAST(a@0 AS Float64), 2) as pow(t1.a,Int64(2))], metrics=[output_rows=100, elapsed_compute=<slt:ignore>, output_bytes=1600.0 B, output_batches=1, expr_0_eval_time=<slt:ignore>, expr_1_eval_time=<slt:ignore>]
<slt:ignore>
# common expressions
diff --git a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt
index 6774d4f3a0..8dd763205f 100644
--- a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt
+++ b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt
@@ -268,7 +268,7 @@ EXPLAIN ANALYZE SELECT * FROM topk_single_col ORDER BY b DESC LIMIT 1;
----
Plan with Metrics
01)SortExec: TopK(fetch=1), expr=[b@1 DESC], preserve_partitioning=[false], filter=[b@1 IS NULL OR b@1 > bd], metrics=[output_rows=1, output_batches=1, row_replacements=1]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_single_col.parquet]]}, projection=[a, b, c], file_type=parquet, predicate=DynamicFilter [ b@1 IS NULL OR b@1 > bd ], sort_order_for_reorder=[b@1 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible, pruning_predicate=b_null_count@0 > 0 OR b_null_count@0 != row_count@2 AND b_max@1 > bd, required_guarantees=[], metrics=[output_rows=4, output_batches=1, [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_single_col.parquet]]}, projection=[a, b, c], file_type=parquet, predicate=DynamicFilter [ b@1 IS NULL OR b@1 > bd ], sort_order_for_reorder=[b@1 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible, pruning_predicate=b_null_count@0 > 0 OR b_null_count@0 != row_count@2 AND b_max@1 > bd, required_guarantees=[], metrics=[output_rows=4, output_batches=1, [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -319,7 +319,7 @@ EXPLAIN ANALYZE SELECT * FROM topk_multi_col ORDER BY b ASC NULLS LAST, a DESC L
----
Plan with Metrics
01)SortExec: TopK(fetch=2), expr=[b@1 ASC NULLS LAST, a@0 DESC], preserve_partitioning=[false], filter=[b@1 < bb OR b@1 = bb AND (a@0 IS NULL OR a@0 > ac)], metrics=[output_rows=2, output_batches=1, row_replacements=2]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_multi_col.parquet]]}, projection=[a, b, c], file_type=parquet, predicate=DynamicFilter [ b@1 < bb OR b@1 = bb AND (a@0 IS NULL OR a@0 > ac) ], sort_order_for_reorder=[b@1 ASC NULLS LAST, a@0 DESC], dynamic_rg_pruning=eligible, pruning_predicate=b_null_count@1 != row_count@2 AND b_min@0 < bb OR b_null_count@1 != row_count@2 AND b_min@0 <= bb AND bb <= b_max@ [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_multi_col.parquet]]}, projection=[a, b, c], file_type=parquet, predicate=DynamicFilter [ b@1 < bb OR b@1 = bb AND (a@0 IS NULL OR a@0 > ac) ], sort_order_for_reorder=[b@1 ASC NULLS LAST, a@0 DESC], dynamic_rg_pruning=eligible, pruning_predicate=b_null_count@1 != row_count@2 AND b_min@0 < bb OR b_null_count@1 != row_count@2 AND b_min@0 <= bb AND bb <= b_max@ [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -388,8 +388,8 @@ FROM join_probe p INNER JOIN join_build AS build
----
Plan with Metrics
01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0), (b@1, b@1)], projection=[a@3, b@4, c@2, e@5], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/join_build.parquet]]}, projection=[a, b, c], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_ro [...]
-03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/join_probe.parquet]]}, projection=[a, b, e], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}, {c0:ab,c1:bb}]) ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab AND b_nul [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/join_build.parquet]]}, projection=[a, b, c], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_ro [...]
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/join_probe.parquet]]}, projection=[a, b, e], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}, {c0:ab,c1:bb}]) ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab AND b_nul [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -474,9 +474,9 @@ INNER JOIN nested_t3 ON nested_t2.c = nested_t3.d;
Plan with Metrics
01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(c@3, d@0)], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)]
02)--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, b@0)], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)]
-03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nested_t1.parquet]]}, projection=[a, x], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_ [...]
-04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nested_t2.parquet]]}, projection=[b, c, y], file_type=parquet, predicate=DynamicFilter [ b@0 >= aa AND b@0 <= ab AND b@0 IN (SET) ([aa, ab]) ], dynamic_rg_pruning=eligible, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 >= aa AND b_null_count@1 != row_count@2 AND b_min@3 <= ab AND (b_null_count@1 != row_count@2 AND b_min@3 <= aa AND aa <= b_max@0 [...]
-05)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nested_t3.parquet]]}, projection=[d, z], file_type=parquet, predicate=DynamicFilter [ d@0 >= ca AND d@0 <= cb AND hash_lookup ], dynamic_rg_pruning=eligible, pruning_predicate=d_null_count@1 != row_count@2 AND d_max@0 >= ca AND d_null_count@1 != row_count@2 AND d_min@3 <= cb, required_guarantees=[], metrics=[output_rows=2, output_batches=1, files_ranges_pruned_s [...]
+03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nested_t1.parquet]]}, projection=[a, x], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_ [...]
+04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nested_t2.parquet]]}, projection=[b, c, y], file_type=parquet, predicate=DynamicFilter [ b@0 >= aa AND b@0 <= ab AND b@0 IN (SET) ([aa, ab]) ], dynamic_rg_pruning=eligible, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 >= aa AND b_null_count@1 != row_count@2 AND b_min@3 <= ab AND (b_null_count@1 != row_count@2 AND b_min@3 <= aa AND aa <= b_max@0 [...]
+05)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nested_t3.parquet]]}, projection=[d, z], file_type=parquet, predicate=DynamicFilter [ d@0 >= ca AND d@0 <= cb AND hash_lookup ], dynamic_rg_pruning=eligible, pruning_predicate=d_null_count@1 != row_count@2 AND d_max@0 >= ca AND d_null_count@1 != row_count@2 AND d_min@3 <= cb, required_guarantees=[], metrics=[output_rows=2, output_batches=1, files_ranges_pruned_s [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -605,8 +605,8 @@ LIMIT 2;
Plan with Metrics
01)SortExec: TopK(fetch=2), expr=[e@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[e@0 < bb], metrics=[output_rows=2, output_batches=1, row_replacements=2]
02)--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, d@0)], projection=[e@2], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)]
-03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_join_build.parquet]]}, projection=[a], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_r [...]
-04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_join_probe.parquet]]}, projection=[d, e], file_type=parquet, predicate=DynamicFilter [ d@0 >= aa AND d@0 <= ab AND d@0 IN (SET) ([aa, ab]) ] AND DynamicFilter [ e@1 < bb ], dynamic_rg_pruning=eligible, pruning_predicate=d_null_count@1 != row_count@2 AND d_max@0 >= aa AND d_null_count@1 != row_count@2 AND d_min@3 <= ab AND (d_null_count@1 != row_count@2 AN [...]
+03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_join_build.parquet]]}, projection=[a], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_r [...]
+04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_join_probe.parquet]]}, projection=[d, e], file_type=parquet, predicate=DynamicFilter [ d@0 >= aa AND d@0 <= ab AND d@0 IN (SET) ([aa, ab]) ] AND DynamicFilter [ e@1 < bb ], dynamic_rg_pruning=eligible, pruning_predicate=d_null_count@1 != row_count@2 AND d_max@0 >= aa AND d_null_count@1 != row_count@2 AND d_min@3 <= ab AND (d_null_count@1 != row_count@2 AN [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -656,7 +656,7 @@ EXPLAIN ANALYZE SELECT b, a FROM topk_proj ORDER BY a LIMIT 2;
Plan with Metrics
01)ProjectionExec: expr=[b@1 as b, a@0 as a], metrics=[output_rows=2, output_batches=1]
02)--SortExec: TopK(fetch=2), expr=[a@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[a@0 < 2], metrics=[output_rows=2, output_batches=1, row_replacements=2]
-03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_proj.parquet]]}, projection=[a, b], file_type=parquet, predicate=DynamicFilter [ a@0 < 2 ], sort_order_for_reorder=[a@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_min@0 < 2, required_guarantees=[], metrics=[output_rows=3, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_grou [...]
+03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_proj.parquet]]}, projection=[a, b], file_type=parquet, predicate=DynamicFilter [ a@0 < 2 ], sort_order_for_reorder=[a@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_min@0 < 2, required_guarantees=[], metrics=[output_rows=3, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_grou [...]
# Case 2: prune — `SELECT a` — filter stays as `a < 2` on the scan.
query TT
@@ -664,7 +664,7 @@ EXPLAIN ANALYZE SELECT a FROM topk_proj ORDER BY a LIMIT 2;
----
Plan with Metrics
01)SortExec: TopK(fetch=2), expr=[a@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[a@0 < 2], metrics=[output_rows=2, output_batches=1, row_replacements=2]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_proj.parquet]]}, projection=[a], file_type=parquet, predicate=DynamicFilter [ a@0 < 2 ], sort_order_for_reorder=[a@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_min@0 < 2, required_guarantees=[], metrics=[output_rows=3, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pr [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_proj.parquet]]}, projection=[a], file_type=parquet, predicate=DynamicFilter [ a@0 < 2 ], sort_order_for_reorder=[a@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_min@0 < 2, required_guarantees=[], metrics=[output_rows=3, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pr [...]
# Case 3: expression — `SELECT a+1 AS a_plus_1` — the TopK filter is on
# `a_plus_1`, the scan predicate must read `a@0 + 1`.
@@ -673,7 +673,7 @@ EXPLAIN ANALYZE SELECT a + 1 AS a_plus_1, b FROM topk_proj ORDER BY a_plus_1 LIM
----
Plan with Metrics
01)SortExec: TopK(fetch=2), expr=[a_plus_1@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[a_plus_1@0 < 3], metrics=[output_rows=2, output_batches=1, row_replacements=2]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_proj.parquet]]}, projection=[CAST(a@0 AS Int64) + 1 as a_plus_1, b], file_type=parquet, predicate=DynamicFilter [ CAST(a@0 AS Int64) + 1 < 3 ], dynamic_rg_pruning=eligible, metrics=[output_rows=3, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_proj.parquet]]}, projection=[CAST(a@0 AS Int64) + 1 as a_plus_1, b], file_type=parquet, predicate=DynamicFilter [ CAST(a@0 AS Int64) + 1 < 3 ], dynamic_rg_pruning=eligible, metrics=[output_rows=3, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 [...]
# Case 4: alias shadowing — `SELECT a+1 AS a` — the projection renames
# `a+1` to `a`, so the TopK's `a < 3` must still be rewritten to
@@ -683,7 +683,7 @@ EXPLAIN ANALYZE SELECT a + 1 AS a, b FROM topk_proj ORDER BY a LIMIT 2;
----
Plan with Metrics
01)SortExec: TopK(fetch=2), expr=[a@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[a@0 < 3], metrics=[output_rows=2, output_batches=1, row_replacements=2]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_proj.parquet]]}, projection=[CAST(a@0 AS Int64) + 1 as a, b], file_type=parquet, predicate=DynamicFilter [ CAST(a@0 AS Int64) + 1 < 3 ], sort_order_for_reorder=[a@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, metrics=[output_rows=3, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_ [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/topk_proj.parquet]]}, projection=[CAST(a@0 AS Int64) + 1 as a, b], file_type=parquet, predicate=DynamicFilter [ CAST(a@0 AS Int64) + 1 < 3 ], sort_order_for_reorder=[a@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, metrics=[output_rows=3, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_ [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -745,7 +745,7 @@ Plan with Metrics
04)----AggregateExec: mode=FinalPartitioned, gby=[a@0 as a], aggr=[min(join_agg_probe.value)], metrics=[output_rows=2, output_batches=2, spill_count=0, spilled_rows=0]
05)------RepartitionExec: partitioning=Hash([a@0], 4), input_partitions=1, metrics=[output_rows=2, output_batches=2, spill_count=0, spilled_rows=0]
06)--------AggregateExec: mode=Partial, gby=[a@0 as a], aggr=[min(join_agg_probe.value)], metrics=[output_rows=2, output_batches=1, spill_count=0, spilled_rows=0, skipped_aggregation_rows=0, reduction_factor=100% (2/2)]
-07)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/join_agg_probe.parquet]]}, projection=[a, value], file_type=parquet, predicate=DynamicFilter [ a@0 >= h1 AND a@0 <= h2 AND a@0 IN (SET) ([h1, h2]) ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= h1 AND a_null_count@1 != row_count@2 AND a_min@3 <= h2 AND (a_null_count@1 != row_count@2 AND a_min@3 <= h1 AND h1 [...]
+07)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/join_agg_probe.parquet]]}, projection=[a, value], file_type=parquet, predicate=DynamicFilter [ a@0 >= h1 AND a@0 <= h2 AND a@0 IN (SET) ([h1, h2]) ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= h1 AND a_null_count@1 != row_count@2 AND a_min@3 <= h2 AND (a_null_count@1 != row_count@2 AND a_min@3 <= h1 AND h1 [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -808,7 +808,7 @@ ON nulls_build.a = nulls_probe.a AND nulls_build.b = nulls_probe.b;
Plan with Metrics
01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0), (b@1, b@1)], metrics=[output_rows=1, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=3, input_batches=1, input_rows=1, avg_fanout=100% (1/1), probe_hit_rate=100% (1/1)]
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nulls_build.parquet]]}, projection=[a, b], file_type=parquet, metrics=[output_rows=3, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_ [...]
-03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nulls_probe.parquet]]}, projection=[a, b, c], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= 1 AND b@1 <= 2 AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:1}, {c0:,c1:2}, {c0:ab,c1:}]) ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab A [...]
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nulls_probe.parquet]]}, projection=[a, b, c], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= 1 AND b@1 <= 2 AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:1}, {c0:,c1:2}, {c0:ab,c1:}]) ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab A [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -873,8 +873,8 @@ ON lj_build.a = lj_probe.a AND lj_build.b = lj_probe.b;
----
Plan with Metrics
01)HashJoinExec: mode=CollectLeft, join_type=Left, on=[(a@0, a@0), (b@1, b@1)], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=2, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/lj_build.parquet]]}, projection=[a, b, c], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_ [...]
-03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/lj_probe.parquet]]}, projection=[a, b, e], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}, {c0:ab,c1:bb}]) ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab AND b_null_ [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/lj_build.parquet]]}, projection=[a, b, c], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_ [...]
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/lj_probe.parquet]]}, projection=[a, b, e], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}, {c0:ab,c1:bb}]) ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab AND b_null_ [...]
# LEFT SEMI JOIN: only matching build rows are returned; probe scan still
# receives the dynamic filter.
@@ -889,8 +889,8 @@ WHERE EXISTS (
----
Plan with Metrics
01)HashJoinExec: mode=CollectLeft, join_type=LeftSemi, on=[(a@0, a@0), (b@1, b@1)], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=2, input_rows=4, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/lj_build.parquet]]}, projection=[a, b, c], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_ [...]
-03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/lj_probe.parquet]]}, projection=[a, b], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}, {c0:ab,c1:bb}]) ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab AND b_null_cou [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/lj_build.parquet]]}, projection=[a, b, c], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_ [...]
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/lj_probe.parquet]]}, projection=[a, b], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}, {c0:ab,c1:bb}]) ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab AND b_null_cou [...]
statement ok
reset datafusion.explain.analyze_categories;
@@ -959,8 +959,8 @@ FROM hl_probe p INNER JOIN hl_build AS build
----
Plan with Metrics
01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0), (b@1, b@1)], projection=[a@3, b@4, c@2, e@5], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/hl_build.parquet]]}, projection=[a, b, c], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_ [...]
-03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/hl_probe.parquet]]}, projection=[a, b, e], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND hash_lookup ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab AND b_null_count@5 != row_count@2 AND b_max@4 >= ba AND b_ [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/hl_build.parquet]]}, projection=[a, b, c], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_ [...]
+03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/hl_probe.parquet]]}, projection=[a, b, e], file_type=parquet, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND hash_lookup ], dynamic_rg_pruning=eligible, pruning_predicate=a_null_count@1 != row_count@2 AND a_max@0 >= aa AND a_null_count@1 != row_count@2 AND a_min@3 <= ab AND b_null_count@5 != row_count@2 AND b_max@4 >= ba AND b_ [...]
statement ok
drop table hl_build;
@@ -1008,7 +1008,7 @@ FROM int_build b INNER JOIN int_probe p
----
Plan with Metrics
01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id1@0, id1@0), (id2@1, id2@1)], projection=[id1@0, id2@1, value@2, data@5], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)]
-02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/int_build.parquet]]}, projection=[id1, id2, value], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pr [...]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/int_build.parquet]]}, projection=[id1, id2, value], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pr [...]
03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/int_probe.parquet]]}, projection=[id1, id2, data], file_type=parquet, predicate=DynamicFilter [ id1@0 >= 1 AND id1@0 <= 2 AND id2@1 >= 10 AND id2@1 <= 20 AND hash_lookup ], dynamic_rg_pruning=eligible, pruning_predicate=id1_null_count@1 != row_count@2 AND id1_max@0 >= 1 AND id1_null_count@1 != row_count@2 AND id1_min@3 <= 2 AND id2_null_count@5 != row_count@2 AN [...]
statement ok
diff --git a/datafusion/sqllogictest/test_files/scalar.slt b/datafusion/sqllogictest/test_files/scalar.slt
index 1e4ab20dbc..35705a12a9 100644
--- a/datafusion/sqllogictest/test_files/scalar.slt
+++ b/datafusion/sqllogictest/test_files/scalar.slt
@@ -1128,7 +1128,7 @@ select arrow_typeof(round(cast(500 as decimal(10,-2)), -3)),
round(cast(400 as decimal(10,-2)), -3),
round(cast(-500 as decimal(10,-2)), -3);
----
-Decimal128(10, -3) 1000 0000 -1000
+Decimal128(10, -3) 1000 0 -1000
# round decimal with negative scale and carry-over
query TR
diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md
index 7c3e3fff67..01c6799e96 100644
--- a/docs/source/library-user-guide/upgrading/56.0.0.md
+++ b/docs/source/library-user-guide/upgrading/56.0.0.md
@@ -25,6 +25,46 @@
in this section pertains to features and changes that have already been merged
to the main branch and are awaiting release in this version.
+### Upgrade arrow/parquet to 60.0.0 and object_store to 0.14.2
+
+DataFusion 56.0.0 uses `arrow` and `parquet` 60.0.0, and `object_store` 0.14.2.
+This may require updates to your Cargo.toml if you have direct dependencies on
+these crates.
+
+See the [Arrow 60.0.0 release notes] and the [object_store 0.14.2 upgrade guide] for details on breaking changes in those versions.
+
+[arrow 60.0.0 release notes]: https://github.com/apache/arrow-rs/releases/tag/60.0.0
+[object_store 0.14.2 upgrade guide]: https://github.com/apache/arrow-rs-object-store/blob/v0.14.2/CHANGELOG.md
+
+#### Field and schema metadata now use arrow's `Metadata` type
+
+Arrow 60 introduced a dedicated [`Metadata`] type for field and schema metadata
+in place of `HashMap<String, String>`. Following that change, several DataFusion
+APIs now use `Metadata` as well, such as `DFSchema::metadata` and
+`ExprSchema::metadata` return `&Metadata`. `CastExpr::target_metadata` and
+`TryCastExpr::target_metadata` return `Option<&Metadata>`.
+
+**Migration guide:**
+
+```rust,ignore
+// Before
+let meta: &HashMap<String, String> = schema.metadata();
+let field = Field::new("a", DataType::Int64, true)
+ .with_metadata([("k".to_string(), "v".to_string())].into());
+
+// After
+let meta: &Metadata = schema.metadata();
+let field = Field::new("a", DataType::Int64, true)
+ .with_metadata(Metadata::new().with("k", "v"));
+```
+
+`Metadata` supports `.get()`, `.iter()`, `.is_empty()` and `.extend()` like
+`HashMap`, and converts to and from `HashMap<String, String>` and
+`BTreeMap<String, String>` via `From`, so most call sites need only a type
+change. Use `FieldMetadata::to_hashmap` where a `HashMap` is still required.
+
+[`metadata`]: https://docs.rs/arrow/60.0.0/arrow/datatypes/struct.Metadata.html
+
### Missing Parquet null counts are treated as unknown
DataFusion now preserves an omitted Parquet `null_count` statistic as unknown.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org