You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@auron.apache.org by ri...@apache.org on 2026/07/27 03:56:33 UTC
(auron) 01/01: Vendor third-party dependencies via setup-vendors.sh with patch series
This is an automated email from the ASF dual-hosted git repository.
richox pushed a commit to branch dev-setup-vendors
in repository https://gitbox.apache.org/repos/asf/auron.git
commit 8fd5a5ec00a699bc6e0f9a0072ed500818375fe2
Author: diliulou <wa...@kuaishou.com>
AuthorDate: Mon Jul 27 11:55:43 2026 +0800
Vendor third-party dependencies via setup-vendors.sh with patch series
Replace direct git/rev dependencies for datafusion, arrow-rs, orc-rust,
and serde_json with a vendor+patch approach:
- Add dev/vendors/setup-vendors.sh to clone upstream repos at pinned tags
and apply local patch series
- Add patch files under dev/vendors/patches/{datafusion,arrow-rs,orc-rust,serde_json}/
- Update Cargo.toml: use relative path patches to vendored packages,
exclude vendor workspaces from the main workspace
- Update .gitignore to ignore dev/vendors/packages/
- Update .rat-excludes for dev/vendors/**
- Call setup-vendors.sh all from build-native.sh before cargo build
---
.gitignore | 3 +
.rat-excludes | 2 +
Cargo.lock | 77 +--
Cargo.toml | 53 ++-
dev/mvn-build-helper/build-native.sh | 3 +
.../0001-blaze-fix-NaN-comparison-semantic.patch | 49 ++
.../arrow-rs/0002-blaze-add-get_dictionary.patch | 391 +++++++++++++++
.../0003-blaze-skip-parquet-utf-8-validating.patch | 125 +++++
...004-blaze-map-field-name-fixed-to-entries.patch | 43 ++
...x-statistics-null-count-is-negative-error.patch | 26 +
...-of-AsyncReader-ArrowReaderBuilder-to-pub.patch | 70 +++
...ad-page-header-avoid-reading-in-fragments.patch | 41 ++
...upports-processing-old-style-array-struct.patch | 49 ++
.../0001-blaze-update-arrow-dependency.patch | 35 ++
.../0002-blaze-exclude-testing-submodules.patch | 64 +++
...rts-parquet-scan-with-adaptive-batch-size.patch | 51 ++
...l-cases-when-pruning-float-column-with-Na.patch | 182 +++++++
...aze-supports-parquet-dictionary-filtering.patch | 454 ++++++++++++++++++
...allow-nullable-parquet-scan-partition-key.patch | 30 ++
.../0007-blaze-make-scatter-public.patch | 25 +
...cated-min-max-values-only-when-min-==-max.patch | 70 +++
...ll-isnotnull-pruning-for-nested-data-type.patch | 26 +
...nsitive-comparing-in-processing-parquet-f.patch | 43 ++
...1-blaze-supports-pruning-with-starts_with.patch | 86 ++++
...12-blaze-supports-short-circuiting-And-Or.patch | 529 +++++++++++++++++++++
...aze-make-parquet-reader-components-public.patch | 75 +++
...014-blaze-implement-TestMemoryExec-as-any.patch | 25 +
...tr-to-v45.0.0-for-unsupported-string-view.patch | 361 ++++++++++++++
...e-adapt-arrow-dependency-to-v55.2.0-blaze.patch | 36 ++
.../0001-blaze-ignore-invalid-escape.patch | 46 ++
...-do-not-force-escaping-control-characters.patch | 31 ++
dev/vendors/setup-vendors.sh | 162 +++++++
.../apache/spark/sql/auron/AuronConverters.scala | 6 +-
.../sql/auron/AuronSparkSessionExtension.scala | 2 +-
.../org/apache/spark/sql/auron/NativeHelper.scala | 2 +-
.../spark/sql/auron/util/TaskContextHelper.scala | 2 +-
36 files changed, 3192 insertions(+), 83 deletions(-)
diff --git a/.gitignore b/.gitignore
index df61d2fa..749cb769 100644
--- a/.gitignore
+++ b/.gitignore
@@ -65,3 +65,6 @@ dependency-reduced-pom.xml
#lsp
*.prefs
+
+# vendored third-party source trees (generated by dev/vendors/setup-vendors.sh)
+dev/vendors/packages/
diff --git a/.rat-excludes b/.rat-excludes
index 09a244b2..6c2b08e1 100644
--- a/.rat-excludes
+++ b/.rat-excludes
@@ -35,3 +35,5 @@ build/apache-maven-*/**
src/main/resources/auron-build-info.properties
**/tpcds-plan-stability/**
**/tpcds-queries/**
+dev/vendors/**
+tmp/**
diff --git a/Cargo.lock b/Cargo.lock
index c232f4a8..60f61d89 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -100,7 +100,6 @@ checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50"
[[package]]
name = "arrow"
version = "55.2.0"
-source = "git+https://github.com/auron-project/arrow-rs.git?rev=5de02520c#5de02520ca6c9f8f092b4b364906dc0507002cbb"
dependencies = [
"arrow-arith",
"arrow-array",
@@ -120,7 +119,6 @@ dependencies = [
[[package]]
name = "arrow-arith"
version = "55.2.0"
-source = "git+https://github.com/auron-project/arrow-rs.git?rev=5de02520c#5de02520ca6c9f8f092b4b364906dc0507002cbb"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -133,7 +131,6 @@ dependencies = [
[[package]]
name = "arrow-array"
version = "55.2.0"
-source = "git+https://github.com/auron-project/arrow-rs.git?rev=5de02520c#5de02520ca6c9f8f092b4b364906dc0507002cbb"
dependencies = [
"ahash",
"arrow-buffer",
@@ -149,7 +146,6 @@ dependencies = [
[[package]]
name = "arrow-buffer"
version = "55.2.0"
-source = "git+https://github.com/auron-project/arrow-rs.git?rev=5de02520c#5de02520ca6c9f8f092b4b364906dc0507002cbb"
dependencies = [
"bytes 1.12.1",
"half",
@@ -159,7 +155,6 @@ dependencies = [
[[package]]
name = "arrow-cast"
version = "55.2.0"
-source = "git+https://github.com/auron-project/arrow-rs.git?rev=5de02520c#5de02520ca6c9f8f092b4b364906dc0507002cbb"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -179,7 +174,6 @@ dependencies = [
[[package]]
name = "arrow-csv"
version = "55.2.0"
-source = "git+https://github.com/auron-project/arrow-rs.git?rev=5de02520c#5de02520ca6c9f8f092b4b364906dc0507002cbb"
dependencies = [
"arrow-array",
"arrow-cast",
@@ -193,7 +187,6 @@ dependencies = [
[[package]]
name = "arrow-data"
version = "55.2.0"
-source = "git+https://github.com/auron-project/arrow-rs.git?rev=5de02520c#5de02520ca6c9f8f092b4b364906dc0507002cbb"
dependencies = [
"arrow-buffer",
"arrow-schema",
@@ -204,7 +197,6 @@ dependencies = [
[[package]]
name = "arrow-ipc"
version = "55.2.0"
-source = "git+https://github.com/auron-project/arrow-rs.git?rev=5de02520c#5de02520ca6c9f8f092b4b364906dc0507002cbb"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -218,7 +210,6 @@ dependencies = [
[[package]]
name = "arrow-json"
version = "55.2.0"
-source = "git+https://github.com/auron-project/arrow-rs.git?rev=5de02520c#5de02520ca6c9f8f092b4b364906dc0507002cbb"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -239,7 +230,6 @@ dependencies = [
[[package]]
name = "arrow-ord"
version = "55.2.0"
-source = "git+https://github.com/auron-project/arrow-rs.git?rev=5de02520c#5de02520ca6c9f8f092b4b364906dc0507002cbb"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -251,7 +241,6 @@ dependencies = [
[[package]]
name = "arrow-row"
version = "55.2.0"
-source = "git+https://github.com/auron-project/arrow-rs.git?rev=5de02520c#5de02520ca6c9f8f092b4b364906dc0507002cbb"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -263,7 +252,6 @@ dependencies = [
[[package]]
name = "arrow-schema"
version = "55.2.0"
-source = "git+https://github.com/auron-project/arrow-rs.git?rev=5de02520c#5de02520ca6c9f8f092b4b364906dc0507002cbb"
dependencies = [
"bitflags 2.9.1",
"serde",
@@ -273,7 +261,6 @@ dependencies = [
[[package]]
name = "arrow-select"
version = "55.2.0"
-source = "git+https://github.com/auron-project/arrow-rs.git?rev=5de02520c#5de02520ca6c9f8f092b4b364906dc0507002cbb"
dependencies = [
"ahash",
"arrow-array",
@@ -286,7 +273,6 @@ dependencies = [
[[package]]
name = "arrow-string"
version = "55.2.0"
-source = "git+https://github.com/auron-project/arrow-rs.git?rev=5de02520c#5de02520ca6c9f8f092b4b364906dc0507002cbb"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -697,11 +683,12 @@ dependencies = [
[[package]]
name = "comfy-table"
-version = "7.1.4"
+version = "7.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4a65ebfec4fb190b6f90e944a817d60499ee0744e582530e2c9900a22e591d9a"
+checksum = "e0d05af1e006a2407bedef5af410552494ce5be9090444dbbcb57258c1af3d56"
dependencies = [
- "unicode-segmentation",
+ "strum",
+ "strum_macros",
"unicode-width",
]
@@ -830,7 +817,6 @@ dependencies = [
[[package]]
name = "datafusion"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"arrow",
"arrow-ipc",
@@ -884,7 +870,6 @@ dependencies = [
[[package]]
name = "datafusion-catalog"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"arrow",
"async-trait",
@@ -909,7 +894,6 @@ dependencies = [
[[package]]
name = "datafusion-catalog-listing"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"arrow",
"async-trait",
@@ -931,7 +915,6 @@ dependencies = [
[[package]]
name = "datafusion-common"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"ahash",
"arrow",
@@ -956,7 +939,6 @@ dependencies = [
[[package]]
name = "datafusion-common-runtime"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"futures",
"log",
@@ -966,7 +948,6 @@ dependencies = [
[[package]]
name = "datafusion-datasource"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"arrow",
"async-compression",
@@ -1001,7 +982,6 @@ dependencies = [
[[package]]
name = "datafusion-datasource-csv"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"arrow",
"async-trait",
@@ -1025,7 +1005,6 @@ dependencies = [
[[package]]
name = "datafusion-datasource-json"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"arrow",
"async-trait",
@@ -1049,7 +1028,6 @@ dependencies = [
[[package]]
name = "datafusion-datasource-parquet"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"arrow",
"async-trait",
@@ -1081,12 +1059,10 @@ dependencies = [
[[package]]
name = "datafusion-doc"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
[[package]]
name = "datafusion-execution"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"arrow",
"dashmap",
@@ -1104,7 +1080,6 @@ dependencies = [
[[package]]
name = "datafusion-expr"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"arrow",
"async-trait",
@@ -1125,7 +1100,6 @@ dependencies = [
[[package]]
name = "datafusion-expr-common"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"arrow",
"datafusion-common",
@@ -1247,7 +1221,6 @@ dependencies = [
[[package]]
name = "datafusion-functions"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"arrow",
"arrow-buffer",
@@ -1275,7 +1248,6 @@ dependencies = [
[[package]]
name = "datafusion-functions-aggregate"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"ahash",
"arrow",
@@ -1295,7 +1267,6 @@ dependencies = [
[[package]]
name = "datafusion-functions-aggregate-common"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"ahash",
"arrow",
@@ -1307,7 +1278,6 @@ dependencies = [
[[package]]
name = "datafusion-functions-nested"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"arrow",
"arrow-ord",
@@ -1328,7 +1298,6 @@ dependencies = [
[[package]]
name = "datafusion-functions-table"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"arrow",
"async-trait",
@@ -1343,7 +1312,6 @@ dependencies = [
[[package]]
name = "datafusion-functions-window"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"arrow",
"datafusion-common",
@@ -1360,7 +1328,6 @@ dependencies = [
[[package]]
name = "datafusion-functions-window-common"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"datafusion-common",
"datafusion-physical-expr-common",
@@ -1369,7 +1336,6 @@ dependencies = [
[[package]]
name = "datafusion-macros"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"datafusion-expr",
"quote",
@@ -1379,7 +1345,6 @@ dependencies = [
[[package]]
name = "datafusion-optimizer"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"arrow",
"chrono",
@@ -1398,7 +1363,6 @@ dependencies = [
[[package]]
name = "datafusion-physical-expr"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"ahash",
"arrow",
@@ -1419,7 +1383,6 @@ dependencies = [
[[package]]
name = "datafusion-physical-expr-common"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"ahash",
"arrow",
@@ -1432,7 +1395,6 @@ dependencies = [
[[package]]
name = "datafusion-physical-optimizer"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"arrow",
"datafusion-common",
@@ -1451,7 +1413,6 @@ dependencies = [
[[package]]
name = "datafusion-physical-plan"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"ahash",
"arrow",
@@ -1480,7 +1441,6 @@ dependencies = [
[[package]]
name = "datafusion-pruning"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"arrow",
"arrow-schema",
@@ -1500,7 +1460,6 @@ dependencies = [
[[package]]
name = "datafusion-session"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"arrow",
"async-trait",
@@ -1523,7 +1482,6 @@ dependencies = [
[[package]]
name = "datafusion-spark"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"arrow",
"datafusion-catalog",
@@ -1538,7 +1496,6 @@ dependencies = [
[[package]]
name = "datafusion-sql"
version = "49.0.0"
-source = "git+https://github.com/auron-project/datafusion.git?rev=9034aeffb#9034aeffbc280ab11e81f670efec4e7217dc357a"
dependencies = [
"arrow",
"bigdecimal",
@@ -2777,8 +2734,7 @@ checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50"
[[package]]
name = "orc-rust"
-version = "0.7.0"
-source = "git+https://github.com/auron-project/datafusion-orc.git?rev=9beb12c#9beb12cbff3de5a6b8b4f6e4f4b602f9a762c9c7"
+version = "0.7.1"
dependencies = [
"arrow",
"async-trait",
@@ -2790,8 +2746,6 @@ dependencies = [
"flate2",
"futures",
"futures-util",
- "iana-time-zone",
- "log",
"lz4_flex 0.11.5",
"lzokay-native",
"num",
@@ -2843,7 +2797,6 @@ dependencies = [
[[package]]
name = "parquet"
version = "55.2.0"
-source = "git+https://github.com/auron-project/arrow-rs.git?rev=5de02520c#5de02520ca6c9f8f092b4b364906dc0507002cbb"
dependencies = [
"ahash",
"arrow-array",
@@ -3702,7 +3655,6 @@ dependencies = [
[[package]]
name = "serde_json"
version = "1.0.96"
-source = "git+https://github.com/auron-project/json?rev=95fa6cb#95fa6cbac6b7146a8883f5ce761316d11282399b"
dependencies = [
"itoa",
"ryu",
@@ -3935,6 +3887,25 @@ version = "0.2.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fe895eb47f22e2ddd4dabc02bce419d2e643c8e3b585c78158b349195bc24d82"
+[[package]]
+name = "strum"
+version = "0.26.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8fec0f0aef304996cf250b31b5a10dee7980c85da9d759361292b8bca5a18f06"
+
+[[package]]
+name = "strum_macros"
+version = "0.26.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4c6bee85a5a24955dc440386795aa378cd9cf82acd5f764469152d2270e581be"
+dependencies = [
+ "heck",
+ "proc-macro2",
+ "quote",
+ "rustversion",
+ "syn 2.0.104",
+]
+
[[package]]
name = "subtle"
version = "2.6.1"
diff --git a/Cargo.toml b/Cargo.toml
index aff57e00..ae9f44f4 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -27,6 +27,11 @@ members = [
"native-engine/auron-planner",
"native-engine/auron-memmgr",
]
+exclude = [
+ "dev/vendors/packages/datafusion",
+ "dev/vendors/packages/arrow-rs",
+ "dev/vendors/packages/serde_json",
+]
[workspace.package]
version = "8.0.0"
@@ -198,31 +203,31 @@ rdkafka = { version = "0.36.0", features = ["tokio"] }
[patch.crates-io]
# datafusion: branch=v49.0.0-blaze
-datafusion = { git = "https://github.com/auron-project/datafusion.git", rev = "9034aeffb"}
-datafusion-common = { git = "https://github.com/auron-project/datafusion.git", rev = "9034aeffb"}
-datafusion-expr = { git = "https://github.com/auron-project/datafusion.git", rev = "9034aeffb"}
-datafusion-datasource = { git = "https://github.com/auron-project/datafusion.git", rev = "9034aeffb"}
-datafusion-datasource-parquet = { git = "https://github.com/auron-project/datafusion.git", rev = "9034aeffb"}
-datafusion-execution = { git = "https://github.com/auron-project/datafusion.git", rev = "9034aeffb"}
-datafusion-optimizer = { git = "https://github.com/auron-project/datafusion.git", rev = "9034aeffb"}
-datafusion-physical-expr = { git = "https://github.com/auron-project/datafusion.git", rev = "9034aeffb"}
-datafusion-spark = { git = "https://github.com/auron-project/datafusion.git", rev = "9034aeffb"}
-orc-rust = { git = "https://github.com/auron-project/datafusion-orc.git", rev = "9beb12c"}
+datafusion = { path = "./dev/vendors/packages/datafusion/datafusion/core"}
+datafusion-common = { path = "./dev/vendors/packages/datafusion/datafusion/common"}
+datafusion-expr = { path = "./dev/vendors/packages/datafusion/datafusion/expr"}
+datafusion-datasource = { path = "./dev/vendors/packages/datafusion/datafusion/datasource"}
+datafusion-datasource-parquet = { path = "./dev/vendors/packages/datafusion/datafusion/datasource-parquet"}
+datafusion-execution = { path = "./dev/vendors/packages/datafusion/datafusion/execution"}
+datafusion-optimizer = { path = "./dev/vendors/packages/datafusion/datafusion/optimizer"}
+datafusion-physical-expr = { path = "./dev/vendors/packages/datafusion/datafusion/physical-expr"}
+datafusion-spark = { path = "./dev/vendors/packages/datafusion/datafusion/spark"}
+orc-rust = { path = "./dev/vendors/packages/orc-rust"}
# arrow: branch=v55.2.0-blaze
-arrow = { git = "https://github.com/auron-project/arrow-rs.git", rev = "5de02520c"}
-arrow-arith = { git = "https://github.com/auron-project/arrow-rs.git", rev = "5de02520c"}
-arrow-array = { git = "https://github.com/auron-project/arrow-rs.git", rev = "5de02520c"}
-arrow-buffer = { git = "https://github.com/auron-project/arrow-rs.git", rev = "5de02520c"}
-arrow-cast = { git = "https://github.com/auron-project/arrow-rs.git", rev = "5de02520c"}
-arrow-data = { git = "https://github.com/auron-project/arrow-rs.git", rev = "5de02520c"}
-arrow-ipc = { git = "https://github.com/auron-project/arrow-rs.git", rev = "5de02520c"}
-arrow-ord = { git = "https://github.com/auron-project/arrow-rs.git", rev = "5de02520c"}
-arrow-row = { git = "https://github.com/auron-project/arrow-rs.git", rev = "5de02520c"}
-arrow-schema = { git = "https://github.com/auron-project/arrow-rs.git", rev = "5de02520c"}
-arrow-select = { git = "https://github.com/auron-project/arrow-rs.git", rev = "5de02520c"}
-arrow-string = { git = "https://github.com/auron-project/arrow-rs.git", rev = "5de02520c"}
-parquet = { git = "https://github.com/auron-project/arrow-rs.git", rev = "5de02520c"}
+arrow = { path = "./dev/vendors/packages/arrow-rs/arrow"}
+arrow-arith = { path = "./dev/vendors/packages/arrow-rs/arrow-arith"}
+arrow-array = { path = "./dev/vendors/packages/arrow-rs/arrow-array"}
+arrow-buffer = { path = "./dev/vendors/packages/arrow-rs/arrow-buffer"}
+arrow-cast = { path = "./dev/vendors/packages/arrow-rs/arrow-cast"}
+arrow-data = { path = "./dev/vendors/packages/arrow-rs/arrow-data"}
+arrow-ipc = { path = "./dev/vendors/packages/arrow-rs/arrow-ipc"}
+arrow-ord = { path = "./dev/vendors/packages/arrow-rs/arrow-ord"}
+arrow-row = { path = "./dev/vendors/packages/arrow-rs/arrow-row"}
+arrow-schema = { path = "./dev/vendors/packages/arrow-rs/arrow-schema"}
+arrow-select = { path = "./dev/vendors/packages/arrow-rs/arrow-select"}
+arrow-string = { path = "./dev/vendors/packages/arrow-rs/arrow-string"}
+parquet = { path = "./dev/vendors/packages/arrow-rs/parquet"}
# serde_json: branch=v1.0.96-blaze
-serde_json = { git = "https://github.com/auron-project/json", rev = "95fa6cb" }
+serde_json = { path = "./dev/vendors/packages/serde_json" }
diff --git a/dev/mvn-build-helper/build-native.sh b/dev/mvn-build-helper/build-native.sh
index 84e06c77..89b6838a 100755
--- a/dev/mvn-build-helper/build-native.sh
+++ b/dev/mvn-build-helper/build-native.sh
@@ -25,6 +25,9 @@ _CALLING_DIR="$(pwd)"
PROJECT_DIR="$(cd "`dirname "$0"`/../.."; pwd)"
cd "$PROJECT_DIR"
+# Ensure vendored third-party source trees are ready before building.
+./dev/vendors/setup-vendors.sh all
+
profile="$1"
features_arg=""
if [ -n "$2" ]; then
diff --git a/dev/vendors/patches/arrow-rs/0001-blaze-fix-NaN-comparison-semantic.patch b/dev/vendors/patches/arrow-rs/0001-blaze-fix-NaN-comparison-semantic.patch
new file mode 100644
index 00000000..2007c049
--- /dev/null
+++ b/dev/vendors/patches/arrow-rs/0001-blaze-fix-NaN-comparison-semantic.patch
@@ -0,0 +1,49 @@
+From ed4ace347da4acfe0c3453e1b34ede067de894e2 Mon Sep 17 00:00:00 2001
+From: zhangli20 <zh...@kuaishou.com>
+Date: Sun, 15 Sep 2024 20:54:37 +0800
+Subject: [PATCH] blaze: fix NaN comparison semantic
+
+---
+ arrow-array/src/arithmetic.rs | 19 ++++++++++++++++++-
+ 1 file changed, 18 insertions(+), 1 deletion(-)
+
+diff --git a/arrow-array/src/arithmetic.rs b/arrow-array/src/arithmetic.rs
+index b5f4a106f..c4e88b91a 100644
+--- a/arrow-array/src/arithmetic.rs
++++ b/arrow-array/src/arithmetic.rs
+@@ -390,7 +390,18 @@ macro_rules! native_type_float_op {
+
+ #[inline]
+ fn compare(self, rhs: Self) -> Ordering {
+- <$t>::total_cmp(&self, &rhs)
++ // blaze:
++ // nan-safe version according to semantics where NaN == NaN and
++ // NaN is greater than any non-NaN double.
++ if self.is_nan() && rhs.is_nan() {
++ Ordering::Equal
++ } else if self.is_nan() {
++ Ordering::Greater
++ } else if rhs.is_nan() {
++ Ordering::Less
++ } else {
++ <$t>::total_cmp(&self, &rhs)
++ }
+ }
+
+ #[inline]
+@@ -398,6 +409,12 @@ macro_rules! native_type_float_op {
+ // Equivalent to `self.total_cmp(&rhs).is_eq()`
+ // but LLVM isn't able to realise this is bitwise equality
+ // https://rust.godbolt.org/z/347nWGxoW
++ // blaze:
++ // nan-safe version according to semantics where NaN == NaN and
++ // NaN is greater than any non-NaN double.
++ if self.is_nan() && rhs.is_nan() {
++ return true;
++ }
+ self.to_bits() == rhs.to_bits()
+ }
+ }
+--
+2.50.1
+
diff --git a/dev/vendors/patches/arrow-rs/0002-blaze-add-get_dictionary.patch b/dev/vendors/patches/arrow-rs/0002-blaze-add-get_dictionary.patch
new file mode 100644
index 00000000..f77c615a
--- /dev/null
+++ b/dev/vendors/patches/arrow-rs/0002-blaze-add-get_dictionary.patch
@@ -0,0 +1,391 @@
+From 83e2918853b9519b669a3143293943df9626fd6b Mon Sep 17 00:00:00 2001
+From: zhangli20 <zh...@kuaishou.com>
+Date: Sun, 15 Sep 2024 21:00:43 +0800
+Subject: [PATCH] blaze: add get_dictionary
+
+adgsg
+---
+ parquet/src/arrow/array_reader/mod.rs | 2 +-
+ parquet/src/arrow/async_reader/mod.rs | 12 +-
+ parquet/src/arrow/mod.rs | 4 +-
+ parquet/src/blaze.rs | 274 ++++++++++++++++++++++++++
+ parquet/src/column/reader.rs | 2 +-
+ parquet/src/lib.rs | 4 +
+ 6 files changed, 293 insertions(+), 5 deletions(-)
+ create mode 100644 parquet/src/blaze.rs
+
+diff --git a/parquet/src/arrow/array_reader/mod.rs b/parquet/src/arrow/array_reader/mod.rs
+index 94d61c9ea..06b06c9de 100644
+--- a/parquet/src/arrow/array_reader/mod.rs
++++ b/parquet/src/arrow/array_reader/mod.rs
+@@ -30,7 +30,7 @@ use crate::column::reader::decoder::ColumnValueDecoder;
+ use crate::file::reader::{FilePageIterator, FileReader};
+
+ mod builder;
+-mod byte_array;
++pub mod byte_array;
+ mod byte_array_dictionary;
+ mod byte_view_array;
+ mod empty_array;
+diff --git a/parquet/src/arrow/async_reader/mod.rs b/parquet/src/arrow/async_reader/mod.rs
+index 611d6999e..248de5be3 100644
+--- a/parquet/src/arrow/async_reader/mod.rs
++++ b/parquet/src/arrow/async_reader/mod.rs
+@@ -118,6 +118,12 @@ pub trait AsyncFileReader: Send {
+ &'a mut self,
+ options: Option<&'a ArrowReaderOptions>,
+ ) -> BoxFuture<'a, Result<Arc<ParquetMetaData>>>;
++
++ /// Sync version of get_bytes
++ /// this is only used by blaze, for reading dictionary values for row group pruning
++ fn get_bytes_sync(&mut self, range: Range<u64>) -> Result<Bytes> {
++ unimplemented!("blaze only")
++ }
+ }
+
+ /// This allows Box<dyn AsyncFileReader + '_> to be used as an AsyncFileReader,
+@@ -136,6 +142,10 @@ impl AsyncFileReader for Box<dyn AsyncFileReader + '_> {
+ ) -> BoxFuture<'a, Result<Arc<ParquetMetaData>>> {
+ self.as_mut().get_metadata(options)
+ }
++
++ fn get_bytes_sync(&mut self, range: Range<u64>) -> Result<Bytes> {
++ self.as_mut().get_bytes_sync(range)
++ }
+ }
+
+ impl<T: AsyncFileReader + MetadataFetch + AsyncRead + AsyncSeek + Unpin> MetadataSuffixFetch for T {
+@@ -555,7 +565,7 @@ struct ReaderFactory<T> {
+ /// Optional filter
+ filter: Option<RowFilter>,
+
+- /// Limit to apply to remaining row groups.
++ /// Limit to apply to remaining row groups.
+ limit: Option<usize>,
+
+ /// Offset to apply to the next
+diff --git a/parquet/src/arrow/mod.rs b/parquet/src/arrow/mod.rs
+index e33d6a05a..4afcb9464 100644
+--- a/parquet/src/arrow/mod.rs
++++ b/parquet/src/arrow/mod.rs
+@@ -179,10 +179,10 @@
+ //! assert_eq!(50, record_batch.num_rows());
+ //! ```
+
+-experimental!(mod array_reader);
++experimental!(pub mod array_reader);
+ pub mod arrow_reader;
+ pub mod arrow_writer;
+-mod buffer;
++pub mod buffer;
+ mod decoder;
+
+ #[cfg(feature = "async")]
+diff --git a/parquet/src/blaze.rs b/parquet/src/blaze.rs
+new file mode 100644
+index 000000000..88ab30d1e
+--- /dev/null
++++ b/parquet/src/blaze.rs
+@@ -0,0 +1,274 @@
++use std::io::Read;
++use std::sync::{Arc, Mutex};
++use bytes::Bytes;
++use zstd::zstd_safe::WriteBuf;
++use arrow_array::*;
++use arrow_schema::DataType;
++use crate::arrow::async_reader::AsyncFileReader;
++use crate::arrow::array_reader::byte_array::ByteArrayDecoderPlain;
++use crate::arrow::buffer::offset_buffer::OffsetBuffer;
++use crate::basic::{Encoding, PageType};
++use crate::column::page::{Page, PageReader};
++use crate::column::reader::{ColumnReader, get_column_reader};
++use crate::data_type::*;
++use crate::encodings::decoding::{Decoder, PlainDecoder};
++use crate::file::metadata::RowGroupMetaData;
++use crate::file::reader::{ChunkReader, Length, SerializedPageReader};
++
++pub fn get_dictionary_for_pruning<T: AsyncFileReader + Send + 'static>(
++ input: &mut T,
++ row_group_metadata: &RowGroupMetaData,
++ col_idx: usize,
++) -> crate::errors::Result<Option<ArrayRef>> {
++
++ let column_metadata = row_group_metadata.column(col_idx);
++
++ // check whether dictionary page exists
++ if column_metadata
++ .page_encoding_stats()
++ .iter()
++ .flat_map(|stats| stats.iter())
++ .all(|stat| stat.page_type != PageType::DICTIONARY_PAGE)
++ {
++ // no dictionary page
++ return Ok(None);
++ }
++
++ // check whether dictionary encoded page exists, if so, cannot use for pruning
++ if column_metadata
++ .page_encoding_stats()
++ .iter()
++ .flat_map(|stats| stats.iter())
++ .any(|stat| {
++ !matches!(stat.encoding, Encoding::PLAIN_DICTIONARY | Encoding::RLE_DICTIONARY)
++ })
++ {
++ // at least one non-dictionary encoding is present
++ return Ok(None);
++ }
++
++ // check whether non-dictionary encoding exists in column meta, if so, cannot use for pruning
++ if column_metadata.encodings().contains(&Encoding::PLAIN_DICTIONARY) {
++ if column_metadata
++ .encodings()
++ .iter()
++ .any(|enc| {
++ !matches!(enc, Encoding::PLAIN_DICTIONARY | Encoding::RLE | Encoding::BIT_PACKED)
++ })
++ {
++ // if remove returned true, PLAIN_DICTIONARY was present, which means at
++ // least one page was dictionary encoded and 1.0 encodings are used
++
++ // RLE and BIT_PACKED are only used for repetition or definition levels
++ return Ok(None);
++ }
++ } else {
++ // if PLAIN_DICTIONARY wasn't present, then either the column is not
++ // dictionary-encoded, or the 2.0 encoding, RLE_DICTIONARY, was used.
++ // for 2.0, this cannot determine whether a page fell back without
++ // page encoding stats
++ return Ok(None);
++ }
++
++ let mut page_reader: Box<dyn PageReader> = Box::new(SerializedPageReader::new(
++ Arc::new(WrappedChunkReader {
++ inner: Arc::new(Mutex::new(input)),
++ start: column_metadata.byte_range().0,
++ len: column_metadata.byte_range().1,
++ }),
++ column_metadata,
++ row_group_metadata.num_rows() as usize,
++ None,
++ )?);
++ let page_metadata = page_reader.peek_next_page()?;
++ if !page_metadata.map(|metadata| metadata.is_dict).unwrap_or(false) {
++ return Ok(None);
++ }
++
++ macro_rules! get_prim_dict {
++ ($r:expr, $parquet_type:ty, $array_type:ty) => {{
++ match $r.page_reader.get_next_page()? {
++ Some(Page::DictionaryPage { buf, num_values, .. }) => {
++ let mut dict_values = vec![Default::default(); num_values as usize];
++ let mut dict_decoder = PlainDecoder::<$parquet_type>::new(
++ column_metadata.column_descr().type_length());
++ dict_decoder.set_data(buf, num_values as usize)?;
++ dict_decoder.get(&mut dict_values)?;
++ Ok(Some(
++ Arc::new(<$array_type>::from_iter_values(dict_values))
++ ))
++ }
++ _ => {
++ Ok(None)
++ }
++ }
++ }}
++ }
++
++ // safety:
++ // bypass lifetime checking.
++ // page_reader is guarenteed not to be used outside its lifetime.
++ let unsafe_page_reader = unsafe {
++ std::mem::transmute::<_, Box<dyn PageReader>>(page_reader)
++ };
++ match get_column_reader(column_metadata.column_descr_ptr(), unsafe_page_reader) {
++ ColumnReader::BoolColumnReader(_) => {
++ return Ok(None);
++ },
++ ColumnReader::Int32ColumnReader(mut r) => {
++ return get_prim_dict!(r, Int32Type, Int32Array);
++ },
++ ColumnReader::Int64ColumnReader(mut r) => {
++ return get_prim_dict!(r, Int64Type, Int64Array);
++ },
++ ColumnReader::Int96ColumnReader(_) => {
++ return Ok(None);
++ },
++ ColumnReader::FloatColumnReader(mut r) => {
++ return get_prim_dict!(r, FloatType, Float32Array);
++ },
++ ColumnReader::DoubleColumnReader(mut r) => {
++ return get_prim_dict!(r, DoubleType, Float64Array);
++ },
++ ColumnReader::ByteArrayColumnReader(mut r) => {
++ match r.page_reader.get_next_page()? {
++ Some(Page::DictionaryPage { buf, num_values, .. }) => {
++ let mut buffer = OffsetBuffer::<i32>::default();
++ let mut decoder = ByteArrayDecoderPlain::new(
++ buf,
++ num_values as usize,
++ Some(num_values as usize),
++ false,
++ );
++ decoder.read(&mut buffer, usize::MAX)?;
++ return Ok(Some(buffer.into_array(None, DataType::Binary)));
++ }
++ _ => return Ok(None),
++ }
++ },
++ ColumnReader::FixedLenByteArrayColumnReader(_) => {
++ return Ok(None);
++ }
++ }
++}
++
++struct WrappedChunkReader<'a, T: AsyncFileReader + Send + 'static> {
++ inner: Arc<Mutex<& 'a mut T>>,
++ start: u64,
++ len: u64,
++}
++unsafe impl<T: AsyncFileReader + Send + 'static> Sync for WrappedChunkReader<'_, T> {}
++
++impl <T: AsyncFileReader + Send + 'static> Read for WrappedChunkReader<'_, T> {
++ fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
++ buf.copy_from_slice(self.get_bytes(0, buf.len())?.as_slice());
++ self.start += buf.len() as u64;
++ Ok(buf.len())
++ }
++}
++
++impl<T: AsyncFileReader + Send + 'static> Length for WrappedChunkReader<'_, T> {
++ fn len(&self) -> u64 {
++ self.len
++ }
++}
++impl <T: AsyncFileReader + Send + 'static> ChunkReader for WrappedChunkReader<'_, T> {
++ type T = Self;
++
++ fn get_read(&self, start: u64) -> crate::errors::Result<Self::T> {
++ Ok(Self {
++ inner: self.inner.clone(),
++ start,
++ len: self.len,
++ })
++ }
++
++ fn get_bytes(&self, start: u64, length: usize) -> crate::errors::Result<Bytes> {
++ let start = self.start + start;
++ let end = start + length as u64;
++ let mut inner = self.inner.lock().unwrap();
++ Ok(inner.get_bytes_sync(start..end)?)
++ }
++}
++
++#[cfg(test)]
++mod test {
++ use std::fs::File;
++ use std::io::{Read, Seek, SeekFrom};
++ use std::ops::Range;
++ use std::sync::Arc;
++ use bytes::Bytes;
++ use futures::future::BoxFuture;
++ use arrow_array::Array;
++ use arrow_array::cast::as_string_array;
++ use arrow_cast::cast;
++ use arrow_schema::DataType;
++ use crate::arrow::arrow_reader::ArrowReaderOptions;
++ use crate::arrow::async_reader::AsyncFileReader;
++ use crate::arrow::ParquetRecordBatchStreamBuilder;
++ use crate::blaze::get_dictionary_for_pruning;
++ use crate::errors::ParquetError;
++ use crate::file::footer::{decode_footer, decode_metadata};
++ use crate::file::FOOTER_SIZE;
++ use crate::file::metadata::ParquetMetaData;
++ use crate::file::reader::ChunkReader;
++
++ struct AsyncLocalFileReader(File);
++ impl AsyncFileReader for AsyncLocalFileReader {
++ fn get_bytes(&mut self, range: Range<u64>) -> BoxFuture<'_, crate::errors::Result<Bytes>> {
++ Box::pin(async move {
++ Ok(self.0.get_bytes(range.start, (range.end - range.start) as usize)?)
++ })
++ }
++
++ fn get_metadata<'a>(
++ &'a mut self,
++ options: Option<&'a ArrowReaderOptions>,
++ ) -> BoxFuture<'a, Result<Arc<ParquetMetaData>, ParquetError>> {
++ Box::pin(async move {
++ self.0.seek(SeekFrom::End(-(FOOTER_SIZE as i64)))?;
++
++ let mut buf = [0_u8; FOOTER_SIZE];
++ self.0.read_exact(&mut buf)?;
++
++ let metadata_len = decode_footer(&buf)?;
++ self.0.seek(SeekFrom::End(-(FOOTER_SIZE as i64) - metadata_len as i64))?;
++
++ let mut buf = Vec::with_capacity(metadata_len);
++ self.0.try_clone()?.take(metadata_len as _).read_to_end(&mut buf)?;
++
++ let metadata = decode_metadata(&buf)?;
++ Ok(Arc::new(metadata))
++ })
++ }
++ }
++
++ #[tokio::test]
++ async fn test_get_dictionary() -> crate::errors::Result<()> {
++ let input = AsyncLocalFileReader(
++ File::open("/Volumes/Workspace/blaze-init/arrow-rs/with-dict-2.parquet")?
++ );
++ let mut builder = ParquetRecordBatchStreamBuilder::new_with_options(
++ input,
++ ArrowReaderOptions::new().with_page_index(true),
++ ).await?;
++
++ let col_idx = builder
++ .schema()
++ .fields()
++ .iter()
++ .position(|f| f.name().eq_ignore_ascii_case("status"))
++ .unwrap();
++
++ let dict = get_dictionary_for_pruning(&mut builder.input.0, builder.metadata.row_group(0), col_idx)
++ .unwrap()
++ .expect("no dictionary");
++ let dict = cast(&dict, &DataType::Utf8)?;
++ let dict_strs = as_string_array(&dict);
++
++ eprintln!("got dictionary: {:?}", dict_strs);
++ eprintln!("dict bytes: {}", dict_strs.offsets()[dict_strs.len()] - dict_strs.offsets()[0]);
++ assert!(!dict.is_empty());
++ Ok(())
++ }
++}
+diff --git a/parquet/src/column/reader.rs b/parquet/src/column/reader.rs
+index b69980578..9242f2232 100644
+--- a/parquet/src/column/reader.rs
++++ b/parquet/src/column/reader.rs
+@@ -114,7 +114,7 @@ pub type ColumnReaderImpl<T> = GenericColumnReader<
+ pub struct GenericColumnReader<R, D, V> {
+ descr: ColumnDescPtr,
+
+- page_reader: Box<dyn PageReader>,
++ pub page_reader: Box<dyn PageReader>,
+
+ /// The total number of values stored in the data page.
+ num_buffered_values: usize,
+diff --git a/parquet/src/lib.rs b/parquet/src/lib.rs
+index f814ddeb0..b69d85ba8 100644
+--- a/parquet/src/lib.rs
++++ b/parquet/src/lib.rs
+@@ -173,3 +173,7 @@ pub mod record;
+ pub mod schema;
+
+ pub mod thrift;
++
++#[cfg(feature = "arrow")]
++#[cfg(feature = "async")]
++pub mod blaze;
+--
+2.50.1
+
diff --git a/dev/vendors/patches/arrow-rs/0003-blaze-skip-parquet-utf-8-validating.patch b/dev/vendors/patches/arrow-rs/0003-blaze-skip-parquet-utf-8-validating.patch
new file mode 100644
index 00000000..e26623ec
--- /dev/null
+++ b/dev/vendors/patches/arrow-rs/0003-blaze-skip-parquet-utf-8-validating.patch
@@ -0,0 +1,125 @@
+From 0ed0f11556f78301050d71d4c40b77d46ecba52a Mon Sep 17 00:00:00 2001
+From: zhangli20 <zh...@kuaishou.com>
+Date: Sun, 15 Sep 2024 21:14:08 +0800
+Subject: [PATCH] blaze: skip parquet utf-8 validating
+
+---
+ arrow-array/src/types.rs | 40 +++++++++++------------
+ arrow-data/src/data.rs | 4 +--
+ parquet/src/arrow/buffer/offset_buffer.rs | 27 ++++++++-------
+ 3 files changed, 37 insertions(+), 34 deletions(-)
+
+diff --git a/arrow-array/src/types.rs b/arrow-array/src/types.rs
+index 3d8cfcdb1..605dc10d6 100644
+--- a/arrow-array/src/types.rs
++++ b/arrow-array/src/types.rs
+@@ -1409,26 +1409,26 @@ impl<O: OffsetSizeTrait> ByteArrayType for GenericStringType<O> {
+ };
+
+ fn validate(offsets: &OffsetBuffer<Self::Offset>, values: &Buffer) -> Result<(), ArrowError> {
+- // Verify that the slice as a whole is valid UTF-8
+- let validated = std::str::from_utf8(values).map_err(|e| {
+- ArrowError::InvalidArgumentError(format!("Encountered non UTF-8 data: {e}"))
+- })?;
+-
+- // Verify each offset is at a valid character boundary in this UTF-8 array
+- for offset in offsets.iter() {
+- let o = offset.as_usize();
+- if !validated.is_char_boundary(o) {
+- if o < validated.len() {
+- return Err(ArrowError::InvalidArgumentError(format!(
+- "Split UTF-8 codepoint at offset {o}"
+- )));
+- }
+- return Err(ArrowError::InvalidArgumentError(format!(
+- "Offset of {o} exceeds length of values {}",
+- validated.len()
+- )));
+- }
+- }
++ // // Verify that the slice as a whole is valid UTF-8
++ // let validated = std::str::from_utf8(values).map_err(|e| {
++ // ArrowError::InvalidArgumentError(format!("Encountered non UTF-8 data: {e}"))
++ // })?;
++
++ // // Verify each offset is at a valid character boundary in this UTF-8 array
++ // for offset in offsets.iter() {
++ // let o = offset.as_usize();
++ // if !validated.is_char_boundary(o) {
++ // if o < validated.len() {
++ // return Err(ArrowError::InvalidArgumentError(format!(
++ // "Split UTF-8 codepoint at offset {o}"
++ // )));
++ // }
++ // return Err(ArrowError::InvalidArgumentError(format!(
++ // "Offset of {o} exceeds length of values {}",
++ // validated.len()
++ // )));
++ // }
++ // }
+ Ok(())
+ }
+ }
+diff --git a/arrow-data/src/data.rs b/arrow-data/src/data.rs
+index 4c117184d..1d7a39e9d 100644
+--- a/arrow-data/src/data.rs
++++ b/arrow-data/src/data.rs
+@@ -1300,8 +1300,8 @@ impl ArrayData {
+ /// 1. Union type_ids are valid see [#85](https://github.com/apache/arrow-rs/issues/85)
+ pub fn validate_values(&self) -> Result<(), ArrowError> {
+ match &self.data_type {
+- DataType::Utf8 => self.validate_utf8::<i32>(),
+- DataType::LargeUtf8 => self.validate_utf8::<i64>(),
++ DataType::Utf8 => /* self.validate_utf8::<i32>() */ Ok(()),
++ DataType::LargeUtf8 => /* self.validate_utf8::<i64>() */ Ok(()),
+ DataType::Binary => self.validate_offsets_full::<i32>(self.buffers[1].len()),
+ DataType::LargeBinary => self.validate_offsets_full::<i64>(self.buffers[1].len()),
+ DataType::BinaryView => {
+diff --git a/parquet/src/arrow/buffer/offset_buffer.rs b/parquet/src/arrow/buffer/offset_buffer.rs
+index 5051dce12..802df6ed0 100644
+--- a/parquet/src/arrow/buffer/offset_buffer.rs
++++ b/parquet/src/arrow/buffer/offset_buffer.rs
+@@ -60,17 +60,18 @@ impl<I: OffsetSizeTrait> OffsetBuffer<I> {
+ /// UTF-8. This should be done by calling [`Self::check_valid_utf8`] after
+ /// all data has been written
+ pub fn try_push(&mut self, data: &[u8], validate_utf8: bool) -> Result<()> {
+- if validate_utf8 {
+- if let Some(&b) = data.first() {
+- // A valid code-point iff it does not start with 0b10xxxxxx
+- // Bit-magic taken from `std::str::is_char_boundary`
+- if (b as i8) < -0x40 {
+- return Err(ParquetError::General(
+- "encountered non UTF-8 data".to_string(),
+- ));
+- }
+- }
+- }
++ // skip utf-8 validating
++ // if validate_utf8 {
++ // if let Some(&b) = data.first() {
++ // // A valid code-point iff it does not start with 0b10xxxxxx
++ // // Bit-magic taken from `std::str::is_char_boundary`
++ // if (b as i8) < -0x40 {
++ // return Err(ParquetError::General(
++ // "encountered non UTF-8 data".to_string(),
++ // ));
++ // }
++ // }
++ // }
+
+ self.values.extend_from_slice(data);
+
+@@ -118,7 +119,9 @@ impl<I: OffsetSizeTrait> OffsetBuffer<I> {
+ ///
+ /// [`Self::try_push`] can perform this validation check on insertion
+ pub fn check_valid_utf8(&self, start_offset: usize) -> Result<()> {
+- check_valid_utf8(&self.values.as_slice()[start_offset..])
++ // blaze: ignore utf8 validation
++ // check_valid_utf8(&self.values.as_slice()[start_offset..])
++ Ok(())
+ }
+
+ /// Converts this into an [`ArrayRef`] with the provided `data_type` and `null_buffer`
+--
+2.50.1
+
diff --git a/dev/vendors/patches/arrow-rs/0004-blaze-map-field-name-fixed-to-entries.patch b/dev/vendors/patches/arrow-rs/0004-blaze-map-field-name-fixed-to-entries.patch
new file mode 100644
index 00000000..f6066e0a
--- /dev/null
+++ b/dev/vendors/patches/arrow-rs/0004-blaze-map-field-name-fixed-to-entries.patch
@@ -0,0 +1,43 @@
+From ac1cb670bf1d82289c9a7aacca46a438905a8fcd Mon Sep 17 00:00:00 2001
+From: zhangli20 <zh...@kuaishou.com>
+Date: Sun, 15 Sep 2024 21:21:14 +0800
+Subject: [PATCH] blaze: map_field's name fixed to 'entries'
+
+---
+ parquet/src/arrow/async_reader/metadata.rs | 6 ++++++
+ parquet/src/arrow/schema/complex.rs | 2 +-
+ 2 files changed, 7 insertions(+), 1 deletion(-)
+
+diff --git a/parquet/src/arrow/async_reader/metadata.rs b/parquet/src/arrow/async_reader/metadata.rs
+index e0f7bdbbe..9a985e885 100644
+--- a/parquet/src/arrow/async_reader/metadata.rs
++++ b/parquet/src/arrow/async_reader/metadata.rs
+@@ -84,6 +84,12 @@ pub trait MetadataSuffixFetch: MetadataFetch {
+ /// Note the returned type is a boxed future, often created by
+ /// [FutureExt::boxed]. See the trait documentation for an example
+ fn fetch_suffix(&mut self, suffix: usize) -> BoxFuture<'_, Result<Bytes>>;
++
++ /// Sync version of get_bytes
++ /// this is only used by blaze, for reading dictionary values for row group pruning
++ fn get_bytes_sync(&mut self, range: Range<usize>) -> Result<Bytes> {
++ unimplemented!("blaze only")
++ }
+ }
+
+ /// An asynchronous interface to load [`ParquetMetaData`] from an async source
+diff --git a/parquet/src/arrow/schema/complex.rs b/parquet/src/arrow/schema/complex.rs
+index 16d46bd85..af70b8684 100644
+--- a/parquet/src/arrow/schema/complex.rs
++++ b/parquet/src/arrow/schema/complex.rs
+@@ -365,7 +365,7 @@ impl Visitor {
+ };
+
+ let map_field = Field::new_struct(
+- map_key_value.name(),
++ "entries", // blaze: map_field's name fixed to 'entries'
+ [key_field, value_field],
+ false, // The inner map field is always non-nullable (#1697)
+ )
+--
+2.50.1
+
diff --git a/dev/vendors/patches/arrow-rs/0005-blaze-fix-statistics-null-count-is-negative-error.patch b/dev/vendors/patches/arrow-rs/0005-blaze-fix-statistics-null-count-is-negative-error.patch
new file mode 100644
index 00000000..d72c36fb
--- /dev/null
+++ b/dev/vendors/patches/arrow-rs/0005-blaze-fix-statistics-null-count-is-negative-error.patch
@@ -0,0 +1,26 @@
+From 308b3c77cd28c654a84b38df46871cf0a88444f3 Mon Sep 17 00:00:00 2001
+From: zhangli20 <zh...@kuaishou.com>
+Date: Sun, 15 Sep 2024 21:24:51 +0800
+Subject: [PATCH] blaze: fix 'Statistics null count is negative (-1)' error
+
+---
+ parquet/src/file/statistics.rs | 3 ++-
+ 1 file changed, 2 insertions(+), 1 deletion(-)
+
+diff --git a/parquet/src/file/statistics.rs b/parquet/src/file/statistics.rs
+index b7522a76f..b7849e4f9 100644
+--- a/parquet/src/file/statistics.rs
++++ b/parquet/src/file/statistics.rs
+@@ -129,7 +129,8 @@ pub fn from_thrift(
+ // Number of nulls recorded, when it is not available, we just mark it as 0.
+ // TODO this should be `None` if there is no information about NULLS.
+ // see https://github.com/apache/arrow-rs/pull/6216/files
+- let null_count = stats.null_count.unwrap_or(0);
++ let null_count = stats.null_count.unwrap_or(0)
++ .max(0); // blaze - fix 'Statistics null count is negative (-1)' error
+
+ if null_count < 0 {
+ return Err(ParquetError::General(format!(
+--
+2.50.1
+
diff --git a/dev/vendors/patches/arrow-rs/0006-blaze-expose-fields-of-AsyncReader-ArrowReaderBuilder-to-pub.patch b/dev/vendors/patches/arrow-rs/0006-blaze-expose-fields-of-AsyncReader-ArrowReaderBuilder-to-pub.patch
new file mode 100644
index 00000000..c160fce8
--- /dev/null
+++ b/dev/vendors/patches/arrow-rs/0006-blaze-expose-fields-of-AsyncReader-ArrowReaderBuilder-to-pub.patch
@@ -0,0 +1,70 @@
+From 0f9132f81a3a6d69ed2b893a5bc5d7820f819a4d Mon Sep 17 00:00:00 2001
+From: zhangli20 <zh...@kuaishou.com>
+Date: Mon, 16 Sep 2024 00:16:57 +0800
+Subject: [PATCH] blaze: expose fields of AsyncReader/ArrowReaderBuilder to
+ public
+
+---
+ parquet/src/arrow/arrow_reader/mod.rs | 22 +++++++++++-----------
+ parquet/src/arrow/async_reader/mod.rs | 2 +-
+ 2 files changed, 12 insertions(+), 12 deletions(-)
+
+diff --git a/parquet/src/arrow/arrow_reader/mod.rs b/parquet/src/arrow/arrow_reader/mod.rs
+index 9127423ef..027b62b44 100644
+--- a/parquet/src/arrow/arrow_reader/mod.rs
++++ b/parquet/src/arrow/arrow_reader/mod.rs
+@@ -91,27 +91,27 @@ pub mod statistics;
+ /// [`StatisticsConverter`]: statistics::StatisticsConverter
+ /// [Querying Parquet with Millisecond Latency]: https://arrow.apache.org/blog/2022/12/26/querying-parquet-with-millisecond-latency/
+ pub struct ArrowReaderBuilder<T> {
+- pub(crate) input: T,
++ pub input: T,
+
+- pub(crate) metadata: Arc<ParquetMetaData>,
++ pub metadata: Arc<ParquetMetaData>,
+
+- pub(crate) schema: SchemaRef,
++ pub schema: SchemaRef,
+
+- pub(crate) fields: Option<Arc<ParquetField>>,
++ pub fields: Option<Arc<ParquetField>>,
+
+- pub(crate) batch_size: usize,
++ pub batch_size: usize,
+
+- pub(crate) row_groups: Option<Vec<usize>>,
++ pub row_groups: Option<Vec<usize>>,
+
+- pub(crate) projection: ProjectionMask,
++ pub projection: ProjectionMask,
+
+- pub(crate) filter: Option<RowFilter>,
++ pub filter: Option<RowFilter>,
+
+- pub(crate) selection: Option<RowSelection>,
++ pub selection: Option<RowSelection>,
+
+- pub(crate) limit: Option<usize>,
++ pub limit: Option<usize>,
+
+- pub(crate) offset: Option<usize>,
++ pub offset: Option<usize>,
+ }
+
+ impl<T: Debug> Debug for ArrowReaderBuilder<T> {
+diff --git a/parquet/src/arrow/async_reader/mod.rs b/parquet/src/arrow/async_reader/mod.rs
+index 248de5be3..b9152f924 100644
+--- a/parquet/src/arrow/async_reader/mod.rs
++++ b/parquet/src/arrow/async_reader/mod.rs
+@@ -215,7 +215,7 @@ impl ArrowReaderMetadata {
+ ///
+ /// Allows sharing the same builder for both the sync and async versions, whilst also not
+ /// breaking the pre-existing ParquetRecordBatchStreamBuilder API
+-pub struct AsyncReader<T>(T);
++pub struct AsyncReader<T>(pub T);
+
+ /// A builder for reading parquet files from an `async` source as [`ParquetRecordBatchStream`]
+ ///
+--
+2.50.1
+
diff --git a/dev/vendors/patches/arrow-rs/0007-blaze-use-bufreader-to-read-page-header-avoid-reading-in-fragments.patch b/dev/vendors/patches/arrow-rs/0007-blaze-use-bufreader-to-read-page-header-avoid-reading-in-fragments.patch
new file mode 100644
index 00000000..fa8bee6d
--- /dev/null
+++ b/dev/vendors/patches/arrow-rs/0007-blaze-use-bufreader-to-read-page-header-avoid-reading-in-fragments.patch
@@ -0,0 +1,41 @@
+From 417f165b72b99a25cc8f910fde9b43c1d5e06587 Mon Sep 17 00:00:00 2001
+From: zhangli20 <zh...@kuaishou.com>
+Date: Tue, 19 Nov 2024 10:52:22 +0800
+Subject: [PATCH] blaze: use bufreader to read page header, avoid reading in
+ fragments
+
+---
+ parquet/src/arrow/async_reader/metadata.rs | 2 +-
+ parquet/src/bin/parquet-layout.rs | 3 ++-
+ 2 files changed, 3 insertions(+), 2 deletions(-)
+
+diff --git a/parquet/src/arrow/async_reader/metadata.rs b/parquet/src/arrow/async_reader/metadata.rs
+index 9a985e885..2c994b9f0 100644
+--- a/parquet/src/arrow/async_reader/metadata.rs
++++ b/parquet/src/arrow/async_reader/metadata.rs
+@@ -87,7 +87,7 @@ pub trait MetadataSuffixFetch: MetadataFetch {
+
+ /// Sync version of get_bytes
+ /// this is only used by blaze, for reading dictionary values for row group pruning
+- fn get_bytes_sync(&mut self, range: Range<usize>) -> Result<Bytes> {
++ fn get_bytes_sync(&mut self, range: Range<u64>) -> Result<Bytes> {
+ unimplemented!("blaze only")
+ }
+ }
+diff --git a/parquet/src/bin/parquet-layout.rs b/parquet/src/bin/parquet-layout.rs
+index 46a231a7d..26e22e1b2 100644
+--- a/parquet/src/bin/parquet-layout.rs
++++ b/parquet/src/bin/parquet-layout.rs
+@@ -174,7 +174,8 @@ fn read_page_header<C: ChunkReader>(reader: &C, offset: u64) -> Result<(usize, P
+ }
+ }
+
+- let input = reader.get_read(offset)?;
++ // blaze: use BufReader to avoid reading in fragments
++ let input = std::io::BufReader::new(reader.get_read(offset)?);
+ let mut tracked = TrackedRead(input, 0);
+ let mut prot = TCompactInputProtocol::new(&mut tracked);
+ let header = PageHeader::read_from_in_protocol(&mut prot)?;
+--
+2.50.1
+
diff --git a/dev/vendors/patches/arrow-rs/0008-blaze-parquet-supports-processing-old-style-array-struct.patch b/dev/vendors/patches/arrow-rs/0008-blaze-parquet-supports-processing-old-style-array-struct.patch
new file mode 100644
index 00000000..45da8ae4
--- /dev/null
+++ b/dev/vendors/patches/arrow-rs/0008-blaze-parquet-supports-processing-old-style-array-struct.patch
@@ -0,0 +1,49 @@
+From 5de02520ca6c9f8f092b4b364906dc0507002cbb Mon Sep 17 00:00:00 2001
+From: zhangli20 <zh...@kuaishou.com>
+Date: Sat, 21 Dec 2024 02:16:53 +0800
+Subject: [PATCH] blaze: parquet supports processing old-style
+ array<struct<..>> represented as `REPEATED group field_nme {..}`
+
+---
+ parquet/src/arrow/schema/complex.rs | 25 +++++++++++++++++++++----
+ 1 file changed, 21 insertions(+), 4 deletions(-)
+
+diff --git a/parquet/src/arrow/schema/complex.rs b/parquet/src/arrow/schema/complex.rs
+index af70b8684..bf497fb77 100644
+--- a/parquet/src/arrow/schema/complex.rs
++++ b/parquet/src/arrow/schema/complex.rs
+@@ -185,10 +185,27 @@ impl Visitor {
+ Some(fields)
+ }
+ Some(d) => {
+- return Err(arrow_err!(
+- "incompatible arrow schema, expected struct got {}",
+- d
+- ))
++ // blaze: match old-style array<struct<..>> represented as
++ // `REPEATED group field_nme {..}`
++ let mut old_style_fields = None;
++ if let DataType::List(f) = d {
++ if f.name() == struct_type.name() {
++ if let DataType::Struct(fields) = &f.data_type() {
++ if fields.len() == parquet_fields.len() {
++ old_style_fields = Some(fields);
++ }
++ }
++ }
++ }
++ match old_style_fields {
++ Some(fields) => Some(fields),
++ None => {
++ return Err(arrow_err!(
++ "incompatible arrow schema, expected struct got {}",
++ d
++ ));
++ }
++ }
+ }
+ None => None,
+ };
+--
+2.50.1
+
diff --git a/dev/vendors/patches/datafusion/0001-blaze-update-arrow-dependency.patch b/dev/vendors/patches/datafusion/0001-blaze-update-arrow-dependency.patch
new file mode 100644
index 00000000..ea396fbd
--- /dev/null
+++ b/dev/vendors/patches/datafusion/0001-blaze-update-arrow-dependency.patch
@@ -0,0 +1,35 @@
+From 0015 Mon Sep 17 00:00:00 2001
+From: blaze <bl...@example.com>
+Date: Mon, 27 Jul 2026 02:26:00 +0800
+Subject: [PATCH] blaze: update arrow dependency to vendored v55.2.0-blaze
+
+---
+ Cargo.toml | 16 ++++++++++++++++
+ 1 file changed, 16 insertions(+)
+
+diff --git a/Cargo.toml b/Cargo.toml
+index 11cd3c637..b2fbf0df3 100644
+--- a/Cargo.toml
++++ b/Cargo.toml
+@@ -230,3 +230,19 @@ unexpected_cfgs = { level = "warn", check-cfg = [
+ "cfg(tarpaulin_include)",
+ ] }
+ unused_qualifications = "deny"
++
++[patch.crates-io]
++# arrow: vendored from v55.2.0-blaze
++arrow = { path = "../arrow-rs/arrow" }
++arrow-arith = { path = "../arrow-rs/arrow-arith" }
++arrow-array = { path = "../arrow-rs/arrow-array" }
++arrow-buffer = { path = "../arrow-rs/arrow-buffer" }
++arrow-cast = { path = "../arrow-rs/arrow-cast" }
++arrow-data = { path = "../arrow-rs/arrow-data" }
++arrow-ipc = { path = "../arrow-rs/arrow-ipc" }
++arrow-ord = { path = "../arrow-rs/arrow-ord" }
++arrow-row = { path = "../arrow-rs/arrow-row" }
++arrow-schema = { path = "../arrow-rs/arrow-schema" }
++arrow-select = { path = "../arrow-rs/arrow-select" }
++arrow-string = { path = "../arrow-rs/arrow-string" }
++parquet = { path = "../arrow-rs/parquet" }
+--
+2.50.1
diff --git a/dev/vendors/patches/datafusion/0002-blaze-exclude-testing-submodules.patch b/dev/vendors/patches/datafusion/0002-blaze-exclude-testing-submodules.patch
new file mode 100644
index 00000000..3cd59bad
--- /dev/null
+++ b/dev/vendors/patches/datafusion/0002-blaze-exclude-testing-submodules.patch
@@ -0,0 +1,64 @@
+From a3f528a1ea0d3e64992dbda9d764774fbc6185b0 Mon Sep 17 00:00:00 2001
+From: zhangli20 <ri...@gmail.com>
+Date: Wed, 13 Aug 2025 11:34:56 +0800
+Subject: [PATCH] blaze: exclude testing submodules
+
+---
+ .gitmodules | 20 ++++++++++----------
+ datafusion-testing | 1 -
+ parquet-testing | 1 -
+ testing | 1 -
+ 4 files changed, 10 insertions(+), 13 deletions(-)
+ delete mode 160000 datafusion-testing
+ delete mode 160000 parquet-testing
+ delete mode 160000 testing
+
+diff --git a/.gitmodules b/.gitmodules
+index 037accdbe..41cef1bba 100644
+--- a/.gitmodules
++++ b/.gitmodules
+@@ -1,10 +1,10 @@
+-[submodule "parquet-testing"]
+- path = parquet-testing
+- url = https://github.com/apache/parquet-testing.git
+-[submodule "testing"]
+- path = testing
+- url = https://github.com/apache/arrow-testing
+-[submodule "datafusion-testing"]
+- path = datafusion-testing
+- url = https://github.com/apache/datafusion-testing.git
+- branch = main
++# [submodule "parquet-testing"]
++# path = parquet-testing
++# url = https://github.com/apache/parquet-testing.git
++# [submodule "testing"]
++# path = testing
++# url = https://github.com/apache/arrow-testing
++# [submodule "datafusion-testing"]
++# path = datafusion-testing
++# url = https://github.com/apache/datafusion-testing.git
++# branch = main
+diff --git a/datafusion-testing b/datafusion-testing
+deleted file mode 160000
+index e9f9e22cc..000000000
+--- a/datafusion-testing
++++ /dev/null
+@@ -1 +0,0 @@
+-Subproject commit e9f9e22ccf09145a7368f80fd6a871f11e2b4481
+diff --git a/parquet-testing b/parquet-testing
+deleted file mode 160000
+index 107b36603..000000000
+--- a/parquet-testing
++++ /dev/null
+@@ -1 +0,0 @@
+-Subproject commit 107b36603e051aee26bd93e04b871034f6c756c0
+diff --git a/testing b/testing
+deleted file mode 160000
+index d2a137123..000000000
+--- a/testing
++++ /dev/null
+@@ -1 +0,0 @@
+-Subproject commit d2a13712303498963395318a4eb42872e66aead7
+--
+2.50.1
+
diff --git a/dev/vendors/patches/datafusion/0003-blaze-supports-parquet-scan-with-adaptive-batch-size.patch b/dev/vendors/patches/datafusion/0003-blaze-supports-parquet-scan-with-adaptive-batch-size.patch
new file mode 100644
index 00000000..dd47a23e
--- /dev/null
+++ b/dev/vendors/patches/datafusion/0003-blaze-supports-parquet-scan-with-adaptive-batch-size.patch
@@ -0,0 +1,51 @@
+From feeca02d670231014a2144293331e311494e31c6 Mon Sep 17 00:00:00 2001
+From: zhangli20 <zh...@kuaishou.com>
+Date: Sun, 15 Sep 2024 18:43:15 +0800
+Subject: [PATCH] blaze: supports parquet scan with adaptive batch size
+
+---
+ datafusion/datasource-parquet/src/opener.rs | 28 +++++++++++++++++++++
+ 1 file changed, 28 insertions(+)
+
+diff --git a/datafusion/datasource-parquet/src/opener.rs b/datafusion/datasource-parquet/src/opener.rs
+index 7c208d142..ffeaaf63c 100644
+--- a/datafusion/datasource-parquet/src/opener.rs
++++ b/datafusion/datasource-parquet/src/opener.rs
+@@ -393,6 +393,34 @@ impl FileOpener for ParquetOpener {
+ builder = builder.with_limit(limit)
+ }
+
++ // blaze:
++ // target batch size is usually 1MB ~ 16MB as there are lightweight compressions
++ // like RLE/DICT encodings
++ let adaptive_mem_size = 1048576;
++ let total_uncompressed_size = builder
++ .metadata()
++ .row_groups()
++ .iter()
++ .map(|rg| rg
++ .columns()
++ .iter()
++ .enumerate()
++ .filter(|(idx, _)| mask.leaf_included(*idx))
++ .map(|(_, col)| col.uncompressed_size())
++ .sum::<i64>())
++ .sum::<i64>();
++ let total_num_rows = builder
++ .metadata()
++ .row_groups()
++ .iter()
++ .map(|rg| rg.num_rows())
++ .sum::<i64>();
++ let adaptive_batch_size = batch_size
++ .min((adaptive_mem_size * total_num_rows / total_uncompressed_size.max(1)) as usize)
++ .max(1);
++ log::info!("executing parquet scan with adaptive batch size: {adaptive_batch_size}");
++ let batch_size = adaptive_batch_size;
++
+ let stream = builder
+ .with_projection(mask)
+ .with_batch_size(batch_size)
+--
+2.50.1
+
diff --git a/dev/vendors/patches/datafusion/0004-blaze-handle-special-cases-when-pruning-float-column-with-Na.patch b/dev/vendors/patches/datafusion/0004-blaze-handle-special-cases-when-pruning-float-column-with-Na.patch
new file mode 100644
index 00000000..527e908d
--- /dev/null
+++ b/dev/vendors/patches/datafusion/0004-blaze-handle-special-cases-when-pruning-float-column-with-Na.patch
@@ -0,0 +1,182 @@
+From c9fda37e2cfd77d13698f59f3ff594dd17b38d6a Mon Sep 17 00:00:00 2001
+From: zhangli20 <zh...@kuaishou.com>
+Date: Sun, 15 Sep 2024 22:59:40 +0800
+Subject: [PATCH] blaze: handle special cases when pruning float column with
+ NaN values
+
+---
+ datafusion/pruning/src/pruning_predicate.rs | 134 +++++++++++++++++++-
+ 1 file changed, 132 insertions(+), 2 deletions(-)
+
+diff --git a/datafusion/pruning/src/pruning_predicate.rs b/datafusion/pruning/src/pruning_predicate.rs
+index e863f57f8..e8e404862 100644
+--- a/datafusion/pruning/src/pruning_predicate.rs
++++ b/datafusion/pruning/src/pruning_predicate.rs
+@@ -24,10 +24,11 @@ use std::sync::Arc;
+
+ use arrow::array::AsArray;
+ use arrow::{
+- array::{new_null_array, ArrayRef, BooleanArray},
++ array::{new_null_array, Array, ArrayRef, ArrowNativeTypeOp, BooleanArray},
+ datatypes::{DataType, Field, Schema, SchemaRef},
+ record_batch::{RecordBatch, RecordBatchOptions},
+ };
++use arrow::array::types::{Float32Type, Float64Type};
+ // pub use for backwards compatibility
+ pub use datafusion_common::pruning::PruningStatistics;
+ use datafusion_physical_expr::simplifier::PhysicalExprSimplifier;
+@@ -42,6 +43,7 @@ use datafusion_common::{
+ ScalarValue,
+ };
+ use datafusion_common::{Column, DFSchema};
++use datafusion_common::cast::{as_float32_array, as_float64_array};
+ use datafusion_expr_common::operator::Operator;
+ use datafusion_physical_expr::utils::{collect_columns, Guarantee, LiteralGuarantee};
+ use datafusion_physical_expr::{expressions as phys_expr, PhysicalExprRef};
+@@ -917,6 +919,134 @@ fn build_statistics_record_batch<S: PruningStatistics + ?Sized>(
+ StatisticsType::NullCount => statistics.null_counts(&column),
+ StatisticsType::RowCount => statistics.row_counts(&column),
+ };
++
++ let array = array.and_then(|array| -> Option<ArrayRef> {
++ if array.data_type().is_floating() {
++ let min_values = statistics.min_values(&column);
++ let max_values = statistics.max_values(&column);
++
++ let min_is_nan = match &min_values {
++ Some(array) => {
++ match array.data_type() {
++ DataType::Float32 => as_float32_array(&array)
++ .ok()?
++ .iter()
++ .flatten()
++ .any(|v| v.is_nan()),
++ DataType::Float64 => as_float64_array(&array)
++ .ok()?
++ .iter()
++ .flatten()
++ .any(|v| v.is_nan()),
++ _ => false,
++ }
++ }
++ None => false,
++ };
++ let max_is_nan = match &max_values {
++ Some(array) => {
++ match array.data_type() {
++ DataType::Float32 => as_float32_array(&array)
++ .ok()?
++ .iter()
++ .flatten()
++ .any(|v| v.is_nan()),
++ DataType::Float64 => as_float64_array(&array)
++ .ok()?
++ .iter()
++ .flatten()
++ .any(|v| v.is_nan()),
++ _ => false,
++ }
++ }
++ None => false,
++ };
++ if min_is_nan || max_is_nan {
++ return None;
++ }
++
++ let min_is_positive_0 = match &min_values {
++ Some(array) => {
++ match array.data_type() {
++ DataType::Float32 => as_float32_array(&array)
++ .ok()?
++ .iter()
++ .flatten()
++ .any(|v| v.is_sign_positive() && v.is_zero()),
++ DataType::Float64 => as_float64_array(&array)
++ .ok()?
++ .iter()
++ .flatten()
++ .any(|v| v.is_sign_positive() && v.is_zero()),
++ _ => false,
++ }
++ },
++ _ => false,
++ };
++ let max_is_negative_0 = match &max_values {
++ Some(array) => {
++ match array.data_type() {
++ DataType::Float32 => as_float32_array(&array)
++ .ok()?
++ .iter()
++ .flatten()
++ .any(|v| v.is_sign_negative() && v.is_zero()),
++ DataType::Float64 => as_float64_array(&array)
++ .ok()?
++ .iter()
++ .flatten()
++ .any(|v| v.is_sign_negative() && v.is_zero()),
++ _ => false,
++ }
++ },
++ _ => false,
++ };
++
++ if min_is_positive_0 && *statistics_type == StatisticsType::Min {
++ match array.data_type() {
++ DataType::Float32 => return Some(Arc::new(
++ arrow::compute::unary::<_, _, Float32Type>(
++ as_float32_array(&array).ok()?,
++ |v| if v.is_sign_positive() && v.is_zero() {
++ 0.0f32
++ } else {
++ v
++ }))),
++ DataType::Float64 => return Some(Arc::new(
++ arrow::compute::unary::<_, _, Float64Type>(
++ as_float64_array(&array).ok()?,
++ |v| if v.is_sign_positive() && v.is_zero() {
++ 0.0f64
++ } else {
++ v
++ }))),
++ _ => return None,
++ };
++ }
++ if max_is_negative_0 && *statistics_type == StatisticsType::Max {
++ match array.data_type() {
++ DataType::Float32 => return Some(Arc::new(
++ arrow::compute::unary::<_, _, Float32Type>(
++ as_float32_array(&array).ok()?,
++ |v| if v.is_sign_negative() && v.is_zero() {
++ 0.0f32
++ } else {
++ v
++ }))),
++ DataType::Float64 => return Some(Arc::new(
++ arrow::compute::unary::<_, _, Float64Type>(
++ as_float64_array(&array).ok()?,
++ |v| if v.is_sign_negative() && v.is_zero() {
++ 0.0f64
++ } else {
++ v
++ }))),
++ _ => return None,
++ };
++ }
++ }
++ Some(array)
++ });
+ let array = array.unwrap_or_else(|| new_null_array(data_type, num_containers));
+
+ if num_containers != array.len() {
+@@ -4422,7 +4552,7 @@ mod tests {
+ true,
+ // s1 ["AB", "A\u{10ffff}\u{10ffff}\u{10ffff}"] ==> some rows could pass (must keep)
+ true,
+- // s1 ["A\u{10ffff}\u{10ffff}", "A\u{10ffff}\u{10ffff}"] ==> no row match. (min, max) maybe truncate
++ // s1 ["A\u{10ffff}\u{10ffff}", "A\u{10ffff}\u{10ffff}"] ==> no row match. (min, max) maybe truncate
+ // orignal (min, max) maybe ("A\u{10ffff}\u{10ffff}\u{10ffff}", "A\u{10ffff}\u{10ffff}\u{10ffff}\u{10ffff}")
+ true,
+ ];
+--
+2.50.1
+
diff --git a/dev/vendors/patches/datafusion/0005-blaze-supports-parquet-dictionary-filtering.patch b/dev/vendors/patches/datafusion/0005-blaze-supports-parquet-dictionary-filtering.patch
new file mode 100644
index 00000000..88c2741b
--- /dev/null
+++ b/dev/vendors/patches/datafusion/0005-blaze-supports-parquet-dictionary-filtering.patch
@@ -0,0 +1,454 @@
+From a887b7659f10ec62ed9102aa9a3e9fd5d2e9ea0c Mon Sep 17 00:00:00 2001
+From: zhangli20 <zh...@kuaishou.com>
+Date: Sun, 15 Sep 2024 22:02:24 +0800
+Subject: [PATCH] blaze: supports parquet dictionary filtering
+
+---
+ datafusion-examples/examples/pruning.rs | 4 +
+ datafusion/common/src/pruning.rs | 5 +
+ datafusion/datasource-parquet/src/opener.rs | 2 +-
+ .../datasource-parquet/src/page_filter.rs | 4 +
+ .../src/row_group_filter.rs | 151 +++++++++++++-----
+ datafusion/pruning/Cargo.toml | 2 +
+ datafusion/pruning/src/pruning_predicate.rs | 62 ++++++-
+ 7 files changed, 189 insertions(+), 41 deletions(-)
+
+diff --git a/datafusion-examples/examples/pruning.rs b/datafusion-examples/examples/pruning.rs
+index b2d2fa13b..430520e73 100644
+--- a/datafusion-examples/examples/pruning.rs
++++ b/datafusion-examples/examples/pruning.rs
+@@ -184,6 +184,10 @@ impl PruningStatistics for MyCatalog {
+ // but we do not illustrate that here
+ None
+ }
++
++ fn dictionary_values(&self, column: &Column) -> Option<ArrayRef> {
++ None
++ }
+ }
+
+ fn create_pruning_predicate(expr: Expr, schema: &SchemaRef) -> PruningPredicate {
+diff --git a/datafusion/common/src/pruning.rs b/datafusion/common/src/pruning.rs
+index 48750e3c9..151a99abb 100644
+--- a/datafusion/common/src/pruning.rs
++++ b/datafusion/common/src/pruning.rs
+@@ -126,6 +126,11 @@ pub trait PruningStatistics {
+ column: &Column,
+ values: &HashSet<ScalarValue>,
+ ) -> Option<BooleanArray>;
++
++ /// return dictionary values for the named column as ListArray<?>
++ fn dictionary_values(&self, _column: &Column) -> Option<ArrayRef> {
++ None
++ }
+ }
+
+ /// Prune files based on their partition values.
+diff --git a/datafusion/datasource-parquet/src/opener.rs b/datafusion/datasource-parquet/src/opener.rs
+index ffeaaf63c..e31e33cfa 100644
+--- a/datafusion/datasource-parquet/src/opener.rs
++++ b/datafusion/datasource-parquet/src/opener.rs
+@@ -345,8 +345,8 @@ impl FileOpener for ParquetOpener {
+ if let Some(predicate) = predicate.as_ref() {
+ if enable_row_group_stats_pruning {
+ row_groups.prune_by_statistics(
++ &mut builder,
+ &physical_file_schema,
+- builder.parquet_schema(),
+ rg_metadata,
+ predicate,
+ &file_metrics,
+diff --git a/datafusion/datasource-parquet/src/page_filter.rs b/datafusion/datasource-parquet/src/page_filter.rs
+index 5f3e05747..c01f6f277 100644
+--- a/datafusion/datasource-parquet/src/page_filter.rs
++++ b/datafusion/datasource-parquet/src/page_filter.rs
+@@ -507,4 +507,8 @@ impl PruningStatistics for PagesPruningStatistics<'_> {
+ ) -> Option<BooleanArray> {
+ None
+ }
++
++ fn dictionary_values(&self, _column: &datafusion_common::Column) -> Option<ArrayRef> {
++ None
++ }
+ }
+diff --git a/datafusion/datasource-parquet/src/row_group_filter.rs b/datafusion/datasource-parquet/src/row_group_filter.rs
+index 51d50d780..982eed293 100644
+--- a/datafusion/datasource-parquet/src/row_group_filter.rs
++++ b/datafusion/datasource-parquet/src/row_group_filter.rs
+@@ -15,12 +15,15 @@
+ // specific language governing permissions and limitations
+ // under the License.
+
++use std::cell::{OnceCell, RefCell};
+ use std::collections::{HashMap, HashSet};
+ use std::sync::Arc;
+
+ use super::{ParquetAccessPlan, ParquetFileMetrics};
+-use arrow::array::{ArrayRef, BooleanArray};
+-use arrow::datatypes::Schema;
++use arrow::array::{Array, ArrayRef, BooleanArray, ListArray};
++use arrow::buffer::{OffsetBuffer, ScalarBuffer};
++use arrow::datatypes::{Field, Schema};
++use parking_lot::Mutex;
+ use datafusion_common::pruning::PruningStatistics;
+ use datafusion_common::{Column, Result, ScalarValue};
+ use datafusion_datasource::FileRange;
+@@ -29,7 +32,6 @@ use parquet::arrow::arrow_reader::statistics::StatisticsConverter;
+ use parquet::arrow::parquet_column;
+ use parquet::basic::Type;
+ use parquet::data_type::Decimal;
+-use parquet::schema::types::SchemaDescriptor;
+ use parquet::{
+ arrow::{async_reader::AsyncFileReader, ParquetRecordBatchStreamBuilder},
+ bloom_filter::Sbbf,
+@@ -102,10 +104,10 @@ impl RowGroupAccessPlanFilter {
+ ///
+ /// # Panics
+ /// if `groups.len() != self.len()`
+- pub fn prune_by_statistics(
++ pub fn prune_by_statistics<T: AsyncFileReader + Send + 'static>(
+ &mut self,
++ builder: &mut ParquetRecordBatchStreamBuilder<T>,
+ arrow_schema: &Schema,
+- parquet_schema: &SchemaDescriptor,
+ groups: &[RowGroupMetaData],
+ predicate: &PruningPredicate,
+ metrics: &ParquetFileMetrics,
+@@ -115,35 +117,47 @@ impl RowGroupAccessPlanFilter {
+
+ assert_eq!(groups.len(), self.access_plan.len());
+ // Indexes of row groups still to scan
++ // blaze: for less reading of dictionary pages, we still prune in separated passes
+ let row_group_indexes = self.access_plan.row_group_indexes();
+- let row_group_metadatas = row_group_indexes
+- .iter()
+- .map(|&i| &groups[i])
+- .collect::<Vec<_>>();
+-
+- let pruning_stats = RowGroupPruningStatistics {
+- parquet_schema,
+- row_group_metadatas,
+- arrow_schema,
+- };
++ for &rg_idx in &row_group_indexes {
++ let row_group_metadatas = vec![&groups[rg_idx]];
++ let mut pruning_stats = RowGroupPruningStatistics {
++ row_group_metadatas,
++ arrow_schema,
++ use_dictionary_filtering: false,
++ cached_dictionaries: Mutex::new(vec![]),
++ builder: RefCell::new(builder),
++ };
+
+- // try to prune the row groups in a single call
+- match predicate.prune(&pruning_stats) {
+- Ok(values) => {
+- // values[i] is false means the predicate could not be true for row group i
+- for (idx, &value) in row_group_indexes.iter().zip(values.iter()) {
+- if !value {
+- self.access_plan.skip(*idx);
+- metrics.row_groups_pruned_statistics.add(1);
+- } else {
+- metrics.row_groups_matched_statistics.add(1);
++ // prune using two runs because pruning with dictionary filtering is more expensive
++ // first run: prune without dictionary filtering
++ match predicate.prune(&pruning_stats) {
++ Ok(values) => {
++ if !values[0] {
++ self.access_plan.skip(rg_idx);
++ continue; // no need to prune with dictionary
+ }
+ }
++ Err(e) => {
++ log::info!("Error evaluating row group predicate values without dictionary {e}");
++ metrics.predicate_evaluation_errors.add(1);
++ continue; // no need to prune with dictionary
++ }
+ }
+- // stats filter array could not be built, so we can't prune
+- Err(e) => {
+- log::debug!("Error evaluating row group predicate values {e}");
+- metrics.predicate_evaluation_errors.add(1);
++
++ // second run: prune with dictionary filtering
++ pruning_stats.use_dictionary_filtering = true;
++ pruning_stats.cached_dictionaries = Mutex::new(vec![]);
++ match predicate.prune(&pruning_stats) {
++ Ok(values) => {
++ if !values[0] {
++ self.access_plan.skip(rg_idx);
++ }
++ }
++ Err(e) => {
++ log::info!("Error evaluating row group predicate values with dictionary {e}");
++ metrics.predicate_evaluation_errors.add(1);
++ }
+ }
+ }
+ }
+@@ -359,34 +373,51 @@ impl PruningStatistics for BloomFilterStatistics {
+
+ Some(BooleanArray::from(vec![contains]))
+ }
++
++ fn dictionary_values(&self, _column: &Column) -> Option<ArrayRef> {
++ None
++ }
+ }
+
+ /// Wraps a slice of [`RowGroupMetaData`] in a way that implements [`PruningStatistics`]
+-struct RowGroupPruningStatistics<'a> {
+- parquet_schema: &'a SchemaDescriptor,
++struct RowGroupPruningStatistics<'a, T: AsyncFileReader + Send + 'static> {
+ row_group_metadatas: Vec<&'a RowGroupMetaData>,
+ arrow_schema: &'a Schema,
++ use_dictionary_filtering: bool,
++ cached_dictionaries: Mutex<Vec<OnceCell<Option<ArrayRef>>>>,
++ builder: RefCell<&'a mut ParquetRecordBatchStreamBuilder<T>>,
+ }
+
+-impl<'a> RowGroupPruningStatistics<'a> {
++impl<'a, T: AsyncFileReader + Send + 'static> RowGroupPruningStatistics<'a, T> {
+ /// Return an iterator over the row group metadata
+- fn metadata_iter(&'a self) -> impl Iterator<Item = &'a RowGroupMetaData> + 'a {
+- self.row_group_metadatas.iter().copied()
++ fn metadata_iter(&self) -> impl Iterator<Item = &'a RowGroupMetaData> + 'a {
++ // blaze: bypass lifetime checker
++ let unchecked_self: &'a Self = unsafe {
++ std::mem::transmute(self)
++ };
++ unchecked_self.row_group_metadatas.iter().copied()
+ }
+
+ fn statistics_converter<'b>(
+- &'a self,
++ &self,
+ column: &'b Column,
+ ) -> Result<StatisticsConverter<'a>> {
++ // blaze: bypass lifetime checker
++ let unchecked_self: &'a Self = unsafe {
++ std::mem::transmute(self)
++ };
+ Ok(StatisticsConverter::try_new(
+ &column.name,
+- self.arrow_schema,
+- self.parquet_schema,
++ unchecked_self.arrow_schema,
++ unsafe {
++ // blaze: bypass lifetime checker
++ std::mem::transmute(unchecked_self.builder.borrow_mut().parquet_schema())
++ }
+ )?)
+ }
+ }
+
+-impl PruningStatistics for RowGroupPruningStatistics<'_> {
++impl<'a, T: AsyncFileReader + Send + 'static> PruningStatistics for RowGroupPruningStatistics<'a, T> {
+ fn min_values(&self, column: &Column) -> Option<ArrayRef> {
+ self.statistics_converter(column)
+ .and_then(|c| Ok(c.row_group_mins(self.metadata_iter())?))
+@@ -426,6 +457,50 @@ impl PruningStatistics for RowGroupPruningStatistics<'_> {
+ ) -> Option<BooleanArray> {
+ None
+ }
++
++ fn dictionary_values(&self, column: &Column) -> Option<ArrayRef> {
++ if !self.use_dictionary_filtering {
++ return None;
++ }
++
++ let col_idx = self
++ .row_group_metadatas
++ .first()
++ .unwrap()
++ .columns()
++ .iter()
++ .position(|c| c.column_path().string().eq_ignore_ascii_case(&column.flat_name()))?;
++
++ let mut cached_dictionaries = self.cached_dictionaries.lock();
++ cached_dictionaries.resize_with(col_idx + 1, || Default::default());
++ cached_dictionaries[col_idx].get_or_init(|| {
++ let mut values = vec![];
++ for row_group_metadata in self.metadata_iter() {
++ let dict_values =
++ parquet::blaze::get_dictionary_for_pruning(
++ &mut self.builder.borrow_mut().input.0,
++ row_group_metadata,
++ col_idx,
++ )
++ .ok()
++ .flatten()?;
++
++ let dict_array =
++ ListArray::try_new(
++ Arc::new(Field::new("item", dict_values.data_type().clone(), true)),
++ OffsetBuffer::new(ScalarBuffer::from(vec![0, dict_values.len() as i32])),
++ dict_values,
++ None,
++ )
++ .ok()?;
++ values.push(dict_array);
++ }
++ arrow::compute::concat(&values.iter()
++ .map(|c| c as &dyn Array)
++ .collect::<Vec<_>>()
++ ).ok()
++ }).as_ref().cloned()
++ }
+ }
+
+ #[cfg(test)]
+diff --git a/datafusion/pruning/Cargo.toml b/datafusion/pruning/Cargo.toml
+index 6acf178e4..90b31b933 100644
+--- a/datafusion/pruning/Cargo.toml
++++ b/datafusion/pruning/Cargo.toml
+@@ -16,7 +16,9 @@ arrow = { workspace = true }
+ arrow-schema = { workspace = true }
+ datafusion-common = { workspace = true, default-features = true }
+ datafusion-datasource = { workspace = true }
++datafusion-expr = { workspace = true }
+ datafusion-expr-common = { workspace = true, default-features = true }
++datafusion-functions-nested = { workspace = true }
+ datafusion-physical-expr = { workspace = true }
+ datafusion-physical-expr-common = { workspace = true }
+ datafusion-physical-plan = { workspace = true }
+diff --git a/datafusion/pruning/src/pruning_predicate.rs b/datafusion/pruning/src/pruning_predicate.rs
+index e8e404862..f8818292b 100644
+--- a/datafusion/pruning/src/pruning_predicate.rs
++++ b/datafusion/pruning/src/pruning_predicate.rs
+@@ -45,8 +45,10 @@ use datafusion_common::{
+ use datafusion_common::{Column, DFSchema};
+ use datafusion_common::cast::{as_float32_array, as_float64_array};
+ use datafusion_expr_common::operator::Operator;
++use datafusion_expr::ScalarUDF;
++use datafusion_functions_nested::array_has::ArrayHas;
+ use datafusion_physical_expr::utils::{collect_columns, Guarantee, LiteralGuarantee};
+-use datafusion_physical_expr::{expressions as phys_expr, PhysicalExprRef};
++use datafusion_physical_expr::{expressions as phys_expr, PhysicalExprRef, ScalarFunctionExpr};
+ use datafusion_physical_expr_common::physical_expr::snapshot_physical_expr;
+ use datafusion_physical_plan::{ColumnarValue, PhysicalExpr};
+
+@@ -814,6 +816,7 @@ impl RequiredColumns {
+ StatisticsType::Max => format!("{column_name}_max"),
+ StatisticsType::NullCount => format!("{column_name}_null_count"),
+ StatisticsType::RowCount => "row_count".to_string(),
++ StatisticsType::Dictionary => "dict".to_string(),
+ };
+
+ let stat_column = phys_expr::Column::new(&stat_column_name, idx);
+@@ -868,6 +871,26 @@ impl RequiredColumns {
+ ) -> Result<Arc<dyn PhysicalExpr>> {
+ self.stat_column_expr(column, column_expr, field, StatisticsType::RowCount)
+ }
++
++
++ /// rewrite col --> col_dict
++ fn dict_column_expr(
++ &mut self,
++ column: &phys_expr::Column,
++ column_expr: &Arc<dyn PhysicalExpr>,
++ field: &Field,
++ ) -> Result<Arc<dyn PhysicalExpr>> {
++
++ let list_type = DataType::List(Arc::new(
++ Field::new("item", field.data_type().clone(), true)
++ ));
++ self.stat_column_expr(
++ column,
++ column_expr,
++ &field.clone().with_data_type(list_type),
++ StatisticsType::Dictionary,
++ )
++ }
+ }
+
+ impl From<Vec<(phys_expr::Column, StatisticsType, Field)>> for RequiredColumns {
+@@ -918,6 +941,7 @@ fn build_statistics_record_batch<S: PruningStatistics + ?Sized>(
+ StatisticsType::Max => statistics.max_values(&column),
+ StatisticsType::NullCount => statistics.null_counts(&column),
+ StatisticsType::RowCount => statistics.row_counts(&column),
++ StatisticsType::Dictionary => statistics.dictionary_values(&column),
+ };
+
+ let array = array.and_then(|array| -> Option<ArrayRef> {
+@@ -1190,6 +1214,20 @@ impl<'a> PruningExpressionBuilder<'a> {
+ row_count_field,
+ )
+ }
++
++ fn dict_column_expr(&mut self) -> Result<Arc<dyn PhysicalExpr>> {
++ self.required_columns
++ .dict_column_expr(&self.column, &self.column_expr, self.field)
++ }
++
++ fn dict_column_contains_expr(&mut self) -> Result<Arc<dyn PhysicalExpr>> {
++ Ok(Arc::new(ScalarFunctionExpr::new(
++ "array_has",
++ Arc::new(ScalarUDF::from(ArrayHas::new())),
++ vec![self.dict_column_expr()?, self.scalar_expr().clone()],
++ Arc::new(Field::new("result", DataType::Boolean, true)),
++ )))
++ }
+ }
+
+ /// This function is designed to rewrite the column_expr to
+@@ -1689,7 +1727,7 @@ fn build_statistics_expr(
+ // (column / 2) = 4 => (column_min / 2) <= 4 && 4 <= (column_max / 2)
+ let min_column_expr = expr_builder.min_column_expr()?;
+ let max_column_expr = expr_builder.max_column_expr()?;
+- Arc::new(phys_expr::BinaryExpr::new(
++ let stat_pred_expr = Arc::new(phys_expr::BinaryExpr::new(
+ Arc::new(phys_expr::BinaryExpr::new(
+ min_column_expr,
+ Operator::LtEq,
+@@ -1701,6 +1739,17 @@ fn build_statistics_expr(
+ Operator::LtEq,
+ max_column_expr,
+ )),
++ ));
++
++ // append dict_pred expr (stat && dictionary_contains(column))
++ let dict = expr_builder.dict_column_expr()?;
++ let has_no_stat_pred = Arc::new(phys_expr::IsNullExpr::new(stat_pred_expr.clone()));
++ let has_no_dict = Arc::new(phys_expr::IsNullExpr::new(dict));
++ let dict_contained = expr_builder.dict_column_contains_expr()?;
++ Arc::new(phys_expr::BinaryExpr::new(
++ Arc::new(phys_expr::BinaryExpr::new(has_no_stat_pred, Operator::Or, stat_pred_expr)),
++ Operator::And,
++ Arc::new(phys_expr::BinaryExpr::new(has_no_dict, Operator::Or, dict_contained)),
+ ))
+ }
+ Operator::NotLikeMatch => build_not_like_match(expr_builder)?,
+@@ -1989,6 +2038,7 @@ pub(crate) enum StatisticsType {
+ Max,
+ NullCount,
+ RowCount,
++ Dictionary,
+ }
+
+ #[cfg(test)]
+@@ -2350,6 +2400,10 @@ mod tests {
+ .get(column)
+ .and_then(|container_stats| container_stats.contained(values))
+ }
++
++ fn dictionary_values(&self, column: &Column) -> Option<ArrayRef> {
++ None
++ }
+ }
+
+ /// Returns the specified min/max container values
+@@ -2387,6 +2441,10 @@ mod tests {
+ ) -> Option<BooleanArray> {
+ None
+ }
++
++ fn dictionary_values(&self, column: &Column) -> Option<ArrayRef> {
++ None
++ }
+ }
+
+ /// Row count should only be referenced once in the pruning expression, even if we need the row count
+--
+2.50.1
+
diff --git a/dev/vendors/patches/datafusion/0006-blaze-allow-nullable-parquet-scan-partition-key.patch b/dev/vendors/patches/datafusion/0006-blaze-allow-nullable-parquet-scan-partition-key.patch
new file mode 100644
index 00000000..451dfb0c
--- /dev/null
+++ b/dev/vendors/patches/datafusion/0006-blaze-allow-nullable-parquet-scan-partition-key.patch
@@ -0,0 +1,30 @@
+From 534651739be8f95b6dedf34b06fe7c355611af70 Mon Sep 17 00:00:00 2001
+From: zhangli20 <zh...@kuaishou.com>
+Date: Mon, 16 Sep 2024 10:03:29 +0800
+Subject: [PATCH] blaze: allow nullable parquet scan partition key
+
+---
+ datafusion/datasource/src/file_scan_config.rs | 7 ++++---
+ 1 file changed, 4 insertions(+), 3 deletions(-)
+
+diff --git a/datafusion/datasource/src/file_scan_config.rs b/datafusion/datasource/src/file_scan_config.rs
+index 959a2dd33..2606c5387 100644
+--- a/datafusion/datasource/src/file_scan_config.rs
++++ b/datafusion/datasource/src/file_scan_config.rs
+@@ -770,9 +770,10 @@ impl FileScanConfig {
+ self.file_schema.field(idx).clone()
+ } else {
+ let partition_idx = idx - self.file_schema.fields().len();
+- Arc::unwrap_or_clone(Arc::clone(
+- &self.table_partition_cols[partition_idx],
+- ))
++ self.table_partition_cols[partition_idx]
++ .as_ref()
++ .clone()
++ .with_nullable(true) // blaze: allow nullable parquet scan partition key
+ }
+ })
+ .collect();
+--
+2.50.1
+
diff --git a/dev/vendors/patches/datafusion/0007-blaze-make-scatter-public.patch b/dev/vendors/patches/datafusion/0007-blaze-make-scatter-public.patch
new file mode 100644
index 00000000..fd284c21
--- /dev/null
+++ b/dev/vendors/patches/datafusion/0007-blaze-make-scatter-public.patch
@@ -0,0 +1,25 @@
+From b781bce65fef561bb3e4006c874d8f88291b637d Mon Sep 17 00:00:00 2001
+From: zhangli20 <zh...@kuaishou.com>
+Date: Mon, 16 Sep 2024 10:04:54 +0800
+Subject: [PATCH] blaze: make scatter() public
+
+---
+ datafusion/physical-expr-common/src/physical_expr.rs | 2 +-
+ 1 file changed, 1 insertion(+), 1 deletion(-)
+
+diff --git a/datafusion/physical-expr-common/src/physical_expr.rs b/datafusion/physical-expr-common/src/physical_expr.rs
+index b4cb08715..a328f146e 100644
+--- a/datafusion/physical-expr-common/src/physical_expr.rs
++++ b/datafusion/physical-expr-common/src/physical_expr.rs
+@@ -21,7 +21,7 @@ use std::fmt::{Debug, Display, Formatter};
+ use std::hash::{Hash, Hasher};
+ use std::sync::Arc;
+
+-use crate::utils::scatter;
++pub use crate::utils::scatter;
+
+ use arrow::array::BooleanArray;
+ use arrow::compute::filter_record_batch;
+--
+2.50.1
+
diff --git a/dev/vendors/patches/datafusion/0008-blaze-use-deprecated-min-max-values-only-when-min-==-max.patch b/dev/vendors/patches/datafusion/0008-blaze-use-deprecated-min-max-values-only-when-min-==-max.patch
new file mode 100644
index 00000000..5252c200
--- /dev/null
+++ b/dev/vendors/patches/datafusion/0008-blaze-use-deprecated-min-max-values-only-when-min-==-max.patch
@@ -0,0 +1,70 @@
+From da84929cc253eeb4b14b20f6f141145da2a14d05 Mon Sep 17 00:00:00 2001
+From: zhangli20 <zh...@kuaishou.com>
+Date: Mon, 16 Sep 2024 11:15:48 +0800
+Subject: [PATCH] blaze: use deprecated min/max values only when min == max
+
+---
+ .../src/row_group_filter.rs | 40 +++++++++++++++++--
+ 1 file changed, 36 insertions(+), 4 deletions(-)
+
+diff --git a/datafusion/datasource-parquet/src/row_group_filter.rs b/datafusion/datasource-parquet/src/row_group_filter.rs
+index 982eed293..0453ffa29 100644
+--- a/datafusion/datasource-parquet/src/row_group_filter.rs
++++ b/datafusion/datasource-parquet/src/row_group_filter.rs
+@@ -417,17 +417,49 @@ impl<'a, T: AsyncFileReader + Send + 'static> RowGroupPruningStatistics<'a, T> {
+ }
+ }
+
++macro_rules! is_min_max_deprecated {
++ ($self:expr, $column:expr) => {{
++ $self.row_group_metadatas
++ .iter()
++ .any(|row_group_metadata| row_group_metadata
++ .columns()
++ .iter()
++ .find(|c| c.column_path().string().eq_ignore_ascii_case(&$column.flat_name()))
++ .and_then(|c| c.statistics())
++ .map(|stat| stat.is_min_max_deprecated())
++ .unwrap_or(false))
++ }}
++}
++
+ impl<'a, T: AsyncFileReader + Send + 'static> PruningStatistics for RowGroupPruningStatistics<'a, T> {
+ fn min_values(&self, column: &Column) -> Option<ArrayRef> {
+- self.statistics_converter(column)
++ let min_values = self.statistics_converter(column)
+ .and_then(|c| Ok(c.row_group_mins(self.metadata_iter())?))
+- .ok()
++ .ok();
++ if is_min_max_deprecated!(self, column) {
++ let max_values = self.statistics_converter(column)
++ .and_then(|c| Ok(c.row_group_maxes(self.metadata_iter())?))
++ .ok();
++ if min_values != max_values {
++ return None;
++ }
++ }
++ min_values
+ }
+
+ fn max_values(&self, column: &Column) -> Option<ArrayRef> {
+- self.statistics_converter(column)
++ let max_values = self.statistics_converter(column)
+ .and_then(|c| Ok(c.row_group_maxes(self.metadata_iter())?))
+- .ok()
++ .ok();
++ if is_min_max_deprecated!(self, column) {
++ let min_values = self.statistics_converter(column)
++ .and_then(|c| Ok(c.row_group_mins(self.metadata_iter())?))
++ .ok();
++ if min_values != max_values {
++ return None;
++ }
++ }
++ max_values
+ }
+
+ fn num_containers(&self) -> usize {
+--
+2.50.1
+
diff --git a/dev/vendors/patches/datafusion/0009-blaze-disable-isnull-isnotnull-pruning-for-nested-data-type.patch b/dev/vendors/patches/datafusion/0009-blaze-disable-isnull-isnotnull-pruning-for-nested-data-type.patch
new file mode 100644
index 00000000..bc1bf10f
--- /dev/null
+++ b/dev/vendors/patches/datafusion/0009-blaze-disable-isnull-isnotnull-pruning-for-nested-data-type.patch
@@ -0,0 +1,26 @@
+From 4d8a76a18ae3eeef66e9d064fa29eae13ee7c2c9 Mon Sep 17 00:00:00 2001
+From: zhangli20 <zh...@kuaishou.com>
+Date: Mon, 16 Sep 2024 11:22:36 +0800
+Subject: [PATCH] blaze: disable isnull/isnotnull pruning for nested data type
+
+---
+ datafusion/pruning/src/pruning_predicate.rs | 3 +++
+ 1 file changed, 3 insertions(+)
+
+diff --git a/datafusion/pruning/src/pruning_predicate.rs b/datafusion/pruning/src/pruning_predicate.rs
+index f8818292b..43e3d616f 100644
+--- a/datafusion/pruning/src/pruning_predicate.rs
++++ b/datafusion/pruning/src/pruning_predicate.rs
+@@ -1449,6 +1449,9 @@ fn build_is_null_column_expr(
+ ) -> Option<Arc<dyn PhysicalExpr>> {
+ if let Some(col) = expr.as_any().downcast_ref::<phys_expr::Column>() {
+ let field = schema.field_with_name(col.name()).ok()?;
++ if field.data_type().is_nested() {
++ return None;
++ }
+
+ let null_count_field = &Field::new(field.name(), DataType::UInt64, true);
+ if with_not {
+--
+2.50.1
+
diff --git a/dev/vendors/patches/datafusion/0010-blaze-use-case-insensitive-comparing-in-processing-parquet-f.patch b/dev/vendors/patches/datafusion/0010-blaze-use-case-insensitive-comparing-in-processing-parquet-f.patch
new file mode 100644
index 00000000..c6b87471
--- /dev/null
+++ b/dev/vendors/patches/datafusion/0010-blaze-use-case-insensitive-comparing-in-processing-parquet-f.patch
@@ -0,0 +1,43 @@
+From 80f119b76b2e6a24b5eb7f4dd1639dff838caa97 Mon Sep 17 00:00:00 2001
+From: zhangli20 <zh...@kuaishou.com>
+Date: Mon, 16 Sep 2024 11:27:42 +0800
+Subject: [PATCH] blaze: use case-insensitive comparing in processing parquet
+ file schema
+
+---
+ datafusion/common/src/dfschema.rs | 2 +-
+ datafusion/datasource/src/schema_adapter.rs | 5 ++++-
+ 2 files changed, 5 insertions(+), 2 deletions(-)
+
+diff --git a/datafusion/common/src/dfschema.rs b/datafusion/common/src/dfschema.rs
+index 88303bbcd..1e9b71ac2 100644
+--- a/datafusion/common/src/dfschema.rs
++++ b/datafusion/common/src/dfschema.rs
+@@ -571,7 +571,7 @@ impl DFSchema {
+ .fields
+ .iter()
+ .zip(arrow_schema.fields().iter())
+- .all(|(dffield, arrowfield)| dffield.name() == arrowfield.name())
++ .all(|(dffield, arrowfield)| dffield.name().eq_ignore_ascii_case(arrowfield.name()))
+ }
+
+ /// Check to see if fields in 2 Arrow schemas are compatible
+diff --git a/datafusion/datasource/src/schema_adapter.rs b/datafusion/datasource/src/schema_adapter.rs
+index 5e743a3f0..12722b433 100644
+--- a/datafusion/datasource/src/schema_adapter.rs
++++ b/datafusion/datasource/src/schema_adapter.rs
+@@ -276,7 +276,10 @@ impl SchemaAdapter for DefaultSchemaAdapter {
+ /// Panics if index is not in range for the table schema
+ fn map_column_index(&self, index: usize, file_schema: &Schema) -> Option<usize> {
+ let field = self.projected_table_schema.field(index);
+- Some(file_schema.fields.find(field.name())?.0)
++ file_schema
++ .fields
++ .iter()
++ .position(|f| f.name().eq_ignore_ascii_case(field.name()))
+ }
+
+ /// Creates a `SchemaMapping` for casting or mapping the columns from the
+--
+2.50.1
+
diff --git a/dev/vendors/patches/datafusion/0011-blaze-supports-pruning-with-starts_with.patch b/dev/vendors/patches/datafusion/0011-blaze-supports-pruning-with-starts_with.patch
new file mode 100644
index 00000000..c470ab4b
--- /dev/null
+++ b/dev/vendors/patches/datafusion/0011-blaze-supports-pruning-with-starts_with.patch
@@ -0,0 +1,86 @@
+From 533456f6af70c9f51256b108f3bf2a189133d0f8 Mon Sep 17 00:00:00 2001
+From: zhangli20 <zh...@kuaishou.com>
+Date: Mon, 16 Sep 2024 11:43:47 +0800
+Subject: [PATCH] blaze: supports pruning with starts_with
+
+---
+ datafusion/pruning/src/pruning_predicate.rs | 56 +++++++++++++++++++++
+ 1 file changed, 56 insertions(+)
+
+diff --git a/datafusion/pruning/src/pruning_predicate.rs b/datafusion/pruning/src/pruning_predicate.rs
+index 43e3d616f..648d32374 100644
+--- a/datafusion/pruning/src/pruning_predicate.rs
++++ b/datafusion/pruning/src/pruning_predicate.rs
+@@ -1490,6 +1490,48 @@ fn build_is_null_column_expr(
+ }
+ }
+
++fn build_starts_with_expr(
++ expr: &Arc<dyn PhysicalExpr>,
++ prefix: &str,
++ schema: &Schema,
++ required_columns: &mut RequiredColumns,
++) -> Option<Arc<dyn PhysicalExpr>> {
++ if let Some(col) = expr.as_any().downcast_ref::<phys_expr::Column>() {
++ let field = schema.field_with_name(col.name()).ok()?;
++
++ let min_field = &Field::new(field.name(), DataType::UInt64, true);
++ let min_column_expr = required_columns.min_column_expr(col, expr, min_field).ok()?;
++
++ let max_field = &min_field.clone();
++ let max_column_expr = required_columns.max_column_expr(col, expr, max_field).ok()?;
++
++ let lit_prefix = phys_expr::lit(prefix.to_owned());
++ let lit_1 = phys_expr::lit(1i64);
++ let lit_len = phys_expr::lit(prefix.len() as i64);
++
++ let min_prefix = Arc::new(ScalarFunctionExpr::new(
++ "substr",
++ crate::functions::unicode::substr(),
++ vec![min_column_expr, lit_1.clone(), lit_len.clone()],
++ DataType::Utf8,
++ ));
++ let max_prefix = Arc::new(ScalarFunctionExpr::new(
++ "substr",
++ crate::functions::unicode::substr(),
++ vec![max_column_expr, lit_1.clone(), lit_len.clone()],
++ DataType::Utf8,
++ ));
++
++ Some(Arc::new(phys_expr::BinaryExpr::new(
++ phys_expr::binary(min_prefix, Operator::LtEq, lit_prefix.clone(), schema).ok()?,
++ Operator::And,
++ phys_expr::binary(max_prefix, Operator::GtEq, lit_prefix.clone(), schema).ok()?,
++ )))
++ } else {
++ None
++ }
++}
++
+ /// The maximum number of entries in an `InList` that might be rewritten into
+ /// an OR chain
+ const MAX_LIST_VALUE_SIZE_REWRITE: usize = 20;
+@@ -1684,6 +1726,20 @@ fn build_predicate_expression(
+ };
+ return expr;
+ }
++ if let Some(scalar_fn) = expr_any.downcast_ref::<ScalarFunctionExpr>() {
++ if scalar_fn.name() == "starts_with" {
++ let arg1 = &scalar_fn.args()[0];
++ let arg2 = &scalar_fn.args()[1];
++ if let Some(literal) = arg2.as_any().downcast_ref::<phys_expr::Literal>() {
++ if let ScalarValue::Utf8(Some(prefix)) = literal.value() {
++ return build_starts_with_expr(arg1, prefix, schema, required_columns)
++ .unwrap_or(unhandled);
++ }
++ }
++ }
++ return unhandled;
++ }
++
+
+ let expr_builder =
+ PruningExpressionBuilder::try_new(&left, &right, op, schema, required_columns);
+--
+2.50.1
+
diff --git a/dev/vendors/patches/datafusion/0012-blaze-supports-short-circuiting-And-Or.patch b/dev/vendors/patches/datafusion/0012-blaze-supports-short-circuiting-And-Or.patch
new file mode 100644
index 00000000..290d1571
--- /dev/null
+++ b/dev/vendors/patches/datafusion/0012-blaze-supports-short-circuiting-And-Or.patch
@@ -0,0 +1,529 @@
+From 4ddde6d86cd299607efcf177fd24784f5ca1ede0 Mon Sep 17 00:00:00 2001
+From: zhangli20 <zh...@kuaishou.com>
+Date: Mon, 16 Sep 2024 14:17:08 +0800
+Subject: [PATCH] blaze: supports short circuiting And/Or
+
+---
+ Cargo.lock | 1 +
+ .../physical-expr/src/expressions/mod.rs | 2 +
+ .../src/expressions/short_circuiting.rs | 352 ++++++++++++++++++
+ datafusion/pruning/Cargo.toml | 1 +
+ datafusion/pruning/src/pruning_predicate.rs | 67 ++--
+ 5 files changed, 399 insertions(+), 24 deletions(-)
+ create mode 100644 datafusion/physical-expr/src/expressions/short_circuiting.rs
+
+diff --git a/Cargo.lock b/Cargo.lock
+index a088005a0..ed5cb3d45 100644
+--- a/Cargo.lock
++++ b/Cargo.lock
+@@ -2571,6 +2571,7 @@ dependencies = [
+ "datafusion-datasource",
+ "datafusion-expr",
+ "datafusion-expr-common",
++ "datafusion-functions",
+ "datafusion-functions-nested",
+ "datafusion-physical-expr",
+ "datafusion-physical-expr-common",
+diff --git a/datafusion/physical-expr/src/expressions/mod.rs b/datafusion/physical-expr/src/expressions/mod.rs
+index 8f46133ed..bee04e5da 100644
+--- a/datafusion/physical-expr/src/expressions/mod.rs
++++ b/datafusion/physical-expr/src/expressions/mod.rs
+@@ -31,6 +31,7 @@ mod literal;
+ mod negative;
+ mod no_op;
+ mod not;
++mod short_circuiting;
+ mod try_cast;
+ mod unknown_column;
+
+@@ -52,5 +53,6 @@ pub use literal::{lit, Literal};
+ pub use negative::{negative, NegativeExpr};
+ pub use no_op::NoOp;
+ pub use not::{not, NotExpr};
++pub use short_circuiting::{SCAndExpr, SCOrExpr};
+ pub use try_cast::{try_cast, TryCastExpr};
+ pub use unknown_column::UnKnownColumn;
+diff --git a/datafusion/physical-expr/src/expressions/short_circuiting.rs b/datafusion/physical-expr/src/expressions/short_circuiting.rs
+new file mode 100644
+index 000000000..f47a65f90
+--- /dev/null
++++ b/datafusion/physical-expr/src/expressions/short_circuiting.rs
+@@ -0,0 +1,352 @@
++// Copyright 2022 The Blaze Authors
++//
++// Licensed under the Apache License, Version 2.0 (the "License");
++// you may not use this file except in compliance with the License.
++// You may obtain a copy of the License at
++//
++// http://www.apache.org/licenses/LICENSE-2.0
++//
++// Unless required by applicable law or agreed to in writing, software
++// distributed under the License is distributed on an "AS IS" BASIS,
++// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
++// See the License for the specific language governing permissions and
++// limitations under the License.
++
++use std::any::Any;
++use std::fmt::{Display, Formatter};
++use std::sync::Arc;
++use arrow::array::*;
++use arrow::compute;
++use arrow::datatypes::*;
++use arrow::record_batch::RecordBatch;
++use datafusion_common::{DataFusionError, Result, ScalarValue};
++use datafusion_expr::ColumnarValue;
++use crate::PhysicalExpr;
++
++/// Computes logical OR with short circuiting
++#[derive(Debug, Eq, Hash)]
++pub struct SCAndExpr {
++ pub left: Arc<dyn PhysicalExpr>,
++ pub right: Arc<dyn PhysicalExpr>,
++}
++
++impl PartialEq for SCAndExpr {
++ fn eq(&self, other: &Self) -> bool {
++ self.left.eq(&other.left) && self.right.eq(&other.right)
++ }
++}
++
++impl SCAndExpr {
++ pub fn new(left: Arc<dyn PhysicalExpr>, right: Arc<dyn PhysicalExpr>) -> Self {
++ Self {left, right}
++ }
++}
++
++impl Display for SCAndExpr {
++ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
++ write!(f, "({} &&(SC) {})", self.left, self.right)
++ }
++}
++
++impl PhysicalExpr for SCAndExpr {
++ fn as_any(&self) -> &dyn Any {
++ self
++ }
++
++ fn data_type(&self, _input_schema: &Schema) -> Result<DataType> {
++ Ok(DataType::Boolean)
++ }
++
++ fn nullable(&self, _input_schema: &Schema) -> Result<bool> {
++ Ok(true)
++ }
++
++ fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
++ evaluate_and(&self.left, &self.right, batch)
++ }
++
++ fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
++ vec![&self.left, &self.right]
++ }
++
++ fn with_new_children(
++ self: Arc<Self>,
++ children: Vec<Arc<dyn PhysicalExpr>>,
++ ) -> Result<Arc<dyn PhysicalExpr>> {
++ Ok(Arc::new(Self::new(
++ children[0].clone(),
++ children[1].clone(),
++ )))
++ }
++
++ fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
++ self.left.fmt_sql(f)?;
++ write!(f, " &&(SC) ")?;
++ self.right.fmt_sql(f)?;
++ Ok(())
++ }
++}
++
++/// Computes logical AND with short circuiting
++#[derive(Debug, Eq, Hash)]
++pub struct SCOrExpr {
++ pub left: Arc<dyn PhysicalExpr>,
++ pub right: Arc<dyn PhysicalExpr>,
++}
++
++impl PartialEq for SCOrExpr {
++ fn eq(&self, other: &Self) -> bool {
++ self.left.eq(&other.left) && self.right.eq(&other.right)
++ }
++}
++
++impl SCOrExpr {
++ pub fn new(left: Arc<dyn PhysicalExpr>, right: Arc<dyn PhysicalExpr>) -> Self {
++ Self {left, right}
++ }
++}
++
++impl Display for SCOrExpr {
++ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
++ write!(f, "({} ||(SC) {})", self.left, self.right)
++ }
++}
++
++impl PhysicalExpr for SCOrExpr {
++ fn as_any(&self) -> &dyn Any {
++ self
++ }
++
++ fn data_type(&self, _input_schema: &Schema) -> Result<DataType> {
++ Ok(DataType::Boolean)
++ }
++
++ fn nullable(&self, _input_schema: &Schema) -> Result<bool> {
++ Ok(true)
++ }
++
++ fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
++ evaluate_or(&self.left, &self.right, batch)
++ }
++
++ fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
++ vec![&self.left, &self.right]
++ }
++
++ fn with_new_children(
++ self: Arc<Self>,
++ children: Vec<Arc<dyn PhysicalExpr>>,
++ ) -> Result<Arc<dyn PhysicalExpr>> {
++ Ok(Arc::new(Self::new(
++ children[0].clone(),
++ children[1].clone(),
++ )))
++ }
++
++ fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
++ self.left.fmt_sql(f)?;
++ write!(f, " ||(SC) ")?;
++ self.right.fmt_sql(f)?;
++ Ok(())
++ }
++}
++
++fn evaluate_and(
++ left: &Arc<dyn PhysicalExpr>,
++ right: &Arc<dyn PhysicalExpr>,
++ batch: &RecordBatch,
++) -> Result<ColumnarValue> {
++ Ok(match left.evaluate(batch)? {
++ ColumnarValue::Scalar(ScalarValue::Boolean(Some(true))) => right.evaluate(batch)?,
++ ColumnarValue::Scalar(ScalarValue::Boolean(Some(false))) => {
++ ColumnarValue::Scalar(ScalarValue::Boolean(Some(false)))
++ }
++
++ ColumnarValue::Scalar(s) if s.is_null() => match right.evaluate(batch)? {
++ ColumnarValue::Scalar(ScalarValue::Boolean(Some(true))) => {
++ ColumnarValue::Scalar(ScalarValue::Boolean(None))
++ }
++ ColumnarValue::Scalar(ScalarValue::Boolean(Some(false))) => {
++ ColumnarValue::Scalar(ScalarValue::Boolean(Some(false)))
++ }
++ ColumnarValue::Scalar(v) if v.is_null() => {
++ ColumnarValue::Scalar(ScalarValue::Boolean(None))
++ }
++ ColumnarValue::Array(array) => {
++ ColumnarValue::Array(compute::nullif(&array, as_boolean_array(&array))?)
++ }
++ _ => {
++ return Err(DataFusionError::Internal(
++ "AND: invalid operands".to_string(),
++ ))
++ }
++ },
++ ColumnarValue::Array(left) => {
++ let left_prim = as_boolean_array(&left);
++ let right_selected = if left_prim.null_count() > 0 {
++ right.evaluate_selection(
++ batch,
++ &compute::not(&compute::prep_null_mask_filter(&compute::not(left_prim)?))?,
++ )?
++ } else {
++ right.evaluate_selection(batch, left_prim)?
++ };
++ let right = right_selected.into_array(left.len())?;
++ let right_prim = as_boolean_array(&right);
++ ColumnarValue::Array(Arc::new(compute::and_kleene(left_prim, right_prim)?))
++ }
++ _ => {
++ return Err(DataFusionError::Internal(
++ "AND: invalid operands".to_string(),
++ ))
++ }
++ })
++}
++
++fn evaluate_or(
++ left: &Arc<dyn PhysicalExpr>,
++ right: &Arc<dyn PhysicalExpr>,
++ batch: &RecordBatch,
++) -> Result<ColumnarValue> {
++ Ok(match left.evaluate(batch)? {
++ ColumnarValue::Scalar(ScalarValue::Boolean(Some(true))) => {
++ ColumnarValue::Scalar(ScalarValue::Boolean(Some(true)))
++ }
++ ColumnarValue::Scalar(ScalarValue::Boolean(Some(false))) => right.evaluate(batch)?,
++
++ ColumnarValue::Scalar(s) if s.is_null() => match right.evaluate(batch)? {
++ ColumnarValue::Scalar(ScalarValue::Boolean(Some(true))) => {
++ ColumnarValue::Scalar(ScalarValue::Boolean(Some(true)))
++ }
++ ColumnarValue::Scalar(ScalarValue::Boolean(Some(false))) => {
++ ColumnarValue::Scalar(ScalarValue::Boolean(None))
++ }
++ ColumnarValue::Scalar(v) if v.is_null() => {
++ ColumnarValue::Scalar(ScalarValue::Boolean(None))
++ }
++ ColumnarValue::Array(array) => ColumnarValue::Array(compute::nullif(
++ &array,
++ &compute::not(as_boolean_array(&array))?,
++ )?),
++ _ => {
++ return Err(DataFusionError::Internal(
++ "OR: invalid operands".to_string(),
++ ))
++ }
++ },
++ ColumnarValue::Array(left) => {
++ let left_prim = as_boolean_array(&left);
++ let right_selected = if left_prim.null_count() > 0 {
++ right.evaluate_selection(
++ batch,
++ &compute::not(&compute::prep_null_mask_filter(left_prim))?,
++ )?
++ } else {
++ right.evaluate_selection(batch, &compute::not(left_prim)?)?
++ };
++ let right = right_selected.into_array(left.len())?;
++ let right_prim = as_boolean_array(&right);
++ ColumnarValue::Array(Arc::new(compute::or_kleene(left_prim, right_prim)?))
++ }
++ _ => {
++ return Err(DataFusionError::Internal(
++ "OR: invalid operands".to_string(),
++ ))
++ }
++ })
++}
++
++#[cfg(test)]
++mod test {
++ use arrow::array::*;
++ use arrow::record_batch::RecordBatch;
++ use std::sync::Arc;
++ use crate::expressions::Column;
++ use crate::expressions::short_circuiting::{SCAndExpr, SCOrExpr};
++ use crate::PhysicalExpr;
++
++ #[test]
++ fn test() {
++ let arg1: ArrayRef = Arc::new(BooleanArray::from_iter(&[
++ Some(true),
++ Some(true),
++ Some(true),
++ Some(false),
++ Some(false),
++ Some(false),
++ None,
++ None,
++ None,
++ ]));
++ let arg2: ArrayRef = Arc::new(BooleanArray::from_iter(&[
++ Some(true),
++ Some(false),
++ None,
++ Some(true),
++ Some(false),
++ None,
++ Some(true),
++ Some(false),
++ None,
++ ]));
++ let batch =
++ RecordBatch::try_from_iter_with_nullable([("a", arg1, true), ("b", arg2, true)])
++ .unwrap();
++
++ // +---------+---------+---------+---------+
++ // | AND | TRUE | FALSE | UNKNOWN |
++ // +---------+---------+---------+---------+
++ // | TRUE | TRUE | FALSE | UNKNOWN |
++ // | FALSE | FALSE | FALSE | FALSE |
++ // | UNKNOWN | UNKNOWN | FALSE | UNKNOWN |
++ // +---------+---------+---------+---------+
++ let output = SCAndExpr::new(Arc::new(Column::new("a", 0)), Arc::new(Column::new("b", 1)))
++ .evaluate(&batch)
++ .unwrap()
++ .into_array(9)
++ .unwrap();
++
++ assert_eq!(
++ as_boolean_array(&output).into_iter().collect::<Vec<_>>(),
++ vec![
++ Some(true),
++ Some(false),
++ None,
++ Some(false),
++ Some(false),
++ Some(false),
++ None,
++ Some(false),
++ None,
++ ]
++ );
++
++ // +---------+---------+---------+---------+
++ // | OR | TRUE | FALSE | UNKNOWN |
++ // +---------+---------+---------+---------+
++ // | TRUE | TRUE | TRUE | TRUE |
++ // | FALSE | TRUE | FALSE | UNKNOWN |
++ // | UNKNOWN | TRUE | UNKNOWN | UNKNOWN |
++ // +---------+---------+---------+---------+
++ let output = SCOrExpr::new(Arc::new(Column::new("a", 0)), Arc::new(Column::new("b", 1)))
++ .evaluate(&batch)
++ .unwrap()
++ .into_array(9)
++ .unwrap();
++
++ assert_eq!(
++ as_boolean_array(&output).into_iter().collect::<Vec<_>>(),
++ vec![
++ Some(true),
++ Some(true),
++ Some(true),
++ Some(true),
++ Some(false),
++ None,
++ Some(true),
++ None,
++ None,
++ ]
++ );
++ }
++}
+diff --git a/datafusion/pruning/Cargo.toml b/datafusion/pruning/Cargo.toml
+index 90b31b933..dc8b8fd57 100644
+--- a/datafusion/pruning/Cargo.toml
++++ b/datafusion/pruning/Cargo.toml
+@@ -18,6 +18,7 @@ datafusion-common = { workspace = true, default-features = true }
+ datafusion-datasource = { workspace = true }
+ datafusion-expr = { workspace = true }
+ datafusion-expr-common = { workspace = true, default-features = true }
++datafusion-functions = { workspace = true }
+ datafusion-functions-nested = { workspace = true }
+ datafusion-physical-expr = { workspace = true }
+ datafusion-physical-expr-common = { workspace = true }
+diff --git a/datafusion/pruning/src/pruning_predicate.rs b/datafusion/pruning/src/pruning_predicate.rs
+index 648d32374..0e4953039 100644
+--- a/datafusion/pruning/src/pruning_predicate.rs
++++ b/datafusion/pruning/src/pruning_predicate.rs
+@@ -1511,20 +1511,19 @@ fn build_starts_with_expr(
+
+ let min_prefix = Arc::new(ScalarFunctionExpr::new(
+ "substr",
+- crate::functions::unicode::substr(),
++ datafusion_functions::unicode::substr(),
+ vec![min_column_expr, lit_1.clone(), lit_len.clone()],
+- DataType::Utf8,
++ Arc::new(Field::new("result", DataType::Utf8, true)),
+ ));
+ let max_prefix = Arc::new(ScalarFunctionExpr::new(
+ "substr",
+- crate::functions::unicode::substr(),
++ datafusion_functions::unicode::substr(),
+ vec![max_column_expr, lit_1.clone(), lit_len.clone()],
+- DataType::Utf8,
++ Arc::new(Field::new("result", DataType::Utf8, true)),
+ ));
+
+- Some(Arc::new(phys_expr::BinaryExpr::new(
++ Some(Arc::new(phys_expr::SCAndExpr::new(
+ phys_expr::binary(min_prefix, Operator::LtEq, lit_prefix.clone(), schema).ok()?,
+- Operator::And,
+ phys_expr::binary(max_prefix, Operator::GtEq, lit_prefix.clone(), schema).ok()?,
+ )))
+ } else {
+@@ -1673,6 +1672,38 @@ fn build_predicate_expression(
+ }
+ }
+
++ if let Some(sc_and) = expr_any.downcast_ref::<phys_expr::SCAndExpr>() {
++ let left_expr = build_predicate_expression(&sc_and.left, schema, required_columns, unhandled_hook);
++ let right_expr = build_predicate_expression(&sc_and.right, schema, required_columns, unhandled_hook);
++ return match (&left_expr, &right_expr) {
++ (l, _) if is_always_true(l) => right_expr,
++ (_, r) if is_always_true(r) => left_expr,
++ (_, _) => Arc::new(phys_expr::SCAndExpr::new(left_expr, right_expr)),
++ };
++ }
++ if let Some(sc_or) = expr_any.downcast_ref::<phys_expr::SCOrExpr>() {
++ let left_expr = build_predicate_expression(&sc_or.left, schema, required_columns, unhandled_hook);
++ let right_expr = build_predicate_expression(&sc_or.right, schema, required_columns, unhandled_hook);
++ if is_always_true(&left_expr) || is_always_true(&right_expr) {
++ return unhandled_hook.handle(expr);
++ } else {
++ return Arc::new(phys_expr::SCOrExpr::new(left_expr, right_expr));
++ }
++ }
++ if let Some(scalar_fn) = expr_any.downcast_ref::<ScalarFunctionExpr>() {
++ if scalar_fn.name() == "starts_with" {
++ let arg1 = &scalar_fn.args()[0];
++ let arg2 = &scalar_fn.args()[1];
++ if let Some(literal) = arg2.as_any().downcast_ref::<phys_expr::Literal>() {
++ if let ScalarValue::Utf8(Some(prefix)) = literal.value() {
++ return build_starts_with_expr(arg1, prefix, schema, required_columns)
++ .unwrap_or_else(|| unhandled_hook.handle(expr));
++ }
++ }
++ }
++ return unhandled_hook.handle(expr);
++ }
++
+ let (left, op, right) = {
+ if let Some(bin_expr) = expr_any.downcast_ref::<phys_expr::BinaryExpr>() {
+ (
+@@ -1722,24 +1753,13 @@ fn build_predicate_expression(
+ (left, Operator::Or, _) if is_always_false(left) => right_expr,
+ (_, Operator::Or, right) if is_always_false(right) => left_expr,
+
++ (_, Operator::And, _) => Arc::new(phys_expr::SCAndExpr::new(left_expr, right_expr)),
++ (_, Operator::Or, _) => Arc::new(phys_expr::SCOrExpr::new(left_expr, right_expr)),
+ _ => Arc::new(phys_expr::BinaryExpr::new(left_expr, op, right_expr)),
++
+ };
+ return expr;
+ }
+- if let Some(scalar_fn) = expr_any.downcast_ref::<ScalarFunctionExpr>() {
+- if scalar_fn.name() == "starts_with" {
+- let arg1 = &scalar_fn.args()[0];
+- let arg2 = &scalar_fn.args()[1];
+- if let Some(literal) = arg2.as_any().downcast_ref::<phys_expr::Literal>() {
+- if let ScalarValue::Utf8(Some(prefix)) = literal.value() {
+- return build_starts_with_expr(arg1, prefix, schema, required_columns)
+- .unwrap_or(unhandled);
+- }
+- }
+- }
+- return unhandled;
+- }
+-
+
+ let expr_builder =
+ PruningExpressionBuilder::try_new(&left, &right, op, schema, required_columns);
+@@ -1805,10 +1825,9 @@ fn build_statistics_expr(
+ let has_no_stat_pred = Arc::new(phys_expr::IsNullExpr::new(stat_pred_expr.clone()));
+ let has_no_dict = Arc::new(phys_expr::IsNullExpr::new(dict));
+ let dict_contained = expr_builder.dict_column_contains_expr()?;
+- Arc::new(phys_expr::BinaryExpr::new(
+- Arc::new(phys_expr::BinaryExpr::new(has_no_stat_pred, Operator::Or, stat_pred_expr)),
+- Operator::And,
+- Arc::new(phys_expr::BinaryExpr::new(has_no_dict, Operator::Or, dict_contained)),
++ Arc::new(phys_expr::SCAndExpr::new(
++ Arc::new(phys_expr::SCOrExpr::new(has_no_stat_pred, stat_pred_expr)),
++ Arc::new(phys_expr::SCOrExpr::new(has_no_dict, dict_contained)),
+ ))
+ }
+ Operator::NotLikeMatch => build_not_like_match(expr_builder)?,
+--
+2.50.1
+
diff --git a/dev/vendors/patches/datafusion/0013-blaze-make-parquet-reader-components-public.patch b/dev/vendors/patches/datafusion/0013-blaze-make-parquet-reader-components-public.patch
new file mode 100644
index 00000000..260a9c4d
--- /dev/null
+++ b/dev/vendors/patches/datafusion/0013-blaze-make-parquet-reader-components-public.patch
@@ -0,0 +1,75 @@
+From 948d5bc4cd7e4a63c99d0927dc8070420534ba0d Mon Sep 17 00:00:00 2001
+From: zhangli20 <zh...@kuaishou.com>
+Date: Tue, 17 Sep 2024 21:01:01 +0800
+Subject: [PATCH] blaze: make parquet reader components public
+
+---
+ datafusion/datasource-parquet/src/mod.rs | 4 ++--
+ datafusion/datasource-parquet/src/opener.rs | 4 ++--
+ datafusion/datasource/src/schema_adapter.rs | 6 +++---
+ 3 files changed, 7 insertions(+), 7 deletions(-)
+
+diff --git a/datafusion/datasource-parquet/src/mod.rs b/datafusion/datasource-parquet/src/mod.rs
+index 0b4e86240..78152459f 100644
+--- a/datafusion/datasource-parquet/src/mod.rs
++++ b/datafusion/datasource-parquet/src/mod.rs
+@@ -22,8 +22,8 @@
+ pub mod access_plan;
+ pub mod file_format;
+ mod metrics;
+-mod opener;
+-mod page_filter;
++pub mod opener;
++pub mod page_filter;
+ mod reader;
+ mod row_filter;
+ mod row_group_filter;
+diff --git a/datafusion/datasource-parquet/src/opener.rs b/datafusion/datasource-parquet/src/opener.rs
+index e31e33cfa..914e40074 100644
+--- a/datafusion/datasource-parquet/src/opener.rs
++++ b/datafusion/datasource-parquet/src/opener.rs
+@@ -52,7 +52,7 @@ use parquet::arrow::{ParquetRecordBatchStreamBuilder, ProjectionMask};
+ use parquet::file::metadata::ParquetMetaDataReader;
+
+ /// Implements [`FileOpener`] for a parquet file
+-pub(super) struct ParquetOpener {
++pub struct ParquetOpener {
+ /// Execution partition index
+ pub partition_index: usize,
+ /// Column indexes in `table_schema` needed by the query
+@@ -95,7 +95,7 @@ pub(super) struct ParquetOpener {
+ /// Optional parquet FileDecryptionProperties
+ pub file_decryption_properties: Option<Arc<FileDecryptionProperties>>,
+ /// Rewrite expressions in the context of the file schema
+- pub(crate) expr_adapter_factory: Option<Arc<dyn PhysicalExprAdapterFactory>>,
++ pub expr_adapter_factory: Option<Arc<dyn PhysicalExprAdapterFactory>>,
+ }
+
+ impl FileOpener for ParquetOpener {
+diff --git a/datafusion/datasource/src/schema_adapter.rs b/datafusion/datasource/src/schema_adapter.rs
+index 12722b433..1773b6011 100644
+--- a/datafusion/datasource/src/schema_adapter.rs
++++ b/datafusion/datasource/src/schema_adapter.rs
+@@ -355,16 +355,16 @@ where
+ pub struct SchemaMapping {
+ /// The schema of the table. This is the expected schema after conversion
+ /// and it should match the schema of the query result.
+- projected_table_schema: SchemaRef,
++ pub projected_table_schema: SchemaRef,
+ /// Mapping from field index in `projected_table_schema` to index in
+ /// projected file_schema.
+ ///
+ /// They are Options instead of just plain `usize`s because the table could
+ /// have fields that don't exist in the file.
+- field_mappings: Vec<Option<usize>>,
++ pub field_mappings: Vec<Option<usize>>,
+ /// Function used to adapt a column from the file schema to the table schema
+ /// when it exists in both schemas
+- cast_column: Arc<CastColumnFn>,
++ pub cast_column: Arc<CastColumnFn>,
+ }
+
+ impl Debug for SchemaMapping {
+--
+2.50.1
+
diff --git a/dev/vendors/patches/datafusion/0014-blaze-implement-TestMemoryExec-as-any.patch b/dev/vendors/patches/datafusion/0014-blaze-implement-TestMemoryExec-as-any.patch
new file mode 100644
index 00000000..85738dd5
--- /dev/null
+++ b/dev/vendors/patches/datafusion/0014-blaze-implement-TestMemoryExec-as-any.patch
@@ -0,0 +1,25 @@
+From 833c0abf2721625258ed9cc08bb8546cdc7fdb64 Mon Sep 17 00:00:00 2001
+From: zhangli20 <zh...@zhangli20deMacBook-Pro.local>
+Date: Tue, 12 Aug 2025 16:45:29 +0800
+Subject: [PATCH] blaze: implement TestMemoryExec.as_any()
+
+---
+ datafusion/physical-plan/src/test.rs | 2 +-
+ 1 file changed, 1 insertion(+), 1 deletion(-)
+
+diff --git a/datafusion/physical-plan/src/test.rs b/datafusion/physical-plan/src/test.rs
+index be921e058..349f9955b 100644
+--- a/datafusion/physical-plan/src/test.rs
++++ b/datafusion/physical-plan/src/test.rs
+@@ -131,7 +131,7 @@ impl ExecutionPlan for TestMemoryExec {
+ }
+
+ fn as_any(&self) -> &dyn Any {
+- unimplemented!()
++ self
+ }
+
+ fn properties(&self) -> &PlanProperties {
+--
+2.50.1
+
diff --git a/dev/vendors/patches/datafusion/0015-blaze-revert-substr-to-v45.0.0-for-unsupported-string-view.patch b/dev/vendors/patches/datafusion/0015-blaze-revert-substr-to-v45.0.0-for-unsupported-string-view.patch
new file mode 100644
index 00000000..dbcf4b1b
--- /dev/null
+++ b/dev/vendors/patches/datafusion/0015-blaze-revert-substr-to-v45.0.0-for-unsupported-string-view.patch
@@ -0,0 +1,361 @@
+From 634a09f0df511ef69ec0026a43e16d70188166d6 Mon Sep 17 00:00:00 2001
+From: zhangli20 <ri...@gmail.com>
+Date: Wed, 13 Aug 2025 15:53:27 +0800
+Subject: [PATCH] blaze: revert substr() to v45.0.0 for unsupported string view
+ types
+
+---
+ datafusion/functions/src/unicode/substr.rs | 129 ++++++++++-----------
+ 1 file changed, 64 insertions(+), 65 deletions(-)
+
+diff --git a/datafusion/functions/src/unicode/substr.rs b/datafusion/functions/src/unicode/substr.rs
+index 4dcbea480..d69b5d04b 100644
+--- a/datafusion/functions/src/unicode/substr.rs
++++ b/datafusion/functions/src/unicode/substr.rs
+@@ -19,18 +19,16 @@ use std::any::Any;
+ use std::sync::Arc;
+
+ use crate::strings::make_and_append_view;
+-use crate::utils::make_scalar_function;
++use crate::utils::{make_scalar_function, utf8_to_str_type};
+ use arrow::array::{
+- Array, ArrayIter, ArrayRef, AsArray, Int64Array, NullBufferBuilder, StringArrayType,
+- StringViewArray, StringViewBuilder,
++ Array, ArrayIter, ArrayRef, AsArray, GenericStringBuilder, Int64Array,
++ OffsetSizeTrait, StringArrayType, StringViewArray,
+ };
+-use arrow::buffer::ScalarBuffer;
+ use arrow::datatypes::DataType;
++use arrow_buffer::{NullBufferBuilder, ScalarBuffer};
+ use datafusion_common::cast::as_int64_array;
+ use datafusion_common::{exec_err, plan_err, Result};
+-use datafusion_expr::{
+- ColumnarValue, Documentation, ScalarUDFImpl, Signature, Volatility,
+-};
++use datafusion_expr::{ColumnarValue, Documentation, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility};
+ use datafusion_macros::user_doc;
+
+ #[user_doc(
+@@ -90,15 +88,15 @@ impl ScalarUDFImpl for SubstrFunc {
+ &self.signature
+ }
+
+- // `SubstrFunc` always generates `Utf8View` output for its efficiency.
+- fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+- Ok(DataType::Utf8View)
++ fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
++ if arg_types[0] == DataType::Utf8View {
++ Ok(DataType::Utf8View)
++ } else {
++ utf8_to_str_type(&arg_types[0], "substr")
++ }
+ }
+
+- fn invoke_with_args(
+- &self,
+- args: datafusion_expr::ScalarFunctionArgs,
+- ) -> Result<ColumnarValue> {
++ fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
+ make_scalar_function(substr, vec![])(&args.args)
+ }
+
+@@ -185,11 +183,11 @@ pub fn substr(args: &[ArrayRef]) -> Result<ArrayRef> {
+ match args[0].data_type() {
+ DataType::Utf8 => {
+ let string_array = args[0].as_string::<i32>();
+- string_substr::<_>(string_array, &args[1..])
++ string_substr::<_, i32>(string_array, &args[1..])
+ }
+ DataType::LargeUtf8 => {
+ let string_array = args[0].as_string::<i64>();
+- string_substr::<_>(string_array, &args[1..])
++ string_substr::<_, i64>(string_array, &args[1..])
+ }
+ DataType::Utf8View => {
+ let string_array = args[0].as_string_view();
+@@ -425,9 +423,10 @@ fn string_view_substr(
+ }
+ }
+
+-fn string_substr<'a, V>(string_array: V, args: &[ArrayRef]) -> Result<ArrayRef>
++fn string_substr<'a, V, T>(string_array: V, args: &[ArrayRef]) -> Result<ArrayRef>
+ where
+ V: StringArrayType<'a>,
++ T: OffsetSizeTrait,
+ {
+ let start_array = as_int64_array(&args[0])?;
+ let count_array_opt = if args.len() == 2 {
+@@ -442,7 +441,7 @@ where
+ match args.len() {
+ 1 => {
+ let iter = ArrayIter::new(string_array);
+- let mut result_builder = StringViewBuilder::new();
++ let mut result_builder = GenericStringBuilder::<T>::new();
+ for (string, start) in iter.zip(start_array.iter()) {
+ match (string, start) {
+ (Some(string), Some(start)) => {
+@@ -465,7 +464,7 @@ where
+ 2 => {
+ let iter = ArrayIter::new(string_array);
+ let count_array = count_array_opt.unwrap();
+- let mut result_builder = StringViewBuilder::new();
++ let mut result_builder = GenericStringBuilder::<T>::new();
+
+ for ((string, start), count) in
+ iter.zip(start_array.iter()).zip(count_array.iter())
+@@ -507,8 +506,8 @@ where
+
+ #[cfg(test)]
+ mod tests {
+- use arrow::array::{Array, StringViewArray};
+- use arrow::datatypes::DataType::Utf8View;
++ use arrow::array::{Array, StringArray, StringViewArray};
++ use arrow::datatypes::DataType::{Utf8, Utf8View};
+
+ use datafusion_common::{exec_err, Result, ScalarValue};
+ use datafusion_expr::{ColumnarValue, ScalarUDFImpl};
+@@ -618,8 +617,8 @@ mod tests {
+ ],
+ Ok(Some("alphabet")),
+ &str,
+- Utf8View,
+- StringViewArray
++ Utf8,
++ StringArray
+ );
+ test_function!(
+ SubstrFunc::new(),
+@@ -629,8 +628,8 @@ mod tests {
+ ],
+ Ok(Some("ésoj")),
+ &str,
+- Utf8View,
+- StringViewArray
++ Utf8,
++ StringArray
+ );
+ test_function!(
+ SubstrFunc::new(),
+@@ -640,8 +639,8 @@ mod tests {
+ ],
+ Ok(Some("joséésoj")),
+ &str,
+- Utf8View,
+- StringViewArray
++ Utf8,
++ StringArray
+ );
+ test_function!(
+ SubstrFunc::new(),
+@@ -651,8 +650,8 @@ mod tests {
+ ],
+ Ok(Some("alphabet")),
+ &str,
+- Utf8View,
+- StringViewArray
++ Utf8,
++ StringArray
+ );
+ test_function!(
+ SubstrFunc::new(),
+@@ -662,8 +661,8 @@ mod tests {
+ ],
+ Ok(Some("lphabet")),
+ &str,
+- Utf8View,
+- StringViewArray
++ Utf8,
++ StringArray
+ );
+ test_function!(
+ SubstrFunc::new(),
+@@ -673,8 +672,8 @@ mod tests {
+ ],
+ Ok(Some("phabet")),
+ &str,
+- Utf8View,
+- StringViewArray
++ Utf8,
++ StringArray
+ );
+ test_function!(
+ SubstrFunc::new(),
+@@ -684,8 +683,8 @@ mod tests {
+ ],
+ Ok(Some("alphabet")),
+ &str,
+- Utf8View,
+- StringViewArray
++ Utf8,
++ StringArray
+ );
+ test_function!(
+ SubstrFunc::new(),
+@@ -695,8 +694,8 @@ mod tests {
+ ],
+ Ok(Some("")),
+ &str,
+- Utf8View,
+- StringViewArray
++ Utf8,
++ StringArray
+ );
+ test_function!(
+ SubstrFunc::new(),
+@@ -706,8 +705,8 @@ mod tests {
+ ],
+ Ok(None),
+ &str,
+- Utf8View,
+- StringViewArray
++ Utf8,
++ StringArray
+ );
+ test_function!(
+ SubstrFunc::new(),
+@@ -718,8 +717,8 @@ mod tests {
+ ],
+ Ok(Some("ph")),
+ &str,
+- Utf8View,
+- StringViewArray
++ Utf8,
++ StringArray
+ );
+ test_function!(
+ SubstrFunc::new(),
+@@ -730,8 +729,8 @@ mod tests {
+ ],
+ Ok(Some("phabet")),
+ &str,
+- Utf8View,
+- StringViewArray
++ Utf8,
++ StringArray
+ );
+ test_function!(
+ SubstrFunc::new(),
+@@ -742,8 +741,8 @@ mod tests {
+ ],
+ Ok(Some("alph")),
+ &str,
+- Utf8View,
+- StringViewArray
++ Utf8,
++ StringArray
+ );
+ // starting from 5 (10 + -5)
+ test_function!(
+@@ -755,8 +754,8 @@ mod tests {
+ ],
+ Ok(Some("alph")),
+ &str,
+- Utf8View,
+- StringViewArray
++ Utf8,
++ StringArray
+ );
+ // starting from -1 (4 + -5)
+ test_function!(
+@@ -768,8 +767,8 @@ mod tests {
+ ],
+ Ok(Some("")),
+ &str,
+- Utf8View,
+- StringViewArray
++ Utf8,
++ StringArray
+ );
+ // starting from 0 (5 + -5)
+ test_function!(
+@@ -781,8 +780,8 @@ mod tests {
+ ],
+ Ok(Some("")),
+ &str,
+- Utf8View,
+- StringViewArray
++ Utf8,
++ StringArray
+ );
+ test_function!(
+ SubstrFunc::new(),
+@@ -793,8 +792,8 @@ mod tests {
+ ],
+ Ok(None),
+ &str,
+- Utf8View,
+- StringViewArray
++ Utf8,
++ StringArray
+ );
+ test_function!(
+ SubstrFunc::new(),
+@@ -805,8 +804,8 @@ mod tests {
+ ],
+ Ok(None),
+ &str,
+- Utf8View,
+- StringViewArray
++ Utf8,
++ StringArray
+ );
+ test_function!(
+ SubstrFunc::new(),
+@@ -817,8 +816,8 @@ mod tests {
+ ],
+ exec_err!("negative substring length not allowed: substr(<str>, 1, -1)"),
+ &str,
+- Utf8View,
+- StringViewArray
++ Utf8,
++ StringArray
+ );
+ test_function!(
+ SubstrFunc::new(),
+@@ -829,8 +828,8 @@ mod tests {
+ ],
+ Ok(Some("és")),
+ &str,
+- Utf8View,
+- StringViewArray
++ Utf8,
++ StringArray
+ );
+ #[cfg(not(feature = "unicode_expressions"))]
+ test_function!(
+@@ -843,8 +842,8 @@ mod tests {
+ "function substr requires compilation with feature flag: unicode_expressions."
+ ),
+ &str,
+- Utf8View,
+- StringViewArray
++ Utf8,
++ StringArray
+ );
+ test_function!(
+ SubstrFunc::new(),
+@@ -854,8 +853,8 @@ mod tests {
+ ],
+ Ok(Some("abc")),
+ &str,
+- Utf8View,
+- StringViewArray
++ Utf8,
++ StringArray
+ );
+ test_function!(
+ SubstrFunc::new(),
+@@ -866,8 +865,8 @@ mod tests {
+ ],
+ exec_err!("negative overflow when calculating skip value"),
+ &str,
+- Utf8View,
+- StringViewArray
++ Utf8,
++ StringArray
+ );
+
+ Ok(())
+--
+2.50.1
+
diff --git a/dev/vendors/patches/orc-rust/0001-blaze-adapt-arrow-dependency-to-v55.2.0-blaze.patch b/dev/vendors/patches/orc-rust/0001-blaze-adapt-arrow-dependency-to-v55.2.0-blaze.patch
new file mode 100644
index 00000000..9bf85a2d
--- /dev/null
+++ b/dev/vendors/patches/orc-rust/0001-blaze-adapt-arrow-dependency-to-v55.2.0-blaze.patch
@@ -0,0 +1,36 @@
+From de680c77ada706215ef25b9a2c8c949fb0315bd0 Mon Sep 17 00:00:00 2001
+From: zhangli20 <zh...@kuaishou.com>
+Date: Mon, 27 Jul 2026 02:04:47 +0800
+Subject: [PATCH] blaze: adapt arrow dependency to v55.2.0-blaze
+
+---
+ Cargo.toml | 6 +++---
+ 1 file changed, 3 insertions(+), 3 deletions(-)
+
+diff --git a/Cargo.toml b/Cargo.toml
+index 3c2683a..e9ffa4c 100644
+--- a/Cargo.toml
++++ b/Cargo.toml
+@@ -32,7 +32,7 @@ rust-version = "1.73"
+ all-features = true
+
+ [dependencies]
+-arrow = { version = "57.0", features = ["prettyprint", "chrono-tz", "ipc_compression"] }
++arrow = { version = "55.2.0", features = ["prettyprint", "chrono-tz", "ipc_compression"] }
+ bytemuck = { version = "1.22.0", features = ["must_cast"] }
+ bytes = "1.10.1"
+ chrono = { version = "0.4.41", default-features = false, features = ["std"] }
+@@ -68,8 +68,8 @@ clap = { version = "4.5.4", features = ["derive"], optional = true }
+ opendal = { version = "0.53", optional = true, default-features = false }
+
+ [dev-dependencies]
+-arrow-ipc = { version = "57.0", features = ["lz4"] }
+-arrow-json = "57.0"
++arrow-ipc = { version = "55.2.0", features = ["lz4"] }
++arrow-json = "55.2.0"
+ criterion = { version = "0.5", default-features = false, features = ["async_tokio"] }
+ opendal = { version = "0.53", default-features = false, features = ["services-memory"] }
+ pretty_assertions = "1.3.0"
+--
+2.50.1
+
diff --git a/dev/vendors/patches/serde_json/0001-blaze-ignore-invalid-escape.patch b/dev/vendors/patches/serde_json/0001-blaze-ignore-invalid-escape.patch
new file mode 100644
index 00000000..39acd534
--- /dev/null
+++ b/dev/vendors/patches/serde_json/0001-blaze-ignore-invalid-escape.patch
@@ -0,0 +1,46 @@
+From 11ac9a213d7e8dedfe7e94af30d93dfa823b3217 Mon Sep 17 00:00:00 2001
+From: zhangli20 <zh...@kuaishou.com>
+Date: Wed, 17 May 2023 15:48:45 +0800
+Subject: [PATCH] blaze: ignore invalid escape
+
+---
+ src/read.rs | 16 ++++++++++++++++
+ 1 file changed, 16 insertions(+)
+
+diff --git a/src/read.rs b/src/read.rs
+index fc3a3ca..cac8132 100644
+--- a/src/read.rs
++++ b/src/read.rs
+@@ -849,6 +849,15 @@ fn parse_escape<'de, R: Read<'de>>(
+ validate: bool,
+ scratch: &mut Vec<u8>,
+ ) -> Result<()> {
++
++ // blaze:
++ // invalid escape case: decode the backslash char as b'\\'
++ let peek_ch = tri!(peek_or_eof(read));
++ if ![b'"', b'\\', b'/', b'b', b'f', b'n', b'r', b't', b'u'].contains(&peek_ch) {
++ scratch.push(b'\\');
++ return Ok(());
++ }
++
+ let ch = tri!(next_or_eof(read));
+
+ match ch {
+@@ -950,6 +959,13 @@ fn ignore_escape<'de, R>(read: &mut R) -> Result<()>
+ where
+ R: ?Sized + Read<'de>,
+ {
++ // blaze:
++ // invalid escape case: decode the backslash char as b'\\'
++ let peek_ch = tri!(peek_or_eof(read));
++ if ![b'"', b'\\', b'/', b'b', b'f', b'n', b'r', b't', b'u'].contains(&peek_ch) {
++ return Ok(());
++ }
++
+ let ch = tri!(next_or_eof(read));
+
+ match ch {
+--
+2.50.1
+
diff --git a/dev/vendors/patches/serde_json/0002-blaze-do-not-force-escaping-control-characters.patch b/dev/vendors/patches/serde_json/0002-blaze-do-not-force-escaping-control-characters.patch
new file mode 100644
index 00000000..da70d28f
--- /dev/null
+++ b/dev/vendors/patches/serde_json/0002-blaze-do-not-force-escaping-control-characters.patch
@@ -0,0 +1,31 @@
+From 95fa6cbac6b7146a8883f5ce761316d11282399b Mon Sep 17 00:00:00 2001
+From: zhangli20 <zh...@kuaishou.com>
+Date: Thu, 18 May 2023 14:26:01 +0800
+Subject: [PATCH] blaze do not force escaping control characters
+
+---
+ src/read.rs | 8 ++++++--
+ 1 file changed, 6 insertions(+), 2 deletions(-)
+
+diff --git a/src/read.rs b/src/read.rs
+index cac8132..b88606d 100644
+--- a/src/read.rs
++++ b/src/read.rs
+@@ -791,8 +791,12 @@ static ESCAPE: [bool; 256] = {
+ const __: bool = false; // allow unescaped
+ [
+ // 1 2 3 4 5 6 7 8 9 A B C D E F
+- CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, // 0
+- CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, // 1
++ // CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, // 0
++ // CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, CT, // 1
++
++ // blaze: will not escape any characters other than \0
++ CT, __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, // 0
++ __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, // 1
+ __, __, QU, __, __, __, __, __, __, __, __, __, __, __, __, __, // 2
+ __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, // 3
+ __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, // 4
+--
+2.50.1
+
diff --git a/dev/vendors/setup-vendors.sh b/dev/vendors/setup-vendors.sh
new file mode 100755
index 00000000..59a07a9e
--- /dev/null
+++ b/dev/vendors/setup-vendors.sh
@@ -0,0 +1,162 @@
+#!/usr/bin/env bash
+
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# =============================================================================
+# Setup vendored third-party source trees.
+#
+# This script clones upstream projects at pinned tags/commits into
+# dev/vendors/packages/<name> and applies the local patch series stored under
+# dev/vendors/patches/<name>/.
+#
+# Usage:
+# ./dev/vendors/setup-vendors.sh <name> [name ...]
+# ./dev/vendors/setup-vendors.sh all
+# =============================================================================
+
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO_ROOT="$(cd "${SCRIPT_DIR}/../.." && pwd)"
+PACKAGES_DIR="${SCRIPT_DIR}/packages"
+PATCHES_DIR="${SCRIPT_DIR}/patches"
+
+# -----------------------------------------------------------------------------
+# Vendor definitions: name -> upstream repo url, tag/commit
+#
+# To add a new vendor, append "name|repo_url|tag" to VENDOR_DEFS.
+# -----------------------------------------------------------------------------
+VENDOR_DEFS=(
+ "datafusion|https://github.com/apache/datafusion.git|49.0.0"
+ "arrow-rs|https://github.com/apache/arrow-rs.git|55.2.0"
+ "orc-rust|https://github.com/datafusion-contrib/orc-rust.git|v0.7.1"
+ "serde_json|https://github.com/serde-rs/json.git|v1.0.96"
+)
+
+vendor_lookup() {
+ local key="$1" field="$2"
+ local def
+ for def in "${VENDOR_DEFS[@]}"; do
+ local name repo tag
+ name="${def%%|*}"
+ rest="${def#*|}"
+ repo="${rest%%|*}"
+ tag="${rest#*|}"
+ if [[ "${name}" == "${key}" ]]; then
+ case "${field}" in
+ repo) echo "${repo}" ;;
+ tag) echo "${tag}" ;;
+ esac
+ return 0
+ fi
+ done
+ return 1
+}
+
+# -----------------------------------------------------------------------------
+# Apply a series of numbered patch files to the working tree via git apply.
+# -----------------------------------------------------------------------------
+apply_patches() {
+ local name="$1"
+ local patch_dir="${PATCHES_DIR}/${name}"
+ local target_dir="${PACKAGES_DIR}/${name}"
+
+ if [[ ! -d "${patch_dir}" ]]; then
+ echo "[WARN] No patch directory found for ${name} at ${patch_dir}, skipping patches."
+ return 0
+ fi
+
+ local patches=()
+ local f
+ while IFS= read -r f; do patches+=("$f"); done < <(ls -1 "${patch_dir}"/*.patch 2>/dev/null | sort)
+
+ if [[ ${#patches[@]} -eq 0 ]]; then
+ echo "[INFO] No patches to apply for ${name}."
+ return 0
+ fi
+
+ echo "[INFO] Applying ${#patches[@]} patch(es) for ${name} ..."
+ local i=0
+ for patch in "${patches[@]}"; do
+ i=$((i + 1))
+ echo " [${i}/${#patches[@]}] $(basename "${patch}")"
+ git -C "${target_dir}" apply --3way "${patch}"
+ done
+}
+
+# -----------------------------------------------------------------------------
+# Setup a single vendored project.
+# -----------------------------------------------------------------------------
+setup_vendor() {
+ local name="$1"
+ local repo tag target_dir
+ repo="$(vendor_lookup "${name}" repo)" || true
+ tag="$(vendor_lookup "${name}" tag)" || true
+ target_dir="${PACKAGES_DIR}/${name}"
+
+ if [[ -z "${repo:-}" ]]; then
+ echo "[ERROR] Unknown vendor: ${name}" >&2
+ exit 1
+ fi
+
+ if [[ -f "${target_dir}/Cargo.toml" ]]; then
+ echo "[INFO] ${name} already set up at ${target_dir}, skipping."
+ return 0
+ fi
+
+ echo "[INFO] Setting up vendor: ${name}"
+ echo "[INFO] repo: ${repo}"
+ echo "[INFO] tag: ${tag}"
+ echo "[INFO] dest: ${target_dir}"
+
+ if [[ -d "${target_dir}" ]]; then
+ echo "[INFO] Target directory already exists, removing: ${target_dir}"
+ rm -rf "${target_dir}"
+ fi
+
+ mkdir -p "${PACKAGES_DIR}"
+
+ echo "[INFO] Cloning ${repo} (shallow, tag ${tag}) ..."
+ git clone --quiet --depth 1 --branch "${tag}" "${repo}" "${target_dir}"
+
+ apply_patches "${name}"
+
+ echo "[INFO] ${name} setup complete."
+}
+
+# -----------------------------------------------------------------------------
+# Main
+# -----------------------------------------------------------------------------
+main() {
+ local names=("$@")
+ if [[ $# -eq 0 ]] || [[ "$1" == "all" ]]; then
+ names=()
+ local def
+ for def in "${VENDOR_DEFS[@]}"; do
+ names+=("${def%%|*}")
+ done
+ fi
+
+ for name in "${names[@]}"; do
+ setup_vendor "${name}"
+ done
+
+ echo "[INFO] All vendors setup complete."
+}
+
+main ${1+"$@"}
diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronConverters.scala b/spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronConverters.scala
index 979aafaf..a7c9fd2f 100644
--- a/spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronConverters.scala
+++ b/spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronConverters.scala
@@ -509,7 +509,7 @@ object AuronConverters extends Logging {
assert(
!exec.requiredSchema.exists(e => existTimestampType(e.dataType)),
s"Parquet scan with timestamp type is not supported for table: ${tableIdentifier
- .getOrElse("unknown")}. " +
+ .getOrElse("unknown")}. " +
"Set spark.auron.enable.scan.parquet.timestamp=true to enable timestamp support " +
"or remove timestamp columns from the query.")
}
@@ -520,7 +520,7 @@ object AuronConverters extends Logging {
assert(
!exec.requiredSchema.exists(e => existTimestampType(e.dataType)),
s"ORC scan with timestamp type is not supported for tableIdentifier: ${tableIdentifier
- .getOrElse("unknown")}. " +
+ .getOrElse("unknown")}. " +
"Set spark.auron.enable.scan.orc.timestamp=true to enable timestamp support " +
"or remove timestamp columns from the query.")
}
@@ -528,7 +528,7 @@ object AuronConverters extends Logging {
case p =>
throw new NotImplementedError(
s"Cannot convert FileSourceScanExec tableIdentifier: ${tableIdentifier.getOrElse(
- "unknown")}, class: ${p.getClass.getName}")
+ "unknown")}, class: ${p.getClass.getName}")
}
}
diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronSparkSessionExtension.scala b/spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronSparkSessionExtension.scala
index b68b0495..47492aa3 100644
--- a/spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronSparkSessionExtension.scala
+++ b/spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronSparkSessionExtension.scala
@@ -91,7 +91,7 @@ case class AuronColumnarOverrides(sparkSession: SparkSession) extends ColumnarRu
dumpSimpleSparkPlanTreeNode(sparkPlanTransformed)
logInfo(s"Transformed spark plan after preColumnarTransitions:\n${sparkPlanTransformed
- .treeString(verbose = true, addSuffix = true)}")
+ .treeString(verbose = true, addSuffix = true)}")
// post-transform
Shims.get.postTransform(sparkPlanTransformed, sparkSession.sparkContext)
diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeHelper.scala b/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeHelper.scala
index 87cd2829..26e5e8ab 100644
--- a/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeHelper.scala
+++ b/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeHelper.scala
@@ -74,7 +74,7 @@ object NativeHelper extends Logging {
val heapMemory = Runtime.getRuntime.maxMemory()
val offheapMemory = totalMemory - heapMemory
logWarning(s"memory total: ${Utils.bytesToString(totalMemory)}, onheap: ${Utils.bytesToString(
- heapMemory)}, offheap: ${Utils.bytesToString(offheapMemory)}")
+ heapMemory)}, offheap: ${Utils.bytesToString(offheapMemory)}")
offheapMemory
}
diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/auron/util/TaskContextHelper.scala b/spark-extension/src/main/scala/org/apache/spark/sql/auron/util/TaskContextHelper.scala
index 309139eb..37d012cc 100644
--- a/spark-extension/src/main/scala/org/apache/spark/sql/auron/util/TaskContextHelper.scala
+++ b/spark-extension/src/main/scala/org/apache/spark/sql/auron/util/TaskContextHelper.scala
@@ -47,7 +47,7 @@ object TaskContextHelper extends Logging {
val thread = Thread.currentThread()
val threadPrefix = if (context != null) {
s"auron native task ${context.partitionId()}.${context.attemptNumber()} in stage ${context
- .stageId()}.${context.stageAttemptNumber()} (TID ${context.taskAttemptId()})"
+ .stageId()}.${context.stageAttemptNumber()} (TID ${context.taskAttemptId()})"
} else {
"auron native task"
}