diff --git a/Cargo.toml b/Cargo.toml
index 877cead93673..c155e475a026 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -69,22 +69,22 @@ version = "41.0.0"
 ahash = { version = "0.8", default-features = false, features = [
     "runtime-rng",
 ] }
-arrow = { version = "52.2.0", features = [
+arrow = { version = "53.0.0", features = [
     "prettyprint",
 ] }
-arrow-array = { version = "52.2.0", default-features = false, features = [
+arrow-array = { version = "53.0.0", default-features = false, features = [
     "chrono-tz",
 ] }
-arrow-buffer = { version = "52.2.0", default-features = false }
-arrow-flight = { version = "52.2.0", features = [
+arrow-buffer = { version = "53.0.0", default-features = false }
+arrow-flight = { version = "53.0.0", features = [
     "flight-sql-experimental",
 ] }
-arrow-ipc = { version = "52.2.0", default-features = false, features = [
+arrow-ipc = { version = "53.0.0", default-features = false, features = [
     "lz4",
 ] }
-arrow-ord = { version = "52.2.0", default-features = false }
-arrow-schema = { version = "52.2.0", default-features = false }
-arrow-string = { version = "52.2.0", default-features = false }
+arrow-ord = { version = "53.0.0", default-features = false }
+arrow-schema = { version = "53.0.0", default-features = false }
+arrow-string = { version = "53.0.0", default-features = false }
 async-trait = "0.1.73"
 bigdecimal = "=0.4.1"
 bytes = "1.4"
@@ -122,15 +122,17 @@ indexmap = "2.0.0"
 itertools = "0.13"
 log = "^0.4"
 num_cpus = "1.13.0"
-object_store = { version = "0.10.2", default-features = false }
+object_store = { version = "0.11.0", default-features = false }
 parking_lot = "0.12"
-parquet = { version = "52.2.0", default-features = false, features = [
+parquet = { version = "53.0.0", default-features = false, features = [
     "arrow",
     "async",
     "object_store",
 ] }
+pbjson = { version = "0.7.0" }
 # Should match arrow-flight's version of prost.
-prost = "0.12.3"
+prost = "0.13.1"
+prost-derive = "0.13.1"
 rand = "0.8"
 regex = "1.8"
 rstest = "0.22.0"
diff --git a/datafusion-cli/Cargo.lock b/datafusion-cli/Cargo.lock
index ddc6242977d3..039f3fb9a6aa 100644
--- a/datafusion-cli/Cargo.lock
+++ b/datafusion-cli/Cargo.lock
@@ -179,9 +179,9 @@ checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50"
 
 [[package]]
 name = "arrow"
-version = "52.2.0"
+version = "53.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "05048a8932648b63f21c37d88b552ccc8a65afb6dfe9fc9f30ce79174c2e7a85"
+checksum = "45aef0d9cf9a039bf6cd1acc451b137aca819977b0928dece52bd92811b640ba"
 dependencies = [
  "arrow-arith",
  "arrow-array",
@@ -200,9 +200,9 @@ dependencies = [
 
 [[package]]
 name = "arrow-arith"
-version = "52.2.0"
+version = "53.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1d8a57966e43bfe9a3277984a14c24ec617ad874e4c0e1d2a1b083a39cfbf22c"
+checksum = "03675e42d1560790f3524800e41403b40d0da1c793fe9528929fde06d8c7649a"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -215,9 +215,9 @@ dependencies = [
 
 [[package]]
 name = "arrow-array"
-version = "52.2.0"
+version = "53.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "16f4a9468c882dc66862cef4e1fd8423d47e67972377d85d80e022786427768c"
+checksum = "cd2bf348cf9f02a5975c5962c7fa6dee107a2009a7b41ac5fb1a027e12dc033f"
 dependencies = [
  "ahash",
  "arrow-buffer",
@@ -232,9 +232,9 @@ dependencies = [
 
 [[package]]
 name = "arrow-buffer"
-version = "52.2.0"
+version = "53.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c975484888fc95ec4a632cdc98be39c085b1bb518531b0c80c5d462063e5daa1"
+checksum = "3092e37715f168976012ce52273c3989b5793b0db5f06cbaa246be25e5f0924d"
 dependencies = [
  "bytes",
  "half",
@@ -243,9 +243,9 @@ dependencies = [
 
 [[package]]
 name = "arrow-cast"
-version = "52.2.0"
+version = "53.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "da26719e76b81d8bc3faad1d4dbdc1bcc10d14704e63dc17fc9f3e7e1e567c8e"
+checksum = "7ce1018bb710d502f9db06af026ed3561552e493e989a79d0d0f5d9cf267a785"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -264,9 +264,9 @@ dependencies = [
 
 [[package]]
 name = "arrow-csv"
-version = "52.2.0"
+version = "53.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c13c36dc5ddf8c128df19bab27898eea64bf9da2b555ec1cd17a8ff57fba9ec2"
+checksum = "fd178575f45624d045e4ebee714e246a05d9652e41363ee3f57ec18cca97f740"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -283,9 +283,9 @@ dependencies = [
 
 [[package]]
 name = "arrow-data"
-version = "52.2.0"
+version = "53.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dd9d6f18c65ef7a2573ab498c374d8ae364b4a4edf67105357491c031f716ca5"
+checksum = "4e4ac0c4ee79150afe067dc4857154b3ee9c1cd52b5f40d59a77306d0ed18d65"
 dependencies = [
  "arrow-buffer",
  "arrow-schema",
@@ -295,9 +295,9 @@ dependencies = [
 
 [[package]]
 name = "arrow-ipc"
-version = "52.2.0"
+version = "53.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e786e1cdd952205d9a8afc69397b317cfbb6e0095e445c69cda7e8da5c1eeb0f"
+checksum = "bb307482348a1267f91b0912e962cd53440e5de0f7fb24c5f7b10da70b38c94a"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -310,9 +310,9 @@ dependencies = [
 
 [[package]]
 name = "arrow-json"
-version = "52.2.0"
+version = "53.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fb22284c5a2a01d73cebfd88a33511a3234ab45d66086b2ca2d1228c3498e445"
+checksum = "d24805ba326758effdd6f2cbdd482fcfab749544f21b134701add25b33f474e6"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -330,9 +330,9 @@ dependencies = [
 
 [[package]]
 name = "arrow-ord"
-version = "52.2.0"
+version = "53.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "42745f86b1ab99ef96d1c0bcf49180848a64fe2c7a7a0d945bc64fa2b21ba9bc"
+checksum = "644046c479d80ae8ed02a7f1e1399072ea344ca6a7b0e293ab2d5d9ed924aa3b"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -345,9 +345,9 @@ dependencies = [
 
 [[package]]
 name = "arrow-row"
-version = "52.2.0"
+version = "53.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4cd09a518c602a55bd406bcc291a967b284cfa7a63edfbf8b897ea4748aad23c"
+checksum = "a29791f8eb13b340ce35525b723f5f0df17ecb955599e11f65c2a94ab34e2efb"
 dependencies = [
  "ahash",
  "arrow-array",
@@ -359,15 +359,15 @@ dependencies = [
 
 [[package]]
 name = "arrow-schema"
-version = "52.2.0"
+version = "53.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9e972cd1ff4a4ccd22f86d3e53e835c2ed92e0eea6a3e8eadb72b4f1ac802cf8"
+checksum = "c85320a3a2facf2b2822b57aa9d6d9d55edb8aee0b6b5d3b8df158e503d10858"
 
 [[package]]
 name = "arrow-select"
-version = "52.2.0"
+version = "53.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "600bae05d43483d216fb3494f8c32fdbefd8aa4e1de237e790dbb3d9f44690a3"
+checksum = "9cc7e6b582e23855fd1625ce46e51647aa440c20ea2e71b1d748e0839dd73cba"
 dependencies = [
  "ahash",
  "arrow-array",
@@ -379,9 +379,9 @@ dependencies = [
 
 [[package]]
 name = "arrow-string"
-version = "52.2.0"
+version = "53.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f0dc1985b67cb45f6606a248ac2b4a288849f196bab8c657ea5589f47cdd55e6"
+checksum = "0775b6567c66e56ded19b87a954b6b1beffbdd784ef95a3a2b03f59570c1d230"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -430,13 +430,13 @@ dependencies = [
 
 [[package]]
 name = "async-trait"
-version = "0.1.81"
+version = "0.1.82"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6e0c28dcc82d7c8ead5cb13beb15405b57b8546e93215673ff8ca0349a028107"
+checksum = "a27b8a3a6e1a44fa4c8baf1f653e4172e81486d4941f2237e20dc2d0cf4ddff1"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.75",
+ "syn 2.0.77",
 ]
 
 [[package]]
@@ -503,9 +503,9 @@ dependencies = [
 
 [[package]]
 name = "aws-credential-types"
-version = "1.2.0"
+version = "1.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e16838e6c9e12125face1c1eff1343c75e3ff540de98ff7ebd61874a89bcfeb9"
+checksum = "60e8f6b615cb5fc60a98132268508ad104310f0cfb25a1c22eee76efdf9154da"
 dependencies = [
  "aws-smithy-async",
  "aws-smithy-runtime-api",
@@ -515,14 +515,15 @@ dependencies = [
 
 [[package]]
 name = "aws-runtime"
-version = "1.4.0"
+version = "1.4.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f42c2d4218de4dcd890a109461e2f799a1a2ba3bcd2cde9af88360f5df9266c6"
+checksum = "2424565416eef55906f9f8cece2072b6b6a76075e3ff81483ebe938a89a4c05f"
 dependencies = [
  "aws-credential-types",
  "aws-sigv4",
  "aws-smithy-async",
  "aws-smithy-http",
+ "aws-smithy-runtime",
  "aws-smithy-runtime-api",
  "aws-smithy-types",
  "aws-types",
@@ -730,6 +731,7 @@ dependencies = [
  "base64-simd",
  "bytes",
  "bytes-utils",
+ "futures-core",
  "http 0.2.12",
  "http 1.1.0",
  "http-body 0.4.6",
@@ -742,6 +744,8 @@ dependencies = [
  "ryu",
  "serde",
  "time",
+ "tokio",
+ "tokio-util",
 ]
 
 [[package]]
@@ -930,9 +934,9 @@ dependencies = [
 
 [[package]]
 name = "cc"
-version = "1.1.13"
+version = "1.1.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "72db2f7947ecee9b03b510377e8bb9077afa27176fdbff55c51027e976fdcc48"
+checksum = "57b6a275aa2903740dc87da01c62040406b8812552e97129a63ea8850a17c6e6"
 dependencies = [
  "jobserver",
  "libc",
@@ -1011,7 +1015,7 @@ dependencies = [
  "heck 0.5.0",
  "proc-macro2",
  "quote",
- "syn 2.0.75",
+ "syn 2.0.77",
 ]
 
 [[package]]
@@ -1070,9 +1074,9 @@ dependencies = [
 
 [[package]]
 name = "constant_time_eq"
-version = "0.3.0"
+version = "0.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f7144d30dcf0fafbce74250a3963025d8d52177934239851c917d29f1df280c2"
+checksum = "7c74b8349d32d297c9134b8c88677813a227df8f779daa29bfc29c183fe3dca6"
 
 [[package]]
 name = "core-foundation"
@@ -1167,7 +1171,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "edb49164822f3ee45b17acd4a208cfc1251410cf0cad9a833234c9890774dd9f"
 dependencies = [
  "quote",
- "syn 2.0.75",
+ "syn 2.0.77",
 ]
 
 [[package]]
@@ -1268,6 +1272,9 @@ dependencies = [
  "async-trait",
  "aws-config",
  "aws-credential-types",
+ "aws-sdk-sso",
+ "aws-sdk-ssooidc",
+ "aws-sdk-sts",
  "clap",
  "ctor",
  "datafusion",
@@ -1697,9 +1704,9 @@ dependencies = [
 
 [[package]]
 name = "fastrand"
-version = "2.1.0"
+version = "2.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9fc0510504f03c51ada170672ac806f1f105a88aa97a5281117e1ddc3368e51a"
+checksum = "e8c02a5121d4ea3eb16a80748c74f5549a5665e4c21333c6098f283870fbdea6"
 
 [[package]]
 name = "fd-lock"
@@ -1730,9 +1737,9 @@ dependencies = [
 
 [[package]]
 name = "flate2"
-version = "1.0.32"
+version = "1.0.33"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9c0596c1eac1f9e04ed902702e9878208b336edc9d6fddc8a48387349bab3666"
+checksum = "324a1be68054ef05ad64b861cc9eaf1d623d2d8cb25b4bf2cb9cdd902b4bf253"
 dependencies = [
  "crc32fast",
  "miniz_oxide 0.8.0",
@@ -1818,7 +1825,7 @@ checksum = "87750cf4b7a4c0625b1529e4c543c2182106e4dedc60a2a6455e00d212c489ac"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.75",
+ "syn 2.0.77",
 ]
 
 [[package]]
@@ -2136,7 +2143,7 @@ dependencies = [
  "hyper 1.4.1",
  "hyper-util",
  "rustls 0.23.12",
- "rustls-native-certs 0.7.2",
+ "rustls-native-certs 0.7.3",
  "rustls-pki-types",
  "tokio",
  "tokio-rustls 0.26.0",
@@ -2198,9 +2205,9 @@ dependencies = [
 
 [[package]]
 name = "indexmap"
-version = "2.4.0"
+version = "2.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "93ead53efc7ea8ed3cfb0c79fc8023fbb782a5432b52830b6518941cebe6505c"
+checksum = "68b900aa2f7301e21c36462b170ee99994de34dff39a4a6a528e80e7376d07e5"
 dependencies = [
  "equivalent",
  "hashbrown",
@@ -2616,18 +2623,18 @@ dependencies = [
 
 [[package]]
 name = "object"
-version = "0.36.3"
+version = "0.36.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "27b64972346851a39438c60b341ebc01bba47464ae329e55cf343eb93964efd9"
+checksum = "084f1a5821ac4c651660a94a7153d27ac9d8a53736203f58b31945ded098070a"
 dependencies = [
  "memchr",
 ]
 
 [[package]]
 name = "object_store"
-version = "0.10.2"
+version = "0.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e6da452820c715ce78221e8202ccc599b4a52f3e1eb3eedb487b680c81a8e3f3"
+checksum = "25a0c4b3a0e31f8b66f71ad8064521efa773910196e2cde791436f13409f3b45"
 dependencies = [
  "async-trait",
  "base64 0.22.1",
@@ -2706,9 +2713,9 @@ dependencies = [
 
 [[package]]
 name = "parquet"
-version = "52.2.0"
+version = "53.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e977b9066b4d3b03555c22bdc442f3fadebd96a39111249113087d0edb2691cd"
+checksum = "f0fbf928021131daaa57d334ca8e3904fe9ae22f73c56244fc7db9b04eedc3d8"
 dependencies = [
  "ahash",
  "arrow-array",
@@ -2826,7 +2833,7 @@ checksum = "2f38a4412a78282e09a2cf38d195ea5420d15ba0602cb375210efbc877243965"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.75",
+ "syn 2.0.77",
 ]
 
 [[package]]
@@ -2919,9 +2926,9 @@ dependencies = [
 
 [[package]]
 name = "quinn"
-version = "0.11.3"
+version = "0.11.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b22d8e7369034b9a7132bc2008cac12f2013c8132b45e0554e6e20e2617f2156"
+checksum = "8c7c5fdde3cdae7203427dc4f0a68fe0ed09833edc525a03456b153b79828684"
 dependencies = [
  "bytes",
  "pin-project-lite",
@@ -2937,9 +2944,9 @@ dependencies = [
 
 [[package]]
 name = "quinn-proto"
-version = "0.11.6"
+version = "0.11.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ba92fb39ec7ad06ca2582c0ca834dfeadcaf06ddfc8e635c80aa7e1c05315fdd"
+checksum = "fadfaed2cd7f389d0161bb73eeb07b7b78f8691047a6f3e73caaeae55310a4a6"
 dependencies = [
  "bytes",
  "rand",
@@ -2954,22 +2961,22 @@ dependencies = [
 
 [[package]]
 name = "quinn-udp"
-version = "0.5.4"
+version = "0.5.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8bffec3605b73c6f1754535084a85229fa8a30f86014e6c81aeec4abb68b0285"
+checksum = "4fe68c2e9e1a1234e218683dbdf9f9dfcb094113c5ac2b938dfcb9bab4c4140b"
 dependencies = [
  "libc",
  "once_cell",
  "socket2",
  "tracing",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
 ]
 
 [[package]]
 name = "quote"
-version = "1.0.36"
+version = "1.0.37"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0fa76aaf39101c457836aec0ce2316dbdc3ab723cdda1c6bd4e6ad4208acaca7"
+checksum = "b5b9d34b8991d19d98081b46eacdd8eb58c6f2b201139f7c5f643cc155a633af"
 dependencies = [
  "proc-macro2",
 ]
@@ -3095,7 +3102,7 @@ dependencies = [
  "pin-project-lite",
  "quinn",
  "rustls 0.23.12",
- "rustls-native-certs 0.7.2",
+ "rustls-native-certs 0.7.3",
  "rustls-pemfile 2.1.3",
  "rustls-pki-types",
  "serde",
@@ -3175,18 +3182,18 @@ checksum = "583034fd73374156e66797ed8e5b0d5690409c9226b22d87cb7f19821c05d152"
 
 [[package]]
 name = "rustc_version"
-version = "0.4.0"
+version = "0.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bfa0f585226d2e68097d4f95d113b15b83a82e819ab25717ec0590d9584ef366"
+checksum = "cfcb3a22ef46e85b45de6ee7e79d063319ebb6594faafcf1c225ea92ab6e9b92"
 dependencies = [
  "semver",
 ]
 
 [[package]]
 name = "rustix"
-version = "0.38.34"
+version = "0.38.35"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "70dc5ec042f7a43c4a73241207cecc9873a06d45debb38b329f8541d85c2730f"
+checksum = "a85d50532239da68e9addb745ba38ff4612a242c1c7ceea689c4bc7c2f43c36f"
 dependencies = [
  "bitflags 2.6.0",
  "errno",
@@ -3216,7 +3223,7 @@ dependencies = [
  "once_cell",
  "ring",
  "rustls-pki-types",
- "rustls-webpki 0.102.6",
+ "rustls-webpki 0.102.7",
  "subtle",
  "zeroize",
 ]
@@ -3235,9 +3242,9 @@ dependencies = [
 
 [[package]]
 name = "rustls-native-certs"
-version = "0.7.2"
+version = "0.7.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "04182dffc9091a404e0fc069ea5cd60e5b866c3adf881eff99a32d048242dffa"
+checksum = "e5bfb394eeed242e909609f56089eecfe5fda225042e8b171791b9c95f5931e5"
 dependencies = [
  "openssl-probe",
  "rustls-pemfile 2.1.3",
@@ -3283,9 +3290,9 @@ dependencies = [
 
 [[package]]
 name = "rustls-webpki"
-version = "0.102.6"
+version = "0.102.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8e6b52d4fda176fd835fdc55a835d4a89b8499cad995885a21149d5ad62f852e"
+checksum = "84678086bd54edf2b415183ed7a94d0efb049f1b646a33e22a36f3794be6ae56"
 dependencies = [
  "ring",
  "rustls-pki-types",
@@ -3398,29 +3405,29 @@ checksum = "a3f0bf26fd526d2a95683cd0f87bf103b8539e2ca1ef48ce002d67aad59aa0b4"
 
 [[package]]
 name = "serde"
-version = "1.0.208"
+version = "1.0.209"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cff085d2cb684faa248efb494c39b68e522822ac0de72ccf08109abde717cfb2"
+checksum = "99fce0ffe7310761ca6bf9faf5115afbc19688edd00171d81b1bb1b116c63e09"
 dependencies = [
  "serde_derive",
 ]
 
 [[package]]
 name = "serde_derive"
-version = "1.0.208"
+version = "1.0.209"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "24008e81ff7613ed8e5ba0cfaf24e2c2f1e5b8a0495711e44fcd4882fca62bcf"
+checksum = "a5831b979fd7b5439637af1752d535ff49f4860c0f341d1baeb6faf0f4242170"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.75",
+ "syn 2.0.77",
 ]
 
 [[package]]
 name = "serde_json"
-version = "1.0.125"
+version = "1.0.127"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "83c8e735a073ccf5be70aa8066aa984eaf2fa000db6c8d0100ae605b366d31ed"
+checksum = "8043c06d9f82bd7271361ed64f415fe5e12a77fdb52e573e7f06a516dea329ad"
 dependencies = [
  "itoa",
  "memchr",
@@ -3489,24 +3496,23 @@ checksum = "3c5e1a9a646d36c3599cd173a41282daf47c44583ad367b8e6837255952e5c67"
 
 [[package]]
 name = "snafu"
-version = "0.7.5"
+version = "0.8.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e4de37ad025c587a29e8f3f5605c00f70b98715ef90b9061a815b9e59e9042d6"
+checksum = "2b835cb902660db3415a672d862905e791e54d306c6e8189168c7f3d9ae1c79d"
 dependencies = [
- "doc-comment",
  "snafu-derive",
 ]
 
 [[package]]
 name = "snafu-derive"
-version = "0.7.5"
+version = "0.8.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "990079665f075b699031e9c08fd3ab99be5029b96f3b78dc0709e8f77e4efebf"
+checksum = "38d1e02fca405f6280643174a50c942219f0bbf4dbf7d480f1dd864d6f211ae5"
 dependencies = [
- "heck 0.4.1",
+ "heck 0.5.0",
  "proc-macro2",
  "quote",
- "syn 1.0.109",
+ "syn 2.0.77",
 ]
 
 [[package]]
@@ -3549,7 +3555,7 @@ checksum = "01b2e185515564f15375f593fb966b5718bc624ba77fe49fa4616ad619690554"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.75",
+ "syn 2.0.77",
 ]
 
 [[package]]
@@ -3595,7 +3601,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "rustversion",
- "syn 2.0.75",
+ "syn 2.0.77",
 ]
 
 [[package]]
@@ -3608,7 +3614,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "rustversion",
- "syn 2.0.75",
+ "syn 2.0.77",
 ]
 
 [[package]]
@@ -3630,9 +3636,9 @@ dependencies = [
 
 [[package]]
 name = "syn"
-version = "2.0.75"
+version = "2.0.77"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f6af063034fc1935ede7be0122941bafa9bacb949334d090b77ca98b5817c7d9"
+checksum = "9f35bcdf61fd8e7be6caf75f429fdca8beb3ed76584befb503b1569faee373ed"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -3693,7 +3699,7 @@ checksum = "a4558b58466b9ad7ca0f102865eccc95938dca1a74a856f2b57b6629050da261"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.75",
+ "syn 2.0.77",
 ]
 
 [[package]]
@@ -3763,9 +3769,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"
 
 [[package]]
 name = "tokio"
-version = "1.39.3"
+version = "1.40.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9babc99b9923bfa4804bd74722ff02c0381021eafa4db9949217e3be8e84fff5"
+checksum = "e2b070231665d27ad9ec9b8df639893f46727666c6767db40317fbe920a5d998"
 dependencies = [
  "backtrace",
  "bytes",
@@ -3787,7 +3793,7 @@ checksum = "693d596312e88961bc67d7f1f97af8a70227d9f90c31bba5806eec004978d752"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.75",
+ "syn 2.0.77",
 ]
 
 [[package]]
@@ -3870,7 +3876,7 @@ checksum = "34704c8d6ebcbc939824180af020566b01a7c01f80641264eba0999f6c2b6be7"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.75",
+ "syn 2.0.77",
 ]
 
 [[package]]
@@ -3915,7 +3921,7 @@ checksum = "f03ca4cb38206e2bef0700092660bb74d696f808514dae47fa1467cbfe26e96e"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.75",
+ "syn 2.0.77",
 ]
 
 [[package]]
@@ -4064,7 +4070,7 @@ dependencies = [
  "once_cell",
  "proc-macro2",
  "quote",
- "syn 2.0.75",
+ "syn 2.0.77",
  "wasm-bindgen-shared",
 ]
 
@@ -4098,7 +4104,7 @@ checksum = "afc340c74d9005395cf9dd098506f7f44e38f2b4a21c6aaacf9a105ea5e1e836"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.75",
+ "syn 2.0.77",
  "wasm-bindgen-backend",
  "wasm-bindgen-shared",
 ]
@@ -4383,7 +4389,7 @@ checksum = "fa4f8080344d4671fb4e831a13ad1e68092748387dfc4f55e356242fae12ce3e"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.75",
+ "syn 2.0.77",
 ]
 
 [[package]]
@@ -4431,9 +4437,9 @@ dependencies = [
 
 [[package]]
 name = "zstd-sys"
-version = "2.0.12+zstd.1.5.6"
+version = "2.0.13+zstd.1.5.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0a4e40c320c3cb459d9a9ff6de98cff88f4751ee9275d140e2be94a2b74e4c13"
+checksum = "38ff0f21cfee8f97d94cef41359e0c89aa6113028ab0291aa8ca0038995a95aa"
 dependencies = [
  "cc",
  "pkg-config",
diff --git a/datafusion-cli/Cargo.toml b/datafusion-cli/Cargo.toml
index f477bad69a2c..f2f52846ab54 100644
--- a/datafusion-cli/Cargo.toml
+++ b/datafusion-cli/Cargo.toml
@@ -30,9 +30,16 @@ rust-version = "1.76"
 readme = "README.md"
 
 [dependencies]
-arrow = { version = "52.2.0" }
+arrow = { version = "53.0.0" }
 async-trait = "0.1.73"
 aws-config = "1.5.5"
+# begin pin aws-sdk crates otherwise CI MSRV check fails
+# We can't update these libraries yet as it requires Rust 1.78, which is not available until Nov 2024
+# per https://github.com/apache/datafusion?tab=readme-ov-file#rust-version-compatibility-policy
+aws-sdk-sso = "=1.39.0"
+aws-sdk-ssooidc = "=1.40.0"
+aws-sdk-sts = "=1.39.0"
+# end pin aws-sdk crates
 aws-credential-types = "1.2.0"
 clap = { version = "4.5.16", features = ["derive", "cargo"] }
 datafusion = { path = "../datafusion/core", version = "41.0.0", features = [
@@ -49,9 +56,9 @@ dirs = "4.0.0"
 env_logger = "0.9"
 futures = "0.3"
 mimalloc = { version = "0.1", default-features = false }
-object_store = { version = "0.10.1", features = ["aws", "gcp", "http"] }
+object_store = { version = "0.11.0", features = ["aws", "gcp", "http"] }
 parking_lot = { version = "0.12" }
-parquet = { version = "52.2.0", default-features = false }
+parquet = { version = "53.0.0", default-features = false }
 regex = "1.8"
 rustyline = "11.0"
 tokio = { version = "1.24", features = ["macros", "rt", "rt-multi-thread", "sync", "parking_lot", "signal"] }
diff --git a/datafusion-cli/src/functions.rs b/datafusion-cli/src/functions.rs
index a85c43f3576f..99511e969386 100644
--- a/datafusion-cli/src/functions.rs
+++ b/datafusion-cli/src/functions.rs
@@ -32,6 +32,7 @@ use datafusion::physical_plan::memory::MemoryExec;
 use datafusion::physical_plan::ExecutionPlan;
 use datafusion::scalar::ScalarValue;
 use parquet::basic::ConvertedType;
+use parquet::data_type::{ByteArray, FixedLenByteArray};
 use parquet::file::reader::FileReader;
 use parquet::file::serialized_reader::SerializedFileReader;
 use parquet::file::statistics::Statistics;
@@ -250,49 +251,69 @@ impl TableProvider for ParquetMetadataTable {
 fn convert_parquet_statistics(
     value: &Statistics,
     converted_type: ConvertedType,
-) -> (String, String) {
+) -> (Option<String>, Option<String>) {
     match (value, converted_type) {
-        (Statistics::Boolean(val), _) => (val.min().to_string(), val.max().to_string()),
-        (Statistics::Int32(val), _) => (val.min().to_string(), val.max().to_string()),
-        (Statistics::Int64(val), _) => (val.min().to_string(), val.max().to_string()),
-        (Statistics::Int96(val), _) => (val.min().to_string(), val.max().to_string()),
-        (Statistics::Float(val), _) => (val.min().to_string(), val.max().to_string()),
-        (Statistics::Double(val), _) => (val.min().to_string(), val.max().to_string()),
-        (Statistics::ByteArray(val), ConvertedType::UTF8) => {
-            let min_bytes = val.min();
-            let max_bytes = val.max();
-            let min = min_bytes
-                .as_utf8()
-                .map(|v| v.to_string())
-                .unwrap_or_else(|_| min_bytes.to_string());
-
-            let max = max_bytes
-                .as_utf8()
-                .map(|v| v.to_string())
-                .unwrap_or_else(|_| max_bytes.to_string());
-            (min, max)
-        }
-        (Statistics::ByteArray(val), _) => (val.min().to_string(), val.max().to_string()),
-        (Statistics::FixedLenByteArray(val), ConvertedType::UTF8) => {
-            let min_bytes = val.min();
-            let max_bytes = val.max();
-            let min = min_bytes
-                .as_utf8()
-                .map(|v| v.to_string())
-                .unwrap_or_else(|_| min_bytes.to_string());
-
-            let max = max_bytes
-                .as_utf8()
-                .map(|v| v.to_string())
-                .unwrap_or_else(|_| max_bytes.to_string());
-            (min, max)
-        }
-        (Statistics::FixedLenByteArray(val), _) => {
-            (val.min().to_string(), val.max().to_string())
-        }
+        (Statistics::Boolean(val), _) => (
+            val.min_opt().map(|v| v.to_string()),
+            val.max_opt().map(|v| v.to_string()),
+        ),
+        (Statistics::Int32(val), _) => (
+            val.min_opt().map(|v| v.to_string()),
+            val.max_opt().map(|v| v.to_string()),
+        ),
+        (Statistics::Int64(val), _) => (
+            val.min_opt().map(|v| v.to_string()),
+            val.max_opt().map(|v| v.to_string()),
+        ),
+        (Statistics::Int96(val), _) => (
+            val.min_opt().map(|v| v.to_string()),
+            val.max_opt().map(|v| v.to_string()),
+        ),
+        (Statistics::Float(val), _) => (
+            val.min_opt().map(|v| v.to_string()),
+            val.max_opt().map(|v| v.to_string()),
+        ),
+        (Statistics::Double(val), _) => (
+            val.min_opt().map(|v| v.to_string()),
+            val.max_opt().map(|v| v.to_string()),
+        ),
+        (Statistics::ByteArray(val), ConvertedType::UTF8) => (
+            byte_array_to_string(val.min_opt()),
+            byte_array_to_string(val.max_opt()),
+        ),
+        (Statistics::ByteArray(val), _) => (
+            val.min_opt().map(|v| v.to_string()),
+            val.max_opt().map(|v| v.to_string()),
+        ),
+        (Statistics::FixedLenByteArray(val), ConvertedType::UTF8) => (
+            fixed_len_byte_array_to_string(val.min_opt()),
+            fixed_len_byte_array_to_string(val.max_opt()),
+        ),
+        (Statistics::FixedLenByteArray(val), _) => (
+            val.min_opt().map(|v| v.to_string()),
+            val.max_opt().map(|v| v.to_string()),
+        ),
     }
 }
 
+/// Convert to a string if it has utf8 encoding, otherwise print bytes directly
+fn byte_array_to_string(val: Option<&ByteArray>) -> Option<String> {
+    val.map(|v| {
+        v.as_utf8()
+            .map(|s| s.to_string())
+            .unwrap_or_else(|_e| v.to_string())
+    })
+}
+
+/// Convert to a string if it has utf8 encoding, otherwise print bytes directly
+fn fixed_len_byte_array_to_string(val: Option<&FixedLenByteArray>) -> Option<String> {
+    val.map(|v| {
+        v.as_utf8()
+            .map(|s| s.to_string())
+            .unwrap_or_else(|_e| v.to_string())
+    })
+}
+
 pub struct ParquetMetadataFunc {}
 
 impl TableFunctionImpl for ParquetMetadataFunc {
@@ -376,17 +397,13 @@ impl TableFunctionImpl for ParquetMetadataFunc {
                 let converted_type = column.column_descr().converted_type();
 
                 if let Some(s) = column.statistics() {
-                    let (min_val, max_val) = if s.has_min_max_set() {
-                        let (min_val, max_val) =
-                            convert_parquet_statistics(s, converted_type);
-                        (Some(min_val), Some(max_val))
-                    } else {
-                        (None, None)
-                    };
+                    let (min_val, max_val) =
+                        convert_parquet_statistics(s, converted_type);
                     stats_min_arr.push(min_val.clone());
                     stats_max_arr.push(max_val.clone());
-                    stats_null_count_arr.push(Some(s.null_count() as i64));
-                    stats_distinct_count_arr.push(s.distinct_count().map(|c| c as i64));
+                    stats_null_count_arr.push(s.null_count_opt().map(|c| c as i64));
+                    stats_distinct_count_arr
+                        .push(s.distinct_count_opt().map(|c| c as i64));
                     stats_min_value_arr.push(min_val);
                     stats_max_value_arr.push(max_val);
                 } else {
diff --git a/datafusion-examples/Cargo.toml b/datafusion-examples/Cargo.toml
index 57b9930177d4..f430a87e190d 100644
--- a/datafusion-examples/Cargo.toml
+++ b/datafusion-examples/Cargo.toml
@@ -73,13 +73,13 @@ mimalloc = { version = "0.1", default-features = false }
 num_cpus = { workspace = true }
 object_store = { workspace = true, features = ["aws", "http"] }
 prost = { workspace = true }
-prost-derive = { version = "0.13", default-features = false }
+prost-derive = { workspace = true }
 serde = { version = "1.0.136", features = ["derive"] }
 serde_json = { workspace = true }
 tempfile = { workspace = true }
 test-utils = { path = "../test-utils" }
 tokio = { workspace = true, features = ["rt-multi-thread", "parking_lot"] }
-tonic = "0.11"
+tonic = "0.12.1"
 url = { workspace = true }
 uuid = "1.7"
 
diff --git a/datafusion/common/Cargo.toml b/datafusion/common/Cargo.toml
index 79e20ba1215c..1ac27b40c219 100644
--- a/datafusion/common/Cargo.toml
+++ b/datafusion/common/Cargo.toml
@@ -61,7 +61,7 @@ num_cpus = { workspace = true }
 object_store = { workspace = true, optional = true }
 parquet = { workspace = true, optional = true, default-features = true }
 paste = "1.0.15"
-pyo3 = { version = "0.21.0", optional = true }
+pyo3 = { version = "0.22.0", optional = true }
 sqlparser = { workspace = true }
 tokio = { workspace = true }
 
diff --git a/datafusion/common/src/scalar/mod.rs b/datafusion/common/src/scalar/mod.rs
index 5acc2b6f188e..dd0225cfc4db 100644
--- a/datafusion/common/src/scalar/mod.rs
+++ b/datafusion/common/src/scalar/mod.rs
@@ -4331,7 +4331,7 @@ mod tests {
             .strip_backtrace();
         assert_eq!(
             err,
-            "Arrow error: Compute error: Overflow happened on: 2147483647 - -2147483648"
+            "Arrow error: Arithmetic overflow: Overflow happened on: 2147483647 - -2147483648"
         )
     }
 
@@ -4352,7 +4352,7 @@ mod tests {
             .sub_checked(&int_value_2)
             .unwrap_err()
             .strip_backtrace();
-        assert_eq!(err, "Arrow error: Compute error: Overflow happened on: 9223372036854775807 - -9223372036854775808")
+        assert_eq!(err, "Arrow error: Arithmetic overflow: Overflow happened on: 9223372036854775807 - -9223372036854775808")
     }
 
     #[test]
@@ -5868,7 +5868,7 @@ mod tests {
                     let root_err = err.find_root();
                     match  root_err{
                         DataFusionError::ArrowError(
-                            ArrowError::ComputeError(_),
+                            ArrowError::ArithmeticOverflow(_),
                             _,
                         ) => {}
                         _ => return Err(err),
diff --git a/datafusion/core/src/datasource/file_format/parquet.rs b/datafusion/core/src/datasource/file_format/parquet.rs
index 23e765f0f2cd..76e8ad9da559 100644
--- a/datafusion/core/src/datasource/file_format/parquet.rs
+++ b/datafusion/core/src/datasource/file_format/parquet.rs
@@ -2010,7 +2010,7 @@ mod tests {
 
         // test result in int_col
         let int_col_index = page_index.get(4).unwrap();
-        let int_col_offset = offset_index.get(4).unwrap();
+        let int_col_offset = offset_index.get(4).unwrap().page_locations();
 
         // 325 pages in int_col
         assert_eq!(int_col_offset.len(), 325);
diff --git a/datafusion/core/src/datasource/physical_plan/parquet/page_filter.rs b/datafusion/core/src/datasource/physical_plan/parquet/page_filter.rs
index e4d26a460ecd..4e71993b5153 100644
--- a/datafusion/core/src/datasource/physical_plan/parquet/page_filter.rs
+++ b/datafusion/core/src/datasource/physical_plan/parquet/page_filter.rs
@@ -392,13 +392,16 @@ impl<'a> PagesPruningStatistics<'a> {
             trace!("No page offsets for row group {row_group_index}, skipping");
             return None;
         };
-        let Some(page_offsets) = row_group_page_offsets.get(parquet_column_index) else {
+        let Some(offset_index_metadata) =
+            row_group_page_offsets.get(parquet_column_index)
+        else {
             trace!(
                 "No page offsets for column {:?} in row group {row_group_index}, skipping",
                 converter.arrow_field()
             );
             return None;
         };
+        let page_offsets = offset_index_metadata.page_locations();
 
         Some(Self {
             row_group_index,
diff --git a/datafusion/core/src/datasource/physical_plan/parquet/row_group_filter.rs b/datafusion/core/src/datasource/physical_plan/parquet/row_group_filter.rs
index 6a6910748fc8..ccd77d90be57 100644
--- a/datafusion/core/src/datasource/physical_plan/parquet/row_group_filter.rs
+++ b/datafusion/core/src/datasource/physical_plan/parquet/row_group_filter.rs
@@ -487,11 +487,23 @@ mod tests {
         let schema_descr = get_test_schema_descr(vec![field]);
         let rgm1 = get_row_group_meta_data(
             &schema_descr,
-            vec![ParquetStatistics::int32(Some(1), Some(10), None, 0, false)],
+            vec![ParquetStatistics::int32(
+                Some(1),
+                Some(10),
+                None,
+                Some(0),
+                false,
+            )],
         );
         let rgm2 = get_row_group_meta_data(
             &schema_descr,
-            vec![ParquetStatistics::int32(Some(11), Some(20), None, 0, false)],
+            vec![ParquetStatistics::int32(
+                Some(11),
+                Some(20),
+                None,
+                Some(0),
+                false,
+            )],
         );
 
         let metrics = parquet_file_metrics();
@@ -520,11 +532,17 @@ mod tests {
         let schema_descr = get_test_schema_descr(vec![field]);
         let rgm1 = get_row_group_meta_data(
             &schema_descr,
-            vec![ParquetStatistics::int32(None, None, None, 0, false)],
+            vec![ParquetStatistics::int32(None, None, None, Some(0), false)],
         );
         let rgm2 = get_row_group_meta_data(
             &schema_descr,
-            vec![ParquetStatistics::int32(Some(11), Some(20), None, 0, false)],
+            vec![ParquetStatistics::int32(
+                Some(11),
+                Some(20),
+                None,
+                Some(0),
+                false,
+            )],
         );
         let metrics = parquet_file_metrics();
         // missing statistics for first row group mean that the result from the predicate expression
@@ -560,15 +578,15 @@ mod tests {
         let rgm1 = get_row_group_meta_data(
             &schema_descr,
             vec![
-                ParquetStatistics::int32(Some(1), Some(10), None, 0, false),
-                ParquetStatistics::int32(Some(1), Some(10), None, 0, false),
+                ParquetStatistics::int32(Some(1), Some(10), None, Some(0), false),
+                ParquetStatistics::int32(Some(1), Some(10), None, Some(0), false),
             ],
         );
         let rgm2 = get_row_group_meta_data(
             &schema_descr,
             vec![
-                ParquetStatistics::int32(Some(11), Some(20), None, 0, false),
-                ParquetStatistics::int32(Some(11), Some(20), None, 0, false),
+                ParquetStatistics::int32(Some(11), Some(20), None, Some(0), false),
+                ParquetStatistics::int32(Some(11), Some(20), None, Some(0), false),
             ],
         );
 
@@ -633,16 +651,16 @@ mod tests {
         let rgm1 = get_row_group_meta_data(
             &schema_descr,
             vec![
-                ParquetStatistics::int32(Some(-10), Some(-1), None, 0, false), // c2
-                ParquetStatistics::int32(Some(1), Some(10), None, 0, false),
+                ParquetStatistics::int32(Some(-10), Some(-1), None, Some(0), false), // c2
+                ParquetStatistics::int32(Some(1), Some(10), None, Some(0), false),
             ],
         );
         // rg1 has c2 greater than zero, c1 less than zero
         let rgm2 = get_row_group_meta_data(
             &schema_descr,
             vec![
-                ParquetStatistics::int32(Some(1), Some(10), None, 0, false),
-                ParquetStatistics::int32(Some(-10), Some(-1), None, 0, false),
+                ParquetStatistics::int32(Some(1), Some(10), None, Some(0), false),
+                ParquetStatistics::int32(Some(-10), Some(-1), None, Some(0), false),
             ],
         );
 
@@ -669,15 +687,15 @@ mod tests {
         let rgm1 = get_row_group_meta_data(
             &schema_descr,
             vec![
-                ParquetStatistics::int32(Some(1), Some(10), None, 0, false),
-                ParquetStatistics::boolean(Some(false), Some(true), None, 0, false),
+                ParquetStatistics::int32(Some(1), Some(10), None, Some(0), false),
+                ParquetStatistics::boolean(Some(false), Some(true), None, Some(0), false),
             ],
         );
         let rgm2 = get_row_group_meta_data(
             &schema_descr,
             vec![
-                ParquetStatistics::int32(Some(11), Some(20), None, 0, false),
-                ParquetStatistics::boolean(Some(false), Some(true), None, 1, false),
+                ParquetStatistics::int32(Some(11), Some(20), None, Some(0), false),
+                ParquetStatistics::boolean(Some(false), Some(true), None, Some(1), false),
             ],
         );
         vec![rgm1, rgm2]
@@ -775,7 +793,7 @@ mod tests {
                 Some(100),
                 Some(600),
                 None,
-                0,
+                Some(0),
                 false,
             )],
         );
@@ -783,13 +801,25 @@ mod tests {
             &schema_descr,
             // [0.1, 0.2]
             // c1 > 5, this row group will not be included in the results.
-            vec![ParquetStatistics::int32(Some(10), Some(20), None, 0, false)],
+            vec![ParquetStatistics::int32(
+                Some(10),
+                Some(20),
+                None,
+                Some(0),
+                false,
+            )],
         );
         let rgm3 = get_row_group_meta_data(
             &schema_descr,
             // [1, None]
             // c1 > 5, this row group can not be filtered out, so will be included in the results.
-            vec![ParquetStatistics::int32(Some(100), None, None, 0, false)],
+            vec![ParquetStatistics::int32(
+                Some(100),
+                None,
+                None,
+                Some(0),
+                false,
+            )],
         );
         let metrics = parquet_file_metrics();
         let mut row_groups = RowGroupAccessPlanFilter::new(ParquetAccessPlan::new_all(3));
@@ -837,7 +867,7 @@ mod tests {
                 Some(100),
                 Some(600),
                 None,
-                0,
+                Some(0),
                 false,
             )],
         );
@@ -845,30 +875,62 @@ mod tests {
             &schema_descr,
             // [10, 20]
             // c1 > 5, this row group will be included in the results.
-            vec![ParquetStatistics::int32(Some(10), Some(20), None, 0, false)],
+            vec![ParquetStatistics::int32(
+                Some(10),
+                Some(20),
+                None,
+                Some(0),
+                false,
+            )],
         );
         let rgm3 = get_row_group_meta_data(
             &schema_descr,
             // [0, 2]
             // c1 > 5, this row group will not be included in the results.
-            vec![ParquetStatistics::int32(Some(0), Some(2), None, 0, false)],
+            vec![ParquetStatistics::int32(
+                Some(0),
+                Some(2),
+                None,
+                Some(0),
+                false,
+            )],
         );
         let rgm4 = get_row_group_meta_data(
             &schema_descr,
             // [None, 2]
-            // c1 > 5, this row group can not be filtered out, so will be included in the results.
-            vec![ParquetStatistics::int32(None, Some(2), None, 0, false)],
+            // c1 > 5, this row group will also not be included in the results
+            // (the min value is unknown, but the max value is 2, so no values can be greater than 5)
+            vec![ParquetStatistics::int32(
+                None,
+                Some(2),
+                None,
+                Some(0),
+                false,
+            )],
+        );
+        let rgm5 = get_row_group_meta_data(
+            &schema_descr,
+            // [2, None]
+            // c1 > 5, this row group must be included
+            // (the min value is 2, but the max value is unknown, so it may have values greater than 5)
+            vec![ParquetStatistics::int32(
+                Some(2),
+                None,
+                None,
+                Some(0),
+                false,
+            )],
         );
         let metrics = parquet_file_metrics();
-        let mut row_groups = RowGroupAccessPlanFilter::new(ParquetAccessPlan::new_all(4));
+        let mut row_groups = RowGroupAccessPlanFilter::new(ParquetAccessPlan::new_all(5));
         row_groups.prune_by_statistics(
             &schema,
             &schema_descr,
-            &[rgm1, rgm2, rgm3, rgm4],
+            &[rgm1, rgm2, rgm3, rgm4, rgm5],
             &pruning_predicate,
             &metrics,
         );
-        assert_pruned(row_groups, ExpectedPruning::Some(vec![0, 1, 3]));
+        assert_pruned(row_groups, ExpectedPruning::Some(vec![0, 1, 4]));
     }
     #[test]
     fn row_group_pruning_predicate_decimal_type3() {
@@ -896,19 +958,25 @@ mod tests {
                 Some(600),
                 Some(800),
                 None,
-                0,
+                Some(0),
                 false,
             )],
         );
         let rgm2 = get_row_group_meta_data(
             &schema_descr,
             // [0.1, 0.2]
-            vec![ParquetStatistics::int64(Some(10), Some(20), None, 0, false)],
+            vec![ParquetStatistics::int64(
+                Some(10),
+                Some(20),
+                None,
+                Some(0),
+                false,
+            )],
         );
         let rgm3 = get_row_group_meta_data(
             &schema_descr,
             // [0.1, 0.2]
-            vec![ParquetStatistics::int64(None, None, None, 0, false)],
+            vec![ParquetStatistics::int64(None, None, None, Some(0), false)],
         );
         let metrics = parquet_file_metrics();
         let mut row_groups = RowGroupAccessPlanFilter::new(ParquetAccessPlan::new_all(3));
@@ -957,7 +1025,7 @@ mod tests {
                     8000i128.to_be_bytes().to_vec(),
                 ))),
                 None,
-                0,
+                Some(0),
                 false,
             )],
         );
@@ -973,7 +1041,7 @@ mod tests {
                     20000i128.to_be_bytes().to_vec(),
                 ))),
                 None,
-                0,
+                Some(0),
                 false,
             )],
         );
@@ -981,7 +1049,11 @@ mod tests {
         let rgm3 = get_row_group_meta_data(
             &schema_descr,
             vec![ParquetStatistics::fixed_len_byte_array(
-                None, None, None, 0, false,
+                None,
+                None,
+                None,
+                Some(0),
+                false,
             )],
         );
         let metrics = parquet_file_metrics();
@@ -1027,7 +1099,7 @@ mod tests {
                 // 80.00
                 Some(ByteArray::from(8000i128.to_be_bytes().to_vec())),
                 None,
-                0,
+                Some(0),
                 false,
             )],
         );
@@ -1039,13 +1111,19 @@ mod tests {
                 // 200.00
                 Some(ByteArray::from(20000i128.to_be_bytes().to_vec())),
                 None,
-                0,
+                Some(0),
                 false,
             )],
         );
         let rgm3 = get_row_group_meta_data(
             &schema_descr,
-            vec![ParquetStatistics::byte_array(None, None, None, 0, false)],
+            vec![ParquetStatistics::byte_array(
+                None,
+                None,
+                None,
+                Some(0),
+                false,
+            )],
         );
         let metrics = parquet_file_metrics();
         let mut row_groups = RowGroupAccessPlanFilter::new(ParquetAccessPlan::new_all(3));
diff --git a/datafusion/functions/src/regex/regexpreplace.rs b/datafusion/functions/src/regex/regexpreplace.rs
index d28c6cd36d65..0b0f7287e1ec 100644
--- a/datafusion/functions/src/regex/regexpreplace.rs
+++ b/datafusion/functions/src/regex/regexpreplace.rs
@@ -401,8 +401,7 @@ fn _regexp_replace_static_pattern_replace<T: OffsetSizeTrait>(
         DataType::Utf8View => {
             let string_view_array = as_string_view_array(&args[0])?;
 
-            let mut builder = StringViewBuilder::with_capacity(string_view_array.len())
-                .with_block_size(1024 * 1024 * 2);
+            let mut builder = StringViewBuilder::with_capacity(string_view_array.len());
 
             for val in string_view_array.iter() {
                 if let Some(val) = val {
diff --git a/datafusion/physical-expr-common/src/binary_view_map.rs b/datafusion/physical-expr-common/src/binary_view_map.rs
index 18bc6801aa60..bdcf7bbacc69 100644
--- a/datafusion/physical-expr-common/src/binary_view_map.rs
+++ b/datafusion/physical-expr-common/src/binary_view_map.rs
@@ -149,7 +149,7 @@ where
             output_type,
             map: hashbrown::raw::RawTable::with_capacity(INITIAL_MAP_CAPACITY),
             map_size: 0,
-            builder: GenericByteViewBuilder::new().with_block_size(2 * 1024 * 1024),
+            builder: GenericByteViewBuilder::new(),
             random_state: RandomState::new(),
             hashes_buffer: vec![],
             null: None,
diff --git a/datafusion/physical-plan/src/coalesce/mod.rs b/datafusion/physical-plan/src/coalesce/mod.rs
index ce5a1e53abfd..46875fae94fc 100644
--- a/datafusion/physical-plan/src/coalesce/mod.rs
+++ b/datafusion/physical-plan/src/coalesce/mod.rs
@@ -248,7 +248,7 @@ fn gc_string_view_batch(batch: &RecordBatch) -> RecordBatch {
                 // See https://github.com/apache/arrow-rs/issues/6094 for more details.
                 let mut builder = StringViewBuilder::with_capacity(s.len());
                 if ideal_buffer_size > 0 {
-                    builder = builder.with_block_size(ideal_buffer_size as u32);
+                    builder = builder.with_fixed_block_size(ideal_buffer_size as u32);
                 }
 
                 for v in s.iter() {
@@ -580,7 +580,8 @@ mod tests {
     impl StringViewTest {
         /// Create a `StringViewArray` with the parameters specified in this struct
         fn build(self) -> StringViewArray {
-            let mut builder = StringViewBuilder::with_capacity(100).with_block_size(8192);
+            let mut builder =
+                StringViewBuilder::with_capacity(100).with_fixed_block_size(8192);
             loop {
                 for &v in self.strings.iter() {
                     builder.append_option(v);
diff --git a/datafusion/proto-common/Cargo.toml b/datafusion/proto-common/Cargo.toml
index 33a7ecd2daab..7ba503171520 100644
--- a/datafusion/proto-common/Cargo.toml
+++ b/datafusion/proto-common/Cargo.toml
@@ -44,7 +44,7 @@ arrow = { workspace = true }
 chrono = { workspace = true }
 datafusion-common = { workspace = true }
 object_store = { workspace = true }
-pbjson = { version = "0.6.0", optional = true }
+pbjson = { workspace = true, optional = true }
 prost = { workspace = true }
 serde = { version = "1.0", optional = true }
 serde_json = { workspace = true, optional = true }
diff --git a/datafusion/proto-common/gen/Cargo.toml b/datafusion/proto-common/gen/Cargo.toml
index 54ec0e44694b..cca49dba7ed3 100644
--- a/datafusion/proto-common/gen/Cargo.toml
+++ b/datafusion/proto-common/gen/Cargo.toml
@@ -34,5 +34,5 @@ workspace = true
 
 [dependencies]
 # Pin these dependencies so that the generated output is deterministic
-pbjson-build = "=0.6.2"
-prost-build = "=0.12.6"
+pbjson-build = "=0.7.0"
+prost-build = "=0.13.1"
diff --git a/datafusion/proto/Cargo.toml b/datafusion/proto/Cargo.toml
index 2804ed019b61..32678246c005 100644
--- a/datafusion/proto/Cargo.toml
+++ b/datafusion/proto/Cargo.toml
@@ -52,7 +52,7 @@ datafusion-common = { workspace = true, default-features = true }
 datafusion-expr = { workspace = true }
 datafusion-proto-common = { workspace = true }
 object_store = { workspace = true }
-pbjson = { version = "0.6.0", optional = true }
+pbjson = { workspace = true, optional = true }
 prost = { workspace = true }
 serde = { version = "1.0", optional = true }
 serde_json = { workspace = true, optional = true }
diff --git a/datafusion/proto/gen/Cargo.toml b/datafusion/proto/gen/Cargo.toml
index 401c51c94563..1dc5f7e0dddc 100644
--- a/datafusion/proto/gen/Cargo.toml
+++ b/datafusion/proto/gen/Cargo.toml
@@ -34,5 +34,5 @@ workspace = true
 
 [dependencies]
 # Pin these dependencies so that the generated output is deterministic
-pbjson-build = "=0.6.2"
-prost-build = "=0.12.6"
+pbjson-build = "=0.7.0"
+prost-build = "=0.13.1"
diff --git a/datafusion/proto/src/generated/datafusion_proto_common.rs b/datafusion/proto/src/generated/datafusion_proto_common.rs
index ebc05718a458..3d7b1007b04e 100644
--- a/datafusion/proto/src/generated/datafusion_proto_common.rs
+++ b/datafusion/proto/src/generated/datafusion_proto_common.rs
@@ -45,10 +45,10 @@ pub struct ParquetFormat {
     pub options: ::core::option::Option<TableParquetOptions>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct AvroFormat {}
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct NdJsonFormat {
     #[prost(message, optional, tag = "1")]
     pub options: ::core::option::Option<JsonOptions>,
@@ -89,10 +89,10 @@ pub struct Constraints {
     pub constraints: ::prost::alloc::vec::Vec<Constraint>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct AvroOptions {}
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct ArrowOptions {}
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
@@ -137,7 +137,7 @@ pub struct Timestamp {
     pub timezone: ::prost::alloc::string::String,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct Decimal {
     #[prost(uint32, tag = "3")]
     pub precision: u32,
@@ -145,7 +145,7 @@ pub struct Decimal {
     pub scale: i32,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct Decimal256Type {
     #[prost(uint32, tag = "3")]
     pub precision: u32,
@@ -223,7 +223,7 @@ pub mod scalar_nested_value {
     }
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct ScalarTime32Value {
     #[prost(oneof = "scalar_time32_value::Value", tags = "1, 2")]
     pub value: ::core::option::Option<scalar_time32_value::Value>,
@@ -231,7 +231,7 @@ pub struct ScalarTime32Value {
 /// Nested message and enum types in `ScalarTime32Value`.
 pub mod scalar_time32_value {
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum Value {
         #[prost(int32, tag = "1")]
         Time32SecondValue(i32),
@@ -240,7 +240,7 @@ pub mod scalar_time32_value {
     }
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct ScalarTime64Value {
     #[prost(oneof = "scalar_time64_value::Value", tags = "1, 2")]
     pub value: ::core::option::Option<scalar_time64_value::Value>,
@@ -248,7 +248,7 @@ pub struct ScalarTime64Value {
 /// Nested message and enum types in `ScalarTime64Value`.
 pub mod scalar_time64_value {
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum Value {
         #[prost(int64, tag = "1")]
         Time64MicrosecondValue(i64),
@@ -267,7 +267,7 @@ pub struct ScalarTimestampValue {
 /// Nested message and enum types in `ScalarTimestampValue`.
 pub mod scalar_timestamp_value {
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum Value {
         #[prost(int64, tag = "1")]
         TimeMicrosecondValue(i64),
@@ -288,7 +288,7 @@ pub struct ScalarDictionaryValue {
     pub value: ::core::option::Option<::prost::alloc::boxed::Box<ScalarValue>>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct IntervalDayTimeValue {
     #[prost(int32, tag = "1")]
     pub days: i32,
@@ -296,7 +296,7 @@ pub struct IntervalDayTimeValue {
     pub milliseconds: i32,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct IntervalMonthDayNanoValue {
     #[prost(int32, tag = "1")]
     pub months: i32,
@@ -558,10 +558,10 @@ pub mod arrow_type {
 ///    }
 /// }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct EmptyMessage {}
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct JsonWriterOptions {
     #[prost(enumeration = "CompressionTypeVariant", tag = "1")]
     pub compression: i32,
@@ -655,7 +655,7 @@ pub struct CsvOptions {
 }
 /// Options controlling CSV format
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct JsonOptions {
     /// Compression type
     #[prost(enumeration = "CompressionTypeVariant", tag = "1")]
@@ -720,7 +720,7 @@ pub struct ParquetColumnOptions {
 /// Nested message and enum types in `ParquetColumnOptions`.
 pub mod parquet_column_options {
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum BloomFilterEnabledOpt {
         #[prost(bool, tag = "1")]
         BloomFilterEnabled(bool),
@@ -732,7 +732,7 @@ pub mod parquet_column_options {
         Encoding(::prost::alloc::string::String),
     }
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum DictionaryEnabledOpt {
         #[prost(bool, tag = "3")]
         DictionaryEnabled(bool),
@@ -750,19 +750,19 @@ pub mod parquet_column_options {
         StatisticsEnabled(::prost::alloc::string::String),
     }
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum BloomFilterFppOpt {
         #[prost(double, tag = "6")]
         BloomFilterFpp(f64),
     }
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum BloomFilterNdvOpt {
         #[prost(uint64, tag = "7")]
         BloomFilterNdv(u64),
     }
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum MaxStatisticsSizeOpt {
         #[prost(uint32, tag = "8")]
         MaxStatisticsSize(u32),
@@ -857,7 +857,7 @@ pub struct ParquetOptions {
 /// Nested message and enum types in `ParquetOptions`.
 pub mod parquet_options {
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum MetadataSizeHintOpt {
         #[prost(uint64, tag = "4")]
         MetadataSizeHint(u64),
@@ -869,7 +869,7 @@ pub mod parquet_options {
         Compression(::prost::alloc::string::String),
     }
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum DictionaryEnabledOpt {
         #[prost(bool, tag = "11")]
         DictionaryEnabled(bool),
@@ -881,13 +881,13 @@ pub mod parquet_options {
         StatisticsEnabled(::prost::alloc::string::String),
     }
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum MaxStatisticsSizeOpt {
         #[prost(uint64, tag = "14")]
         MaxStatisticsSize(u64),
     }
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum ColumnIndexTruncateLengthOpt {
         #[prost(uint64, tag = "17")]
         ColumnIndexTruncateLength(u64),
@@ -899,13 +899,13 @@ pub mod parquet_options {
         Encoding(::prost::alloc::string::String),
     }
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum BloomFilterFppOpt {
         #[prost(double, tag = "21")]
         BloomFilterFpp(f64),
     }
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum BloomFilterNdvOpt {
         #[prost(uint64, tag = "22")]
         BloomFilterNdv(u64),
diff --git a/datafusion/proto/src/generated/pbjson.rs b/datafusion/proto/src/generated/pbjson.rs
index cff58d3ddc4a..1c433c2572c4 100644
--- a/datafusion/proto/src/generated/pbjson.rs
+++ b/datafusion/proto/src/generated/pbjson.rs
@@ -12,6 +12,7 @@ impl serde::Serialize for AggLimit {
         let mut struct_ser = serializer.serialize_struct("datafusion.AggLimit", len)?;
         if self.limit != 0 {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("limit", ToString::to_string(&self.limit).as_str())?;
         }
         struct_ser.end()
@@ -613,6 +614,7 @@ impl serde::Serialize for AggregateUdfExprNode {
         }
         if let Some(v) = self.fun_definition.as_ref() {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("funDefinition", pbjson::private::base64::encode(&v).as_str())?;
         }
         struct_ser.end()
@@ -2348,6 +2350,7 @@ impl serde::Serialize for CopyToNode {
         }
         if !self.file_type.is_empty() {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("fileType", pbjson::private::base64::encode(&self.file_type).as_str())?;
         }
         if !self.partition_by.is_empty() {
@@ -3953,6 +3956,7 @@ impl serde::Serialize for CustomTableScanNode {
         }
         if !self.custom_table_data.is_empty() {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("customTableData", pbjson::private::base64::encode(&self.custom_table_data).as_str())?;
         }
         struct_ser.end()
@@ -5037,10 +5041,12 @@ impl serde::Serialize for FileRange {
         let mut struct_ser = serializer.serialize_struct("datafusion.FileRange", len)?;
         if self.start != 0 {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("start", ToString::to_string(&self.start).as_str())?;
         }
         if self.end != 0 {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("end", ToString::to_string(&self.end).as_str())?;
         }
         struct_ser.end()
@@ -5922,6 +5928,7 @@ impl serde::Serialize for GlobalLimitExecNode {
         }
         if self.fetch != 0 {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("fetch", ToString::to_string(&self.fetch).as_str())?;
         }
         struct_ser.end()
@@ -6357,6 +6364,7 @@ impl serde::Serialize for HashRepartition {
         }
         if self.partition_count != 0 {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("partitionCount", ToString::to_string(&self.partition_count).as_str())?;
         }
         struct_ser.end()
@@ -8409,10 +8417,12 @@ impl serde::Serialize for LimitNode {
         }
         if self.skip != 0 {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("skip", ToString::to_string(&self.skip).as_str())?;
         }
         if self.fetch != 0 {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("fetch", ToString::to_string(&self.fetch).as_str())?;
         }
         struct_ser.end()
@@ -9860,6 +9870,7 @@ impl serde::Serialize for LogicalExtensionNode {
         let mut struct_ser = serializer.serialize_struct("datafusion.LogicalExtensionNode", len)?;
         if !self.node.is_empty() {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("node", pbjson::private::base64::encode(&self.node).as_str())?;
         }
         if !self.inputs.is_empty() {
@@ -11982,14 +11993,17 @@ impl serde::Serialize for PartitionStats {
         let mut struct_ser = serializer.serialize_struct("datafusion.PartitionStats", len)?;
         if self.num_rows != 0 {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("numRows", ToString::to_string(&self.num_rows).as_str())?;
         }
         if self.num_batches != 0 {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("numBatches", ToString::to_string(&self.num_batches).as_str())?;
         }
         if self.num_bytes != 0 {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("numBytes", ToString::to_string(&self.num_bytes).as_str())?;
         }
         if !self.column_stats.is_empty() {
@@ -12146,10 +12160,12 @@ impl serde::Serialize for PartitionedFile {
         }
         if self.size != 0 {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("size", ToString::to_string(&self.size).as_str())?;
         }
         if self.last_modified_ns != 0 {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("lastModifiedNs", ToString::to_string(&self.last_modified_ns).as_str())?;
         }
         if !self.partition_values.is_empty() {
@@ -12314,6 +12330,7 @@ impl serde::Serialize for Partitioning {
             match v {
                 partitioning::PartitionMethod::RoundRobin(v) => {
                     #[allow(clippy::needless_borrow)]
+                    #[allow(clippy::needless_borrows_for_generic_args)]
                     struct_ser.serialize_field("roundRobin", ToString::to_string(&v).as_str())?;
                 }
                 partitioning::PartitionMethod::Hash(v) => {
@@ -12321,6 +12338,7 @@ impl serde::Serialize for Partitioning {
                 }
                 partitioning::PartitionMethod::Unknown(v) => {
                     #[allow(clippy::needless_borrow)]
+                    #[allow(clippy::needless_borrows_for_generic_args)]
                     struct_ser.serialize_field("unknown", ToString::to_string(&v).as_str())?;
                 }
             }
@@ -12462,6 +12480,7 @@ impl serde::Serialize for PhysicalAggregateExprNode {
         }
         if let Some(v) = self.fun_definition.as_ref() {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("funDefinition", pbjson::private::base64::encode(&v).as_str())?;
         }
         if let Some(v) = self.aggregate_function.as_ref() {
@@ -13644,6 +13663,7 @@ impl serde::Serialize for PhysicalExtensionExprNode {
         let mut struct_ser = serializer.serialize_struct("datafusion.PhysicalExtensionExprNode", len)?;
         if !self.expr.is_empty() {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("expr", pbjson::private::base64::encode(&self.expr).as_str())?;
         }
         if !self.inputs.is_empty() {
@@ -13755,6 +13775,7 @@ impl serde::Serialize for PhysicalExtensionNode {
         let mut struct_ser = serializer.serialize_struct("datafusion.PhysicalExtensionNode", len)?;
         if !self.node.is_empty() {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("node", pbjson::private::base64::encode(&self.node).as_str())?;
         }
         if !self.inputs.is_empty() {
@@ -13869,6 +13890,7 @@ impl serde::Serialize for PhysicalHashRepartition {
         }
         if self.partition_count != 0 {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("partitionCount", ToString::to_string(&self.partition_count).as_str())?;
         }
         struct_ser.end()
@@ -15085,6 +15107,7 @@ impl serde::Serialize for PhysicalScalarUdfNode {
         }
         if let Some(v) = self.fun_definition.as_ref() {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("funDefinition", pbjson::private::base64::encode(&v).as_str())?;
         }
         if let Some(v) = self.return_type.as_ref() {
@@ -15687,6 +15710,7 @@ impl serde::Serialize for PhysicalWindowExprNode {
         }
         if let Some(v) = self.fun_definition.as_ref() {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("funDefinition", pbjson::private::base64::encode(&v).as_str())?;
         }
         if let Some(v) = self.window_function.as_ref() {
@@ -16901,6 +16925,7 @@ impl serde::Serialize for RepartitionNode {
             match v {
                 repartition_node::PartitionMethod::RoundRobin(v) => {
                     #[allow(clippy::needless_borrow)]
+                    #[allow(clippy::needless_borrows_for_generic_args)]
                     struct_ser.serialize_field("roundRobin", ToString::to_string(&v).as_str())?;
                 }
                 repartition_node::PartitionMethod::Hash(v) => {
@@ -17123,6 +17148,7 @@ impl serde::Serialize for ScalarUdfExprNode {
         }
         if let Some(v) = self.fun_definition.as_ref() {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("funDefinition", pbjson::private::base64::encode(&v).as_str())?;
         }
         struct_ser.end()
@@ -17691,6 +17717,7 @@ impl serde::Serialize for SortExecNode {
         }
         if self.fetch != 0 {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("fetch", ToString::to_string(&self.fetch).as_str())?;
         }
         if self.preserve_partitioning {
@@ -18052,6 +18079,7 @@ impl serde::Serialize for SortNode {
         }
         if self.fetch != 0 {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("fetch", ToString::to_string(&self.fetch).as_str())?;
         }
         struct_ser.end()
@@ -18180,6 +18208,7 @@ impl serde::Serialize for SortPreservingMergeExecNode {
         }
         if self.fetch != 0 {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("fetch", ToString::to_string(&self.fetch).as_str())?;
         }
         struct_ser.end()
@@ -19625,6 +19654,7 @@ impl serde::Serialize for ValuesNode {
         let mut struct_ser = serializer.serialize_struct("datafusion.ValuesNode", len)?;
         if self.n_cols != 0 {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("nCols", ToString::to_string(&self.n_cols).as_str())?;
         }
         if !self.values_list.is_empty() {
@@ -20299,6 +20329,7 @@ impl serde::Serialize for WindowExprNode {
         }
         if let Some(v) = self.fun_definition.as_ref() {
             #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
             struct_ser.serialize_field("funDefinition", pbjson::private::base64::encode(&v).as_str())?;
         }
         if let Some(v) = self.window_function.as_ref() {
diff --git a/datafusion/proto/src/generated/prost.rs b/datafusion/proto/src/generated/prost.rs
index 2ce8004e3248..dbcf7672a48c 100644
--- a/datafusion/proto/src/generated/prost.rs
+++ b/datafusion/proto/src/generated/prost.rs
@@ -239,7 +239,7 @@ pub struct HashRepartition {
     pub partition_count: u64,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct EmptyRelationNode {
     #[prost(bool, tag = "1")]
     pub produce_one_row: bool,
@@ -443,7 +443,7 @@ pub struct UnnestNode {
     pub options: ::core::option::Option<UnnestOptions>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct UnnestOptions {
     #[prost(bool, tag = "1")]
     pub preserve_nulls: bool,
@@ -912,7 +912,7 @@ pub struct WindowFrameBound {
     pub bound_value: ::core::option::Option<super::datafusion_common::ScalarValue>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct FixedSizeBinary {
     #[prost(int32, tag = "1")]
     pub length: i32,
@@ -1474,7 +1474,7 @@ pub struct FileGroup {
     pub files: ::prost::alloc::vec::Vec<PartitionedFile>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct ScanLimit {
     /// wrap into a message to make it optional
     #[prost(uint32, tag = "1")]
@@ -1721,7 +1721,7 @@ pub struct MaybePhysicalSortExprs {
     pub sort_expr: ::prost::alloc::vec::Vec<PhysicalSortExprNode>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct AggLimit {
     /// wrap into a message to make it optional
     #[prost(uint64, tag = "1")]
@@ -1877,7 +1877,7 @@ pub struct JoinFilter {
     pub schema: ::core::option::Option<super::datafusion_common::Schema>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct ColumnIndex {
     #[prost(uint32, tag = "1")]
     pub index: u32,
@@ -1903,7 +1903,7 @@ pub struct PartitionedFile {
     pub statistics: ::core::option::Option<super::datafusion_common::Statistics>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct FileRange {
     #[prost(int64, tag = "1")]
     pub start: i64,
diff --git a/datafusion/sql/src/unparser/expr.rs b/datafusion/sql/src/unparser/expr.rs
index 8ff4773e8e94..efe95dd734eb 100644
--- a/datafusion/sql/src/unparser/expr.rs
+++ b/datafusion/sql/src/unparser/expr.rs
@@ -2035,49 +2035,49 @@ mod tests {
                     "1 YEAR 1 MONTH 1 DAY 3 HOUR 10 MINUTE 20 SECOND",
                 ),
                 IntervalStyle::PostgresVerbose,
-                r#"INTERVAL '0 YEARS 13 MONS 1 DAYS 3 HOURS 10 MINS 20.000000000 SECS'"#,
+                r#"INTERVAL '13 MONS 1 DAYS 3 HOURS 10 MINS 20.000000000 SECS'"#,
             ),
             (
                 interval_month_day_nano_lit("1.5 MONTH"),
                 IntervalStyle::PostgresVerbose,
-                r#"INTERVAL '0 YEARS 1 MONS 15 DAYS 0 HOURS 0 MINS 0.000000000 SECS'"#,
+                r#"INTERVAL '1 MONS 15 DAYS'"#,
             ),
             (
                 interval_month_day_nano_lit("-3 MONTH"),
                 IntervalStyle::PostgresVerbose,
-                r#"INTERVAL '0 YEARS -3 MONS 0 DAYS 0 HOURS 0 MINS 0.000000000 SECS'"#,
+                r#"INTERVAL '-3 MONS'"#,
             ),
             (
                 interval_month_day_nano_lit("1 MONTH")
                     .add(interval_month_day_nano_lit("1 DAY")),
                 IntervalStyle::PostgresVerbose,
-                r#"(INTERVAL '0 YEARS 1 MONS 0 DAYS 0 HOURS 0 MINS 0.000000000 SECS' + INTERVAL '0 YEARS 0 MONS 1 DAYS 0 HOURS 0 MINS 0.000000000 SECS')"#,
+                r#"(INTERVAL '1 MONS' + INTERVAL '1 DAYS')"#,
             ),
             (
                 interval_month_day_nano_lit("1 MONTH")
                     .sub(interval_month_day_nano_lit("1 DAY")),
                 IntervalStyle::PostgresVerbose,
-                r#"(INTERVAL '0 YEARS 1 MONS 0 DAYS 0 HOURS 0 MINS 0.000000000 SECS' - INTERVAL '0 YEARS 0 MONS 1 DAYS 0 HOURS 0 MINS 0.000000000 SECS')"#,
+                r#"(INTERVAL '1 MONS' - INTERVAL '1 DAYS')"#,
             ),
             (
                 interval_datetime_lit("10 DAY 1 HOUR 10 MINUTE 20 SECOND"),
                 IntervalStyle::PostgresVerbose,
-                r#"INTERVAL '0 YEARS 0 MONS 10 DAYS 1 HOURS 10 MINS 20.000 SECS'"#,
+                r#"INTERVAL '10 DAYS 1 HOURS 10 MINS 20.000 SECS'"#,
             ),
             (
                 interval_datetime_lit("10 DAY 1.5 HOUR 10 MINUTE 20 SECOND"),
                 IntervalStyle::PostgresVerbose,
-                r#"INTERVAL '0 YEARS 0 MONS 10 DAYS 1 HOURS 40 MINS 20.000 SECS'"#,
+                r#"INTERVAL '10 DAYS 1 HOURS 40 MINS 20.000 SECS'"#,
             ),
             (
                 interval_year_month_lit("1 YEAR 1 MONTH"),
                 IntervalStyle::PostgresVerbose,
-                r#"INTERVAL '1 YEARS 1 MONS 0 DAYS 0 HOURS 0 MINS 0.00 SECS'"#,
+                r#"INTERVAL '1 YEARS 1 MONS'"#,
             ),
             (
                 interval_year_month_lit("1.5 YEAR 1 MONTH"),
                 IntervalStyle::PostgresVerbose,
-                r#"INTERVAL '1 YEARS 7 MONS 0 DAYS 0 HOURS 0 MINS 0.00 SECS'"#,
+                r#"INTERVAL '1 YEARS 7 MONS'"#,
             ),
             (
                 interval_year_month_lit("1 YEAR 1 MONTH"),
diff --git a/datafusion/sql/tests/cases/plan_to_sql.rs b/datafusion/sql/tests/cases/plan_to_sql.rs
index cdc7bef06afd..d4e189f5f66a 100644
--- a/datafusion/sql/tests/cases/plan_to_sql.rs
+++ b/datafusion/sql/tests/cases/plan_to_sql.rs
@@ -611,7 +611,7 @@ fn sql_round_trip(query: &str, expect: &str) {
 fn test_interval_lhs_eq() {
     sql_round_trip(
         "select interval '2 seconds' = interval '2 seconds'",
-        "SELECT (INTERVAL '0 YEARS 0 MONS 0 DAYS 0 HOURS 0 MINS 2.000000000 SECS' = INTERVAL '0 YEARS 0 MONS 0 DAYS 0 HOURS 0 MINS 2.000000000 SECS')",
+        "SELECT (INTERVAL '2.000000000 SECS' = INTERVAL '2.000000000 SECS')",
     );
 }
 
@@ -619,6 +619,6 @@ fn test_interval_lhs_eq() {
 fn test_interval_lhs_lt() {
     sql_round_trip(
         "select interval '2 seconds' < interval '2 seconds'",
-        "SELECT (INTERVAL '0 YEARS 0 MONS 0 DAYS 0 HOURS 0 MINS 2.000000000 SECS' < INTERVAL '0 YEARS 0 MONS 0 DAYS 0 HOURS 0 MINS 2.000000000 SECS')",
+        "SELECT (INTERVAL '2.000000000 SECS' < INTERVAL '2.000000000 SECS')",
     );
 }
diff --git a/datafusion/sqllogictest/test_files/aggregate.slt b/datafusion/sqllogictest/test_files/aggregate.slt
index 45cb4d4615d7..afb64b6a211f 100644
--- a/datafusion/sqllogictest/test_files/aggregate.slt
+++ b/datafusion/sqllogictest/test_files/aggregate.slt
@@ -1916,7 +1916,7 @@ from values
   (interval '2 month 15 days'),
   (interval '-2 month')
 ----
-Interval(MonthDayNano) 0 years -2 mons 0 days 0 hours 0 mins 0.000000000 secs 0 years 2 mons 15 days 0 hours 0 mins 0.000000000 secs
+Interval(MonthDayNano) -2 mons 2 mons 15 days
 
 # aggregate Interval(DayTime) min/max
 query T??
@@ -1927,7 +1927,7 @@ from values
   (arrow_cast('-3 minutes', 'Interval(DayTime)')),
   (arrow_cast('30 minutes', 'Interval(DayTime)'));
 ----
-Interval(DayTime) 0 years 0 mons 0 days 0 hours -3 mins 0.000 secs 0 years 0 mons 0 days 1 hours 0 mins 0.000 secs
+Interval(DayTime) -3 mins 1 hours
 
 # aggregate Interval(YearMonth) min/max
 query T??
@@ -1938,7 +1938,7 @@ from values
   (arrow_cast('13 months', 'Interval(YearMonth)')),
   (arrow_cast('1 year', 'Interval(YearMonth)'));
 ----
-Interval(YearMonth) -1 years 0 mons 0 days 0 hours 0 mins 0.00 secs 1 years 1 mons 0 days 0 hours 0 mins 0.00 secs
+Interval(YearMonth) -1 years 0 mons 1 years 1 mons
 
 # aggregate
 query II
diff --git a/datafusion/sqllogictest/test_files/arrow_typeof.slt b/datafusion/sqllogictest/test_files/arrow_typeof.slt
index bae6dc33c8cf..77b10b41ccb3 100644
--- a/datafusion/sqllogictest/test_files/arrow_typeof.slt
+++ b/datafusion/sqllogictest/test_files/arrow_typeof.slt
@@ -290,22 +290,22 @@ query ?
 ---
 select arrow_cast(interval '30 minutes', 'Interval(MonthDayNano)');
 ----
-0 years 0 mons 0 days 0 hours 30 mins 0.000000000 secs
+30 mins
 
 query ?
 select arrow_cast('30 minutes', 'Interval(DayTime)');
 ----
-0 years 0 mons 0 days 0 hours 30 mins 0.000 secs
+30 mins
 
 query ?
 select arrow_cast('1 year 5 months', 'Interval(YearMonth)');
 ----
-1 years 5 mons 0 days 0 hours 0 mins 0.00 secs
+1 years 5 mons
 
 query ?
 select arrow_cast('30 minutes', 'Interval(MonthDayNano)');
 ----
-0 years 0 mons 0 days 0 hours 30 mins 0.000000000 secs
+30 mins
 
 
 ## Duration
@@ -432,5 +432,7 @@ MyAwesomeString Utf8View
 
 # Fails until we update to use the arrow-cast release with support for casting utf8 types to BinaryView
 # refer to merge commit https://github.com/apache/arrow-rs/commit/4bd737dab2aa17aca200259347909d48ed793ba1
-query error DataFusion error: This feature is not implemented: Unsupported CAST from Utf8 to BinaryView
+query ?T
 select arrow_cast('MyAwesomeString', 'BinaryView'), arrow_typeof(arrow_cast('MyAwesomeString', 'BinaryView'))
+----
+4d79417765736f6d65537472696e67 BinaryView
diff --git a/datafusion/sqllogictest/test_files/ddl.slt b/datafusion/sqllogictest/test_files/ddl.slt
index 7164425fc0f5..21edb458fe56 100644
--- a/datafusion/sqllogictest/test_files/ddl.slt
+++ b/datafusion/sqllogictest/test_files/ddl.slt
@@ -710,7 +710,7 @@ create table t (i interval, x int) as values (interval '5 days 3 nanoseconds', C
 query ?I
 select * from t;
 ----
-0 years 0 mons 5 days 0 hours 0 mins 0.000000003 secs 1
+5 days 0.000000003 secs 1
 
 statement ok
 drop table t;
diff --git a/datafusion/sqllogictest/test_files/expr.slt b/datafusion/sqllogictest/test_files/expr.slt
index 81ae60f3ba93..002e8db2132d 100644
--- a/datafusion/sqllogictest/test_files/expr.slt
+++ b/datafusion/sqllogictest/test_files/expr.slt
@@ -122,203 +122,203 @@ SELECT
 query ?
 SELECT interval '1'
 ----
-0 years 0 mons 0 days 0 hours 0 mins 1.000000000 secs
+1.000000000 secs
 
 query ?
 SELECT interval '1 second'
 ----
-0 years 0 mons 0 days 0 hours 0 mins 1.000000000 secs
+1.000000000 secs
 
 query ?
 SELECT interval '500 milliseconds'
 ----
-0 years 0 mons 0 days 0 hours 0 mins 0.500000000 secs
+0.500000000 secs
 
 query ?
 SELECT interval '5 second'
 ----
-0 years 0 mons 0 days 0 hours 0 mins 5.000000000 secs
+5.000000000 secs
 
 query ?
 SELECT interval '0.5 minute'
 ----
-0 years 0 mons 0 days 0 hours 0 mins 30.000000000 secs
+30.000000000 secs
 
 query ?
 SELECT interval '.5 minute'
 ----
-0 years 0 mons 0 days 0 hours 0 mins 30.000000000 secs
+30.000000000 secs
 
 query ?
 SELECT interval '5 minute'
 ----
-0 years 0 mons 0 days 0 hours 5 mins 0.000000000 secs
+5 mins
 
 query ?
 SELECT interval '5 minute 1 second'
 ----
-0 years 0 mons 0 days 0 hours 5 mins 1.000000000 secs
+5 mins 1.000000000 secs
 
 query ?
 SELECT interval '1 hour'
 ----
-0 years 0 mons 0 days 1 hours 0 mins 0.000000000 secs
+1 hours
 
 query ?
 SELECT interval '5 hour'
 ----
-0 years 0 mons 0 days 5 hours 0 mins 0.000000000 secs
+5 hours
 
 query ?
 SELECT interval '1 day'
 ----
-0 years 0 mons 1 days 0 hours 0 mins 0.000000000 secs
+1 days
 
 query ?
 SELECT interval '1 week'
 ----
-0 years 0 mons 7 days 0 hours 0 mins 0.000000000 secs
+7 days
 
 query ?
 SELECT interval '2 weeks'
 ----
-0 years 0 mons 14 days 0 hours 0 mins 0.000000000 secs
+14 days
 
 query ?
 SELECT interval '1 day 1'
 ----
-0 years 0 mons 1 days 0 hours 0 mins 1.000000000 secs
+1 days 1.000000000 secs
 
 query ?
 SELECT interval '0.5'
 ----
-0 years 0 mons 0 days 0 hours 0 mins 0.500000000 secs
+0.500000000 secs
 
 query ?
 SELECT interval '0.5 day 1'
 ----
-0 years 0 mons 0 days 12 hours 0 mins 1.000000000 secs
+12 hours 1.000000000 secs
 
 query ?
 SELECT interval '0.49 day'
 ----
-0 years 0 mons 0 days 11 hours 45 mins 36.000000000 secs
+11 hours 45 mins 36.000000000 secs
 
 query ?
 SELECT interval '0.499 day'
 ----
-0 years 0 mons 0 days 11 hours 58 mins 33.600000000 secs
+11 hours 58 mins 33.600000000 secs
 
 query ?
 SELECT interval '0.4999 day'
 ----
-0 years 0 mons 0 days 11 hours 59 mins 51.360000000 secs
+11 hours 59 mins 51.360000000 secs
 
 query ?
 SELECT interval '0.49999 day'
 ----
-0 years 0 mons 0 days 11 hours 59 mins 59.136000000 secs
+11 hours 59 mins 59.136000000 secs
 
 query ?
 SELECT interval '0.49999999999 day'
 ----
-0 years 0 mons 0 days 11 hours 59 mins 59.999999136 secs
+11 hours 59 mins 59.999999136 secs
 
 query ?
 SELECT interval '5 day'
 ----
-0 years 0 mons 5 days 0 hours 0 mins 0.000000000 secs
+5 days
 
 # Hour is ignored, this matches PostgreSQL
 query ?
 SELECT interval '5 day' hour
 ----
-0 years 0 mons 5 days 0 hours 0 mins 0.000000000 secs
+5 days
 
 query ?
 SELECT interval '5 day 4 hours 3 minutes 2 seconds 100 milliseconds'
 ----
-0 years 0 mons 5 days 4 hours 3 mins 2.100000000 secs
+5 days 4 hours 3 mins 2.100000000 secs
 
 query ?
 SELECT interval '0.5 month'
 ----
-0 years 0 mons 15 days 0 hours 0 mins 0.000000000 secs
+15 days
 
 query ?
 SELECT interval '0.5' month
 ----
-0 years 0 mons 15 days 0 hours 0 mins 0.000000000 secs
+15 days
 
 query ?
 SELECT interval '1 month'
 ----
-0 years 1 mons 0 days 0 hours 0 mins 0.000000000 secs
+1 mons
 
 query ?
 SELECT interval '1' MONTH
 ----
-0 years 1 mons 0 days 0 hours 0 mins 0.000000000 secs
+1 mons
 
 query ?
 SELECT interval '5 month'
 ----
-0 years 5 mons 0 days 0 hours 0 mins 0.000000000 secs
+5 mons
 
 query ?
 SELECT interval '13 month'
 ----
-0 years 13 mons 0 days 0 hours 0 mins 0.000000000 secs
+13 mons
 
 query ?
 SELECT interval '0.5 year'
 ----
-0 years 6 mons 0 days 0 hours 0 mins 0.000000000 secs
+6 mons
 
 query ?
 SELECT interval '1 year'
 ----
-0 years 12 mons 0 days 0 hours 0 mins 0.000000000 secs
+12 mons
 
 query ?
 SELECT interval '1 decade'
 ----
-0 years 120 mons 0 days 0 hours 0 mins 0.000000000 secs
+120 mons
 
 query ?
 SELECT interval '2 decades'
 ----
-0 years 240 mons 0 days 0 hours 0 mins 0.000000000 secs
+240 mons
 
 query ?
 SELECT interval '1 century'
 ----
-0 years 1200 mons 0 days 0 hours 0 mins 0.000000000 secs
+1200 mons
 
 query ?
 SELECT interval '2 year'
 ----
-0 years 24 mons 0 days 0 hours 0 mins 0.000000000 secs
+24 mons
 
 query ?
 SELECT interval '1 year 1 day'
 ----
-0 years 12 mons 1 days 0 hours 0 mins 0.000000000 secs
+12 mons 1 days
 
 query ?
 SELECT interval '1 year 1 day 1 hour'
 ----
-0 years 12 mons 1 days 1 hours 0 mins 0.000000000 secs
+12 mons 1 days 1 hours
 
 query ?
 SELECT interval '1 year 1 day 1 hour 1 minute'
 ----
-0 years 12 mons 1 days 1 hours 1 mins 0.000000000 secs
+12 mons 1 days 1 hours 1 mins
 
 query ?
 SELECT interval '1 year 1 day 1 hour 1 minute 1 second'
 ----
-0 years 12 mons 1 days 1 hours 1 mins 1.000000000 secs
+12 mons 1 days 1 hours 1 mins 1.000000000 secs
 
 query I
 SELECT ascii('')
diff --git a/datafusion/sqllogictest/test_files/interval.slt b/datafusion/sqllogictest/test_files/interval.slt
index afb262cf95a5..077f38d5d5bb 100644
--- a/datafusion/sqllogictest/test_files/interval.slt
+++ b/datafusion/sqllogictest/test_files/interval.slt
@@ -45,250 +45,250 @@ Interval(MonthDayNano) Interval(MonthDayNano)
 query ?
 select interval '5' years
 ----
-0 years 0 mons 0 days 0 hours 0 mins 5.000000000 secs
+5.000000000 secs
 
 
 # check all different kinds of intervals
 query ?
 select interval '5' year
 ----
-0 years 60 mons 0 days 0 hours 0 mins 0.000000000 secs
+60 mons
 
 query ?
 select interval '5' month
 ----
-0 years 5 mons 0 days 0 hours 0 mins 0.000000000 secs
+5 mons
 
 query ?
 select interval '5' months
 ----
-0 years 0 mons 0 days 0 hours 0 mins 5.000000000 secs
+5.000000000 secs
 
 query ?
 select interval '5' week
 ----
-0 years 0 mons 35 days 0 hours 0 mins 0.000000000 secs
+35 days
 
 
 query ?
 select interval '5' day
 ----
-0 years 0 mons 5 days 0 hours 0 mins 0.000000000 secs
+5 days
 
 query ?
 select interval '5' hour
 ----
-0 years 0 mons 0 days 5 hours 0 mins 0.000000000 secs
+5 hours
 
 ## This seems wrong (5 mons)
 query ?
 select interval '5' hours
 ----
-0 years 0 mons 0 days 0 hours 0 mins 5.000000000 secs
+5.000000000 secs
 
 query ?
 select interval '5' minute
 ----
-0 years 0 mons 0 days 0 hours 5 mins 0.000000000 secs
+5 mins
 
 query ?
 select interval '5' second
 ----
-0 years 0 mons 0 days 0 hours 0 mins 5.000000000 secs
+5.000000000 secs
 
 query ?
 select interval '5' millisecond
 ----
-0 years 0 mons 0 days 0 hours 0 mins 0.005000000 secs
+0.005000000 secs
 
 query ?
 select interval '5' milliseconds
 ----
-0 years 0 mons 0 days 0 hours 0 mins 0.005000000 secs
+0.005000000 secs
 
 query ?
 select interval '5' microsecond
 ----
-0 years 0 mons 0 days 0 hours 0 mins 0.000005000 secs
+0.000005000 secs
 
 query ?
 select interval '5' microseconds
 ----
-0 years 0 mons 0 days 0 hours 0 mins 0.000005000 secs
+0.000005000 secs
 
 query ?
 select interval '5' nanosecond
 ----
-0 years 0 mons 0 days 0 hours 0 mins 0.000000005 secs
+0.000000005 secs
 
 query ?
 select interval '5' nanoseconds
 ----
-0 years 0 mons 0 days 0 hours 0 mins 0.000000005 secs
+0.000000005 secs
 
 query ?
 select interval '5 YEAR'
 ----
-0 years 60 mons 0 days 0 hours 0 mins 0.000000000 secs
+60 mons
 
 query ?
 select interval '5 MONTH'
 ----
-0 years 5 mons 0 days 0 hours 0 mins 0.000000000 secs
+5 mons
 
 query ?
 select interval '5 WEEK'
 ----
-0 years 0 mons 35 days 0 hours 0 mins 0.000000000 secs
+35 days
 
 query ?
 select interval '5 DAY'
 ----
-0 years 0 mons 5 days 0 hours 0 mins 0.000000000 secs
+5 days
 
 query ?
 select interval '5 HOUR'
 ----
-0 years 0 mons 0 days 5 hours 0 mins 0.000000000 secs
+5 hours
 
 query ?
 select interval '5 HOURS'
 ----
-0 years 0 mons 0 days 5 hours 0 mins 0.000000000 secs
+5 hours
 
 query ?
 select interval '5 MINUTE'
 ----
-0 years 0 mons 0 days 0 hours 5 mins 0.000000000 secs
+5 mins
 
 query ?
 select interval '5 SECOND'
 ----
-0 years 0 mons 0 days 0 hours 0 mins 5.000000000 secs
+5.000000000 secs
 
 query ?
 select interval '5 SECONDS'
 ----
-0 years 0 mons 0 days 0 hours 0 mins 5.000000000 secs
+5.000000000 secs
 
 query ?
 select interval '5 MILLISECOND'
 ----
-0 years 0 mons 0 days 0 hours 0 mins 0.005000000 secs
+0.005000000 secs
 
 query ?
 select interval '5 MILLISECONDS'
 ----
-0 years 0 mons 0 days 0 hours 0 mins 0.005000000 secs
+0.005000000 secs
 
 query ?
 select interval '5 MICROSECOND'
 ----
-0 years 0 mons 0 days 0 hours 0 mins 0.000005000 secs
+0.000005000 secs
 
 query ?
 select interval '5 MICROSECONDS'
 ----
-0 years 0 mons 0 days 0 hours 0 mins 0.000005000 secs
+0.000005000 secs
 
 query ?
 select interval '5 NANOSECOND'
 ----
-0 years 0 mons 0 days 0 hours 0 mins 0.000000005 secs
+0.000000005 secs
 
 query ?
 select interval '5 NANOSECONDS'
 ----
-0 years 0 mons 0 days 0 hours 0 mins 0.000000005 secs
+0.000000005 secs
 
 query ?
 select interval '5 YEAR 5 MONTH 5 DAY 5 HOUR 5 MINUTE 5 SECOND 5 MILLISECOND 5 MICROSECOND 5 NANOSECOND'
 ----
-0 years 65 mons 5 days 5 hours 5 mins 5.005005005 secs
+65 mons 5 days 5 hours 5 mins 5.005005005 secs
 
 # Interval with string literal addition
 query ?
 select interval '1 month' + '1 month'
 ----
-0 years 2 mons 0 days 0 hours 0 mins 0.000000000 secs
+2 mons
 
 # Interval with string literal addition and leading field
 query ?
 select interval '1' + '1' month
 ----
-0 years 2 mons 0 days 0 hours 0 mins 0.000000000 secs
+2 mons
 
 # Interval with nested string literal addition
 query ?
 select interval '1 month' + '1 month' + '1 month'
 ----
-0 years 3 mons 0 days 0 hours 0 mins 0.000000000 secs
+3 mons
 
 # Interval with nested string literal addition and leading field
 query ?
 select interval '1' + '1' + '1' month
 ----
-0 years 3 mons 0 days 0 hours 0 mins 0.000000000 secs
+3 mons
 
 # Interval mega nested string literal addition
 query ?
 select interval '1 year' + '1 month' + '1 day' + '1 hour' + '1 minute' + '1 second' + '1 millisecond' + '1 microsecond' + '1 nanosecond'
 ----
-0 years 13 mons 1 days 1 hours 1 mins 1.001001001 secs
+13 mons 1 days 1 hours 1 mins 1.001001001 secs
 
 # Interval with string literal subtraction
 query ?
 select interval '1 month' - '1 day';
 ----
-0 years 1 mons -1 days 0 hours 0 mins 0.000000000 secs
+1 mons -1 days
 
 # Interval with string literal subtraction and leading field
 query ?
 select interval '5' - '1' - '2' year;
 ----
-0 years 24 mons 0 days 0 hours 0 mins 0.000000000 secs
+24 mons
 
 # Interval with nested string literal subtraction
 query ?
 select interval '1 month' - '1 day' - '1 hour';
 ----
-0 years 1 mons -1 days -1 hours 0 mins 0.000000000 secs
+1 mons -1 days -1 hours
 
 # Interval with nested string literal subtraction and leading field
 query ?
 select interval '10' - '1' - '1' month;
 ----
-0 years 8 mons 0 days 0 hours 0 mins 0.000000000 secs
+8 mons
 
 # Interval mega nested string literal subtraction
 query ?
 select interval '1 year' - '1 month' - '1 day' - '1 hour' - '1 minute' - '1 second' - '1 millisecond' - '1 microsecond' - '1 nanosecond'
 ----
-0 years 11 mons -1 days -1 hours -1 mins -1.001001001 secs
+11 mons -1 days -1 hours -1 mins -1.001001001 secs
 
 # Interval with string literal negation and leading field
 query ?
 select -interval '5' - '1' - '2' year;
 ----
-0 years -96 mons 0 days 0 hours 0 mins 0.000000000 secs
+-96 mons
 
 # Interval with nested string literal negation
 query ?
 select -interval '1 month' + '1 day' + '1 hour';
 ----
-0 years -1 mons 1 days 1 hours 0 mins 0.000000000 secs
+-1 mons 1 days 1 hours
 
 # Interval with nested string literal negation and leading field
 query ?
 select -interval '10' - '1' - '1' month;
 ----
-0 years -12 mons 0 days 0 hours 0 mins 0.000000000 secs
+-12 mons
 
 # Interval mega nested string literal negation
 query ?
 select -interval '1 year' - '1 month' - '1 day' - '1 hour' - '1 minute' - '1 second' - '1 millisecond' - '1 microsecond' - '1 nanosecond'
 ----
-0 years -13 mons -1 days -1 hours -1 mins -1.001001001 secs
+-13 mons -1 days -1 hours -1 mins -1.001001001 secs
 
 # Interval string literal + date
 query D
@@ -343,7 +343,7 @@ select
   arrow_typeof(i)
 from t;
 ----
-0 years 0 mons 5 days 0 hours 0 mins 0.000000003 secs Interval(MonthDayNano)
+5 days 0.000000003 secs Interval(MonthDayNano)
 
 
 statement ok
@@ -359,8 +359,8 @@ insert into t values ('6 days 7 nanoseconds'::interval)
 query ? rowsort
 select -i from t order by 1;
 ----
-0 years 0 mons -5 days 0 hours 0 mins -0.000000003 secs
-0 years 0 mons -6 days 0 hours 0 mins -0.000000007 secs
+-5 days -0.000000003 secs
+-6 days -0.000000007 secs
 
 query ?T rowsort
 select
@@ -368,8 +368,8 @@ select
   arrow_typeof(i)
 from t;
 ----
-0 years 0 mons 5 days 0 hours 0 mins 0.000000003 secs Interval(MonthDayNano)
-0 years 0 mons 6 days 0 hours 0 mins 0.000000007 secs Interval(MonthDayNano)
+5 days 0.000000003 secs Interval(MonthDayNano)
+6 days 0.000000007 secs Interval(MonthDayNano)
 
 statement ok
 drop table t;
diff --git a/datafusion/sqllogictest/test_files/math.slt b/datafusion/sqllogictest/test_files/math.slt
index 6884d762612d..eece56942317 100644
--- a/datafusion/sqllogictest/test_files/math.slt
+++ b/datafusion/sqllogictest/test_files/math.slt
@@ -673,7 +673,7 @@ query error DataFusion error: Arrow error: Compute error: Signed integer overflo
 select lcm(2, 9223372036854775803);
 
 
-query error DataFusion error: Arrow error: Compute error: Overflow happened on: 2107754225 \^ 1221660777
+query error DataFusion error: Arrow error: Arithmetic overflow: Overflow happened on: 2107754225 \^ 1221660777
 select power(2107754225, 1221660777);
 
 # factorial overflow
diff --git a/datafusion/sqllogictest/test_files/repartition_scan.slt b/datafusion/sqllogictest/test_files/repartition_scan.slt
index 6b9cb521f5f8..4c86312f9e51 100644
--- a/datafusion/sqllogictest/test_files/repartition_scan.slt
+++ b/datafusion/sqllogictest/test_files/repartition_scan.slt
@@ -61,7 +61,7 @@ logical_plan
 physical_plan
 01)CoalesceBatchesExec: target_batch_size=8192
 02)--FilterExec: column1@0 != 42
-03)----ParquetExec: file_groups={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:0..104], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:104..208], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:208..312], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:312..414]]}, projection=[column1], predicate=column1@0 != 42, pruning_predicate=CASE WHEN column1_null_count@2 = column1_row_count@3 THEN false ELSE column1_min@0 != 42 OR 42 != column1_max@1 END, required_guarantees=[column1 not in (42)]
+03)----ParquetExec: file_groups={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:0..87], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:87..174], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:174..261], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:261..347]]}, projection=[column1], predicate=column1@0 != 42, pruning_predicate=CASE WHEN column1_null_count@2 = column1_row_count@3 THEN false ELSE column1_min@0 != 42 OR 42 != column1_max@1 END, required_guarantees=[column1 not in (42)]
 
 # disable round robin repartitioning
 statement ok
@@ -77,7 +77,7 @@ logical_plan
 physical_plan
 01)CoalesceBatchesExec: target_batch_size=8192
 02)--FilterExec: column1@0 != 42
-03)----ParquetExec: file_groups={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:0..104], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:104..208], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:208..312], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:312..414]]}, projection=[column1], predicate=column1@0 != 42, pruning_predicate=CASE WHEN column1_null_count@2 = column1_row_count@3 THEN false ELSE column1_min@0 != 42 OR 42 != column1_max@1 END, required_guarantees=[column1 not in (42)]
+03)----ParquetExec: file_groups={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:0..87], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:87..174], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:174..261], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:261..347]]}, projection=[column1], predicate=column1@0 != 42, pruning_predicate=CASE WHEN column1_null_count@2 = column1_row_count@3 THEN false ELSE column1_min@0 != 42 OR 42 != column1_max@1 END, required_guarantees=[column1 not in (42)]
 
 # enable round robin repartitioning again
 statement ok
@@ -102,7 +102,7 @@ physical_plan
 02)--SortExec: expr=[column1@0 ASC NULLS LAST], preserve_partitioning=[true]
 03)----CoalesceBatchesExec: target_batch_size=8192
 04)------FilterExec: column1@0 != 42
-05)--------ParquetExec: file_groups={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/1.parquet:0..205], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/1.parquet:205..405, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:0..5], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:5..210], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:210..414]]}, projection=[column1], predicate=column1@0 != 42, pruning_predicate=CASE WHEN column1_null_count@2 = column1_row_count@3 THEN false ELSE column1_min@0 != 42 OR 42 != column1_max@1 END, required_guarantees=[column1 not in (42)]
+05)--------ParquetExec: file_groups={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/1.parquet:0..172], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/1.parquet:172..338, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:0..6], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:6..178], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:178..347]]}, projection=[column1], predicate=column1@0 != 42, pruning_predicate=CASE WHEN column1_null_count@2 = column1_row_count@3 THEN false ELSE column1_min@0 != 42 OR 42 != column1_max@1 END, required_guarantees=[column1 not in (42)]
 
 
 ## Read the files as though they are ordered
@@ -138,7 +138,7 @@ physical_plan
 01)SortPreservingMergeExec: [column1@0 ASC NULLS LAST]
 02)--CoalesceBatchesExec: target_batch_size=8192
 03)----FilterExec: column1@0 != 42
-04)------ParquetExec: file_groups={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/1.parquet:0..202], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:0..207], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:207..414], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/1.parquet:202..405]]}, projection=[column1], output_ordering=[column1@0 ASC NULLS LAST], predicate=column1@0 != 42, pruning_predicate=CASE WHEN column1_null_count@2 = column1_row_count@3 THEN false ELSE column1_min@0 != 42 OR 42 != column1_max@1 END, required_guarantees=[column1 not in (42)]
+04)------ParquetExec: file_groups={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/1.parquet:0..169], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:0..173], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:173..347], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/1.parquet:169..338]]}, projection=[column1], output_ordering=[column1@0 ASC NULLS LAST], predicate=column1@0 != 42, pruning_predicate=CASE WHEN column1_null_count@2 = column1_row_count@3 THEN false ELSE column1_min@0 != 42 OR 42 != column1_max@1 END, required_guarantees=[column1 not in (42)]
 
 # Cleanup
 statement ok
diff --git a/datafusion/sqllogictest/test_files/timestamps.slt b/datafusion/sqllogictest/test_files/timestamps.slt
index fb0fd8397f2d..4b11e338da70 100644
--- a/datafusion/sqllogictest/test_files/timestamps.slt
+++ b/datafusion/sqllogictest/test_files/timestamps.slt
@@ -1509,19 +1509,19 @@ SELECT val, ts1 - ts2 FROM foo ORDER BY ts2 - ts1;
 query ?
 SELECT i1 - i2 FROM bar;
 ----
-0 years 0 mons -1 days 0 hours 0 mins 0.000000000 secs
-0 years 2 mons -13 days 0 hours 0 mins 0.000000000 secs
-0 years 0 mons 1 days 2 hours 56 mins 0.000000000 secs
-0 years 0 mons 1 days 0 hours 0 mins -3.999999993 secs
+-1 days
+2 mons -13 days
+1 days 2 hours 56 mins
+1 days -3.999999993 secs
 
 # Interval + Interval
 query ?
 SELECT i1 + i2 FROM bar;
 ----
-0 years 0 mons 3 days 0 hours 0 mins 0.000000000 secs
-0 years 2 mons 13 days 0 hours 0 mins 0.000000000 secs
-0 years 0 mons 1 days 3 hours 4 mins 0.000000000 secs
-0 years 0 mons 1 days 0 hours 0 mins 4.000000007 secs
+3 days
+2 mons 13 days
+1 days 3 hours 4 mins
+1 days 4.000000007 secs
 
 # Timestamp - Interval
 query P