diff --git a/native/Cargo.lock b/native/Cargo.lock
index 1658c480a8..8295c99453 100644
--- a/native/Cargo.lock
+++ b/native/Cargo.lock
@@ -419,7 +419,7 @@ version = "58.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f633dbfdf39c039ada1bf9e34c694816eb71fbb7dc78f613993b7245e078a1ed"
 dependencies = [
- "bitflags 2.13.0",
+ "bitflags 2.11.1",
  "serde_core",
  "serde_json",
 ]
@@ -624,9 +624,9 @@ checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0"
 
 [[package]]
 name = "autocfg"
-version = "1.5.1"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53"
+checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8"
 
 [[package]]
 name = "aws-config"
@@ -650,7 +650,7 @@ dependencies = [
  "bytes",
  "fastrand",
  "hex",
- "http 1.4.2",
+ "http 1.4.0",
  "sha1 0.10.6",
  "time",
  "tokio",
@@ -673,19 +673,20 @@ dependencies = [
 
 [[package]]
 name = "aws-lc-rs"
-version = "1.17.0"
+version = "1.16.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5ec2f1fc3ec205783a5da9a7e6c1509cc69dedf09a1949e412c1e18469326d00"
+checksum = "0ec6fb3fe69024a75fa7e1bfb48aa6cf59706a101658ea01bfd33b2b248a038f"
 dependencies = [
  "aws-lc-sys",
+ "untrusted 0.7.1",
  "zeroize",
 ]
 
 [[package]]
 name = "aws-lc-sys"
-version = "0.41.0"
+version = "0.40.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1a2f9779ce85b93ab6170dd940ad0169b5766ff848247aff13bb788b832fe3f4"
+checksum = "f50037ee5e1e41e7b8f9d161680a725bd1626cb6f8c7e901f91f942850852fe7"
 dependencies = [
  "cc",
  "cmake",
@@ -710,7 +711,7 @@ dependencies = [
  "bytes",
  "bytes-utils",
  "fastrand",
- "http 1.4.2",
+ "http 1.4.0",
  "http-body 1.0.1",
  "percent-encoding",
  "pin-project-lite",
@@ -738,7 +739,7 @@ dependencies = [
  "bytes",
  "fastrand",
  "http 0.2.12",
- "http 1.4.2",
+ "http 1.4.0",
  "regex-lite",
  "tracing",
 ]
@@ -763,7 +764,7 @@ dependencies = [
  "bytes",
  "fastrand",
  "http 0.2.12",
- "http 1.4.2",
+ "http 1.4.0",
  "regex-lite",
  "tracing",
 ]
@@ -789,7 +790,7 @@ dependencies = [
  "aws-types",
  "fastrand",
  "http 0.2.12",
- "http 1.4.2",
+ "http 1.4.0",
  "regex-lite",
  "tracing",
 ]
@@ -809,7 +810,7 @@ dependencies = [
  "hex",
  "hmac 0.13.0",
  "http 0.2.12",
- "http 1.4.2",
+ "http 1.4.0",
  "percent-encoding",
  "sha2 0.11.0",
  "time",
@@ -839,7 +840,7 @@ dependencies = [
  "bytes-utils",
  "futures-core",
  "futures-util",
- "http 1.4.2",
+ "http 1.4.0",
  "http-body 1.0.1",
  "http-body-util",
  "percent-encoding",
@@ -858,7 +859,7 @@ dependencies = [
  "aws-smithy-runtime-api",
  "aws-smithy-types",
  "h2",
- "http 1.4.2",
+ "http 1.4.0",
  "hyper",
  "hyper-rustls",
  "hyper-util",
@@ -918,7 +919,7 @@ dependencies = [
  "bytes",
  "fastrand",
  "http 0.2.12",
- "http 1.4.2",
+ "http 1.4.0",
  "http-body 0.4.6",
  "http-body 1.0.1",
  "http-body-util",
@@ -939,7 +940,7 @@ dependencies = [
  "aws-smithy-types",
  "bytes",
  "http 0.2.12",
- "http 1.4.2",
+ "http 1.4.0",
  "pin-project-lite",
  "tokio",
  "tracing",
@@ -965,7 +966,7 @@ checksum = "7442cb268338f0eb8278140a107c046756aa01093d8ef5e99628d34ae09c94f5"
 dependencies = [
  "aws-smithy-runtime-api",
  "aws-smithy-types",
- "http 1.4.2",
+ "http 1.4.0",
 ]
 
 [[package]]
@@ -978,7 +979,7 @@ dependencies = [
  "bytes",
  "bytes-utils",
  "http 0.2.12",
- "http 1.4.2",
+ "http 1.4.0",
  "http-body 0.4.6",
  "http-body 1.0.1",
  "http-body-util",
@@ -1111,7 +1112,7 @@ version = "0.72.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "993776b509cfb49c750f11b8f07a46fa23e0a1386ffc01fb1e7d343efc387895"
 dependencies = [
- "bitflags 2.13.0",
+ "bitflags 2.11.1",
  "cexpr",
  "clang-sys",
  "itertools 0.13.0",
@@ -1133,9 +1134,9 @@ checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
 
 [[package]]
 name = "bitflags"
-version = "2.13.0"
+version = "2.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b4388bee8683e3d04af747c73422af53102d2bd24d9eadb6cbc100baef4b43f8"
+checksum = "c4512299f36f043ab09a583e57bceb5a5aab7a73db1805848e8fef3c9e8c78b3"
 
 [[package]]
 name = "blake2"
@@ -1267,9 +1268,9 @@ dependencies = [
 
 [[package]]
 name = "bumpalo"
-version = "3.20.3"
+version = "3.20.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649"
+checksum = "5d20789868f4b01b2f2caec9f5c4e0213b41e3e5702a50157d699ae31ced2fcb"
 
 [[package]]
 name = "bytemuck"
@@ -1375,9 +1376,9 @@ dependencies = [
 
 [[package]]
 name = "chrono"
-version = "0.4.45"
+version = "0.4.44"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327"
+checksum = "c673075a2e0e5f4a1dde27ce9dee1ea4558c7ffe648f576438a20ca1d2acc4b0"
 dependencies = [
  "iana-time-zone",
  "js-sys",
@@ -1496,9 +1497,9 @@ dependencies = [
 
 [[package]]
 name = "cmov"
-version = "0.5.4"
+version = "0.5.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a"
+checksum = "3f88a43d011fc4a6876cb7344703e297c71dda42494fee094d5f7c76bf13f746"
 
 [[package]]
 name = "colorchoice"
@@ -2881,9 +2882,9 @@ dependencies = [
 
 [[package]]
 name = "displaydoc"
-version = "0.2.6"
+version = "0.2.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1ac70aa55017e108007fbaf5aa0f54b021c98f92ff8af59d42eda9da96e3dd4f"
+checksum = "97369cbbc041bc366949bc74d34658d6cda5621039731c6310521892a3a20ae0"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -2919,9 +2920,9 @@ checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555"
 
 [[package]]
 name = "either"
-version = "1.16.0"
+version = "1.15.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e"
+checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719"
 
 [[package]]
 name = "equator"
@@ -3009,9 +3010,9 @@ dependencies = [
 
 [[package]]
 name = "fastnum"
-version = "0.7.5"
+version = "0.7.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "020d1b59a944bc239d79903fbac2eda2365138b44890c27979562f6592059dcd"
+checksum = "4089ab2dfd45d8ddc92febb5ca80644389d5ebb954f40231274a3f18341762e2"
 dependencies = [
  "bnum",
  "num-integer",
@@ -3055,7 +3056,7 @@ version = "25.12.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "35f6839d7b3b98adde531effaf34f0c2badc6f4735d26fe74709d8e513a96ef3"
 dependencies = [
- "bitflags 2.13.0",
+ "bitflags 2.11.1",
  "rustc_version",
 ]
 
@@ -3312,7 +3313,7 @@ dependencies = [
  "fnv",
  "futures-core",
  "futures-sink",
- "http 1.4.2",
+ "http 1.4.0",
  "indexmap 2.14.0",
  "slab",
  "tokio",
@@ -3446,9 +3447,9 @@ dependencies = [
 
 [[package]]
 name = "http"
-version = "1.4.2"
+version = "1.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6970f50e31d6fc17d3fa27329444bfa74e196cf62e95052a3f6fee181dba6425"
+checksum = "e3ba2a386d7f85a81f119ad7498ebe444d2e22c2af0b86b069416ace48b3311a"
 dependencies = [
  "bytes",
  "itoa",
@@ -3472,7 +3473,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1efedce1fb8e6913f23e0c92de8e62cd5b772a67e7b3946df930a62566c93184"
 dependencies = [
  "bytes",
- "http 1.4.2",
+ "http 1.4.0",
 ]
 
 [[package]]
@@ -3483,7 +3484,7 @@ checksum = "b021d93e26becf5dc7e1b75b1bed1fd93124b374ceb73f43d4d4eafec896a64a"
 dependencies = [
  "bytes",
  "futures-core",
- "http 1.4.2",
+ "http 1.4.0",
  "http-body 1.0.1",
  "pin-project-lite",
 ]
@@ -3511,16 +3512,16 @@ dependencies = [
 
 [[package]]
 name = "hyper"
-version = "1.10.1"
+version = "1.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "55281c53a1894c864990125767da440a4e630446785086f52523b20033b74498"
+checksum = "6299f016b246a94207e63da54dbe807655bf9e00044f73ded42c3ac5305fbcca"
 dependencies = [
  "atomic-waker",
  "bytes",
  "futures-channel",
  "futures-core",
  "h2",
- "http 1.4.2",
+ "http 1.4.0",
  "http-body 1.0.1",
  "httparse",
  "itoa",
@@ -3536,7 +3537,7 @@ version = "0.27.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "33ca68d021ef39cf6463ab54c1d0f5daf03377b70561305bb89a8f83aab66e0f"
 dependencies = [
- "http 1.4.2",
+ "http 1.4.0",
  "hyper",
  "hyper-util",
  "rustls",
@@ -3556,7 +3557,7 @@ dependencies = [
  "bytes",
  "futures-channel",
  "futures-util",
- "http 1.4.2",
+ "http 1.4.0",
  "http-body 1.0.1",
  "hyper",
  "ipnet",
@@ -3595,8 +3596,8 @@ dependencies = [
 
 [[package]]
 name = "iceberg"
-version = "0.9.0"
-source = "git+https://github.com/apache/iceberg-rust?rev=80a30d3#80a30d3440928e1456fd116db934d829844a1a42"
+version = "0.10.0"
+source = "git+https://github.com/parthchandra/iceberg-rust?branch=metadata-columns-test#7d049d2e1528340f69f95a53c8128e99881e8280"
 dependencies = [
  "aes-gcm",
  "anyhow",
@@ -3651,8 +3652,8 @@ dependencies = [
 
 [[package]]
 name = "iceberg-storage-opendal"
-version = "0.9.0"
-source = "git+https://github.com/apache/iceberg-rust?rev=80a30d3#80a30d3440928e1456fd116db934d829844a1a42"
+version = "0.10.0"
+source = "git+https://github.com/apache/iceberg-rust?rev=77c5d710#77c5d7101a658d645c9ed9adbfb45005b3f31e9f"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -3769,9 +3770,9 @@ dependencies = [
 
 [[package]]
 name = "idna_adapter"
-version = "1.2.2"
+version = "1.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cb68373c0d6620ef8105e855e7745e18b0d00d3bdb07fb532e434244cdb9a714"
+checksum = "3acae9609540aa318d1bc588455225fb2085b9ed0c4f6bd0d9d5bcd86f1a0344"
 dependencies = [
  "icu_normalizer",
  "icu_properties",
@@ -3849,6 +3850,16 @@ version = "2.12.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d98f6fed1fde3f8c21bc40a1abb88dd75e67924f9cffc3ef95607bad8017f8e2"
 
+[[package]]
+name = "iri-string"
+version = "0.7.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "25e659a4bb38e810ebc252e53b5814ff908a8c58c2a9ce2fae1bbec24cbf4e20"
+dependencies = [
+ "memchr",
+ "serde",
+]
+
 [[package]]
 name = "is-terminal"
 version = "0.4.17"
@@ -4049,6 +4060,24 @@ dependencies = [
  "wasm-bindgen",
 ]
 
+[[package]]
+name = "jsonwebtoken"
+version = "10.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "eba32bfb4ffdeaca3e34431072faf01745c9b26d25504aa7a6cf5684334fc4fc"
+dependencies = [
+ "aws-lc-rs",
+ "base64",
+ "getrandom 0.2.17",
+ "js-sys",
+ "pem",
+ "serde",
+ "serde_json",
+ "signature",
+ "simple_asn1",
+ "zeroize",
+]
+
 [[package]]
 name = "kv-log-macro"
 version = "1.0.7"
@@ -4308,9 +4337,9 @@ dependencies = [
 
 [[package]]
 name = "mea"
-version = "0.6.4"
+version = "0.6.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2640d335e7273dacdcf51044026139b2e269c3bb0dfc3f8cb3496b85e3f6a42c"
+checksum = "6747f54621d156e1b47eb6b25f39a941b9fc347f98f67d25d8881ff99e8ed832"
 dependencies = [
  "slab",
 ]
@@ -4357,9 +4386,9 @@ dependencies = [
 
 [[package]]
 name = "mio"
-version = "1.2.1"
+version = "1.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "02bd0af71c67b473010cbbc60715ee815645a4dc942899111f494b4b737d6fda"
+checksum = "50b7e5b27aa02a74bac8c3f23f448f8d87ff11f92d3aac1a6ed369ee08cc56c1"
 dependencies = [
  "libc",
  "wasi",
@@ -4555,7 +4584,7 @@ dependencies = [
  "futures-channel",
  "futures-core",
  "futures-util",
- "http 1.4.2",
+ "http 1.4.0",
  "http-body-util",
  "httparse",
  "humantime",
@@ -4652,7 +4681,7 @@ dependencies = [
  "base64",
  "bytes",
  "futures",
- "http 1.4.2",
+ "http 1.4.0",
  "http-body 1.0.1",
  "jiff",
  "log",
@@ -4661,7 +4690,7 @@ dependencies = [
  "percent-encoding",
  "quick-xml 0.39.4",
  "reqsign-core",
- "reqwest 0.13.4",
+ "reqwest 0.13.3",
  "serde",
  "serde_json",
  "tokio",
@@ -4677,7 +4706,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0d6f81ba6960e3fae1882f253b114b21d7e444e1534f209c7737a79f6243eb6f"
 dependencies = [
  "futures",
- "http 1.4.2",
+ "http 1.4.0",
  "mea",
  "opendal-core",
 ]
@@ -4721,7 +4750,7 @@ checksum = "6dea4908d490143a9b0b7f7a790e139ff829b06a023f670455ed3d44f664b361"
 dependencies = [
  "base64",
  "bytes",
- "http 1.4.2",
+ "http 1.4.0",
  "log",
  "opendal-core",
  "opendal-service-azure-common",
@@ -4739,7 +4768,7 @@ version = "0.57.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9b489f13c42e69d69bdd72952b634356ec43a7881a20259b38b540fcecdf4051"
 dependencies = [
- "http 1.4.2",
+ "http 1.4.0",
  "opendal-core",
 ]
 
@@ -4765,7 +4794,7 @@ checksum = "48de101aac565ed06af4b47903c24eafd249075553ec1fb18256751c45148d47"
 dependencies = [
  "async-trait",
  "bytes",
- "http 1.4.2",
+ "http 1.4.0",
  "log",
  "opendal-core",
  "percent-encoding",
@@ -4800,7 +4829,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "328fa55e8888cbdfe00826bfea2a79042422b720e8369e9e021e46121dea5ace"
 dependencies = [
  "bytes",
- "http 1.4.2",
+ "http 1.4.0",
  "log",
  "opendal-core",
  "quick-xml 0.39.4",
@@ -4819,7 +4848,7 @@ dependencies = [
  "base64",
  "bytes",
  "crc32c",
- "http 1.4.2",
+ "http 1.4.0",
  "log",
  "md-5 0.11.0",
  "opendal-core",
@@ -5213,9 +5242,9 @@ checksum = "c33a9471896f1c69cecef8d20cbe2f7accd12527ce60845ff44c153bb2a21b49"
 
 [[package]]
 name = "portable-atomic-util"
-version = "0.2.7"
+version = "0.2.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c2a106d1259c23fac8e543272398ae0e3c0b8d33c88ed73d0cc71b0f1d902618"
+checksum = "091397be61a01d4be58e7841595bd4bfedb15f1cd54977d79b8271e94ed799a3"
 dependencies = [
  "portable-atomic",
 ]
@@ -5313,7 +5342,7 @@ version = "0.18.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "25485360a54d6861439d60facef26de713b1e126bf015ec8f98239467a2b82f7"
 dependencies = [
- "bitflags 2.13.0",
+ "bitflags 2.11.1",
  "chrono",
  "flate2",
  "procfs-core",
@@ -5326,7 +5355,7 @@ version = "0.18.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e6401bf7b6af22f78b563665d15a22e9aef27775b79b149a66ca022468a4e405"
 dependencies = [
- "bitflags 2.13.0",
+ "bitflags 2.11.1",
  "chrono",
  "hex",
 ]
@@ -5407,16 +5436,6 @@ dependencies = [
  "serde",
 ]
 
-[[package]]
-name = "quick-xml"
-version = "0.40.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2474bd2e5029e7ccb6abb2ba48cf2383a333851dedf495901544281590c7da7f"
-dependencies = [
- "memchr",
- "serde",
-]
-
 [[package]]
 name = "quinn"
 version = "0.11.11"
@@ -5470,7 +5489,7 @@ dependencies = [
  "once_cell",
  "socket2",
  "tracing",
- "windows-sys 0.60.2",
+ "windows-sys 0.59.0",
 ]
 
 [[package]]
@@ -5595,7 +5614,7 @@ version = "0.5.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d"
 dependencies = [
- "bitflags 2.13.0",
+ "bitflags 2.11.1",
 ]
 
 [[package]]
@@ -5620,9 +5639,9 @@ dependencies = [
 
 [[package]]
 name = "regex"
-version = "1.12.4"
+version = "1.12.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f1292b7759ae1cb9ec195452d1390a074f0cd8541ab7a5a8c31cd6db45d4a6ba"
+checksum = "e10754a14b9137dd7b1e3e5b0493cc9171fdd105e0ab477f51b72e7f3ac0e276"
 dependencies = [
  "aho-corasick",
  "memchr",
@@ -5649,19 +5668,19 @@ checksum = "cab834c73d247e67f4fae452806d17d3c7501756d98c8808d7c9c7aa7d18f973"
 
 [[package]]
 name = "regex-syntax"
-version = "0.8.11"
+version = "0.8.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4"
+checksum = "dc897dd8d9e8bd1ed8cdad82b5966c3e0ecae09fb1907d58efaa013543185d0a"
 
 [[package]]
 name = "reqsign-aliyun-oss"
-version = "3.1.0"
+version = "3.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "372266b4733756738eeb199a98188037d27a0989980e2600ae7ce1faf00a867d"
+checksum = "57ac2757f3140aa2e213b554148ae0b52733e624fc6723f0cc6bb3d440176c95"
 dependencies = [
  "anyhow",
  "form_urlencoded",
- "http 1.4.2",
+ "http 1.4.0",
  "log",
  "percent-encoding",
  "reqsign-core",
@@ -5672,37 +5691,37 @@ dependencies = [
 
 [[package]]
 name = "reqsign-aws-v4"
-version = "3.0.1"
+version = "3.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7b75624bd8a466e37ddc0a7b6c33ac859a85347c153a916e1dd9d0b68338f74a"
+checksum = "44eaca382e94505a49f1a4849658d153aebf79d9c1a58e5dd3b10361511e9f43"
 dependencies = [
  "anyhow",
  "bytes",
  "form_urlencoded",
- "hex",
- "http 1.4.2",
+ "http 1.4.0",
  "log",
  "percent-encoding",
- "quick-xml 0.40.1",
+ "quick-xml 0.39.4",
  "reqsign-core",
  "rust-ini",
  "serde",
  "serde_json",
  "serde_urlencoded",
- "sha1 0.11.0",
+ "sha1 0.10.6",
 ]
 
 [[package]]
 name = "reqsign-azure-storage"
-version = "3.0.1"
+version = "3.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "62b96928e73ad984de1d99e382749d09e5dab7dd707b767974f7e40aa926b82f"
+checksum = "7a321980405d596bd34aaf95c4722a3de4128a67fd19e74a81a83aa3fdf082e6"
 dependencies = [
  "anyhow",
  "base64",
  "bytes",
  "form_urlencoded",
- "http 1.4.2",
+ "http 1.4.0",
+ "jsonwebtoken",
  "log",
  "pem",
  "percent-encoding",
@@ -5710,14 +5729,14 @@ dependencies = [
  "rsa",
  "serde",
  "serde_json",
- "sha1 0.11.0",
+ "sha1 0.10.6",
 ]
 
 [[package]]
 name = "reqsign-core"
-version = "3.0.1"
+version = "3.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a5fa5cb48808693614d1701fcd3db0b30fa292e0f18e122ae068b6d32eaeed3f"
+checksum = "b10302cf0a7d7e7352ba211fc92c3c5bebf1286153e49cc5aa87348078a8e102"
 dependencies = [
  "anyhow",
  "base64",
@@ -5725,24 +5744,21 @@ dependencies = [
  "form_urlencoded",
  "futures",
  "hex",
- "hmac 0.13.0",
- "http 1.4.2",
+ "hmac 0.12.1",
+ "http 1.4.0",
  "jiff",
  "log",
  "percent-encoding",
- "rsa",
- "serde",
- "serde_json",
- "sha1 0.11.0",
- "sha2 0.11.0",
+ "sha1 0.10.6",
+ "sha2 0.10.9",
  "windows-sys 0.61.2",
 ]
 
 [[package]]
 name = "reqsign-file-read-tokio"
-version = "3.0.1"
+version = "3.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6a4b6f3a3fd29ffcc99a90aec585a65217783badfd73acddf847b63ae683bda9"
+checksum = "e2d89295b3d17abea31851cc8de55d843d89c52132c864963c38d41920613dc5"
 dependencies = [
  "anyhow",
  "reqsign-core",
@@ -5751,12 +5767,13 @@ dependencies = [
 
 [[package]]
 name = "reqsign-google"
-version = "3.0.1"
+version = "3.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "eb215d0876a18b6bd9cdd380b589e5292aaa638ca15266de794b1122d898b6b2"
+checksum = "35cc609b49c69e76ecaceb775a03f792d1ed3e7755ab3548d4534fd801e3242e"
 dependencies = [
  "form_urlencoded",
- "http 1.4.2",
+ "http 1.4.0",
+ "jsonwebtoken",
  "log",
  "percent-encoding",
  "reqsign-aws-v4",
@@ -5764,6 +5781,7 @@ dependencies = [
  "rsa",
  "serde",
  "serde_json",
+ "sha2 0.10.9",
  "tokio",
 ]
 
@@ -5778,7 +5796,7 @@ dependencies = [
  "futures-core",
  "futures-util",
  "h2",
- "http 1.4.2",
+ "http 1.4.0",
  "http-body 1.0.1",
  "http-body-util",
  "hyper",
@@ -5811,15 +5829,15 @@ dependencies = [
 
 [[package]]
 name = "reqwest"
-version = "0.13.4"
+version = "0.13.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "219c5811de6525e5416c7d5d53bb656d3afdbc6c5af816e0802bcfa42dbdc1c3"
+checksum = "62e0021ea2c22aed41653bc7e1419abb2c97e038ff2c33d0e1309e49a97deec0"
 dependencies = [
  "base64",
  "bytes",
  "futures-core",
  "futures-util",
- "http 1.4.2",
+ "http 1.4.0",
  "http-body 1.0.1",
  "http-body-util",
  "hyper",
@@ -5866,15 +5884,15 @@ dependencies = [
  "cfg-if",
  "getrandom 0.2.17",
  "libc",
- "untrusted",
+ "untrusted 0.9.0",
  "windows-sys 0.52.0",
 ]
 
 [[package]]
 name = "roaring"
-version = "0.11.4"
+version = "0.11.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1dedc5658c6ecb3bdb5ef5f3295bb9253f42dcf3fd1402c03f6b1f7659c3c4a9"
+checksum = "8ba9ce64a8f45d7fc86358410bb1a82e8c987504c0d4900e9141d69a9f26c885"
 dependencies = [
  "bytemuck",
  "byteorder",
@@ -5944,7 +5962,7 @@ version = "0.38.44"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "fdb5bc1ae2baa591800df16c9ca78619bf65c0488b41b96ccec5d11220d8c154"
 dependencies = [
- "bitflags 2.13.0",
+ "bitflags 2.11.1",
  "errno",
  "libc",
  "linux-raw-sys 0.4.15",
@@ -5957,7 +5975,7 @@ version = "1.1.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190"
 dependencies = [
- "bitflags 2.13.0",
+ "bitflags 2.11.1",
  "errno",
  "libc",
  "linux-raw-sys 0.12.1",
@@ -5981,9 +5999,9 @@ dependencies = [
 
 [[package]]
 name = "rustls-native-certs"
-version = "0.8.4"
+version = "0.8.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dab5152771c58876a2146916e53e35057e1a4dfa2b9df0f0305b07f611fdea4d"
+checksum = "612460d5f7bea540c490b2b6395d8e34a953e52b491accd6c86c8164c5932a63"
 dependencies = [
  "openssl-probe",
  "rustls-pki-types",
@@ -5993,9 +6011,9 @@ dependencies = [
 
 [[package]]
 name = "rustls-pki-types"
-version = "1.14.1"
+version = "1.14.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "30a7197ae7eb376e574fe940d068c30fe0462554a3ddbe4eca7838e049c937a9"
+checksum = "be040f8b0a225e40375822a563fa9524378b9d63112f53e19ffff34df5d33fdd"
 dependencies = [
  "web-time",
  "zeroize",
@@ -6037,7 +6055,7 @@ dependencies = [
  "aws-lc-rs",
  "ring",
  "rustls-pki-types",
- "untrusted",
+ "untrusted 0.9.0",
 ]
 
 [[package]]
@@ -6126,7 +6144,7 @@ version = "3.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b7f4bc775c73d9a02cde8bf7b2ec4c9d12743edf609006c7facc23998404cd1d"
 dependencies = [
- "bitflags 2.13.0",
+ "bitflags 2.11.1",
  "core-foundation",
  "core-foundation-sys",
  "libc",
@@ -6394,11 +6412,23 @@ version = "0.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e3a9fe34e3e7a50316060351f37187a3f546bce95496156754b601a5fa71b76e"
 
+[[package]]
+name = "simple_asn1"
+version = "0.6.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0d585997b0ac10be3c5ee635f1bab02d512760d14b7c468801ac8a01d9ae5f1d"
+dependencies = [
+ "num-bigint",
+ "num-traits",
+ "thiserror 2.0.18",
+ "time",
+]
+
 [[package]]
 name = "siphasher"
-version = "1.0.3"
+version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8ee5873ec9cce0195efcb7a4e9507a04cd49aec9c83d0389df45b1ef7ba2e649"
+checksum = "b2aa850e253778c88a04c3d7323b043aeda9d3e30d5971937c1855769763678e"
 
 [[package]]
 name = "slab"
@@ -6420,9 +6450,9 @@ checksum = "1b6b67fb9a61334225b5b790716f609cd58395f895b3fe8b328786812a40bc3b"
 
 [[package]]
 name = "socket2"
-version = "0.6.4"
+version = "0.6.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51"
+checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e"
 dependencies = [
  "libc",
  "windows-sys 0.61.2",
@@ -6849,20 +6879,20 @@ dependencies = [
 
 [[package]]
 name = "tower-http"
-version = "0.6.11"
+version = "0.6.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840"
+checksum = "d4e6559d53cc268e5031cd8429d05415bc4cb4aefc4aa5d6cc35fbf5b924a1f8"
 dependencies = [
- "bitflags 2.13.0",
+ "bitflags 2.11.1",
  "bytes",
  "futures-util",
- "http 1.4.2",
+ "http 1.4.0",
  "http-body 1.0.1",
+ "iri-string",
  "pin-project-lite",
  "tower",
  "tower-layer",
  "tower-service",
- "url",
 ]
 
 [[package]]
@@ -6966,9 +6996,9 @@ checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20"
 
 [[package]]
 name = "typetag"
-version = "0.2.22"
+version = "0.2.21"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c5a897b12c6c1151ad0b138b8db50252dc301f93bc3b027db05eec82aeed298c"
+checksum = "be2212c8a9b9bcfca32024de14998494cf9a5dfa59ea1b829de98bac374b86bf"
 dependencies = [
  "erased-serde",
  "inventory",
@@ -6979,9 +7009,9 @@ dependencies = [
 
 [[package]]
 name = "typetag-impl"
-version = "0.2.22"
+version = "0.2.21"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cf808357c6ed7e13ba0f3277ec8d8f21b2d501274895104263985330c726c1c5"
+checksum = "27a7a9b72ba121f6f1f6c3632b85604cac41aedb5ddc70accbebb6cac83de846"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -6996,9 +7026,9 @@ checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75"
 
 [[package]]
 name = "unicode-segmentation"
-version = "1.13.3"
+version = "1.13.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c6f5d3c3b1bf09027a88a6bc961fc00497d651009560b5463668dc81b0fa87a8"
+checksum = "9629274872b2bfaf8d66f5f15725007f635594914870f65218920345aa11aa8c"
 
 [[package]]
 name = "unicode-width"
@@ -7037,6 +7067,12 @@ version = "0.2.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "673aac59facbab8a9007c7f6108d11f63b603f7cabff99fabf650fea5c32b861"
 
+[[package]]
+name = "untrusted"
+version = "0.7.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a156c684c91ea7d62626509bce3cb4e1d9ed5c4d978f7b4352658f96a4c26b4a"
+
 [[package]]
 name = "untrusted"
 version = "0.9.0"
@@ -7376,15 +7412,6 @@ dependencies = [
  "windows-targets 0.52.6",
 ]
 
-[[package]]
-name = "windows-sys"
-version = "0.60.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f2f500e4d28234f72040990ec9d39e3a6b950f9f22d3dba18416c35882612bcb"
-dependencies = [
- "windows-targets 0.53.5",
-]
-
 [[package]]
 name = "windows-sys"
 version = "0.61.2"
@@ -7418,30 +7445,13 @@ dependencies = [
  "windows_aarch64_gnullvm 0.52.6",
  "windows_aarch64_msvc 0.52.6",
  "windows_i686_gnu 0.52.6",
- "windows_i686_gnullvm 0.52.6",
+ "windows_i686_gnullvm",
  "windows_i686_msvc 0.52.6",
  "windows_x86_64_gnu 0.52.6",
  "windows_x86_64_gnullvm 0.52.6",
  "windows_x86_64_msvc 0.52.6",
 ]
 
-[[package]]
-name = "windows-targets"
-version = "0.53.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4945f9f551b88e0d65f3db0bc25c33b8acea4d9e41163edf90dcd0b19f9069f3"
-dependencies = [
- "windows-link",
- "windows_aarch64_gnullvm 0.53.1",
- "windows_aarch64_msvc 0.53.1",
- "windows_i686_gnu 0.53.1",
- "windows_i686_gnullvm 0.53.1",
- "windows_i686_msvc 0.53.1",
- "windows_x86_64_gnu 0.53.1",
- "windows_x86_64_gnullvm 0.53.1",
- "windows_x86_64_msvc 0.53.1",
-]
-
 [[package]]
 name = "windows_aarch64_gnullvm"
 version = "0.42.2"
@@ -7454,12 +7464,6 @@ version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3"
 
-[[package]]
-name = "windows_aarch64_gnullvm"
-version = "0.53.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a9d8416fa8b42f5c947f8482c43e7d89e73a173cead56d044f6a56104a6d1b53"
-
 [[package]]
 name = "windows_aarch64_msvc"
 version = "0.42.2"
@@ -7472,12 +7476,6 @@ version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469"
 
-[[package]]
-name = "windows_aarch64_msvc"
-version = "0.53.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b9d782e804c2f632e395708e99a94275910eb9100b2114651e04744e9b125006"
-
 [[package]]
 name = "windows_i686_gnu"
 version = "0.42.2"
@@ -7490,24 +7488,12 @@ version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b"
 
-[[package]]
-name = "windows_i686_gnu"
-version = "0.53.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "960e6da069d81e09becb0ca57a65220ddff016ff2d6af6a223cf372a506593a3"
-
 [[package]]
 name = "windows_i686_gnullvm"
 version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66"
 
-[[package]]
-name = "windows_i686_gnullvm"
-version = "0.53.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fa7359d10048f68ab8b09fa71c3daccfb0e9b559aed648a8f95469c27057180c"
-
 [[package]]
 name = "windows_i686_msvc"
 version = "0.42.2"
@@ -7520,12 +7506,6 @@ version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66"
 
-[[package]]
-name = "windows_i686_msvc"
-version = "0.53.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1e7ac75179f18232fe9c285163565a57ef8d3c89254a30685b57d83a38d326c2"
-
 [[package]]
 name = "windows_x86_64_gnu"
 version = "0.42.2"
@@ -7538,12 +7518,6 @@ version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78"
 
-[[package]]
-name = "windows_x86_64_gnu"
-version = "0.53.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9c3842cdd74a865a8066ab39c8a7a473c0778a3f29370b5fd6b4b9aa7df4a499"
-
 [[package]]
 name = "windows_x86_64_gnullvm"
 version = "0.42.2"
@@ -7556,12 +7530,6 @@ version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d"
 
-[[package]]
-name = "windows_x86_64_gnullvm"
-version = "0.53.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0ffa179e2d07eee8ad8f57493436566c7cc30ac536a3379fdf008f47f6bb7ae1"
-
 [[package]]
 name = "windows_x86_64_msvc"
 version = "0.42.2"
@@ -7574,12 +7542,6 @@ version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec"
 
-[[package]]
-name = "windows_x86_64_msvc"
-version = "0.53.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650"
-
 [[package]]
 name = "wit-bindgen"
 version = "0.57.1"
@@ -7610,9 +7572,9 @@ checksum = "66fee0b777b0f5ac1c69bb06d361268faafa61cd4682ae064a171c16c433e9e4"
 
 [[package]]
 name = "yoke"
-version = "0.8.3"
+version = "0.8.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "709fe23a0424b6a435d82152b1bd3fdfb0833487d5fa90d05d42762a9891fef5"
+checksum = "abe8c5fda708d9ca3df187cae8bfb9ceda00dd96231bed36e445a1a48e66f9ca"
 dependencies = [
  "stable_deref_trait",
  "yoke-derive",
@@ -7633,18 +7595,18 @@ dependencies = [
 
 [[package]]
 name = "zerocopy"
-version = "0.8.52"
+version = "0.8.48"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ce1022995ff5ff5d841ad7d994facc23098cd40152f2c1d11cd607c6f530653f"
+checksum = "eed437bf9d6692032087e337407a86f04cd8d6a16a37199ed57949d415bd68e9"
 dependencies = [
  "zerocopy-derive",
 ]
 
 [[package]]
 name = "zerocopy-derive"
-version = "0.8.52"
+version = "0.8.48"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1ae7f38b72ec2a254e2b87ef277cf2cd4fb97cbebf944faa6f33354da0867930"
+checksum = "70e3cd084b1788766f53af483dd21f93881ff30d7320490ec3ef7526d203bad4"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -7653,9 +7615,9 @@ dependencies = [
 
 [[package]]
 name = "zerofrom"
-version = "0.1.8"
+version = "0.1.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0ec05a11813ea801ff6d75110ad09cd0824ddba17dfe17128ea0d5f68e6c5272"
+checksum = "69faa1f2a1ea75661980b013019ed6687ed0e83d069bc1114e2cc74c6c04c4df"
 dependencies = [
  "zerofrom-derive",
 ]
@@ -7677,6 +7639,20 @@ name = "zeroize"
 version = "1.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e"
+dependencies = [
+ "zeroize_derive",
+]
+
+[[package]]
+name = "zeroize_derive"
+version = "1.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3c50655cbb0fe3fc43170059e702f1ce5e19b84cec58dc87b037a09935c2f328"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.118",
+]
 
 [[package]]
 name = "zerotrie"
diff --git a/native/Cargo.toml b/native/Cargo.toml
index 3e797eb968..3368b34795 100644
--- a/native/Cargo.toml
+++ b/native/Cargo.toml
@@ -58,10 +58,13 @@ object_store = { version = "0.13.2", features = ["gcp", "azure", "aws", "http"]
 url = "2.2"
 aws-config = "1.8.18"
 aws-credential-types = "1.2.13"
-iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "80a30d3" }
-iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "80a30d3", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] }
+iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "77c5d710" }
+iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "77c5d710", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] }
 reqsign-core = "3"
 
+[patch."https://github.com/apache/iceberg-rust"]
+iceberg = { git = "https://github.com/parthchandra/iceberg-rust", branch = "metadata-columns-test" }
+
 [profile.release]
 debug = true
 overflow-checks = false
diff --git a/native/core/src/execution/operators/iceberg_scan.rs b/native/core/src/execution/operators/iceberg_scan.rs
index e1220d4085..989c899d84 100644
--- a/native/core/src/execution/operators/iceberg_scan.rs
+++ b/native/core/src/execution/operators/iceberg_scan.rs
@@ -651,6 +651,7 @@ mod tests {
             partition: None,
             partition_spec: None,
             name_mapping: None,
+            unified_partition_type: None,
             case_sensitive: false,
         }
     }
diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs
index 25162332fd..893ef9e011 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -3678,6 +3678,41 @@ fn parse_file_scan_tasks_from_common(
         })
         .collect::<Result<Vec<_>, _>>()?;
 
+    // Compute unified partition type from the partition_type_pool.
+    // Each entry is a StructType JSON with the resolved field types for one partition spec.
+    // Merge all specs into a single unified type (dedup by field_id).
+    //
+    // NOTE: The type information here comes from Iceberg Java's PartitionSpec.partitionType()
+    // (via Scala reflection). This is the same type computation as iceberg-rust's
+    // Transform::result_type() -- both implement the Iceberg spec's transform result rules.
+    // When iceberg-rust's own scan planning is used (not Comet's proto path), it computes
+    // this via compute_unified_partition_type(specs, schema) instead.
+    let unified_partition_type = {
+        let mut seen_field_ids = std::collections::HashSet::new();
+        let mut struct_fields: Vec<iceberg::spec::NestedFieldRef> = Vec::new();
+
+        for type_json in &proto_common.partition_type_pool {
+            match serde_json::from_str::<iceberg::spec::StructType>(type_json) {
+                Ok(struct_type) => {
+                    for field in struct_type.fields() {
+                        if seen_field_ids.insert(field.id) {
+                            struct_fields.push(Arc::clone(field));
+                        }
+                    }
+                }
+                Err(e) => {
+                    return Err(ExecutionError::GeneralError(format!(
+                        "Failed to deserialize partition type JSON from pool: {e}"
+                    )));
+                }
+            }
+        }
+
+        iceberg::spec::StructType::new(struct_fields)
+    };
+
+    let unified_partition_type_arc = Arc::new(unified_partition_type);
+
     let results: Result<Vec<_>, _> = proto_tasks
         .iter()
         .map(|proto_task| {
@@ -3778,6 +3813,14 @@ fn parse_file_scan_tasks_from_common(
                 .field_ids
                 .clone();
 
+            let unified_partition_type_for_task = if project_field_ids
+                .contains(&iceberg::metadata_columns::RESERVED_FIELD_ID_PARTITION)
+            {
+                Some(Arc::clone(&unified_partition_type_arc))
+            } else {
+                None
+            };
+
             Ok(iceberg::scan::FileScanTask {
                 file_size_in_bytes: proto_task.file_size_in_bytes,
                 data_file_path: proto_task.data_file_path.clone(),
@@ -3792,6 +3835,7 @@ fn parse_file_scan_tasks_from_common(
                 partition,
                 partition_spec,
                 name_mapping,
+                unified_partition_type: unified_partition_type_for_task,
                 case_sensitive: false,
             })
         })
@@ -5408,4 +5452,21 @@ mod tests {
             }
         });
     }
+
+    #[test]
+    fn test_metadata_field_id_constants_match_iceberg_rust() {
+        // These constants are duplicated in Scala (CometIcebergNativeScan.MetadataFieldIds)
+        // as Int.MaxValue - 1 and Int.MaxValue - 5. This test ensures the Rust constants
+        // haven't drifted, which would cause a silent mismatch with the Scala side.
+        assert_eq!(
+            iceberg::metadata_columns::RESERVED_FIELD_ID_FILE,
+            i32::MAX - 1,
+            "RESERVED_FIELD_ID_FILE must be i32::MAX - 1 to match Scala MetadataFieldIds"
+        );
+        assert_eq!(
+            iceberg::metadata_columns::RESERVED_FIELD_ID_PARTITION,
+            i32::MAX - 5,
+            "RESERVED_FIELD_ID_PARTITION must be i32::MAX - 5 to match Scala MetadataFieldIds"
+        );
+    }
 }
diff --git a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
index bc02e22da9..253b72afcf 100644
--- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
+++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
@@ -125,7 +125,7 @@ case class CometScanRule(session: SparkSession)
       case scan if !CometConf.COMET_NATIVE_SCAN_ENABLED.get(conf) =>
         withFallbackReason(scan, "Comet Scan is not enabled")
 
-      case scan if metadataCols(scan).nonEmpty =>
+      case scan: FileSourceScanExec if metadataCols(scan).nonEmpty =>
         withFallbackReason(
           scan,
           s"Metadata column(s) ${metadataCols(scan).mkString(", ")} is not supported")
@@ -297,6 +297,11 @@ case class CometScanRule(session: SparkSession)
 
     scanExec.scan match {
       case scan: CSVScan if COMET_CSV_V2_NATIVE_ENABLED.get() =>
+        if (scanExec.output.exists(_.isMetadataCol)) {
+          return withFallbackReason(
+            scanExec,
+            "Metadata columns are not supported for CSV V2 scans")
+        }
         val fallbackReasons = new ListBuffer[String]()
         val schemaSupported =
           CometBatchScanExec.isSchemaSupported(scan.readDataSchema, fallbackReasons)
@@ -356,9 +361,30 @@ case class CometScanRule(session: SparkSession)
           return withFallbackReasons(scanExec, fallbackReasons.toSet)
         }
 
+        // Check for unsupported metadata columns in Iceberg scans
+        val unsupportedMetadataCols = scanExec.output.filter(_.isMetadataCol).filterNot { attr =>
+          CometIcebergNativeScan.MetadataFieldIds.keySet.contains(attr.name)
+        }
+        if (unsupportedMetadataCols.nonEmpty) {
+          fallbackReasons += "Unsupported Iceberg metadata columns: " +
+            unsupportedMetadataCols.map(_.name).mkString(", ")
+          return withFallbackReasons(scanExec, fallbackReasons.toSet)
+        }
+
         val typeChecker = CometScanTypeChecker()
+        // Filter out metadata columns from schema check -- their types are handled
+        // by iceberg-rust directly (e.g., _partition can be an empty struct for
+        // unpartitioned tables which the general type checker rejects).
+        val metadataColNames = scanExec.output.filter(_.isMetadataCol).map(_.name).toSet
+        val dataSchema = if (metadataColNames.nonEmpty) {
+          val filtered =
+            scanExec.scan.readSchema().filter(f => !metadataColNames.contains(f.name))
+          new org.apache.spark.sql.types.StructType(filtered.toArray)
+        } else {
+          scanExec.scan.readSchema()
+        }
         val schemaSupported =
-          typeChecker.isSchemaSupported(scanExec.scan.readSchema(), fallbackReasons)
+          typeChecker.isSchemaSupported(dataSchema, fallbackReasons)
 
         if (!schemaSupported) {
           fallbackReasons += "Comet extension is not enabled for " +
@@ -446,15 +472,18 @@ case class CometScanRule(session: SparkSession)
         }
 
         // Now perform all validation using the pre-extracted metadata
-        // Check if table uses a FileIO implementation compatible with iceberg-rust
-
+        // Check if table uses a FileIO implementation compatible with iceberg-rust.
+        // Comet's native reader uses object_store (Rust) for I/O, bypassing Iceberg Java's
+        // FileIO entirely. Only allow known-compatible implementations whose underlying
+        // storage object_store can reach via standard URL schemes.
         val fileIOCompatible = IcebergReflection.getFileIO(metadata.table) match {
           case Some(fileIO)
-              if fileIO.getClass.getName == "org.apache.iceberg.inmemory.InMemoryFileIO" =>
-            fallbackReasons += "InMemoryFileIO is not supported by Comet's native reader"
-            false
-          case Some(_) =>
+              if CometScanRule.CompatibleFileIOClasses.contains(fileIO.getClass.getName) =>
             true
+          case Some(fileIO) =>
+            fallbackReasons += s"FileIO ${fileIO.getClass.getName} is not supported by " +
+              "Comet's native reader (object_store bypasses Iceberg Java FileIO)"
+            false
           case None =>
             fallbackReasons += "Could not check FileIO compatibility"
             false
@@ -772,6 +801,18 @@ case class CometScanTypeChecker() extends DataTypeSupport with CometTypeShim {
 
 object CometScanRule extends Logging {
 
+  // Iceberg FileIO implementations whose underlying storage object_store can reach.
+  // Custom/test FileIO classes (e.g. CustomFileIO in TestSparkExecutorCache) are not compatible
+  // because Comet's native reader bypasses Java FileIO entirely.
+  private val CompatibleFileIOClasses: Set[String] = Set(
+    "org.apache.iceberg.hadoop.HadoopFileIO",
+    "org.apache.iceberg.aws.s3.S3FileIO",
+    "org.apache.iceberg.gcp.gcs.GCSFileIO",
+    "org.apache.iceberg.io.ResolvingFileIO",
+    "org.apache.iceberg.spark.SparkFileIO",
+    "org.apache.iceberg.azure.adlsv2.ADLSFileIO",
+    "org.apache.iceberg.CachingFileIO")
+
   // Per-scheme memo of `NativeBase.isObjectStoreSchemeSupported`. The answer depends only on the
   // URL scheme, so we cache by scheme and never re-cross the JNI boundary for a repeated scheme.
   private val schemeSupportCache =
diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
index b8afc159bb..11987d72c9 100644
--- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
@@ -71,6 +71,14 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
     }
   }
 
+  // Iceberg reserved field IDs for metadata columns.
+  // These must match iceberg-rust's constants in crates/iceberg/src/metadata_columns.rs:
+  //   RESERVED_FIELD_ID_FILE      = i32::MAX - 1  (2147483646)
+  //   RESERVED_FIELD_ID_PARTITION = i32::MAX - 5  (2147483642)
+  // Scala's Int.MaxValue == 2^31 - 1 == Rust's i32::MAX.
+  val MetadataFieldIds: Map[String, Int] =
+    Map("_file" -> (Int.MaxValue - 1), "_partition" -> (Int.MaxValue - 5))
+
   /**
    * Converts an Iceberg partition value to protobuf format. Protobuf is less verbose than JSON.
    * The following types are also serialized as integer values instead of as strings - Timestamps,
@@ -866,14 +874,16 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
 
                 val nameToFieldId = IcebergReflection.buildFieldIdMapping(schema)
 
-                val projectFieldIds = output.flatMap { attr =>
+                val projectFieldIds = output.map { attr =>
                   nameToFieldId
                     .get(attr.name)
                     .orElse(metadata.globalFieldIdMapping.get(attr.name))
-                    .orElse {
-                      logWarning(s"Column '${attr.name}' not found in task or scan schema, " +
-                        "skipping projection")
-                      None
+                    .orElse(CometIcebergNativeScan.MetadataFieldIds.get(attr.name))
+                    .getOrElse {
+                      throw new IllegalStateException(
+                        s"Column '${attr.name}' not found in task schema, global schema, " +
+                          "or metadata field IDs. This indicates a bug in CometScanRule " +
+                          "validation -- all output columns should be resolvable.")
                     }
                 }
 
diff --git a/spark/src/test/resources/sql-tests/iceberg/metadata_column_partition.sql b/spark/src/test/resources/sql-tests/iceberg/metadata_column_partition.sql
new file mode 100644
index 0000000000..0efde9888a
--- /dev/null
+++ b/spark/src/test/resources/sql-tests/iceberg/metadata_column_partition.sql
@@ -0,0 +1,850 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements.  See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership.  The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"); you may not use this file except in compliance
+-- with the License.  You may obtain a copy of the License at
+--
+--   http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied.  See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- Config: spark.sql.catalog.test_cat=org.apache.iceberg.spark.SparkCatalog
+-- Config: spark.sql.catalog.test_cat.type=hadoop
+-- Config: spark.sql.catalog.test_cat.warehouse=/tmp/comet-iceberg-sql-test
+-- Config: spark.comet.enabled=true
+-- Config: spark.comet.exec.enabled=true
+-- Config: spark.comet.iceberg.native.enabled=true
+
+-- All `query` assertions below implicitly validate that Comet's partition type
+-- computation (from partition_type_pool JSON) matches Spark's (from Iceberg Java's
+-- PartitionSpec.partitionType()). If the two diverged, checkSparkAnswerAndOperator
+-- would detect differing struct values. This covers the "dual computation path"
+-- cross-validation concern.
+
+-- =============================================================
+-- Setup: Create table with bucket partitioning (format v2 for MoR)
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.meta_test
+
+statement
+CREATE TABLE test_cat.db.meta_test (id INT, name STRING, value DOUBLE) USING iceberg PARTITIONED BY (bucket(4, id)) TBLPROPERTIES ('format-version' = '2')
+
+statement
+INSERT INTO test_cat.db.meta_test VALUES (1, 'alice', 10.0), (2, 'bob', 20.0), (3, 'charlie', 30.0)
+
+statement
+INSERT INTO test_cat.db.meta_test VALUES (4, 'dave', 40.0), (5, 'eve', 50.0)
+
+-- =============================================================
+-- _file: basic projection (native)
+-- =============================================================
+
+query
+SELECT id, _file FROM test_cat.db.meta_test ORDER BY id
+
+-- Multiple inserts produce multiple files
+query
+SELECT COUNT(DISTINCT _file) > 1 FROM test_cat.db.meta_test
+
+-- _file in a WHERE clause exercises filter pushdown interaction
+query
+SELECT id, name FROM test_cat.db.meta_test WHERE _file LIKE '%.parquet' ORDER BY id
+
+-- =============================================================
+-- _partition: basic struct projection (native)
+-- =============================================================
+
+query
+SELECT id, _partition FROM test_cat.db.meta_test ORDER BY id
+
+-- Access individual partition fields within the struct
+query
+SELECT id, _partition.id_bucket FROM test_cat.db.meta_test ORDER BY id
+
+-- =============================================================
+-- _spec_id: falls back until implemented.
+-- When _spec_id is implemented, change these to plain `query`.
+-- =============================================================
+
+query expect_fallback(Unsupported Iceberg metadata columns)
+SELECT id, _spec_id FROM test_cat.db.meta_test ORDER BY id
+
+-- All rows should have spec_id = 0 (initial spec)
+query expect_fallback(Unsupported Iceberg metadata columns)
+SELECT DISTINCT _spec_id FROM test_cat.db.meta_test
+
+-- =============================================================
+-- _pos: falls back until implemented.
+-- When _pos is implemented, change these to plain `query`.
+-- =============================================================
+
+-- Basic: positions should be 0-based within each file
+query expect_fallback(Unsupported Iceberg metadata columns)
+SELECT id, _file, _pos FROM test_cat.db.meta_test ORDER BY _file, _pos
+
+-- Positions should be contiguous 0..N-1 per file
+query expect_fallback(Unsupported Iceberg metadata columns)
+SELECT _file, MIN(_pos) as min_pos, MAX(_pos) as max_pos, COUNT(*) as cnt FROM test_cat.db.meta_test GROUP BY _file
+
+-- =============================================================
+-- Partition evolution requires IcebergSparkSessionExtensions
+-- (ALTER TABLE ADD PARTITION FIELD / CALL system.add_partition_field).
+-- The SQL file test framework cannot register session extensions via
+-- Config directives. Partition evolution is covered in
+-- CometIcebergNativeSuite which has full session control.
+-- =============================================================
+
+-- =============================================================
+-- Combined: all metadata columns together (falls back due to _pos/_spec_id)
+-- =============================================================
+
+query expect_fallback(Unsupported Iceberg metadata columns)
+SELECT id, _file, _pos, _spec_id, _partition FROM test_cat.db.meta_test ORDER BY _file, _pos
+
+-- Row-level operation simulation: what CoW/MoR would project
+query expect_fallback(Unsupported Iceberg metadata columns)
+SELECT _file, _pos, _spec_id, _partition, id, name FROM test_cat.db.meta_test WHERE id > 3 ORDER BY _file, _pos
+
+-- =============================================================
+-- _file + _partition combined (native -- both supported)
+-- =============================================================
+
+query
+SELECT id, _file, _partition FROM test_cat.db.meta_test ORDER BY id
+
+statement
+DROP TABLE test_cat.db.meta_test
+
+-- =============================================================
+-- Identity partition transform
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.identity_part
+
+statement
+CREATE TABLE test_cat.db.identity_part (id INT, name STRING, dept STRING) USING iceberg PARTITIONED BY (dept)
+
+statement
+INSERT INTO test_cat.db.identity_part VALUES (1, 'Alice', 'eng'), (2, 'Bob', 'sales'), (3, 'Charlie', 'eng')
+
+query
+SELECT id, _partition FROM test_cat.db.identity_part ORDER BY id
+
+query
+SELECT id, _partition.dept FROM test_cat.db.identity_part ORDER BY id
+
+query
+SELECT id, _file, _partition FROM test_cat.db.identity_part ORDER BY id
+
+statement
+DROP TABLE test_cat.db.identity_part
+
+-- =============================================================
+-- Multiple partition fields
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.multi_part
+
+statement
+CREATE TABLE test_cat.db.multi_part (id INT, year INT, month INT, data STRING) USING iceberg PARTITIONED BY (year, month)
+
+statement
+INSERT INTO test_cat.db.multi_part VALUES (1, 2023, 6, 'a'), (2, 2023, 7, 'b'), (3, 2024, 1, 'c')
+
+query
+SELECT id, _partition FROM test_cat.db.multi_part ORDER BY id
+
+statement
+DROP TABLE test_cat.db.multi_part
+
+-- =============================================================
+-- Partition with filter
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.filter_part
+
+statement
+CREATE TABLE test_cat.db.filter_part (id INT, dept STRING, value DOUBLE) USING iceberg PARTITIONED BY (dept)
+
+statement
+INSERT INTO test_cat.db.filter_part VALUES (1, 'eng', 10.5), (2, 'sales', 20.3), (3, 'eng', 30.7)
+
+query
+SELECT id, _partition FROM test_cat.db.filter_part WHERE dept = 'eng' ORDER BY id
+
+statement
+DROP TABLE test_cat.db.filter_part
+
+-- =============================================================
+-- Edge case: Unpartitioned table
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.unpart
+
+statement
+CREATE TABLE test_cat.db.unpart (id INT, val STRING) USING iceberg
+
+statement
+INSERT INTO test_cat.db.unpart VALUES (1, 'a'), (2, 'b')
+
+-- _partition on unpartitioned table is an empty struct (null), which Comet's
+-- shuffle doesn't support. Verify correctness only, not native operator coverage.
+query spark_answer_only
+SELECT id, _partition FROM test_cat.db.unpart ORDER BY id
+
+-- Without ORDER BY (no shuffle), the native scan executes end-to-end
+query
+SELECT id, _partition FROM test_cat.db.unpart
+
+query
+SELECT id, _file FROM test_cat.db.unpart ORDER BY id
+
+-- _spec_id on unpartitioned: should be 0
+query expect_fallback(Unsupported Iceberg metadata columns)
+SELECT id, _spec_id, _partition FROM test_cat.db.unpart ORDER BY id
+
+statement
+DROP TABLE test_cat.db.unpart
+
+-- =============================================================
+-- Edge case: Large batch (tests multi-batch _pos correctness)
+-- Falls back due to _pos, but verifies correctness via Spark parity.
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.large_pos
+
+statement
+CREATE TABLE test_cat.db.large_pos (id INT) USING iceberg
+
+statement
+INSERT INTO test_cat.db.large_pos SELECT id FROM range(10000)
+
+query expect_fallback(Unsupported Iceberg metadata columns)
+SELECT id, _pos FROM test_cat.db.large_pos ORDER BY id LIMIT 5
+
+statement
+DROP TABLE test_cat.db.large_pos
+
+-- =============================================================
+-- Edge case: Multiple row groups (forces _pos offset tracking)
+-- Falls back due to _pos, but verifies correctness via Spark parity.
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.multi_rg
+
+statement
+CREATE TABLE test_cat.db.multi_rg (id INT, data STRING) USING iceberg TBLPROPERTIES ('write.parquet.row-group-size-bytes' = '1024')
+
+statement
+INSERT INTO test_cat.db.multi_rg SELECT id, REPEAT('x', 100) FROM range(1000)
+
+query expect_fallback(Unsupported Iceberg metadata columns)
+SELECT _pos, id FROM test_cat.db.multi_rg WHERE _pos IN (0, 1, 998, 999) ORDER BY _pos
+
+statement
+DROP TABLE test_cat.db.multi_rg
+
+-- =============================================================
+-- DML operations (DELETE, UPDATE, MERGE) internally project metadata columns
+-- (_file, _pos, _spec_id, _partition) to identify rows. Since we don't yet
+-- support all four, the DML scan falls back to Spark's JVM reader. The
+-- subsequent SELECT queries verify that reads after DML return correct results
+-- through the native scan path (applying position deletes, seeing updated data).
+-- Once all metadata columns are implemented, Spark will use the native scan for
+-- the DML read side too (no code change needed here -- the fallback guard lifts
+-- automatically when supportedIcebergMetadataCols covers all projected columns).
+-- =============================================================
+
+-- Copy-on-write DELETE
+statement
+DROP TABLE IF EXISTS test_cat.db.cow_delete
+
+statement
+CREATE TABLE test_cat.db.cow_delete (id INT, name STRING, dept STRING) USING iceberg PARTITIONED BY (dept) TBLPROPERTIES ('write.delete.mode'='copy-on-write')
+
+statement
+INSERT INTO test_cat.db.cow_delete VALUES (1, 'Alice', 'eng'), (2, 'Bob', 'sales'), (3, 'Charlie', 'eng')
+
+statement
+DELETE FROM test_cat.db.cow_delete WHERE id = 2
+
+query
+SELECT id, _partition FROM test_cat.db.cow_delete ORDER BY id
+
+statement
+DROP TABLE test_cat.db.cow_delete
+
+-- Merge-on-read DELETE (native read applies position deletes)
+statement
+DROP TABLE IF EXISTS test_cat.db.mor_delete
+
+statement
+CREATE TABLE test_cat.db.mor_delete (id INT, name STRING, dept STRING) USING iceberg PARTITIONED BY (dept) TBLPROPERTIES ('write.delete.mode'='merge-on-read')
+
+statement
+INSERT INTO test_cat.db.mor_delete VALUES (1, 'Alice', 'eng'), (2, 'Bob', 'sales'), (3, 'Charlie', 'eng')
+
+statement
+DELETE FROM test_cat.db.mor_delete WHERE id = 2
+
+query
+SELECT id, _partition FROM test_cat.db.mor_delete ORDER BY id
+
+-- After MoR delete, _pos still references original file positions (falls back)
+query expect_fallback(Unsupported Iceberg metadata columns)
+SELECT id, _pos FROM test_cat.db.mor_delete ORDER BY id
+
+statement
+DROP TABLE test_cat.db.mor_delete
+
+-- NOTE: Merge-on-read UPDATE is not tested here because UPDATE TABLE is not
+-- supported in Spark 4.0. Covered in CometIcebergNativeSuite instead.
+
+-- =============================================================
+-- Temporal transforms: years() on DATE column
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_years_date
+
+statement
+CREATE TABLE test_cat.db.part_years_date (id INT, event_date DATE, data STRING) USING iceberg PARTITIONED BY (years(event_date))
+
+statement
+INSERT INTO test_cat.db.part_years_date VALUES (1, DATE '2022-03-15', 'a'), (2, DATE '2023-07-20', 'b'), (3, DATE '2022-11-01', 'c')
+
+query
+SELECT id, _partition FROM test_cat.db.part_years_date ORDER BY id
+
+query
+SELECT id, _partition.event_date_year FROM test_cat.db.part_years_date ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_years_date
+
+-- =============================================================
+-- Temporal transforms: months() on DATE column
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_months_date
+
+statement
+CREATE TABLE test_cat.db.part_months_date (id INT, event_date DATE, data STRING) USING iceberg PARTITIONED BY (months(event_date))
+
+statement
+INSERT INTO test_cat.db.part_months_date VALUES (1, DATE '2023-01-10', 'jan'), (2, DATE '2023-02-15', 'feb'), (3, DATE '2023-01-25', 'jan2')
+
+query
+SELECT id, _partition FROM test_cat.db.part_months_date ORDER BY id
+
+query
+SELECT id, _partition.event_date_month FROM test_cat.db.part_months_date ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_months_date
+
+-- =============================================================
+-- Temporal transforms: days() on DATE column
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_days_date
+
+statement
+CREATE TABLE test_cat.db.part_days_date (id INT, event_date DATE, data STRING) USING iceberg PARTITIONED BY (days(event_date))
+
+statement
+INSERT INTO test_cat.db.part_days_date VALUES (1, DATE '2023-06-01', 'x'), (2, DATE '2023-06-02', 'y'), (3, DATE '2023-06-01', 'z')
+
+query
+SELECT id, _partition FROM test_cat.db.part_days_date ORDER BY id
+
+query
+SELECT id, _partition.event_date_day FROM test_cat.db.part_days_date ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_days_date
+
+-- =============================================================
+-- Temporal transforms: hours() on TIMESTAMP column
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_hours_ts
+
+statement
+CREATE TABLE test_cat.db.part_hours_ts (id INT, ts TIMESTAMP, data STRING) USING iceberg PARTITIONED BY (hours(ts))
+
+statement
+INSERT INTO test_cat.db.part_hours_ts VALUES (1, TIMESTAMP '2023-06-15 10:30:00', 'morning'), (2, TIMESTAMP '2023-06-15 14:45:00', 'afternoon'), (3, TIMESTAMP '2023-06-15 10:55:00', 'morning2')
+
+query
+SELECT id, _partition FROM test_cat.db.part_hours_ts ORDER BY id
+
+query
+SELECT id, _partition.ts_hour FROM test_cat.db.part_hours_ts ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_hours_ts
+
+-- =============================================================
+-- Truncate transform: truncate(N, int_col)
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_trunc_int
+
+statement
+CREATE TABLE test_cat.db.part_trunc_int (id INT, value INT, data STRING) USING iceberg PARTITIONED BY (truncate(10, value))
+
+statement
+INSERT INTO test_cat.db.part_trunc_int VALUES (1, 5, 'a'), (2, 15, 'b'), (3, 7, 'c'), (4, 23, 'd')
+
+query
+SELECT id, _partition FROM test_cat.db.part_trunc_int ORDER BY id
+
+query
+SELECT id, _partition.value_trunc FROM test_cat.db.part_trunc_int ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_trunc_int
+
+-- =============================================================
+-- Truncate transform: truncate(N, string_col)
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_trunc_str
+
+statement
+CREATE TABLE test_cat.db.part_trunc_str (id INT, city STRING, data STRING) USING iceberg PARTITIONED BY (truncate(3, city))
+
+statement
+INSERT INTO test_cat.db.part_trunc_str VALUES (1, 'San Francisco', 'x'), (2, 'Santa Cruz', 'y'), (3, 'Boston', 'z')
+
+query
+SELECT id, _partition FROM test_cat.db.part_trunc_str ORDER BY id
+
+query
+SELECT id, _partition.city_trunc FROM test_cat.db.part_trunc_str ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_trunc_str
+
+-- =============================================================
+-- Multiple non-identity transforms on same table
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_multi_transform
+
+statement
+CREATE TABLE test_cat.db.part_multi_transform (id INT, name STRING, ts TIMESTAMP) USING iceberg PARTITIONED BY (bucket(4, id), truncate(2, name), hours(ts))
+
+statement
+INSERT INTO test_cat.db.part_multi_transform VALUES (1, 'Alice', TIMESTAMP '2023-06-15 10:00:00'), (2, 'Bob', TIMESTAMP '2023-06-15 11:00:00'), (3, 'Alice', TIMESTAMP '2023-06-15 10:30:00')
+
+query
+SELECT id, _partition FROM test_cat.db.part_multi_transform ORDER BY id
+
+query
+SELECT id, _partition.id_bucket, _partition.name_trunc, _partition.ts_hour FROM test_cat.db.part_multi_transform ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_multi_transform
+
+-- =============================================================
+-- NULL partition values
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_nulls
+
+statement
+CREATE TABLE test_cat.db.part_nulls (id INT, dept STRING, value DOUBLE) USING iceberg PARTITIONED BY (dept)
+
+statement
+INSERT INTO test_cat.db.part_nulls VALUES (1, 'eng', 10.0), (2, NULL, 20.0), (3, 'sales', 30.0), (4, NULL, 40.0)
+
+query
+SELECT id, _partition FROM test_cat.db.part_nulls ORDER BY id
+
+query
+SELECT id, _partition.dept FROM test_cat.db.part_nulls ORDER BY id
+
+query
+SELECT id, _partition FROM test_cat.db.part_nulls WHERE _partition.dept IS NULL ORDER BY id
+
+query
+SELECT id, _partition FROM test_cat.db.part_nulls WHERE _partition.dept IS NOT NULL ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_nulls
+
+-- =============================================================
+-- _partition in WHERE clause (filter on metadata column itself)
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_where
+
+statement
+CREATE TABLE test_cat.db.part_where (id INT, dept STRING, value DOUBLE) USING iceberg PARTITIONED BY (dept)
+
+statement
+INSERT INTO test_cat.db.part_where VALUES (1, 'eng', 10.0), (2, 'sales', 20.0), (3, 'eng', 30.0), (4, 'hr', 40.0)
+
+query
+SELECT id, value FROM test_cat.db.part_where WHERE _partition.dept = 'eng' ORDER BY id
+
+query
+SELECT id, _partition FROM test_cat.db.part_where WHERE _partition.dept IN ('eng', 'hr') ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_where
+
+-- =============================================================
+-- _partition in GROUP BY
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_groupby
+
+statement
+CREATE TABLE test_cat.db.part_groupby (id INT, dept STRING, salary DOUBLE) USING iceberg PARTITIONED BY (dept)
+
+statement
+INSERT INTO test_cat.db.part_groupby VALUES (1, 'eng', 100.0), (2, 'eng', 120.0), (3, 'sales', 80.0), (4, 'sales', 90.0), (5, 'hr', 70.0)
+
+query
+SELECT _partition.dept, COUNT(*) as cnt, SUM(salary) as total FROM test_cat.db.part_groupby GROUP BY _partition.dept ORDER BY dept
+
+-- ORDER BY on _partition (StructType) falls back because Comet does not support Sort on structs
+query spark_answer_only
+SELECT _partition, COUNT(*) as cnt FROM test_cat.db.part_groupby GROUP BY _partition ORDER BY _partition
+
+statement
+DROP TABLE test_cat.db.part_groupby
+
+-- =============================================================
+-- _partition in JOIN condition
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_join_left
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_join_right
+
+statement
+CREATE TABLE test_cat.db.part_join_left (id INT, dept STRING, value INT) USING iceberg PARTITIONED BY (dept)
+
+statement
+CREATE TABLE test_cat.db.part_join_right (id INT, dept STRING, budget INT) USING iceberg PARTITIONED BY (dept)
+
+statement
+INSERT INTO test_cat.db.part_join_left VALUES (1, 'eng', 10), (2, 'sales', 20), (3, 'eng', 30)
+
+statement
+INSERT INTO test_cat.db.part_join_right VALUES (100, 'eng', 500), (200, 'sales', 300)
+
+query
+SELECT l.id, l.value, r.budget FROM test_cat.db.part_join_left l JOIN test_cat.db.part_join_right r ON l._partition.dept = r._partition.dept ORDER BY l.id
+
+statement
+DROP TABLE test_cat.db.part_join_left
+
+statement
+DROP TABLE test_cat.db.part_join_right
+
+-- =============================================================
+-- _partition in subquery
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_subq
+
+statement
+CREATE TABLE test_cat.db.part_subq (id INT, dept STRING, value INT) USING iceberg PARTITIONED BY (dept)
+
+statement
+INSERT INTO test_cat.db.part_subq VALUES (1, 'eng', 10), (2, 'sales', 20), (3, 'hr', 30), (4, 'eng', 40)
+
+query
+SELECT id, value FROM test_cat.db.part_subq WHERE _partition.dept IN (SELECT _partition.dept FROM test_cat.db.part_subq WHERE value > 25) ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_subq
+
+-- =============================================================
+-- _partition with HAVING clause
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_having
+
+statement
+CREATE TABLE test_cat.db.part_having (id INT, dept STRING, salary DOUBLE) USING iceberg PARTITIONED BY (dept)
+
+statement
+INSERT INTO test_cat.db.part_having VALUES (1, 'eng', 100.0), (2, 'eng', 120.0), (3, 'eng', 110.0), (4, 'sales', 80.0), (5, 'hr', 70.0)
+
+query
+SELECT _partition.dept, COUNT(*) as cnt FROM test_cat.db.part_having GROUP BY _partition.dept HAVING COUNT(*) > 1 ORDER BY dept
+
+statement
+DROP TABLE test_cat.db.part_having
+
+-- =============================================================
+-- _partition with ORDER BY on partition field
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_orderby
+
+statement
+CREATE TABLE test_cat.db.part_orderby (id INT, dept STRING, priority INT) USING iceberg PARTITIONED BY (dept)
+
+statement
+INSERT INTO test_cat.db.part_orderby VALUES (1, 'eng', 3), (2, 'sales', 1), (3, 'hr', 2), (4, 'eng', 4)
+
+query
+SELECT id, _partition.dept FROM test_cat.db.part_orderby ORDER BY _partition.dept, id
+
+statement
+DROP TABLE test_cat.db.part_orderby
+
+-- =============================================================
+-- _partition with DISTINCT
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_distinct
+
+statement
+CREATE TABLE test_cat.db.part_distinct (id INT, region STRING, data STRING) USING iceberg PARTITIONED BY (region)
+
+statement
+INSERT INTO test_cat.db.part_distinct VALUES (1, 'US', 'a'), (2, 'EU', 'b'), (3, 'US', 'c'), (4, 'EU', 'd'), (5, 'APAC', 'e')
+
+query spark_answer_only
+SELECT DISTINCT _partition.region FROM test_cat.db.part_distinct ORDER BY region
+
+query spark_answer_only
+SELECT DISTINCT _partition FROM test_cat.db.part_distinct ORDER BY _partition.region
+
+statement
+DROP TABLE test_cat.db.part_distinct
+
+-- =============================================================
+-- Wide partition struct (many partition fields)
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_wide
+
+statement
+CREATE TABLE test_cat.db.part_wide (id INT, a INT, b INT, c INT, d STRING, e STRING) USING iceberg PARTITIONED BY (a, b, c, d, e)
+
+statement
+INSERT INTO test_cat.db.part_wide VALUES (1, 10, 20, 30, 'x', 'y'), (2, 11, 21, 31, 'p', 'q'), (3, 10, 20, 30, 'x', 'y')
+
+query
+SELECT id, _partition FROM test_cat.db.part_wide ORDER BY id
+
+query
+SELECT id, _partition.a, _partition.b, _partition.c, _partition.d, _partition.e FROM test_cat.db.part_wide ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_wide
+
+-- =============================================================
+-- DATE identity partition
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_date_identity
+
+statement
+CREATE TABLE test_cat.db.part_date_identity (id INT, event_date DATE, data STRING) USING iceberg PARTITIONED BY (event_date)
+
+statement
+INSERT INTO test_cat.db.part_date_identity VALUES (1, DATE '2023-01-15', 'a'), (2, DATE '2023-02-20', 'b'), (3, DATE '2023-01-15', 'c')
+
+query
+SELECT id, _partition FROM test_cat.db.part_date_identity ORDER BY id
+
+query
+SELECT id, _partition.event_date FROM test_cat.db.part_date_identity ORDER BY id
+
+query
+SELECT id, data FROM test_cat.db.part_date_identity WHERE _partition.event_date = DATE '2023-01-15' ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_date_identity
+
+-- =============================================================
+-- String identity with special characters
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_special_chars
+
+statement
+CREATE TABLE test_cat.db.part_special_chars (id INT, tag STRING, data STRING) USING iceberg PARTITIONED BY (tag)
+
+statement
+INSERT INTO test_cat.db.part_special_chars VALUES (1, 'hello world', 'a'), (2, 'foo/bar', 'b'), (3, '', 'c'), (4, 'a=b&c=d', 'd')
+
+query
+SELECT id, _partition FROM test_cat.db.part_special_chars ORDER BY id
+
+query
+SELECT id, _partition.tag FROM test_cat.db.part_special_chars ORDER BY id
+
+query
+SELECT id FROM test_cat.db.part_special_chars WHERE _partition.tag = 'hello world' ORDER BY id
+
+query
+SELECT id FROM test_cat.db.part_special_chars WHERE _partition.tag = '' ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_special_chars
+
+-- =============================================================
+-- Bucket on STRING column
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_bucket_str
+
+statement
+CREATE TABLE test_cat.db.part_bucket_str (id INT, name STRING, value INT) USING iceberg PARTITIONED BY (bucket(8, name))
+
+statement
+INSERT INTO test_cat.db.part_bucket_str VALUES (1, 'Alice', 10), (2, 'Bob', 20), (3, 'Charlie', 30), (4, 'Alice', 40)
+
+query
+SELECT id, _partition FROM test_cat.db.part_bucket_str ORDER BY id
+
+query
+SELECT id, _partition.name_bucket FROM test_cat.db.part_bucket_str ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_bucket_str
+
+-- =============================================================
+-- years() on TIMESTAMP column
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_years_ts
+
+statement
+CREATE TABLE test_cat.db.part_years_ts (id INT, ts TIMESTAMP, data STRING) USING iceberg PARTITIONED BY (years(ts))
+
+statement
+INSERT INTO test_cat.db.part_years_ts VALUES (1, TIMESTAMP '2021-03-15 08:00:00', 'a'), (2, TIMESTAMP '2022-07-20 12:00:00', 'b'), (3, TIMESTAMP '2021-11-01 16:00:00', 'c')
+
+query
+SELECT id, _partition FROM test_cat.db.part_years_ts ORDER BY id
+
+query
+SELECT id, _partition.ts_year FROM test_cat.db.part_years_ts ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_years_ts
+
+-- =============================================================
+-- months() on TIMESTAMP column
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_months_ts
+
+statement
+CREATE TABLE test_cat.db.part_months_ts (id INT, ts TIMESTAMP, data STRING) USING iceberg PARTITIONED BY (months(ts))
+
+statement
+INSERT INTO test_cat.db.part_months_ts VALUES (1, TIMESTAMP '2023-01-10 09:00:00', 'jan'), (2, TIMESTAMP '2023-02-15 10:00:00', 'feb'), (3, TIMESTAMP '2023-01-25 11:00:00', 'jan2')
+
+query
+SELECT id, _partition FROM test_cat.db.part_months_ts ORDER BY id
+
+query
+SELECT id, _partition.ts_month FROM test_cat.db.part_months_ts ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_months_ts
+
+-- =============================================================
+-- Combined: temporal transform + identity on same table
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_mixed
+
+statement
+CREATE TABLE test_cat.db.part_mixed (id INT, event_date DATE, region STRING, data STRING) USING iceberg PARTITIONED BY (years(event_date), region)
+
+statement
+INSERT INTO test_cat.db.part_mixed VALUES (1, DATE '2022-03-01', 'US', 'a'), (2, DATE '2023-07-15', 'EU', 'b'), (3, DATE '2022-09-20', 'US', 'c'), (4, DATE '2023-01-10', 'US', 'd')
+
+query
+SELECT id, _partition FROM test_cat.db.part_mixed ORDER BY id
+
+query
+SELECT id, _partition.event_date_year, _partition.region FROM test_cat.db.part_mixed ORDER BY id
+
+query
+SELECT _partition.event_date_year, COUNT(*) as cnt FROM test_cat.db.part_mixed GROUP BY _partition.event_date_year ORDER BY event_date_year
+
+query
+SELECT id, data FROM test_cat.db.part_mixed WHERE _partition.region = 'US' ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_mixed
+
+-- =============================================================
+-- _partition after multiple inserts (different files, same partition)
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_multi_insert
+
+statement
+CREATE TABLE test_cat.db.part_multi_insert (id INT, dept STRING, value INT) USING iceberg PARTITIONED BY (dept)
+
+statement
+INSERT INTO test_cat.db.part_multi_insert VALUES (1, 'eng', 10), (2, 'sales', 20)
+
+statement
+INSERT INTO test_cat.db.part_multi_insert VALUES (3, 'eng', 30), (4, 'hr', 40)
+
+statement
+INSERT INTO test_cat.db.part_multi_insert VALUES (5, 'sales', 50)
+
+query
+SELECT id, _partition FROM test_cat.db.part_multi_insert ORDER BY id
+
+query
+SELECT _partition.dept, COUNT(DISTINCT _file) as file_cnt, COUNT(*) as row_cnt FROM test_cat.db.part_multi_insert GROUP BY _partition.dept ORDER BY dept
+
+statement
+DROP TABLE test_cat.db.part_multi_insert
diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
index 7a33f40eea..b93b6930be 100644
--- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
@@ -4219,11 +4219,131 @@ class CometIcebergNativeSuite
 
           checkSparkAnswerAndFallbackReason(
             s"SELECT id, value, _spec_id, _pos, _file, _partition FROM $table WHERE id >= 2 ORDER BY id",
-            "Metadata column(s) _spec_id, _partition, _file, _pos is not supported")
+            "Unsupported Iceberg metadata columns: _spec_id, _pos")
         } finally {
           spark.sql(s"DROP TABLE IF EXISTS $table PURGE")
         }
       }
     }
   }
+
+  test("partition evolution - _partition contains fields from all historical specs") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+
+    withTempIcebergDir { warehouseDir =>
+      withSQLConf(
+        "spark.sql.catalog.test_cat" -> "org.apache.iceberg.spark.SparkCatalog",
+        "spark.sql.catalog.test_cat.type" -> "hadoop",
+        "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
+        CometConf.COMET_ENABLED.key -> "true",
+        CometConf.COMET_EXEC_ENABLED.key -> "true",
+        CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+        import org.apache.iceberg.catalog.TableIdentifier
+        import org.apache.iceberg.spark.SparkCatalog
+
+        spark.sql("""
+          CREATE TABLE test_cat.db.part_evolution (
+            id INT, region STRING, category STRING, value DOUBLE
+          ) USING iceberg PARTITIONED BY (region)
+        """)
+
+        spark.sql("""
+          INSERT INTO test_cat.db.part_evolution VALUES
+            (1, 'US', 'A', 10.0), (2, 'EU', 'B', 20.0)
+        """)
+
+        // Add a second partition field via Iceberg Java API (partition evolution -> spec_id 1)
+        val sparkCatalog = spark.sessionState.catalogManager
+          .catalog("test_cat")
+          .asInstanceOf[SparkCatalog]
+        val table = sparkCatalog
+          .icebergCatalog()
+          .loadTable(TableIdentifier.of("db", "part_evolution"))
+        table.updateSpec().addField("category").commit()
+
+        spark.sql("""
+          INSERT INTO test_cat.db.part_evolution VALUES
+            (3, 'US', 'C', 30.0), (4, 'APAC', 'A', 40.0)
+        """)
+
+        // _partition should be a struct with BOTH region and category fields,
+        // covering all historical specs. Files written under spec 0 will have
+        // category=null in _partition; files under spec 1 have both populated.
+        checkIcebergNativeScan(
+          "SELECT id, _partition FROM test_cat.db.part_evolution ORDER BY id")
+
+        // Verify the struct fields are accessible
+        checkIcebergNativeScan(
+          "SELECT id, _partition.region, _partition.category " +
+            "FROM test_cat.db.part_evolution ORDER BY id")
+
+        // Verify correctness: old rows have null category in _partition
+        val result = spark
+          .sql("SELECT id, _partition.category " +
+            "FROM test_cat.db.part_evolution ORDER BY id")
+          .collect()
+        assert(result(0).isNullAt(1), "row 1 (spec 0) should have null _partition.category")
+        assert(result(1).isNullAt(1), "row 2 (spec 0) should have null _partition.category")
+        assert(result(2).getString(1) == "C", "row 3 (spec 1) should have category=C")
+        assert(result(3).getString(1) == "A", "row 4 (spec 1) should have category=A")
+
+        spark.sql("DROP TABLE test_cat.db.part_evolution")
+      }
+    }
+  }
+
+  test("partition evolution - _partition struct fields sorted by field_id ascending") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+
+    withTempIcebergDir { warehouseDir =>
+      withSQLConf(
+        "spark.sql.catalog.test_cat" -> "org.apache.iceberg.spark.SparkCatalog",
+        "spark.sql.catalog.test_cat.type" -> "hadoop",
+        "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
+        CometConf.COMET_ENABLED.key -> "true",
+        CometConf.COMET_EXEC_ENABLED.key -> "true",
+        CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+        import org.apache.iceberg.catalog.TableIdentifier
+        import org.apache.iceberg.spark.SparkCatalog
+
+        // Create table partitioned by region (gets lower field_id in partition spec)
+        spark.sql("""
+          CREATE TABLE test_cat.db.field_order (
+            id INT, region STRING, category STRING
+          ) USING iceberg PARTITIONED BY (region)
+        """)
+
+        spark.sql("""
+          INSERT INTO test_cat.db.field_order VALUES (1, 'US', 'A'), (2, 'EU', 'B')
+        """)
+
+        // Evolve: add category as second partition field (gets higher field_id).
+        // Without sort-by-id, spec-descending traversal would emit category before region.
+        val sparkCatalog = spark.sessionState.catalogManager
+          .catalog("test_cat")
+          .asInstanceOf[SparkCatalog]
+        val table = sparkCatalog
+          .icebergCatalog()
+          .loadTable(TableIdentifier.of("db", "field_order"))
+        table.updateSpec().addField("category").commit()
+
+        spark.sql("""
+          INSERT INTO test_cat.db.field_order VALUES (3, 'APAC', 'C'), (4, 'US', 'D')
+        """)
+
+        // Verify _partition struct field order matches Spark (sorted by field_id ascending).
+        // Spark's Java reader uses buildPartitionProjectionType which sorts by natural key order.
+        checkIcebergNativeScan("SELECT id, _partition FROM test_cat.db.field_order ORDER BY id")
+
+        // Access fields by name to confirm both are present and correctly ordered
+        checkIcebergNativeScan(
+          "SELECT id, _partition.region, _partition.category " +
+            "FROM test_cat.db.field_order ORDER BY id")
+
+        spark.sql("DROP TABLE test_cat.db.field_order")
+      }
+    }
+  }
 }