diff --git a/native/Cargo.lock b/native/Cargo.lock
index 514e34eb49..097d08509b 100644
--- a/native/Cargo.lock
+++ b/native/Cargo.lock
@@ -276,7 +276,7 @@ dependencies = [
  "chrono",
  "chrono-tz",
  "half",
- "hashbrown 0.17.0",
+ "hashbrown 0.17.1",
  "num-complex",
  "num-integer",
  "num-traits",
@@ -357,6 +357,7 @@ dependencies = [
  "arrow-select",
  "flatbuffers",
  "lz4_flex",
+ "zstd",
 ]
 
 [[package]]
@@ -647,7 +648,7 @@ dependencies = [
  "fastrand",
  "hex",
  "http 1.4.0",
- "sha1",
+ "sha1 0.10.6",
  "time",
  "tokio",
  "tracing",
@@ -1859,14 +1860,12 @@ dependencies = [
 
 [[package]]
 name = "datafusion"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "93db0e623840612f7f2cd757f7e8a8922064192363732c88692e0870016e141b"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "arrow",
  "arrow-schema",
  "async-trait",
- "bytes",
  "chrono",
  "datafusion-catalog",
  "datafusion-catalog-listing",
@@ -1894,13 +1893,12 @@ dependencies = [
  "datafusion-session",
  "datafusion-sql",
  "futures",
+ "indexmap 2.14.0",
  "itertools 0.14.0",
  "log",
  "object_store",
  "parking_lot",
  "parquet",
- "rand 0.9.4",
- "regex",
  "sqlparser",
  "tempfile",
  "tokio",
@@ -1910,9 +1908,8 @@ dependencies = [
 
 [[package]]
 name = "datafusion-catalog"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "37cefde60b26a7f4ff61e9d2ff2833322f91df2b568d7238afe67bde5bdffb66"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "arrow",
  "async-trait",
@@ -1935,9 +1932,8 @@ dependencies = [
 
 [[package]]
 name = "datafusion-catalog-listing"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "17e112307715d6a7a331111a4c2330ff54bc237183511c319e3708a4cff431fb"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "arrow",
  "async-trait",
@@ -2131,16 +2127,16 @@ dependencies = [
 
 [[package]]
 name = "datafusion-common"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d72a11ca44a95e1081870d3abb80c717496e8a7acb467a1d3e932bb636af5cc2"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
- "ahash",
  "arrow",
  "arrow-ipc",
+ "arrow-schema",
  "chrono",
+ "foldhash 0.2.0",
  "half",
- "hashbrown 0.16.1",
+ "hashbrown 0.17.1",
  "hex",
  "indexmap 2.14.0",
  "itertools 0.14.0",
@@ -2148,17 +2144,16 @@ dependencies = [
  "log",
  "object_store",
  "parquet",
- "paste",
  "sqlparser",
  "tokio",
+ "uuid",
  "web-time",
 ]
 
 [[package]]
 name = "datafusion-common-runtime"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "89f4afaed29670ec4fd6053643adc749fe3f4bc9d1ce1b8c5679b22c67d12def"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "futures",
  "log",
@@ -2167,9 +2162,8 @@ dependencies = [
 
 [[package]]
 name = "datafusion-datasource"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e9fb386e1691355355a96419978a0022b7947b44d4a24a6ea99f00b6b485cbb6"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "arrow",
  "async-compression",
@@ -2193,6 +2187,7 @@ dependencies = [
  "liblzma",
  "log",
  "object_store",
+ "parking_lot",
  "rand 0.9.4",
  "tokio",
  "tokio-util",
@@ -2202,9 +2197,8 @@ dependencies = [
 
 [[package]]
 name = "datafusion-datasource-arrow"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ffa6c52cfed0734c5f93754d1c0175f558175248bf686c944fb05c373e5fc096"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "arrow",
  "arrow-ipc",
@@ -2226,9 +2220,8 @@ dependencies = [
 
 [[package]]
 name = "datafusion-datasource-csv"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "503f29e0582c1fc189578d665ff57d9300da1f80c282777d7eb67bb79fb8cdca"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "arrow",
  "async-trait",
@@ -2249,9 +2242,8 @@ dependencies = [
 
 [[package]]
 name = "datafusion-datasource-json"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e33804749abc8d0c8cb7473228483cb8070e524c6f6086ee1b85a64debe2b3d2"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "arrow",
  "async-trait",
@@ -2266,16 +2258,14 @@ dependencies = [
  "datafusion-session",
  "futures",
  "object_store",
- "serde_json",
  "tokio",
  "tokio-stream",
 ]
 
 [[package]]
 name = "datafusion-datasource-parquet"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "32a8e0365e0e08e8ff94d912f0ababcf9065a1a304018ba90b1fc83c855b4997"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "arrow",
  "async-trait",
@@ -2285,6 +2275,7 @@ dependencies = [
  "datafusion-datasource",
  "datafusion-execution",
  "datafusion-expr",
+ "datafusion-functions",
  "datafusion-functions-aggregate-common",
  "datafusion-physical-expr",
  "datafusion-physical-expr-adapter",
@@ -2303,20 +2294,17 @@ dependencies = [
 
 [[package]]
 name = "datafusion-doc"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8de6ac0df1662b9148ad3c987978b32cbec7c772f199b1d53520c8fa764a87ee"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 
 [[package]]
 name = "datafusion-execution"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c03c7fbdaefcca4ef6ffe425a5fc2325763bfb426599bb0bf4536466efabe709"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "arrow",
  "arrow-buffer",
  "async-trait",
- "chrono",
  "dashmap",
  "datafusion-common",
  "datafusion-expr",
@@ -2333,11 +2321,11 @@ dependencies = [
 
 [[package]]
 name = "datafusion-expr"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "574b9b6977fedbd2a611cbff12e5caf90f31640ad9dc5870f152836d94bad0dd"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "arrow",
+ "arrow-schema",
  "async-trait",
  "chrono",
  "datafusion-common",
@@ -2348,29 +2336,25 @@ dependencies = [
  "datafusion-physical-expr-common",
  "indexmap 2.14.0",
  "itertools 0.14.0",
- "paste",
  "serde_json",
  "sqlparser",
 ]
 
 [[package]]
 name = "datafusion-expr-common"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7d7c3adf3db8bf61e92eb90cb659c8e8b734593a8f7c8e12a843c7ddba24b87e"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "arrow",
  "datafusion-common",
  "indexmap 2.14.0",
  "itertools 0.14.0",
- "paste",
 ]
 
 [[package]]
 name = "datafusion-functions"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f28aa4e10384e782774b10e72aca4d93ef7b31aa653095d9d4536b0a3dbc51b6"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "arrow",
  "arrow-buffer",
@@ -2385,26 +2369,24 @@ dependencies = [
  "datafusion-expr",
  "datafusion-expr-common",
  "datafusion-macros",
+ "datafusion-physical-expr-common",
  "hex",
  "itertools 0.14.0",
  "log",
- "md-5",
+ "md-5 0.11.0",
  "memchr",
  "num-traits",
  "rand 0.9.4",
  "regex",
- "sha2 0.10.9",
- "unicode-segmentation",
+ "sha2 0.11.0",
  "uuid",
 ]
 
 [[package]]
 name = "datafusion-functions-aggregate"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "00aa6217e56098ba84e0a338176fe52f0a84cca398021512c6c8c5eff806d0ad"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
- "ahash",
  "arrow",
  "datafusion-common",
  "datafusion-doc",
@@ -2414,19 +2396,17 @@ dependencies = [
  "datafusion-macros",
  "datafusion-physical-expr",
  "datafusion-physical-expr-common",
+ "foldhash 0.2.0",
  "half",
  "log",
  "num-traits",
- "paste",
 ]
 
 [[package]]
 name = "datafusion-functions-aggregate-common"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b511250349407db7c43832ab2de63f5557b19a20dfd236b39ca2c04468b50d47"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
- "ahash",
  "arrow",
  "datafusion-common",
  "datafusion-expr-common",
@@ -2435,9 +2415,8 @@ dependencies = [
 
 [[package]]
 name = "datafusion-functions-nested"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ef13a858e20d50f0a9bb5e96e7ac82b4e7597f247515bccca4fdd2992df0212a"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "arrow",
  "arrow-ord",
@@ -2451,34 +2430,32 @@ dependencies = [
  "datafusion-functions-aggregate-common",
  "datafusion-macros",
  "datafusion-physical-expr-common",
- "hashbrown 0.16.1",
+ "hashbrown 0.17.1",
  "itertools 0.14.0",
  "itoa",
  "log",
- "paste",
+ "memchr",
 ]
 
 [[package]]
 name = "datafusion-functions-table"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "72b40d3f5bbb3905f9ccb1ce9485a9595c77b69758a7c24d3ba79e334ff51e7e"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "arrow",
  "async-trait",
  "datafusion-catalog",
  "datafusion-common",
  "datafusion-expr",
+ "datafusion-physical-expr",
  "datafusion-physical-plan",
  "parking_lot",
- "paste",
 ]
 
 [[package]]
 name = "datafusion-functions-window"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d4e88ec9d57c9b685d02f58bfee7be62d72610430ddcedb82a08e5d9925dbfb6"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "arrow",
  "datafusion-common",
@@ -2489,14 +2466,12 @@ dependencies = [
  "datafusion-physical-expr",
  "datafusion-physical-expr-common",
  "log",
- "paste",
 ]
 
 [[package]]
 name = "datafusion-functions-window-common"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8307bb93519b1a91913723a1130cfafeee3f72200d870d88e91a6fc5470ede5c"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "datafusion-common",
  "datafusion-physical-expr-common",
@@ -2504,9 +2479,8 @@ dependencies = [
 
 [[package]]
 name = "datafusion-macros"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2e367e6a71051d0ebdd29b2f85d12059b38b1d1f172c6906e80016da662226bd"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "datafusion-doc",
  "quote",
@@ -2515,9 +2489,8 @@ dependencies = [
 
 [[package]]
 name = "datafusion-optimizer"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e929015451a67f77d9d8b727b2bf3a40c4445fdef6cdc53281d7d97c76888ace"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "arrow",
  "chrono",
@@ -2534,11 +2507,9 @@ dependencies = [
 
 [[package]]
 name = "datafusion-physical-expr"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4b1e68aba7a4b350401cfdf25a3d6f989ad898a7410164afe9ca52080244cb59"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
- "ahash",
  "arrow",
  "datafusion-common",
  "datafusion-expr",
@@ -2546,20 +2517,18 @@ dependencies = [
  "datafusion-functions-aggregate-common",
  "datafusion-physical-expr-common",
  "half",
- "hashbrown 0.16.1",
+ "hashbrown 0.17.1",
  "indexmap 2.14.0",
  "itertools 0.14.0",
  "parking_lot",
- "paste",
  "petgraph",
  "tokio",
 ]
 
 [[package]]
 name = "datafusion-physical-expr-adapter"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ea22315f33cf2e0adc104e8ec42e285f6ed93998d565c65e82fec6a9ee9f9db4"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "arrow",
  "datafusion-common",
@@ -2572,26 +2541,24 @@ dependencies = [
 
 [[package]]
 name = "datafusion-physical-expr-common"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b04b45ea8ad3ac2d78f2ea2a76053e06591c9629c7a603eda16c10649ecf4362"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
- "ahash",
  "arrow",
  "chrono",
  "datafusion-common",
  "datafusion-expr-common",
- "hashbrown 0.16.1",
+ "hashbrown 0.17.1",
  "indexmap 2.14.0",
  "itertools 0.14.0",
  "parking_lot",
+ "pin-project",
 ]
 
 [[package]]
 name = "datafusion-physical-optimizer"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7cb13397809a425918f608dfe8653f332015a3e330004ab191b4404187238b95"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "arrow",
  "datafusion-common",
@@ -2607,12 +2574,12 @@ dependencies = [
 
 [[package]]
 name = "datafusion-physical-plan"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5edc023675791af9d5fb4cc4c24abf5f7bd3bd4dcf9e5bd90ea1eff6976dcc79"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
- "ahash",
  "arrow",
+ "arrow-data",
+ "arrow-ipc",
  "arrow-ord",
  "arrow-schema",
  "async-trait",
@@ -2627,7 +2594,7 @@ dependencies = [
  "datafusion-physical-expr-common",
  "futures",
  "half",
- "hashbrown 0.16.1",
+ "hashbrown 0.17.1",
  "indexmap 2.14.0",
  "itertools 0.14.0",
  "log",
@@ -2639,9 +2606,8 @@ dependencies = [
 
 [[package]]
 name = "datafusion-pruning"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ac8c76860e355616555081cab5968cec1af7a80701ff374510860bcd567e365a"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "arrow",
  "datafusion-common",
@@ -2650,15 +2616,13 @@ dependencies = [
  "datafusion-physical-expr",
  "datafusion-physical-expr-common",
  "datafusion-physical-plan",
- "itertools 0.14.0",
  "log",
 ]
 
 [[package]]
 name = "datafusion-session"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5412111aa48e2424ba926112e192f7a6b7e4ccb450145d25ce5ede9f19dc491e"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "async-trait",
  "datafusion-common",
@@ -2670,9 +2634,8 @@ dependencies = [
 
 [[package]]
 name = "datafusion-spark"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e059dcf8544da0d6598d0235be3cc29c209094a5976b2e4822e4a2cf91c2b5c5"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "arrow",
  "bigdecimal",
@@ -2685,21 +2648,23 @@ dependencies = [
  "datafusion-expr",
  "datafusion-functions",
  "datafusion-functions-aggregate",
+ "datafusion-functions-aggregate-common",
  "datafusion-functions-nested",
  "log",
+ "num-traits",
  "percent-encoding",
  "rand 0.9.4",
  "serde_json",
- "sha1",
- "sha2 0.10.9",
+ "sha1 0.11.0",
+ "sha2 0.11.0",
+ "twox-hash",
  "url",
 ]
 
 [[package]]
 name = "datafusion-sql"
-version = "53.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fa0d133ddf8b9b3b872acac900157f783e7b879fe9a6bccf389abebbfac45ec1"
+version = "54.0.0"
+source = "git+https://github.com/apache/datafusion?branch=branch-54#c8dddb8a03f098b92e0295118d12390381d23fa8"
 dependencies = [
  "arrow",
  "bigdecimal",
@@ -3319,21 +3284,15 @@ dependencies = [
 
 [[package]]
 name = "hashbrown"
-version = "0.16.1"
+version = "0.17.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100"
+checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a"
 dependencies = [
  "allocator-api2",
  "equivalent",
  "foldhash 0.2.0",
 ]
 
-[[package]]
-name = "hashbrown"
-version = "0.17.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4f467dd6dccf739c208452f8014c75c18bb8301b050ad1cfb27153803edb0f51"
-
 [[package]]
 name = "hdfs-sys"
 version = "0.3.0"
@@ -3771,7 +3730,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9"
 dependencies = [
  "equivalent",
- "hashbrown 0.17.0",
+ "hashbrown 0.17.1",
  "serde",
  "serde_core",
 ]
@@ -4282,6 +4241,16 @@ dependencies = [
  "digest 0.10.7",
 ]
 
+[[package]]
+name = "md-5"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98"
+dependencies = [
+ "cfg-if",
+ "digest 0.11.2",
+]
+
 [[package]]
 name = "mea"
 version = "0.6.3"
@@ -4537,7 +4506,7 @@ dependencies = [
  "humantime",
  "hyper",
  "itertools 0.14.0",
- "md-5",
+ "md-5 0.10.6",
  "parking_lot",
  "percent-encoding",
  "quick-xml 0.39.2",
@@ -4615,7 +4584,7 @@ dependencies = [
  "http-body 1.0.1",
  "jiff",
  "log",
- "md-5",
+ "md-5 0.10.6",
  "percent-encoding",
  "quick-xml 0.38.4",
  "reqsign",
@@ -4656,7 +4625,7 @@ dependencies = [
  "http-body 1.0.1",
  "jiff",
  "log",
- "md-5",
+ "md-5 0.10.6",
  "mea",
  "percent-encoding",
  "quick-xml 0.38.4",
@@ -4817,7 +4786,7 @@ dependencies = [
  "flate2",
  "futures",
  "half",
- "hashbrown 0.17.0",
+ "hashbrown 0.17.1",
  "lz4_flex",
  "num-bigint",
  "num-integer",
@@ -5566,7 +5535,7 @@ dependencies = [
  "rust-ini",
  "serde",
  "serde_json",
- "sha1",
+ "sha1 0.10.6",
  "sha2 0.10.9",
  "tokio",
 ]
@@ -5588,7 +5557,7 @@ dependencies = [
  "jiff",
  "log",
  "percent-encoding",
- "sha1",
+ "sha1 0.10.6",
  "sha2 0.10.9",
  "windows-sys 0.61.2",
 ]
@@ -6133,6 +6102,17 @@ dependencies = [
  "digest 0.10.7",
 ]
 
+[[package]]
+name = "sha1"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "aacc4cc499359472b4abe1bf11d0b12e688af9a805fa5e3016f9a386dc2d0214"
+dependencies = [
+ "cfg-if",
+ "cpufeatures 0.3.0",
+ "digest 0.11.2",
+]
+
 [[package]]
 name = "sha2"
 version = "0.10.9"
@@ -6276,9 +6256,9 @@ dependencies = [
 
 [[package]]
 name = "sqlparser"
-version = "0.61.0"
+version = "0.62.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dbf5ea8d4d7c808e1af1cbabebca9a2abe603bcefc22294c5b95018d53200cb7"
+checksum = "13c6d1b651dc4edf07eead2a0c6c78016ce971bc2c10da5266861b13f25e7cec"
 dependencies = [
  "log",
  "sqlparser_derive",
diff --git a/native/Cargo.toml b/native/Cargo.toml
index 4a01bb4f80..c545016792 100644
--- a/native/Cargo.toml
+++ b/native/Cargo.toml
@@ -38,10 +38,10 @@ arrow = { version = "58.3.0", features = ["prettyprint", "ffi", "chrono-tz"] }
 async-trait = { version = "0.1" }
 bytes = { version = "1.11.1" }
 parquet = { version = "58.3.0", default-features = false, features = ["experimental"] }
-datafusion = { version = "53.1.0", default-features = false, features = ["unicode_expressions", "crypto_expressions", "nested_expressions", "parquet"] }
-datafusion-datasource = { version = "53.1.0" }
-datafusion-physical-expr-adapter = { version = "53.1.0" }
-datafusion-spark = { version = "53.1.0", features = ["core"] }
+datafusion = { git = "https://github.com/apache/datafusion", branch = "branch-54", default-features = false, features = ["unicode_expressions", "crypto_expressions", "nested_expressions", "parquet"] }
+datafusion-datasource = { git = "https://github.com/apache/datafusion", branch = "branch-54" }
+datafusion-physical-expr-adapter = { git = "https://github.com/apache/datafusion", branch = "branch-54" }
+datafusion-spark = { git = "https://github.com/apache/datafusion", branch = "branch-54", features = ["core"] }
 datafusion-comet-spark-expr = { path = "spark-expr" }
 datafusion-comet-common = { path = "common" }
 datafusion-comet-jni-bridge = { path = "jni-bridge" }
diff --git a/native/core/Cargo.toml b/native/core/Cargo.toml
index c58d446917..6505e31cf0 100644
--- a/native/core/Cargo.toml
+++ b/native/core/Cargo.toml
@@ -91,7 +91,7 @@ jni = { version = "0.22.4", features = ["invocation"] }
 lazy_static = "1.4"
 assertables = "9"
 hex = "0.4.3"
-datafusion-functions-nested = { version = "53.1.0" }
+datafusion-functions-nested = { git = "https://github.com/apache/datafusion", branch = "branch-54" }
 
 [features]
 backtrace = ["datafusion/backtrace"]
diff --git a/native/core/src/debug/debug_batch_stream.rs b/native/core/src/debug/debug_batch_stream.rs
index b59b6a36f7..37837a5daa 100644
--- a/native/core/src/debug/debug_batch_stream.rs
+++ b/native/core/src/debug/debug_batch_stream.rs
@@ -15,7 +15,6 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::any::Any;
 use std::fmt;
 use std::hash::{Hash, Hasher};
 use std::sync::Arc;
@@ -82,15 +81,20 @@ impl datafusion::physical_plan::ExecutionPlan for DebugExecutionDataStream {
     fn name(&self) -> &str {
         "DebugExecutionDataStream"
     }
-    fn as_any(&self) -> &dyn std::any::Any {
-        self
-    }
     fn properties(&self) -> &Arc<datafusion::physical_plan::PlanProperties> {
         self.inner.properties()
     }
     fn children(&self) -> Vec<&Arc<dyn datafusion::physical_plan::ExecutionPlan>> {
         vec![&self.inner]
     }
+    fn apply_expressions(
+        &self,
+        _f: &mut dyn FnMut(
+            &dyn PhysicalExpr,
+        ) -> Result<datafusion::common::tree_node::TreeNodeRecursion>,
+    ) -> Result<datafusion::common::tree_node::TreeNodeRecursion> {
+        Ok(datafusion::common::tree_node::TreeNodeRecursion::Continue)
+    }
     fn with_new_children(
         self: Arc<Self>,
         children: Vec<Arc<dyn datafusion::physical_plan::ExecutionPlan>>,
@@ -155,9 +159,6 @@ impl Hash for DebugExecutionDataPhyExpr {
 }
 
 impl PhysicalExpr for DebugExecutionDataPhyExpr {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
     fn data_type(&self, input_schema: &Schema) -> Result<DataType> {
         self.inner.data_type(input_schema)
     }
diff --git a/native/core/src/execution/expressions/arithmetic.rs b/native/core/src/execution/expressions/arithmetic.rs
index 320532d773..8d4c59a010 100644
--- a/native/core/src/execution/expressions/arithmetic.rs
+++ b/native/core/src/execution/expressions/arithmetic.rs
@@ -77,10 +77,6 @@ impl Hash for CheckedBinaryExpr {
 }
 
 impl PhysicalExpr for CheckedBinaryExpr {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         self.child.fmt_sql(f)
     }
diff --git a/native/core/src/execution/expressions/list_positions.rs b/native/core/src/execution/expressions/list_positions.rs
index 117aab4d97..11643db4ce 100644
--- a/native/core/src/execution/expressions/list_positions.rs
+++ b/native/core/src/execution/expressions/list_positions.rs
@@ -15,7 +15,6 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::any::Any;
 use std::fmt::{Display, Formatter};
 use std::hash::{Hash, Hasher};
 use std::sync::Arc;
@@ -68,10 +67,6 @@ impl Hash for ListPositionsExpr {
 }
 
 impl PhysicalExpr for ListPositionsExpr {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         Display::fmt(self, f)
     }
diff --git a/native/core/src/execution/expressions/strings.rs b/native/core/src/execution/expressions/strings.rs
index 7219395963..4a7c44cc3e 100644
--- a/native/core/src/execution/expressions/strings.rs
+++ b/native/core/src/execution/expressions/strings.rs
@@ -91,7 +91,7 @@ impl ExpressionBuilder for RlikeBuilder {
         let left = planner.create_expr(expr.left.as_ref().unwrap(), Arc::clone(&input_schema))?;
         let right = planner.create_expr(expr.right.as_ref().unwrap(), input_schema)?;
 
-        match right.as_any().downcast_ref::<Literal>().unwrap().value() {
+        match right.downcast_ref::<Literal>().unwrap().value() {
             ScalarValue::Utf8(Some(pattern)) => Ok(Arc::new(RLike::try_new(left, pattern)?)),
             _ => Err(ExecutionError::GeneralError(
                 "RLike only supports scalar patterns".to_string(),
diff --git a/native/core/src/execution/expressions/subquery.rs b/native/core/src/execution/expressions/subquery.rs
index 9272ede60c..fc7b8104d2 100644
--- a/native/core/src/execution/expressions/subquery.rs
+++ b/native/core/src/execution/expressions/subquery.rs
@@ -29,7 +29,6 @@ use jni::{
     sys::{jboolean, jbyte, jint, jlong, jshort},
 };
 use std::{
-    any::Any,
     fmt::{Display, Formatter},
     hash::Hash,
     sync::Arc,
@@ -63,10 +62,6 @@ impl Display for Subquery {
 }
 
 impl PhysicalExpr for Subquery {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         Display::fmt(self, f)
     }
diff --git a/native/core/src/execution/memory_pools/fair_pool.rs b/native/core/src/execution/memory_pools/fair_pool.rs
index e4a7ceab54..5c4cacbefb 100644
--- a/native/core/src/execution/memory_pools/fair_pool.rs
+++ b/native/core/src/execution/memory_pools/fair_pool.rs
@@ -16,7 +16,7 @@
 // under the License.
 
 use std::{
-    fmt::{Debug, Formatter, Result as FmtResult},
+    fmt::{Debug, Display, Formatter, Result as FmtResult},
     sync::Arc,
 };
 
@@ -83,10 +83,20 @@ impl CometFairMemoryPool {
     }
 }
 
+impl Display for CometFairMemoryPool {
+    fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
+        write!(f, "CometFairMemoryPool")
+    }
+}
+
 unsafe impl Send for CometFairMemoryPool {}
 unsafe impl Sync for CometFairMemoryPool {}
 
 impl MemoryPool for CometFairMemoryPool {
+    fn name(&self) -> &str {
+        "CometFairMemoryPool"
+    }
+
     fn register(&self, _: &MemoryConsumer) {
         let mut state = self.state.lock();
         state.num = state
diff --git a/native/core/src/execution/memory_pools/logging_pool.rs b/native/core/src/execution/memory_pools/logging_pool.rs
index c23672d01a..b05f98c732 100644
--- a/native/core/src/execution/memory_pools/logging_pool.rs
+++ b/native/core/src/execution/memory_pools/logging_pool.rs
@@ -19,6 +19,7 @@ use datafusion::execution::memory_pool::{
     MemoryConsumer, MemoryLimit, MemoryPool, MemoryReservation,
 };
 use log::{info, warn};
+use std::fmt::{self, Display};
 use std::sync::Arc;
 
 #[derive(Debug)]
@@ -36,7 +37,17 @@ impl LoggingMemoryPool {
     }
 }
 
+impl Display for LoggingMemoryPool {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "LoggingMemoryPool(task={})", self.task_attempt_id)
+    }
+}
+
 impl MemoryPool for LoggingMemoryPool {
+    fn name(&self) -> &str {
+        "LoggingMemoryPool"
+    }
+
     fn register(&self, consumer: &MemoryConsumer) {
         info!(
             "[Task {}] MemoryPool[{}].register()",
diff --git a/native/core/src/execution/memory_pools/unified_pool.rs b/native/core/src/execution/memory_pools/unified_pool.rs
index f34418ee94..d26a57d35c 100644
--- a/native/core/src/execution/memory_pools/unified_pool.rs
+++ b/native/core/src/execution/memory_pools/unified_pool.rs
@@ -16,7 +16,7 @@
 // under the License.
 
 use std::{
-    fmt::{Debug, Formatter, Result as FmtResult},
+    fmt::{Debug, Display, Formatter, Result as FmtResult},
     sync::{
         atomic::{AtomicUsize, Ordering::Relaxed},
         Arc,
@@ -93,7 +93,17 @@ impl Drop for CometUnifiedMemoryPool {
 unsafe impl Send for CometUnifiedMemoryPool {}
 unsafe impl Sync for CometUnifiedMemoryPool {}
 
+impl Display for CometUnifiedMemoryPool {
+    fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
+        write!(f, "CometUnifiedMemoryPool(task={})", self.task_attempt_id)
+    }
+}
+
 impl MemoryPool for CometUnifiedMemoryPool {
+    fn name(&self) -> &str {
+        "CometUnifiedMemoryPool"
+    }
+
     fn grow(&self, reservation: &MemoryReservation, additional: usize) {
         self.try_grow(reservation, additional).unwrap();
     }
diff --git a/native/core/src/execution/merge_as_partial.rs b/native/core/src/execution/merge_as_partial.rs
index 1c15ce8329..5ea26115bf 100644
--- a/native/core/src/execution/merge_as_partial.rs
+++ b/native/core/src/execution/merge_as_partial.rs
@@ -26,7 +26,6 @@
 //! outputs state) but redirects `update_batch` calls to `merge_batch`, giving merge
 //! semantics with state output.
 
-use std::any::Any;
 use std::fmt::Debug;
 use std::hash::{Hash, Hasher};
 
@@ -100,10 +99,6 @@ impl MergeAsPartialUDF {
 }
 
 impl AggregateUDFImpl for MergeAsPartialUDF {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         &self.name
     }
diff --git a/native/core/src/execution/operators/expand.rs b/native/core/src/execution/operators/expand.rs
index e06fab23ec..1617712a87 100644
--- a/native/core/src/execution/operators/expand.rs
+++ b/native/core/src/execution/operators/expand.rs
@@ -17,6 +17,7 @@
 
 use arrow::array::{RecordBatch, RecordBatchOptions};
 use arrow::datatypes::SchemaRef;
+use datafusion::common::tree_node::TreeNodeRecursion;
 use datafusion::common::DataFusionError;
 use datafusion::physical_expr::{EquivalenceProperties, PhysicalExpr};
 use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
@@ -29,7 +30,6 @@ use datafusion::{
 };
 use futures::{Stream, StreamExt};
 use std::{
-    any::Any,
     pin::Pin,
     sync::Arc,
     task::{Context, Poll},
@@ -91,10 +91,6 @@ impl DisplayAs for ExpandExec {
 }
 
 impl ExecutionPlan for ExpandExec {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         Arc::clone(&self.schema)
     }
@@ -103,6 +99,19 @@ impl ExecutionPlan for ExpandExec {
         vec![&self.child]
     }
 
+    fn apply_expressions(
+        &self,
+        f: &mut dyn FnMut(&dyn PhysicalExpr) -> datafusion::common::Result<TreeNodeRecursion>,
+    ) -> datafusion::common::Result<TreeNodeRecursion> {
+        let mut tnr = TreeNodeRecursion::Continue;
+        for projection in &self.projections {
+            for expr in projection {
+                tnr = tnr.visit_sibling(|| f(expr.as_ref()))?;
+            }
+        }
+        Ok(tnr)
+    }
+
     fn with_new_children(
         self: Arc<Self>,
         children: Vec<Arc<dyn ExecutionPlan>>,
diff --git a/native/core/src/execution/operators/iceberg_scan.rs b/native/core/src/execution/operators/iceberg_scan.rs
index 55bcbef349..406f5236c9 100644
--- a/native/core/src/execution/operators/iceberg_scan.rs
+++ b/native/core/src/execution/operators/iceberg_scan.rs
@@ -17,7 +17,6 @@
 
 //! Native Iceberg table scan operator using iceberg-rust
 
-use std::any::Any;
 use std::collections::HashMap;
 use std::fmt;
 use std::pin::Pin;
@@ -26,6 +25,7 @@ use std::task::{Context, Poll};
 
 use arrow::array::{ArrayRef, RecordBatch, RecordBatchOptions};
 use arrow::datatypes::SchemaRef;
+use datafusion::common::tree_node::TreeNodeRecursion;
 use datafusion::common::{DataFusionError, Result as DFResult};
 use datafusion::execution::{RecordBatchStream, SendableRecordBatchStream, TaskContext};
 use datafusion::physical_expr::expressions::Column;
@@ -109,10 +109,6 @@ impl ExecutionPlan for IcebergScanExec {
         "IcebergScanExec"
     }
 
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         Arc::clone(&self.output_schema)
     }
@@ -125,6 +121,13 @@ impl ExecutionPlan for IcebergScanExec {
         vec![]
     }
 
+    fn apply_expressions(
+        &self,
+        _f: &mut dyn FnMut(&dyn PhysicalExpr) -> DFResult<TreeNodeRecursion>,
+    ) -> DFResult<TreeNodeRecursion> {
+        Ok(TreeNodeRecursion::Continue)
+    }
+
     fn with_new_children(
         self: Arc<Self>,
         _children: Vec<Arc<dyn ExecutionPlan>>,
diff --git a/native/core/src/execution/operators/parquet_writer.rs b/native/core/src/execution/operators/parquet_writer.rs
index 8ba79098d4..5303888186 100644
--- a/native/core/src/execution/operators/parquet_writer.rs
+++ b/native/core/src/execution/operators/parquet_writer.rs
@@ -18,7 +18,6 @@
 //! Parquet writer operator for writing RecordBatches to Parquet files
 
 use std::{
-    any::Any,
     collections::HashMap,
     fmt,
     fmt::{Debug, Formatter},
@@ -39,9 +38,10 @@ use arrow::datatypes::{Schema, SchemaRef};
 use arrow::record_batch::RecordBatch;
 use async_trait::async_trait;
 use datafusion::{
+    common::tree_node::TreeNodeRecursion,
     error::{DataFusionError, Result},
     execution::context::TaskContext,
-    physical_expr::EquivalenceProperties,
+    physical_expr::{EquivalenceProperties, PhysicalExpr},
     physical_plan::{
         execution_plan::{Boundedness, EmissionType},
         metrics::{ExecutionPlanMetricsSet, MetricsSet},
@@ -404,10 +404,6 @@ impl DisplayAs for ParquetWriterExec {
 
 #[async_trait]
 impl ExecutionPlan for ParquetWriterExec {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "ParquetWriterExec"
     }
@@ -428,6 +424,13 @@ impl ExecutionPlan for ParquetWriterExec {
         vec![&self.input]
     }
 
+    fn apply_expressions(
+        &self,
+        _f: &mut dyn FnMut(&dyn PhysicalExpr) -> Result<TreeNodeRecursion>,
+    ) -> Result<TreeNodeRecursion> {
+        Ok(TreeNodeRecursion::Continue)
+    }
+
     fn with_new_children(
         self: Arc<Self>,
         children: Vec<Arc<dyn ExecutionPlan>>,
diff --git a/native/core/src/execution/operators/scan.rs b/native/core/src/execution/operators/scan.rs
index 90bb741b5e..f3b8f9ad1b 100644
--- a/native/core/src/execution/operators/scan.rs
+++ b/native/core/src/execution/operators/scan.rs
@@ -28,6 +28,7 @@ use arrow::compute::{cast_with_options, take, CastOptions};
 use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
 use arrow::ffi::FFI_ArrowArray;
 use arrow::ffi::FFI_ArrowSchema;
+use datafusion::common::tree_node::TreeNodeRecursion;
 use datafusion::common::{arrow_datafusion_err, DataFusionError, Result as DataFusionResult};
 use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
 use datafusion::physical_plan::metrics::{
@@ -43,7 +44,6 @@ use itertools::Itertools;
 use jni::objects::{Global, JObject, JValue};
 use std::rc::Rc;
 use std::{
-    any::Any,
     pin::Pin,
     sync::{Arc, Mutex},
     task::{Context, Poll},
@@ -383,10 +383,6 @@ fn schema_from_data_types(data_types: &[DataType]) -> SchemaRef {
 }
 
 impl ExecutionPlan for ScanExec {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         Arc::clone(&self.schema)
     }
@@ -395,6 +391,13 @@ impl ExecutionPlan for ScanExec {
         vec![]
     }
 
+    fn apply_expressions(
+        &self,
+        _f: &mut dyn FnMut(&dyn PhysicalExpr) -> DataFusionResult<TreeNodeRecursion>,
+    ) -> DataFusionResult<TreeNodeRecursion> {
+        Ok(TreeNodeRecursion::Continue)
+    }
+
     fn with_new_children(
         self: Arc<Self>,
         _: Vec<Arc<dyn ExecutionPlan>>,
diff --git a/native/core/src/execution/operators/shuffle_scan.rs b/native/core/src/execution/operators/shuffle_scan.rs
index 92c4dc8780..c89814f2af 100644
--- a/native/core/src/execution/operators/shuffle_scan.rs
+++ b/native/core/src/execution/operators/shuffle_scan.rs
@@ -24,6 +24,7 @@ use crate::{
 };
 use arrow::array::ArrayRef;
 use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+use datafusion::common::tree_node::TreeNodeRecursion;
 use datafusion::common::{arrow_datafusion_err, Result as DataFusionResult};
 use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
 use datafusion::physical_plan::metrics::{
@@ -37,7 +38,6 @@ use datafusion::{
 use futures::Stream;
 use jni::objects::{Global, JByteBuffer, JObject};
 use std::{
-    any::Any,
     pin::Pin,
     sync::{Arc, Mutex},
     task::{Context, Poll},
@@ -221,10 +221,6 @@ fn schema_from_data_types(data_types: &[DataType]) -> SchemaRef {
 }
 
 impl ExecutionPlan for ShuffleScanExec {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn schema(&self) -> SchemaRef {
         Arc::clone(&self.schema)
     }
@@ -233,6 +229,13 @@ impl ExecutionPlan for ShuffleScanExec {
         vec![]
     }
 
+    fn apply_expressions(
+        &self,
+        _f: &mut dyn FnMut(&dyn PhysicalExpr) -> DataFusionResult<TreeNodeRecursion>,
+    ) -> DataFusionResult<TreeNodeRecursion> {
+        Ok(TreeNodeRecursion::Continue)
+    }
+
     fn with_new_children(
         self: Arc<Self>,
         _: Vec<Arc<dyn ExecutionPlan>>,
diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs
index 89d118b059..0e33fe78ec 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -265,7 +265,6 @@ impl PhysicalPlanner {
                     let literal =
                         self.create_expr(partition_value, Arc::<Schema>::clone(&empty_schema))?;
                     literal
-                        .as_any()
                         .downcast_ref::<DataFusionLiteral>()
                         .ok_or_else(|| {
                             GeneralError("Expected literal of partition value".to_string())
@@ -459,11 +458,7 @@ impl PhysicalPlanner {
 
                 // WideDecimalBinaryExpr already handles overflow — skip redundant check
                 // but only if its output type matches CheckOverflow's declared type
-                if child
-                    .as_any()
-                    .downcast_ref::<WideDecimalBinaryExpr>()
-                    .is_some()
-                {
+                if child.downcast_ref::<WideDecimalBinaryExpr>().is_some() {
                     let child_type = child.data_type(&input_schema)?;
                     if child_type == data_type {
                         return Ok(child);
@@ -472,7 +467,7 @@ impl PhysicalPlanner {
 
                 // Fuse Cast(Decimal128→Decimal128) + CheckOverflow into single rescale+check
                 // Only fuse when the Cast target type matches the CheckOverflow output type
-                if let Some(cast) = child.as_any().downcast_ref::<Cast>() {
+                if let Some(cast) = child.downcast_ref::<Cast>() {
                     if let (
                         DataType::Decimal128(p_out, s_out),
                         Ok(DataType::Decimal128(_p_in, s_in)),
@@ -1319,12 +1314,10 @@ impl PhysicalPlanner {
                         .iter()
                         .map(|expr| {
                             let literal = self.create_expr(expr, Arc::clone(&required_schema))?;
-                            let df_literal = literal
-                                .as_any()
-                                .downcast_ref::<DataFusionLiteral>()
-                                .ok_or_else(|| {
-                                GeneralError("Expected literal of default value.".to_string())
-                            })?;
+                            let df_literal =
+                                literal.downcast_ref::<DataFusionLiteral>().ok_or_else(|| {
+                                    GeneralError("Expected literal of default value.".to_string())
+                                })?;
                             Ok(df_literal.value().clone())
                         })
                         .collect();
@@ -1892,7 +1885,7 @@ impl PhysicalPlanner {
                         hash_join.as_ref().swap_inputs(PartitionMode::Partitioned)?;
 
                     let mut additional_native_plans = vec![];
-                    if swapped_hash_join.as_any().is::<ProjectionExec>() {
+                    if swapped_hash_join.is::<ProjectionExec>() {
                         // a projection was added to the hash join
                         additional_native_plans.push(Arc::clone(swapped_hash_join.children()[0]));
                     }
@@ -2752,8 +2745,7 @@ impl PhysicalPlanner {
                             &boundary_row.partition_bounds[col_idx],
                             Arc::clone(&input_schema),
                         )?;
-                        let literal_expr =
-                            expr.as_any().downcast_ref::<Literal>().expect("Literal");
+                        let literal_expr = expr.downcast_ref::<Literal>().expect("Literal");
                         col_values.push(literal_expr.value().clone());
                     }
                 }
@@ -2863,12 +2855,7 @@ impl PhysicalPlanner {
                     // TODO this should try and find scalar
                     let arguments = args
                         .iter()
-                        .map(|e| {
-                            e.as_ref()
-                                .as_any()
-                                .downcast_ref::<Literal>()
-                                .map(|lit| lit.value())
-                        })
+                        .map(|e| e.downcast_ref::<Literal>().map(|lit| lit.value()))
                         .collect::<Vec<_>>();
 
                     let args = ReturnFieldArgs {
@@ -2972,7 +2959,7 @@ fn expr_to_columns(
 
     expr.apply(&mut |expr: &Arc<dyn PhysicalExpr>| {
         Ok({
-            if let Some(column) = expr.as_any().downcast_ref::<Column>() {
+            if let Some(column) = expr.downcast_ref::<Column>() {
                 if column.index() > left_field_len + right_field_len {
                     return Err(DataFusionError::Internal(format!(
                         "Column index {} out of range",
@@ -3023,7 +3010,7 @@ impl TreeNodeRewriter for JoinFilterRewriter<'_> {
     type Node = Arc<dyn PhysicalExpr>;
 
     fn f_down(&mut self, node: Self::Node) -> datafusion::common::Result<Transformed<Self::Node>> {
-        if let Some(column) = node.as_any().downcast_ref::<Column>() {
+        if let Some(column) = node.downcast_ref::<Column>() {
             if column.index() < self.left_field_len {
                 // left side
                 let new_index = self
diff --git a/native/core/src/parquet/cast_column.rs b/native/core/src/parquet/cast_column.rs
index 67558b5266..1cc928d1d5 100644
--- a/native/core/src/parquet/cast_column.rs
+++ b/native/core/src/parquet/cast_column.rs
@@ -31,7 +31,6 @@ use datafusion::common::ScalarValue;
 use datafusion::logical_expr::ColumnarValue;
 use datafusion::physical_expr::PhysicalExpr;
 use std::{
-    any::Any,
     fmt::{self, Display},
     hash::Hash,
     sync::Arc,
@@ -250,10 +249,6 @@ impl Display for CometCastColumnExpr {
 }
 
 impl PhysicalExpr for CometCastColumnExpr {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn data_type(&self, _input_schema: &Schema) -> DataFusionResult<DataType> {
         Ok(self.target_field.data_type().clone())
     }
diff --git a/native/core/src/parquet/schema_adapter.rs b/native/core/src/parquet/schema_adapter.rs
index 9a348e5b62..09d5a1dd62 100644
--- a/native/core/src/parquet/schema_adapter.rs
+++ b/native/core/src/parquet/schema_adapter.rs
@@ -33,7 +33,6 @@ use datafusion_physical_expr_adapter::{
     PhysicalExprAdapterFactory,
 };
 use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
-use std::any::Any;
 use std::collections::HashMap;
 use std::fmt::{self, Display};
 use std::hash::{Hash, Hasher};
@@ -454,7 +453,7 @@ impl PhysicalExprAdapter for SparkPhysicalExprAdapter {
             // Walk the expression tree to find Column references
             let mut duplicate_err: Option<DataFusionError> = None;
             let _ = Arc::<dyn PhysicalExpr>::clone(&expr).transform(|e| {
-                if let Some(col) = e.as_any().downcast_ref::<Column>() {
+                if let Some(col) = e.downcast_ref::<Column>() {
                     if let Some((req, matched)) = check_column_duplicate(col.name(), orig_physical)
                     {
                         duplicate_err = Some(DataFusionError::External(Box::new(
@@ -504,7 +503,7 @@ impl PhysicalExprAdapter for SparkPhysicalExprAdapter {
         // the actual parquet stream schema, which uses the original physical names.
         let expr = if let Some(name_map) = &self.logical_to_physical_names {
             expr.transform(|e| {
-                if let Some(col) = e.as_any().downcast_ref::<Column>() {
+                if let Some(col) = e.downcast_ref::<Column>() {
                     if let Some(physical_name) = name_map.get(col.name()) {
                         return Ok(Transformed::yes(Arc::new(Column::new(
                             physical_name,
@@ -533,7 +532,7 @@ impl SparkPhysicalExprAdapter {
         expr: Arc<dyn PhysicalExpr>,
     ) -> DataFusionResult<Arc<dyn PhysicalExpr>> {
         expr.transform(|e| {
-            if let Some(column) = e.as_any().downcast_ref::<Column>() {
+            if let Some(column) = e.downcast_ref::<Column>() {
                 let col_name = column.name();
 
                 // Resolve fields by name because this is the fallback path
@@ -628,14 +627,12 @@ impl SparkPhysicalExprAdapter {
         &self,
         expr: Arc<dyn PhysicalExpr>,
     ) -> DataFusionResult<Transformed<Arc<dyn PhysicalExpr>>> {
-        // Check for CastColumnExpr and replace with spark_expr::Cast
-        // CastColumnExpr is in datafusion_physical_expr::expressions
-        if let Some(cast) = expr
-            .as_any()
-            .downcast_ref::<datafusion::physical_expr::expressions::CastColumnExpr>()
+        // Check for CastExpr and replace with spark_expr::Cast
+        if let Some(cast) = expr.downcast_ref::<datafusion::physical_expr::expressions::CastExpr>()
         {
             let child = Arc::clone(cast.expr());
-            let physical_type = cast.input_field().data_type();
+            let input_field = child.return_field(&self.physical_file_schema)?;
+            let physical_type = input_field.data_type();
             let target_type = cast.target_field().data_type();
 
             // Reject reading a string/binary Parquet column as anything else. Spark's
@@ -647,7 +644,7 @@ impl SparkPhysicalExprAdapter {
             // See #4088 and #4351.
             if is_string_or_binary(physical_type) && !is_string_or_binary(target_type) {
                 return Err(parquet_schema_convert_err(
-                    cast.input_field().name(),
+                    input_field.name(),
                     physical_type,
                     target_type,
                 ));
@@ -672,7 +669,7 @@ impl SparkPhysicalExprAdapter {
                 let rejection = reject_on_non_empty_expr(
                     child,
                     cast.target_field(),
-                    cast.input_field().name(),
+                    input_field.name(),
                     physical_type,
                     target_type,
                 );
@@ -692,7 +689,7 @@ impl SparkPhysicalExprAdapter {
                 let dst_int_precision = i32::from(*dst_p) - i32::from(*dst_s);
                 if dst_s < src_s || dst_int_precision < src_int_precision {
                     return Err(parquet_schema_convert_err(
-                        cast.input_field().name(),
+                        input_field.name(),
                         physical_type,
                         target_type,
                     ));
@@ -716,7 +713,7 @@ impl SparkPhysicalExprAdapter {
                     let dst_int_precision = i32::from(dst_p) - i32::from(dst_s);
                     if dst_int_precision < min_int_precision {
                         return Err(parquet_schema_convert_err(
-                            cast.input_field().name(),
+                            input_field.name(),
                             physical_type,
                             target_type,
                         ));
@@ -740,7 +737,7 @@ impl SparkPhysicalExprAdapter {
                     let rejection = reject_on_non_empty_expr(
                         Arc::clone(&child),
                         cast.target_field(),
-                        cast.input_field().name(),
+                        input_field.name(),
                         physical_type,
                         target_type,
                     );
@@ -800,7 +797,7 @@ impl SparkPhysicalExprAdapter {
                 let rejection = reject_on_non_empty_expr(
                     child,
                     cast.target_field(),
-                    cast.input_field().name(),
+                    input_field.name(),
                     physical_type,
                     target_type,
                 );
@@ -820,7 +817,7 @@ impl SparkPhysicalExprAdapter {
             };
             if is_complex(physical_type) != is_complex(target_type) {
                 return Err(parquet_schema_convert_err(
-                    cast.input_field().name(),
+                    input_field.name(),
                     physical_type,
                     target_type,
                 ));
@@ -842,7 +839,7 @@ impl SparkPhysicalExprAdapter {
                 let comet_cast: Arc<dyn PhysicalExpr> = Arc::new(
                     CometCastColumnExpr::new(
                         child,
-                        Arc::clone(cast.input_field()),
+                        Arc::clone(&input_field),
                         Arc::clone(cast.target_field()),
                         None,
                     )
@@ -996,10 +993,6 @@ impl Display for RejectOnNonEmpty {
 }
 
 impl PhysicalExpr for RejectOnNonEmpty {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn data_type(&self, _input_schema: &Schema) -> DataFusionResult<DataType> {
         Ok(self.target_field.data_type().clone())
     }
diff --git a/native/shuffle/src/shuffle_writer.rs b/native/shuffle/src/shuffle_writer.rs
index 8502c79624..a5de73ae00 100644
--- a/native/shuffle/src/shuffle_writer.rs
+++ b/native/shuffle/src/shuffle_writer.rs
@@ -25,7 +25,8 @@ use crate::partitioners::{
 use crate::{CometPartitioning, CompressionCodec};
 use async_trait::async_trait;
 use datafusion::common::exec_datafusion_err;
-use datafusion::physical_expr::{EquivalenceProperties, Partitioning};
+use datafusion::common::tree_node::TreeNodeRecursion;
+use datafusion::physical_expr::{EquivalenceProperties, Partitioning, PhysicalExpr};
 use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
 use datafusion::physical_plan::EmptyRecordBatchStream;
 use datafusion::{
@@ -40,7 +41,6 @@ use datafusion::{
 };
 use futures::{StreamExt, TryFutureExt, TryStreamExt};
 use std::{
-    any::Any,
     fmt,
     fmt::{Debug, Formatter},
     sync::Arc,
@@ -119,11 +119,6 @@ impl DisplayAs for ShuffleWriterExec {
 
 #[async_trait]
 impl ExecutionPlan for ShuffleWriterExec {
-    /// Return a reference to Any that can be used for downcasting
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "ShuffleWriterExec"
     }
@@ -145,6 +140,27 @@ impl ExecutionPlan for ShuffleWriterExec {
         vec![&self.input]
     }
 
+    fn apply_expressions(
+        &self,
+        f: &mut dyn FnMut(&dyn PhysicalExpr) -> Result<TreeNodeRecursion>,
+    ) -> Result<TreeNodeRecursion> {
+        let mut tnr = TreeNodeRecursion::Continue;
+        match &self.partitioning {
+            CometPartitioning::Hash(exprs, _) => {
+                for expr in exprs {
+                    tnr = tnr.visit_sibling(|| f(expr.as_ref()))?;
+                }
+            }
+            CometPartitioning::RangePartitioning(ordering, _, _, _) => {
+                for sort_expr in ordering.iter() {
+                    tnr = tnr.visit_sibling(|| f(sort_expr.expr.as_ref()))?;
+                }
+            }
+            CometPartitioning::SinglePartition | CometPartitioning::RoundRobin(_, _) => {}
+        }
+        Ok(tnr)
+    }
+
     fn with_new_children(
         self: Arc<Self>,
         children: Vec<Arc<dyn ExecutionPlan>>,
diff --git a/native/spark-expr/src/agg_funcs/avg.rs b/native/spark-expr/src/agg_funcs/avg.rs
index 3760b42504..28c02b6306 100644
--- a/native/spark-expr/src/agg_funcs/avg.rs
+++ b/native/spark-expr/src/agg_funcs/avg.rs
@@ -28,7 +28,7 @@ use datafusion::logical_expr::{
     Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, ReversedUDAF, Signature,
 };
 use datafusion::physical_expr::expressions::format_state_name;
-use std::{any::Any, sync::Arc};
+use std::sync::Arc;
 
 use arrow::array::ArrowNativeTypeOp;
 use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs};
@@ -67,11 +67,6 @@ impl Avg {
 }
 
 impl AggregateUDFImpl for Avg {
-    /// Return a reference to Any that can be used for downcasting
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn accumulator(&self, _acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> {
         // All numeric types use Float64 accumulation after casting
         match (&self.input_data_type, &self.result_data_type) {
@@ -238,8 +233,8 @@ where
 
 impl<T, F> GroupsAccumulator for AvgGroupsAccumulator<T, F>
 where
-    T: ArrowNumericType + Send,
-    F: Fn(T::Native, i64) -> Result<T::Native> + Send,
+    T: ArrowNumericType + Send + 'static,
+    F: Fn(T::Native, i64) -> Result<T::Native> + Send + 'static,
 {
     fn update_batch(
         &mut self,
diff --git a/native/spark-expr/src/agg_funcs/avg_decimal.rs b/native/spark-expr/src/agg_funcs/avg_decimal.rs
index 05d74e1f99..a71bbe44ea 100644
--- a/native/spark-expr/src/agg_funcs/avg_decimal.rs
+++ b/native/spark-expr/src/agg_funcs/avg_decimal.rs
@@ -28,7 +28,7 @@ use datafusion::logical_expr::{
     Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, ReversedUDAF, Signature,
 };
 use datafusion::physical_expr::expressions::format_state_name;
-use std::{any::Any, sync::Arc};
+use std::sync::Arc;
 
 use crate::utils::{build_bool_state, is_valid_decimal_precision, unlikely};
 use crate::{decimal_sum_overflow_error, EvalMode, SparkErrorWithContext};
@@ -108,11 +108,6 @@ impl AvgDecimal {
 }
 
 impl AggregateUDFImpl for AvgDecimal {
-    /// Return a reference to Any that can be used for downcasting
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn accumulator(&self, _acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> {
         match (&self.sum_data_type, &self.result_data_type) {
             (Decimal128(sum_precision, sum_scale), Decimal128(target_precision, target_scale)) => {
diff --git a/native/spark-expr/src/agg_funcs/correlation.rs b/native/spark-expr/src/agg_funcs/correlation.rs
index 7e1d85cf78..b47ce0ba60 100644
--- a/native/spark-expr/src/agg_funcs/correlation.rs
+++ b/native/spark-expr/src/agg_funcs/correlation.rs
@@ -18,7 +18,7 @@
 use arrow::array::{Array, ArrayRef, BooleanArray, Float64Array};
 use arrow::compute::{and, is_not_null};
 use arrow::datatypes::{DataType, Field, FieldRef};
-use std::{any::Any, sync::Arc};
+use std::sync::Arc;
 
 use crate::agg_funcs::covariance::{CovarianceAccumulator, CovarianceGroupsAccumulator};
 use crate::agg_funcs::stddev::StddevAccumulator;
@@ -26,7 +26,6 @@ use crate::agg_funcs::variance::VarianceGroupsAccumulator;
 use arrow::compute::filter;
 use datafusion::common::{Result, ScalarValue};
 use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs};
-use datafusion::logical_expr::type_coercion::aggregates::NUMERICS;
 use datafusion::logical_expr::{
     Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, Signature, Volatility,
 };
@@ -51,18 +50,13 @@ impl Correlation {
         assert!(matches!(data_type, DataType::Float64));
         Self {
             name: name.into(),
-            signature: Signature::uniform(2, NUMERICS.to_vec(), Volatility::Immutable),
+            signature: Signature::numeric(2, Volatility::Immutable),
             null_on_divide_by_zero,
         }
     }
 }
 
 impl AggregateUDFImpl for Correlation {
-    /// Return a reference to Any that can be used for downcasting
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         &self.name
     }
diff --git a/native/spark-expr/src/agg_funcs/covariance.rs b/native/spark-expr/src/agg_funcs/covariance.rs
index 276121c11b..7007bc028e 100644
--- a/native/spark-expr/src/agg_funcs/covariance.rs
+++ b/native/spark-expr/src/agg_funcs/covariance.rs
@@ -23,13 +23,11 @@ use arrow::compute::cast;
 use arrow::datatypes::{DataType, Field, FieldRef, Float64Type};
 use datafusion::common::{downcast_value, unwrap_or_internal_err, Result, ScalarValue};
 use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs};
-use datafusion::logical_expr::type_coercion::aggregates::NUMERICS;
 use datafusion::logical_expr::{
     Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, Signature, Volatility,
 };
 use datafusion::physical_expr::expressions::format_state_name;
 use datafusion::physical_expr::expressions::StatsType;
-use std::any::Any;
 use std::mem::size_of;
 use std::sync::Arc;
 
@@ -66,7 +64,7 @@ impl Covariance {
         assert!(matches!(data_type, DataType::Float64));
         Self {
             name: name.into(),
-            signature: Signature::uniform(2, NUMERICS.to_vec(), Volatility::Immutable),
+            signature: Signature::numeric(2, Volatility::Immutable),
             stats_type,
             null_on_divide_by_zero,
         }
@@ -74,11 +72,6 @@ impl Covariance {
 }
 
 impl AggregateUDFImpl for Covariance {
-    /// Return a reference to Any that can be used for downcasting
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         &self.name
     }
diff --git a/native/spark-expr/src/agg_funcs/stddev.rs b/native/spark-expr/src/agg_funcs/stddev.rs
index 4f971a0dfc..3389a86af6 100644
--- a/native/spark-expr/src/agg_funcs/stddev.rs
+++ b/native/spark-expr/src/agg_funcs/stddev.rs
@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::{any::Any, sync::Arc};
+use std::sync::Arc;
 
 use crate::agg_funcs::variance::{VarianceAccumulator, VarianceGroupsAccumulator};
 use arrow::array::{ArrayRef, AsArray, BooleanArray, Float64Array};
@@ -78,11 +78,6 @@ impl Stddev {
 }
 
 impl AggregateUDFImpl for Stddev {
-    /// Return a reference to Any that can be used for downcasting
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         &self.name
     }
diff --git a/native/spark-expr/src/agg_funcs/sum_decimal.rs b/native/spark-expr/src/agg_funcs/sum_decimal.rs
index 180e884bf0..4ddf13c5d6 100644
--- a/native/spark-expr/src/agg_funcs/sum_decimal.rs
+++ b/native/spark-expr/src/agg_funcs/sum_decimal.rs
@@ -27,7 +27,7 @@ use datafusion::logical_expr::Volatility::Immutable;
 use datafusion::logical_expr::{
     Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, ReversedUDAF, Signature,
 };
-use std::{any::Any, sync::Arc};
+use std::sync::Arc;
 
 #[derive(Debug)]
 pub struct SumDecimal {
@@ -99,10 +99,6 @@ impl SumDecimal {
 }
 
 impl AggregateUDFImpl for SumDecimal {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn accumulator(&self, _args: AccumulatorArgs) -> DFResult<Box<dyn Accumulator>> {
         Ok(Box::new(SumDecimalAccumulator::new(
             self.precision,
diff --git a/native/spark-expr/src/agg_funcs/sum_int.rs b/native/spark-expr/src/agg_funcs/sum_int.rs
index 5ce6066772..59a4371772 100644
--- a/native/spark-expr/src/agg_funcs/sum_int.rs
+++ b/native/spark-expr/src/agg_funcs/sum_int.rs
@@ -29,7 +29,7 @@ use datafusion::logical_expr::Volatility::Immutable;
 use datafusion::logical_expr::{
     Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, ReversedUDAF, Signature,
 };
-use std::{any::Any, sync::Arc};
+use std::sync::Arc;
 
 #[derive(Debug, PartialEq, Eq, Hash)]
 pub struct SumInteger {
@@ -52,10 +52,6 @@ impl SumInteger {
 }
 
 impl AggregateUDFImpl for SumInteger {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "sum"
     }
diff --git a/native/spark-expr/src/agg_funcs/variance.rs b/native/spark-expr/src/agg_funcs/variance.rs
index ebf48c82f2..f1841729cf 100644
--- a/native/spark-expr/src/agg_funcs/variance.rs
+++ b/native/spark-expr/src/agg_funcs/variance.rs
@@ -26,7 +26,6 @@ use datafusion::logical_expr::{
 };
 use datafusion::physical_expr::expressions::format_state_name;
 use datafusion::physical_expr::expressions::StatsType;
-use std::any::Any;
 use std::mem::size_of;
 use std::sync::Arc;
 
@@ -72,11 +71,6 @@ impl Variance {
 }
 
 impl AggregateUDFImpl for Variance {
-    /// Return a reference to Any that can be used for downcasting
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         &self.name
     }
diff --git a/native/spark-expr/src/array_funcs/array_compact.rs b/native/spark-expr/src/array_funcs/array_compact.rs
index 5795ec64f8..18e28b9afa 100644
--- a/native/spark-expr/src/array_funcs/array_compact.rs
+++ b/native/spark-expr/src/array_funcs/array_compact.rs
@@ -33,7 +33,6 @@ use datafusion::common::{exec_err, utils::take_function_args, Result};
 use datafusion::logical_expr::{
     ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, TypeSignature, Volatility,
 };
-use std::any::Any;
 use std::sync::Arc;
 
 #[derive(Debug, PartialEq, Eq, Hash)]
@@ -56,10 +55,6 @@ impl SparkArrayCompact {
 }
 
 impl ScalarUDFImpl for SparkArrayCompact {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "spark_array_compact"
     }
diff --git a/native/spark-expr/src/array_funcs/array_insert.rs b/native/spark-expr/src/array_funcs/array_insert.rs
index bce00483bc..e638c440fd 100644
--- a/native/spark-expr/src/array_funcs/array_insert.rs
+++ b/native/spark-expr/src/array_funcs/array_insert.rs
@@ -30,7 +30,6 @@ use datafusion::logical_expr::ColumnarValue;
 use datafusion::physical_expr::PhysicalExpr;
 use std::hash::Hash;
 use std::{
-    any::Any,
     fmt::{Debug, Display, Formatter},
     sync::Arc,
 };
@@ -92,10 +91,6 @@ impl ArrayInsert {
 }
 
 impl PhysicalExpr for ArrayInsert {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         Display::fmt(self, f)
     }
diff --git a/native/spark-expr/src/array_funcs/array_position.rs b/native/spark-expr/src/array_funcs/array_position.rs
index dbcd8615df..a5841b27d3 100644
--- a/native/spark-expr/src/array_funcs/array_position.rs
+++ b/native/spark-expr/src/array_funcs/array_position.rs
@@ -28,7 +28,6 @@ use datafusion::logical_expr::{
     ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, TypeSignature, Volatility,
 };
 use num::Float;
-use std::any::Any;
 use std::sync::Arc;
 
 /// Spark array_position() function that returns the 1-based position of an element in an array.
@@ -313,10 +312,6 @@ impl SparkArrayPositionFunc {
 }
 
 impl ScalarUDFImpl for SparkArrayPositionFunc {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "spark_array_position"
     }
diff --git a/native/spark-expr/src/array_funcs/arrays_overlap.rs b/native/spark-expr/src/array_funcs/arrays_overlap.rs
index 662186e614..889a0314df 100644
--- a/native/spark-expr/src/array_funcs/arrays_overlap.rs
+++ b/native/spark-expr/src/array_funcs/arrays_overlap.rs
@@ -39,7 +39,6 @@ use datafusion::common::{exec_err, utils::take_function_args, Result, ScalarValu
 use datafusion::logical_expr::{
     ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, TypeSignature, Volatility,
 };
-use std::any::Any;
 use std::sync::Arc;
 
 #[derive(Debug, PartialEq, Eq, Hash)]
@@ -62,10 +61,6 @@ impl SparkArraysOverlap {
 }
 
 impl ScalarUDFImpl for SparkArraysOverlap {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "spark_arrays_overlap"
     }
diff --git a/native/spark-expr/src/array_funcs/arrays_zip.rs b/native/spark-expr/src/array_funcs/arrays_zip.rs
index 2126eb732c..6f0332d3d8 100644
--- a/native/spark-expr/src/array_funcs/arrays_zip.rs
+++ b/native/spark-expr/src/array_funcs/arrays_zip.rs
@@ -27,7 +27,6 @@ use datafusion::common::cast::{as_fixed_size_list_array, as_large_list_array, as
 use datafusion::common::{exec_err, Result, ScalarValue};
 use datafusion::logical_expr::ColumnarValue;
 use datafusion::physical_expr::PhysicalExpr;
-use std::any::Any;
 use std::fmt::{Display, Formatter};
 use std::sync::Arc;
 // TODO: Reuse functions from DF
@@ -74,10 +73,6 @@ impl Display for SparkArraysZipFunc {
 }
 
 impl PhysicalExpr for SparkArraysZipFunc {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn data_type(&self, input_schema: &Schema) -> Result<DataType> {
         let fields = self.fields(input_schema)?;
         Ok(List(Arc::new(Field::new_list_field(
diff --git a/native/spark-expr/src/array_funcs/get_array_struct_fields.rs b/native/spark-expr/src/array_funcs/get_array_struct_fields.rs
index dc05a3b7f0..a873392499 100644
--- a/native/spark-expr/src/array_funcs/get_array_struct_fields.rs
+++ b/native/spark-expr/src/array_funcs/get_array_struct_fields.rs
@@ -27,7 +27,6 @@ use datafusion::logical_expr::ColumnarValue;
 use datafusion::physical_expr::PhysicalExpr;
 use std::hash::Hash;
 use std::{
-    any::Any,
     fmt::{Debug, Display, Formatter},
     sync::Arc,
 };
@@ -75,10 +74,6 @@ impl GetArrayStructFields {
 }
 
 impl PhysicalExpr for GetArrayStructFields {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn data_type(&self, input_schema: &Schema) -> DataFusionResult<DataType> {
         let struct_field = self.child_field(input_schema)?;
         match self.child.data_type(input_schema)? {
diff --git a/native/spark-expr/src/array_funcs/list_extract.rs b/native/spark-expr/src/array_funcs/list_extract.rs
index d3661f496a..83fa23f757 100644
--- a/native/spark-expr/src/array_funcs/list_extract.rs
+++ b/native/spark-expr/src/array_funcs/list_extract.rs
@@ -26,7 +26,6 @@ use datafusion::logical_expr::ColumnarValue;
 use datafusion::physical_expr::PhysicalExpr;
 use std::hash::Hash;
 use std::{
-    any::Any,
     fmt::{Debug, Display, Formatter},
     sync::Arc,
 };
@@ -113,10 +112,6 @@ impl ListExtract {
 }
 
 impl PhysicalExpr for ListExtract {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         Display::fmt(self, f)
     }
diff --git a/native/spark-expr/src/array_funcs/size.rs b/native/spark-expr/src/array_funcs/size.rs
index 9777553341..f206b299d6 100644
--- a/native/spark-expr/src/array_funcs/size.rs
+++ b/native/spark-expr/src/array_funcs/size.rs
@@ -21,7 +21,6 @@ use datafusion::common::{exec_err, DataFusionError, Result as DataFusionResult,
 use datafusion::logical_expr::{
     ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility,
 };
-use std::any::Any;
 use std::sync::Arc;
 
 /// Spark size() function that returns the size of arrays or maps.
@@ -73,10 +72,6 @@ impl SparkSizeFunc {
 }
 
 impl ScalarUDFImpl for SparkSizeFunc {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "size"
     }
diff --git a/native/spark-expr/src/bloom_filter/bloom_filter_agg.rs b/native/spark-expr/src/bloom_filter/bloom_filter_agg.rs
index 4ac236e6fc..03f7223d4c 100644
--- a/native/spark-expr/src/bloom_filter/bloom_filter_agg.rs
+++ b/native/spark-expr/src/bloom_filter/bloom_filter_agg.rs
@@ -17,7 +17,7 @@
 
 use arrow::datatypes::{Field, FieldRef};
 use datafusion::{arrow::datatypes::DataType, logical_expr::Volatility};
-use std::{any::Any, sync::Arc};
+use std::sync::Arc;
 
 use crate::bloom_filter::spark_bloom_filter;
 use crate::bloom_filter::spark_bloom_filter::{SparkBloomFilter, SparkBloomFilterVersion};
@@ -45,7 +45,7 @@ pub struct BloomFilterAgg {
 
 #[inline]
 fn extract_i32_from_literal(expr: Arc<dyn PhysicalExpr>) -> i32 {
-    match expr.as_any().downcast_ref::<Literal>().unwrap().value() {
+    match expr.downcast_ref::<Literal>().unwrap().value() {
         ScalarValue::Int64(scalar_value) => scalar_value.unwrap() as i32,
         _ => {
             unreachable!()
@@ -81,10 +81,6 @@ impl BloomFilterAgg {
 }
 
 impl AggregateUDFImpl for BloomFilterAgg {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "bloom_filter_agg"
     }
diff --git a/native/spark-expr/src/bloom_filter/bloom_filter_might_contain.rs b/native/spark-expr/src/bloom_filter/bloom_filter_might_contain.rs
index ea246dfb25..66168444d9 100644
--- a/native/spark-expr/src/bloom_filter/bloom_filter_might_contain.rs
+++ b/native/spark-expr/src/bloom_filter/bloom_filter_might_contain.rs
@@ -22,7 +22,6 @@ use datafusion::error::DataFusionError;
 use datafusion::logical_expr::{ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility};
 use datafusion::physical_expr::PhysicalExpr;
 use datafusion::physical_plan::ColumnarValue;
-use std::any::Any;
 use std::sync::Arc;
 
 use crate::bloom_filter::spark_bloom_filter::SparkBloomFilter;
@@ -63,10 +62,6 @@ fn evaluate_bloom_filter(
 }
 
 impl ScalarUDFImpl for BloomFilterMightContain {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "might_contain"
     }
diff --git a/native/spark-expr/src/comet_scalar_funcs.rs b/native/spark-expr/src/comet_scalar_funcs.rs
index 7108105dcb..8916cc92d6 100644
--- a/native/spark-expr/src/comet_scalar_funcs.rs
+++ b/native/spark-expr/src/comet_scalar_funcs.rs
@@ -37,7 +37,6 @@ use datafusion::logical_expr::{
     Volatility,
 };
 use datafusion::physical_plan::ColumnarValue;
-use std::any::Any;
 use std::fmt::Debug;
 use std::sync::Arc;
 
@@ -288,10 +287,6 @@ impl CometScalarFunction {
 }
 
 impl ScalarUDFImpl for CometScalarFunction {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         self.name.as_str()
     }
diff --git a/native/spark-expr/src/conditional_funcs/if_expr.rs b/native/spark-expr/src/conditional_funcs/if_expr.rs
index 6b1291fbbe..5e57c54a22 100644
--- a/native/spark-expr/src/conditional_funcs/if_expr.rs
+++ b/native/spark-expr/src/conditional_funcs/if_expr.rs
@@ -24,7 +24,7 @@ use datafusion::logical_expr::ColumnarValue;
 use datafusion::physical_expr::{expressions::CaseExpr, PhysicalExpr};
 use std::fmt::{Display, Formatter};
 use std::hash::Hash;
-use std::{any::Any, sync::Arc};
+use std::sync::Arc;
 
 /// IfExpr is a wrapper around CaseExpr, because `IF(a, b, c)` is semantically equivalent to
 /// `CASE WHEN a THEN b ELSE c END`.
@@ -83,11 +83,6 @@ impl IfExpr {
 }
 
 impl PhysicalExpr for IfExpr {
-    /// Return a reference to Any that can be used for down-casting
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         Display::fmt(self, f)
     }
diff --git a/native/spark-expr/src/conversion_funcs/cast.rs b/native/spark-expr/src/conversion_funcs/cast.rs
index 1f574f1231..dadcd1048e 100644
--- a/native/spark-expr/src/conversion_funcs/cast.rs
+++ b/native/spark-expr/src/conversion_funcs/cast.rs
@@ -63,7 +63,6 @@ use datafusion::common::{internal_err, DataFusionError, Result as DataFusionResu
 use datafusion::physical_expr::PhysicalExpr;
 use datafusion::physical_plan::ColumnarValue;
 use std::{
-    any::Any,
     fmt::{Debug, Display, Formatter},
     hash::Hash,
     sync::Arc,
@@ -740,10 +739,6 @@ impl Display for Cast {
 }
 
 impl PhysicalExpr for Cast {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         Display::fmt(self, f)
     }
diff --git a/native/spark-expr/src/csv_funcs/to_csv.rs b/native/spark-expr/src/csv_funcs/to_csv.rs
index f41cb7f918..01fdc901cb 100644
--- a/native/spark-expr/src/csv_funcs/to_csv.rs
+++ b/native/spark-expr/src/csv_funcs/to_csv.rs
@@ -23,7 +23,6 @@ use arrow::datatypes::{DataType, Schema};
 use datafusion::common::Result;
 use datafusion::logical_expr::ColumnarValue;
 use datafusion::physical_expr::PhysicalExpr;
-use std::any::Any;
 use std::fmt::{Display, Formatter};
 use std::hash::Hash;
 use std::sync::Arc;
@@ -77,10 +76,6 @@ impl Display for ToCsv {
 }
 
 impl PhysicalExpr for ToCsv {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn data_type(&self, _: &Schema) -> Result<DataType> {
         Ok(DataType::Utf8)
     }
diff --git a/native/spark-expr/src/datetime_funcs/date_diff.rs b/native/spark-expr/src/datetime_funcs/date_diff.rs
index ca148c103a..be3c1d3552 100644
--- a/native/spark-expr/src/datetime_funcs/date_diff.rs
+++ b/native/spark-expr/src/datetime_funcs/date_diff.rs
@@ -22,7 +22,6 @@ use datafusion::common::{utils::take_function_args, DataFusionError, Result};
 use datafusion::logical_expr::{
     ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility,
 };
-use std::any::Any;
 use std::sync::Arc;
 
 /// Spark-compatible date_diff function.
@@ -52,10 +51,6 @@ impl Default for SparkDateDiff {
 }
 
 impl ScalarUDFImpl for SparkDateDiff {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "date_diff"
     }
diff --git a/native/spark-expr/src/datetime_funcs/date_from_unix_date.rs b/native/spark-expr/src/datetime_funcs/date_from_unix_date.rs
index 1c88fc47ab..0e624e6472 100644
--- a/native/spark-expr/src/datetime_funcs/date_from_unix_date.rs
+++ b/native/spark-expr/src/datetime_funcs/date_from_unix_date.rs
@@ -21,7 +21,6 @@ use datafusion::common::{utils::take_function_args, DataFusionError, Result, Sca
 use datafusion::logical_expr::{
     ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility,
 };
-use std::any::Any;
 use std::sync::Arc;
 
 /// Spark-compatible date_from_unix_date function.
@@ -48,10 +47,6 @@ impl Default for SparkDateFromUnixDate {
 }
 
 impl ScalarUDFImpl for SparkDateFromUnixDate {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "date_from_unix_date"
     }
diff --git a/native/spark-expr/src/datetime_funcs/date_trunc.rs b/native/spark-expr/src/datetime_funcs/date_trunc.rs
index aeae18e36f..7ceb5234e1 100644
--- a/native/spark-expr/src/datetime_funcs/date_trunc.rs
+++ b/native/spark-expr/src/datetime_funcs/date_trunc.rs
@@ -22,7 +22,6 @@ use datafusion::common::{
 use datafusion::logical_expr::{
     ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility,
 };
-use std::any::Any;
 
 use crate::kernels::temporal::{date_trunc_array_fmt_dyn, date_trunc_dyn};
 
@@ -51,10 +50,6 @@ impl Default for SparkDateTrunc {
 }
 
 impl ScalarUDFImpl for SparkDateTrunc {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "date_trunc"
     }
diff --git a/native/spark-expr/src/datetime_funcs/extract_date_part.rs b/native/spark-expr/src/datetime_funcs/extract_date_part.rs
index acb7d2266e..7344a3953a 100644
--- a/native/spark-expr/src/datetime_funcs/extract_date_part.rs
+++ b/native/spark-expr/src/datetime_funcs/extract_date_part.rs
@@ -22,7 +22,7 @@ use datafusion::common::{internal_datafusion_err, DataFusionError};
 use datafusion::logical_expr::{
     ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility,
 };
-use std::{any::Any, fmt::Debug};
+use std::fmt::Debug;
 
 macro_rules! extract_date_part {
     ($struct_name:ident, $fn_name:expr, $date_part_variant:ident) => {
@@ -44,10 +44,6 @@ macro_rules! extract_date_part {
         }
 
         impl ScalarUDFImpl for $struct_name {
-            fn as_any(&self) -> &dyn Any {
-                self
-            }
-
             fn name(&self) -> &str {
                 $fn_name
             }
diff --git a/native/spark-expr/src/datetime_funcs/hours.rs b/native/spark-expr/src/datetime_funcs/hours.rs
index ea3ef742a4..26ec35ef25 100644
--- a/native/spark-expr/src/datetime_funcs/hours.rs
+++ b/native/spark-expr/src/datetime_funcs/hours.rs
@@ -32,7 +32,7 @@ use datafusion::logical_expr::{
     ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility,
 };
 use num::integer::div_floor;
-use std::{any::Any, fmt::Debug, sync::Arc};
+use std::{fmt::Debug, sync::Arc};
 
 const MICROS_PER_HOUR: i64 = 3_600_000_000;
 
@@ -56,10 +56,6 @@ impl Default for SparkHoursTransform {
 }
 
 impl ScalarUDFImpl for SparkHoursTransform {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "hours_transform"
     }
diff --git a/native/spark-expr/src/datetime_funcs/make_date.rs b/native/spark-expr/src/datetime_funcs/make_date.rs
index 58e4108580..ef29431703 100644
--- a/native/spark-expr/src/datetime_funcs/make_date.rs
+++ b/native/spark-expr/src/datetime_funcs/make_date.rs
@@ -23,7 +23,6 @@ use datafusion::common::{utils::take_function_args, DataFusionError, Result};
 use datafusion::logical_expr::{
     ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility,
 };
-use std::any::Any;
 use std::sync::Arc;
 
 /// Spark-compatible make_date function.
@@ -75,10 +74,6 @@ fn make_date(year: i32, month: i32, day: i32) -> Option<i32> {
 }
 
 impl ScalarUDFImpl for SparkMakeDate {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "make_date"
     }
diff --git a/native/spark-expr/src/datetime_funcs/make_time.rs b/native/spark-expr/src/datetime_funcs/make_time.rs
index 154ef3bf46..ec95d19044 100644
--- a/native/spark-expr/src/datetime_funcs/make_time.rs
+++ b/native/spark-expr/src/datetime_funcs/make_time.rs
@@ -22,7 +22,6 @@ use datafusion::common::{utils::take_function_args, DataFusionError, Result};
 use datafusion::logical_expr::{
     ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility,
 };
-use std::any::Any;
 use std::sync::Arc;
 
 const MICROS_PER_SECOND: i128 = 1_000_000;
@@ -89,10 +88,6 @@ fn make_time(hours: i32, minutes: i32, secs_and_micros_unscaled: i128) -> Result
 }
 
 impl ScalarUDFImpl for SparkMakeTime {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "make_time"
     }
diff --git a/native/spark-expr/src/datetime_funcs/seconds_to_timestamp.rs b/native/spark-expr/src/datetime_funcs/seconds_to_timestamp.rs
index 2da1ac73b0..ab9388692b 100644
--- a/native/spark-expr/src/datetime_funcs/seconds_to_timestamp.rs
+++ b/native/spark-expr/src/datetime_funcs/seconds_to_timestamp.rs
@@ -24,7 +24,6 @@ use datafusion::common::{utils::take_function_args, DataFusionError, Result, Sca
 use datafusion::logical_expr::{
     ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, TypeSignature, Volatility,
 };
-use std::any::Any;
 use std::sync::Arc;
 
 const MICROS_PER_SECOND: i64 = 1_000_000;
@@ -61,10 +60,6 @@ impl Default for SparkSecondsToTimestamp {
 }
 
 impl ScalarUDFImpl for SparkSecondsToTimestamp {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "seconds_to_timestamp"
     }
diff --git a/native/spark-expr/src/datetime_funcs/timestamp_trunc.rs b/native/spark-expr/src/datetime_funcs/timestamp_trunc.rs
index 3435d3ee50..c26c5ef08e 100644
--- a/native/spark-expr/src/datetime_funcs/timestamp_trunc.rs
+++ b/native/spark-expr/src/datetime_funcs/timestamp_trunc.rs
@@ -23,7 +23,6 @@ use datafusion::logical_expr::ColumnarValue;
 use datafusion::physical_expr::PhysicalExpr;
 use std::hash::Hash;
 use std::{
-    any::Any,
     fmt::{Debug, Display, Formatter},
     sync::Arc,
 };
@@ -85,10 +84,6 @@ impl Display for TimestampTruncExpr {
 }
 
 impl PhysicalExpr for TimestampTruncExpr {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         Display::fmt(self, f)
     }
diff --git a/native/spark-expr/src/datetime_funcs/unix_timestamp.rs b/native/spark-expr/src/datetime_funcs/unix_timestamp.rs
index f8e932f396..bd62563a6b 100644
--- a/native/spark-expr/src/datetime_funcs/unix_timestamp.rs
+++ b/native/spark-expr/src/datetime_funcs/unix_timestamp.rs
@@ -24,7 +24,7 @@ use datafusion::logical_expr::{
     ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility,
 };
 use num::integer::div_floor;
-use std::{any::Any, fmt::Debug, sync::Arc};
+use std::{fmt::Debug, sync::Arc};
 
 const MICROS_PER_SECOND: i64 = 1_000_000;
 
@@ -46,10 +46,6 @@ impl SparkUnixTimestamp {
 }
 
 impl ScalarUDFImpl for SparkUnixTimestamp {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "unix_timestamp"
     }
diff --git a/native/spark-expr/src/json_funcs/from_json.rs b/native/spark-expr/src/json_funcs/from_json.rs
index 685ea3c8ec..eaca6db016 100644
--- a/native/spark-expr/src/json_funcs/from_json.rs
+++ b/native/spark-expr/src/json_funcs/from_json.rs
@@ -86,10 +86,6 @@ impl PartialEq<dyn Any> for FromJson {
 }
 
 impl PhysicalExpr for FromJson {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         Display::fmt(self, f)
     }
diff --git a/native/spark-expr/src/json_funcs/to_json.rs b/native/spark-expr/src/json_funcs/to_json.rs
index 3cc827f210..6dcd8bc045 100644
--- a/native/spark-expr/src/json_funcs/to_json.rs
+++ b/native/spark-expr/src/json_funcs/to_json.rs
@@ -79,10 +79,6 @@ impl PartialEq<dyn Any> for ToJson {
 }
 
 impl PhysicalExpr for ToJson {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         Display::fmt(self, f)
     }
diff --git a/native/spark-expr/src/jvm_udf/mod.rs b/native/spark-expr/src/jvm_udf/mod.rs
index 0e3968e60a..0ca603ac9b 100644
--- a/native/spark-expr/src/jvm_udf/mod.rs
+++ b/native/spark-expr/src/jvm_udf/mod.rs
@@ -15,7 +15,6 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::any::Any;
 use std::fmt::{Display, Formatter};
 use std::hash::{Hash, Hasher};
 use std::sync::Arc;
@@ -107,10 +106,6 @@ impl PartialEq for JvmScalarUdfExpr {
 impl Eq for JvmScalarUdfExpr {}
 
 impl PhysicalExpr for JvmScalarUdfExpr {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         Display::fmt(self, f)
     }
diff --git a/native/spark-expr/src/math_funcs/internal/checkoverflow.rs b/native/spark-expr/src/math_funcs/internal/checkoverflow.rs
index f1fb9c2f02..f867fe0c4d 100644
--- a/native/spark-expr/src/math_funcs/internal/checkoverflow.rs
+++ b/native/spark-expr/src/math_funcs/internal/checkoverflow.rs
@@ -28,7 +28,6 @@ use std::hash::Hash;
 
 use crate::SparkError;
 use std::{
-    any::Any,
     fmt::{Display, Formatter},
     sync::Arc,
 };
@@ -91,10 +90,6 @@ impl Display for CheckOverflow {
 }
 
 impl PhysicalExpr for CheckOverflow {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         Display::fmt(self, f)
     }
@@ -274,9 +269,6 @@ mod tests {
     }
 
     impl PhysicalExpr for ScalarChild {
-        fn as_any(&self) -> &dyn Any {
-            self
-        }
         fn data_type(&self, _: &Schema) -> datafusion::common::Result<DataType> {
             Ok(DataType::Decimal128(self.1, self.2))
         }
diff --git a/native/spark-expr/src/math_funcs/internal/decimal_rescale_check.rs b/native/spark-expr/src/math_funcs/internal/decimal_rescale_check.rs
index 1322404951..667f76939a 100644
--- a/native/spark-expr/src/math_funcs/internal/decimal_rescale_check.rs
+++ b/native/spark-expr/src/math_funcs/internal/decimal_rescale_check.rs
@@ -29,7 +29,6 @@ use datafusion::logical_expr::ColumnarValue;
 use datafusion::physical_expr::PhysicalExpr;
 use std::hash::Hash;
 use std::{
-    any::Any,
     fmt::{Display, Formatter},
     sync::Arc,
 };
@@ -154,10 +153,6 @@ fn rescale_and_check(
 }
 
 impl PhysicalExpr for DecimalRescaleCheckOverflow {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         Display::fmt(self, f)
     }
@@ -392,9 +387,6 @@ mod tests {
     }
 
     impl PhysicalExpr for ScalarChild {
-        fn as_any(&self) -> &dyn Any {
-            self
-        }
         fn data_type(&self, _: &Schema) -> datafusion::common::Result<DataType> {
             Ok(DataType::Decimal128(self.1, self.2))
         }
diff --git a/native/spark-expr/src/math_funcs/internal/normalize_nan.rs b/native/spark-expr/src/math_funcs/internal/normalize_nan.rs
index b3838f64f4..165ae3acc9 100644
--- a/native/spark-expr/src/math_funcs/internal/normalize_nan.rs
+++ b/native/spark-expr/src/math_funcs/internal/normalize_nan.rs
@@ -26,7 +26,6 @@ use datafusion::logical_expr::ColumnarValue;
 use datafusion::physical_expr::PhysicalExpr;
 use std::hash::Hash;
 use std::{
-    any::Any,
     fmt::{Display, Formatter},
     sync::Arc,
 };
@@ -57,10 +56,6 @@ impl NormalizeNaNAndZero {
 }
 
 impl PhysicalExpr for NormalizeNaNAndZero {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         Display::fmt(self, f)
     }
diff --git a/native/spark-expr/src/math_funcs/negative.rs b/native/spark-expr/src/math_funcs/negative.rs
index a268894086..650fa401ef 100644
--- a/native/spark-expr/src/math_funcs/negative.rs
+++ b/native/spark-expr/src/math_funcs/negative.rs
@@ -29,7 +29,7 @@ use datafusion::{
 };
 use std::fmt::{Display, Formatter};
 use std::hash::Hash;
-use std::{any::Any, sync::Arc};
+use std::sync::Arc;
 
 pub fn create_negate_expr(
     expr: Arc<dyn PhysicalExpr>,
@@ -96,11 +96,6 @@ impl std::fmt::Display for NegativeExpr {
 }
 
 impl PhysicalExpr for NegativeExpr {
-    /// Return a reference to Any that can be used for downcasting
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn data_type(&self, input_schema: &Schema) -> Result<DataType> {
         self.arg.data_type(input_schema)
     }
diff --git a/native/spark-expr/src/math_funcs/wide_decimal_binary_expr.rs b/native/spark-expr/src/math_funcs/wide_decimal_binary_expr.rs
index 7ff78713be..ca4869357e 100644
--- a/native/spark-expr/src/math_funcs/wide_decimal_binary_expr.rs
+++ b/native/spark-expr/src/math_funcs/wide_decimal_binary_expr.rs
@@ -31,7 +31,7 @@ use datafusion::logical_expr::ColumnarValue;
 use datafusion::physical_expr::PhysicalExpr;
 use std::fmt::{Display, Formatter};
 use std::hash::Hash;
-use std::{any::Any, sync::Arc};
+use std::sync::Arc;
 
 /// The arithmetic operation to perform.
 #[derive(Debug, Hash, PartialEq, Eq, Clone, Copy)]
@@ -165,10 +165,6 @@ fn max_for_precision(precision: u8) -> i256 {
 }
 
 impl PhysicalExpr for WideDecimalBinaryExpr {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn data_type(&self, _input_schema: &Schema) -> Result<DataType> {
         Ok(DataType::Decimal128(
             self.output_precision,
diff --git a/native/spark-expr/src/nondetermenistic_funcs/monotonically_increasing_id.rs b/native/spark-expr/src/nondetermenistic_funcs/monotonically_increasing_id.rs
index 49a5066a38..e7404e00c1 100644
--- a/native/spark-expr/src/nondetermenistic_funcs/monotonically_increasing_id.rs
+++ b/native/spark-expr/src/nondetermenistic_funcs/monotonically_increasing_id.rs
@@ -20,7 +20,6 @@ use arrow::datatypes::{DataType, Schema};
 use datafusion::common::Result;
 use datafusion::logical_expr::ColumnarValue;
 use datafusion::physical_expr::PhysicalExpr;
-use std::any::Any;
 use std::fmt::{Debug, Display, Formatter};
 use std::hash::{Hash, Hasher};
 use std::sync::atomic::{AtomicI64, Ordering};
@@ -66,10 +65,6 @@ impl Hash for MonotonicallyIncreasingId {
 }
 
 impl PhysicalExpr for MonotonicallyIncreasingId {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
         let start = self
             .current_offset
diff --git a/native/spark-expr/src/nondetermenistic_funcs/rand.rs b/native/spark-expr/src/nondetermenistic_funcs/rand.rs
index e23a83d84e..8f07f37344 100644
--- a/native/spark-expr/src/nondetermenistic_funcs/rand.rs
+++ b/native/spark-expr/src/nondetermenistic_funcs/rand.rs
@@ -23,7 +23,6 @@ use arrow::datatypes::{DataType, Schema};
 use datafusion::common::Result;
 use datafusion::logical_expr::ColumnarValue;
 use datafusion::physical_expr::PhysicalExpr;
-use std::any::Any;
 use std::fmt::{Display, Formatter};
 use std::hash::{Hash, Hasher};
 use std::sync::{Arc, Mutex};
@@ -120,10 +119,6 @@ impl Hash for RandExpr {
 }
 
 impl PhysicalExpr for RandExpr {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn data_type(&self, _input_schema: &Schema) -> Result<DataType> {
         Ok(DataType::Float64)
     }
diff --git a/native/spark-expr/src/nondetermenistic_funcs/randn.rs b/native/spark-expr/src/nondetermenistic_funcs/randn.rs
index 40fafedc20..7d50d24811 100644
--- a/native/spark-expr/src/nondetermenistic_funcs/randn.rs
+++ b/native/spark-expr/src/nondetermenistic_funcs/randn.rs
@@ -22,7 +22,6 @@ use arrow::array::RecordBatch;
 use arrow::datatypes::{DataType, Schema};
 use datafusion::logical_expr::ColumnarValue;
 use datafusion::physical_expr::PhysicalExpr;
-use std::any::Any;
 use std::fmt::{Display, Formatter};
 use std::hash::{Hash, Hasher};
 use std::sync::{Arc, Mutex};
@@ -131,10 +130,6 @@ impl Hash for RandnExpr {
 }
 
 impl PhysicalExpr for RandnExpr {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn data_type(&self, _input_schema: &Schema) -> datafusion::common::Result<DataType> {
         Ok(DataType::Float64)
     }
diff --git a/native/spark-expr/src/predicate_funcs/rlike.rs b/native/spark-expr/src/predicate_funcs/rlike.rs
index ed5970a6a2..ee005dd1ac 100644
--- a/native/spark-expr/src/predicate_funcs/rlike.rs
+++ b/native/spark-expr/src/predicate_funcs/rlike.rs
@@ -25,7 +25,6 @@ use datafusion::common::{internal_err, Result, ScalarValue};
 use datafusion::physical_expr::PhysicalExpr;
 use datafusion::physical_plan::ColumnarValue;
 use regex::Regex;
-use std::any::Any;
 use std::fmt::{Display, Formatter};
 use std::hash::{Hash, Hasher};
 use std::sync::Arc;
@@ -102,10 +101,6 @@ impl Display for RLike {
 }
 
 impl PhysicalExpr for RLike {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn data_type(&self, _input_schema: &Schema) -> Result<DataType> {
         Ok(DataType::Boolean)
     }
diff --git a/native/spark-expr/src/string_funcs/contains.rs b/native/spark-expr/src/string_funcs/contains.rs
index bc34ce9cba..537227efdf 100644
--- a/native/spark-expr/src/string_funcs/contains.rs
+++ b/native/spark-expr/src/string_funcs/contains.rs
@@ -27,7 +27,6 @@ use datafusion::common::{exec_err, Result, ScalarValue};
 use datafusion::logical_expr::{
     ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility,
 };
-use std::any::Any;
 use std::sync::Arc;
 
 /// Spark-optimized contains function.
@@ -53,10 +52,6 @@ impl SparkContains {
 }
 
 impl ScalarUDFImpl for SparkContains {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "contains"
     }
diff --git a/native/spark-expr/src/string_funcs/substring.rs b/native/spark-expr/src/string_funcs/substring.rs
index dff32c10d7..9226241d86 100644
--- a/native/spark-expr/src/string_funcs/substring.rs
+++ b/native/spark-expr/src/string_funcs/substring.rs
@@ -24,7 +24,6 @@ use arrow::record_batch::RecordBatch;
 use datafusion::logical_expr::ColumnarValue;
 use datafusion::physical_expr::PhysicalExpr;
 use std::{
-    any::Any,
     fmt::{Display, Formatter},
     hash::Hash,
     sync::Arc,
@@ -68,10 +67,6 @@ impl Display for SubstringExpr {
 }
 
 impl PhysicalExpr for SubstringExpr {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         Display::fmt(self, f)
     }
diff --git a/native/spark-expr/src/struct_funcs/create_named_struct.rs b/native/spark-expr/src/struct_funcs/create_named_struct.rs
index 70e03ad0c0..1a63cb1cc9 100644
--- a/native/spark-expr/src/struct_funcs/create_named_struct.rs
+++ b/native/spark-expr/src/struct_funcs/create_named_struct.rs
@@ -22,7 +22,6 @@ use datafusion::common::Result as DataFusionResult;
 use datafusion::logical_expr::ColumnarValue;
 use datafusion::physical_expr::PhysicalExpr;
 use std::{
-    any::Any,
     fmt::{Display, Formatter},
     hash::Hash,
     sync::Arc,
@@ -53,10 +52,6 @@ impl CreateNamedStruct {
 }
 
 impl PhysicalExpr for CreateNamedStruct {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         Display::fmt(self, f)
     }
diff --git a/native/spark-expr/src/struct_funcs/get_struct_field.rs b/native/spark-expr/src/struct_funcs/get_struct_field.rs
index 7929cea483..6684967c3a 100644
--- a/native/spark-expr/src/struct_funcs/get_struct_field.rs
+++ b/native/spark-expr/src/struct_funcs/get_struct_field.rs
@@ -22,7 +22,6 @@ use datafusion::common::{DataFusionError, Result as DataFusionResult, ScalarValu
 use datafusion::logical_expr::ColumnarValue;
 use datafusion::physical_expr::PhysicalExpr;
 use std::{
-    any::Any,
     fmt::{Display, Formatter},
     hash::Hash,
     sync::Arc,
@@ -62,10 +61,6 @@ impl GetStructField {
 }
 
 impl PhysicalExpr for GetStructField {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         Display::fmt(self, f)
     }
diff --git a/native/spark-expr/src/unbound.rs b/native/spark-expr/src/unbound.rs
index cf0adafa91..69187dbc02 100644
--- a/native/spark-expr/src/unbound.rs
+++ b/native/spark-expr/src/unbound.rs
@@ -59,11 +59,6 @@ impl std::fmt::Display for UnboundColumn {
 }
 
 impl PhysicalExpr for UnboundColumn {
-    /// Return a reference to Any that can be used for downcasting
-    fn as_any(&self) -> &dyn std::any::Any {
-        self
-    }
-
     fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         Display::fmt(self, f)
     }