From 2f9d5ece4b207fae98ce5c65c04c7cc2a3089038 Mon Sep 17 00:00:00 2001 From: Robert Kruszewski Date: Thu, 16 Jul 2026 14:10:36 +0100 Subject: [PATCH] Add support for using vortex in iceberg (#1) Add support for using vortex in iceberg --- Cargo.lock | 1336 ++++++++++++++++- Cargo.toml | 1 + bindings/python/src/data_file.rs | 1 + crates/iceberg/Cargo.toml | 1 + crates/iceberg/public-api.txt | 29 + crates/iceberg/src/arrow/mod.rs | 2 + crates/iceberg/src/arrow/reader/mod.rs | 1 + crates/iceberg/src/arrow/reader/pipeline.rs | 22 +- crates/iceberg/src/arrow/reader/vortex.rs | 1236 +++++++++++++++ crates/iceberg/src/arrow/vortex_util.rs | 56 + crates/iceberg/src/expr/accessor.rs | 4 + crates/iceberg/src/spec/manifest/data_file.rs | 4 + crates/iceberg/src/writer/file_writer/mod.rs | 4 +- .../src/writer/file_writer/vortex_writer.rs | 427 ++++++ 14 files changed, 3057 insertions(+), 67 deletions(-) create mode 100644 crates/iceberg/src/arrow/reader/vortex.rs create mode 100644 crates/iceberg/src/arrow/vortex_util.rs create mode 100644 crates/iceberg/src/writer/file_writer/vortex_writer.rs diff --git a/Cargo.lock b/Cargo.lock index dd33e5c2af..af940aa6c6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -207,6 +207,12 @@ dependencies = [ "rustversion", ] +[[package]] +name = "arcref" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "28f6098a1e8ab66ff91324cce8fea6643101882cf7d09c85acdb1485ecf61e29" + [[package]] name = "array-init" version = "2.1.0" @@ -515,6 +521,18 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "async-channel" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "924ed96dd52d1b75e9c1a3e6275715fd320f5f9439fb5a4a11fa51f4221158d2" +dependencies = [ + "concurrent-queue", + "event-listener-strategy", + "futures-core", + "pin-project-lite", +] + [[package]] name = "async-compression" version = "0.4.42" @@ -527,12 +545,55 @@ dependencies = [ "tokio", ] +[[package]] +name = "async-executor" +version = "1.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c96bf972d85afc50bf5ab8fe2d54d1586b4e0b46c97c50a0c9e71e2f7bcd812a" +dependencies = [ + "async-task", + "concurrent-queue", + "fastrand", + "futures-lite", + "pin-project-lite", + "slab", +] + [[package]] name = "async-ffi" version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f4de21c0feef7e5a556e51af767c953f0501f7f300ba785cc99c47bdc8081a50" +[[package]] +name = "async-fs" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8034a681df4aed8b8edbd7fbe472401ecf009251c8b40556b304567052e294c5" +dependencies = [ + "async-lock", + "blocking", + "futures-lite", +] + +[[package]] +name = "async-io" +version = "2.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "456b8a8feb6f42d237746d4b3e9a178494627745c3c56c6ea55d92ba50d026fc" +dependencies = [ + "autocfg", + "cfg-if 1.0.4", + "concurrent-queue", + "futures-io", + "futures-lite", + "parking", + "polling", + "rustix", + "slab", + "windows-sys 0.61.2", +] + [[package]] name = "async-lock" version = "3.4.2" @@ -544,6 +605,35 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "async-net" +version = "2.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b948000fad4873c1c9339d60f2623323a0cfd3816e5181033c6a5cb68b2accf7" +dependencies = [ + "async-io", + "blocking", + "futures-lite", +] + +[[package]] +name = "async-process" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc50921ec0055cdd8a16de48773bfeec5c972598674347252c0399676be7da75" +dependencies = [ + "async-channel", + "async-io", + "async-lock", + "async-signal", + "async-task", + "blocking", + "cfg-if 1.0.4", + "event-listener", + "futures-lite", + "rustix", +] + [[package]] name = "async-recursion" version = "1.1.1" @@ -552,9 +642,55 @@ checksum = "3b43422f69d8ff38f95f1b2bb76517c91589a924d1559a0e935d7c8ce0274c11" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", +] + +[[package]] +name = "async-signal" +version = "0.2.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52b5aaafa020cf5053a01f2a60e8ff5dccf550f0f77ec54a4e47285ac2bab485" +dependencies = [ + "async-io", + "async-lock", + "atomic-waker", + "cfg-if 1.0.4", + "futures-core", + "futures-io", + "rustix", + "signal-hook-registry", + "slab", + "windows-sys 0.61.2", +] + +[[package]] +name = "async-stream" +version = "0.3.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b5a71a6f37880a80d1d7f19efd781e4b5de42c88f0722cc13bcb6cc2cfe8476" +dependencies = [ + "async-stream-impl", + "futures-core", + "pin-project-lite", +] + +[[package]] +name = "async-stream-impl" +version = "0.3.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.118", ] +[[package]] +name = "async-task" +version = "4.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b75356056920673b02621b35afd0f7dda9306d03c79a30f5c56c44cf256e3de" + [[package]] name = "async-trait" version = "0.1.89" @@ -563,7 +699,7 @@ checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -994,7 +1130,7 @@ checksum = "221eaa237ddf1ca79b60d1372aad77e47f9c0ea5b3ce5099da8c61d027dc77b3" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -1102,6 +1238,12 @@ version = "1.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" +[[package]] +name = "better_io" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef0a3155e943e341e557863e69a708999c94ede624e37865c8e2a91b94efa78f" + [[package]] name = "bigdecimal" version = "0.4.10" @@ -1122,6 +1264,33 @@ version = "0.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "230c5f1ca6a325a32553f8640d31ac9b49f2411e901e427570154868b46da4f7" +[[package]] +name = "bindgen" +version = "0.72.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "993776b509cfb49c750f11b8f07a46fa23e0a1386ffc01fb1e7d343efc387895" +dependencies = [ + "bitflags 2.13.0", + "cexpr", + "clang-sys", + "itertools 0.13.0", + "proc-macro2", + "quote", + "regex", + "rustc-hash", + "shlex 1.3.0", + "syn 2.0.118", +] + +[[package]] +name = "bit-vec" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b71798fca2c1fe1086445a7258a4bc81e6e49dcd24c8d0dd9a1e57395b603f51" +dependencies = [ + "serde", +] + [[package]] name = "bitflags" version = "1.3.2" @@ -1137,6 +1306,18 @@ dependencies = [ "serde_core", ] +[[package]] +name = "bitvec" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ddcec3d12c579d40898fe0a9a358a803c23e9c52ca3c425707f81c9436211837" +dependencies = [ + "funty", + "radium", + "tap", + "wyz", +] + [[package]] name = "blake2" version = "0.10.6" @@ -1187,6 +1368,19 @@ dependencies = [ "generic-array", ] +[[package]] +name = "blocking" +version = "1.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e83f8d02be6967315521be875afa792a316e28d57b5a2d401897e2a7921b7f21" +dependencies = [ + "async-channel", + "async-task", + "futures-io", + "futures-lite", + "piper", +] + [[package]] name = "bnum" version = "0.12.1" @@ -1219,7 +1413,7 @@ dependencies = [ "proc-macro2", "quote", "rustversion", - "syn", + "syn 2.0.118", ] [[package]] @@ -1286,7 +1480,7 @@ checksum = "f9abbd1bc6865053c427f7198e6af43bfdedc55ab791faed4fbd361d789575ff" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -1341,7 +1535,16 @@ dependencies = [ "find-msvc-tools", "jobserver", "libc", - "shlex", + "shlex 2.0.1", +] + +[[package]] +name = "cexpr" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6fac387a98bb7c37292057cffc56d62ecb629900026402633ae9160df93a8766" +dependencies = [ + "nom", ] [[package]] @@ -1407,6 +1610,17 @@ dependencies = [ "inout", ] +[[package]] +name = "clang-sys" +version = "1.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b023947811758c97c59bf9d1c188fd619ad4718dcaa767947df1cadb14f39f4" +dependencies = [ + "glob", + "libc", + "libloading 0.8.9", +] + [[package]] name = "clap" version = "4.6.1" @@ -1438,7 +1652,7 @@ dependencies = [ "heck", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -1586,6 +1800,12 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "18f12cc9948ed9604230cdddc7c86e270f9401ccbe3c2e98a4378c5e7632212f" +[[package]] +name = "const_for" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "988d3bd6bf67b6d7ae2b519c55296fa57411c27c312575e47b2975f8d1d8590b" + [[package]] name = "const_panic" version = "0.2.15" @@ -1627,6 +1847,12 @@ version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" +[[package]] +name = "core_detect" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f8f80099a98041a3d1622845c271458a2d73e688351bf3cb999266764b81d48" + [[package]] name = "countio" version = "0.3.0" @@ -1811,6 +2037,18 @@ dependencies = [ "cmov", ] +[[package]] +name = "custom-labels" +version = "0.4.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4121f4f539659ad975da5bc6702b7d7d6de59ff1bbaec328dfe7dd5058052ef2" +dependencies = [ + "bindgen", + "cc", + "libc", + "pin-project-lite", +] + [[package]] name = "darling" version = "0.20.11" @@ -1842,7 +2080,7 @@ dependencies = [ "proc-macro2", "quote", "strsim", - "syn", + "syn 2.0.118", ] [[package]] @@ -1855,7 +2093,7 @@ dependencies = [ "proc-macro2", "quote", "strsim", - "syn", + "syn 2.0.118", ] [[package]] @@ -1866,7 +2104,7 @@ checksum = "fc34b93ccb385b40dc71c6fceac4b2ad23662c7eeb248cf10d529b7e055b6ead" dependencies = [ "darling_core 0.20.11", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -1877,7 +2115,7 @@ checksum = "ac3984ec7bd6cfa798e62b4a642426a5be0e68f9401cfc2a01e3fa9ea2fcdb8d" dependencies = [ "darling_core 0.23.0", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -2307,7 +2545,7 @@ dependencies = [ "datafusion-proto-common", "datafusion-session", "futures", - "libloading", + "libloading 0.9.0", "log", "prost", "semver", @@ -2456,7 +2694,7 @@ checksum = "587164e03ad68732aa9e7bfe5686e3f25970d4c64fd4bd80790749840892dae5" dependencies = [ "datafusion-doc", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -2746,7 +2984,7 @@ dependencies = [ "defmt-parser", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -2796,7 +3034,7 @@ dependencies = [ "darling 0.20.11", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -2806,7 +3044,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ab63b0e2bf4d5928aff72e83a7dace85d7bba5fe12dcc3c5a572d78caffd3f3c" dependencies = [ "derive_builder_core", - "syn", + "syn 2.0.118", ] [[package]] @@ -2868,7 +3106,7 @@ checksum = "1ac70aa55017e108007fbaf5aa0f54b021c98f92ff8af59d42eda9da96e3dd4f" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -2913,6 +3151,12 @@ version = "0.0.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f678cf4a922c215c63e0de95eb1ff08a958a81d47e485cf9da1e27bf6305cfa5" +[[package]] +name = "dtype_dispatch" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ab23e69df104e2fd85ee63a533a22d2132ef5975dc6b36f9f3e5a7305e4a8ed7" + [[package]] name = "dunce" version = "1.0.5" @@ -2934,7 +3178,7 @@ dependencies = [ "enum-ordinalize", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -2958,6 +3202,26 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "869b0adbda23651a9c5c0c3d270aac9fcb52e8622a8f2b17e57802d7791962f2" +[[package]] +name = "enum-iterator" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4549325971814bda7a44061bf3fe7e487d447cba01e4220a4b454d630d7a016" +dependencies = [ + "enum-iterator-derive", +] + +[[package]] +name = "enum-iterator-derive" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "685adfa4d6f3d765a26bc5dbc936577de9abf756c1feeb3089b01dd395034842" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.118", +] + [[package]] name = "enum-ordinalize" version = "4.4.1" @@ -2975,7 +3239,7 @@ checksum = "42e528e2d34ba8a67a1a650b86beae8ef69fc5fdb638016f386b973226590432" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -3082,6 +3346,48 @@ dependencies = [ "once_cell", ] +[[package]] +name = "ext-trait" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d772df1c1a777963712fb68e014235e80863d6a91a85c4e06ba2d16243a310e5" +dependencies = [ + "ext-trait-proc_macros", +] + +[[package]] +name = "ext-trait-proc_macros" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ab7934152eaf26aa5aa9f7371408ad5af4c31357073c9e84c3b9d7f11ad639a" +dependencies = [ + "proc-macro2", + "quote", + "syn 1.0.109", +] + +[[package]] +name = "extension-traits" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a296e5a895621edf9fa8329c83aa1cb69a964643e36cf54d8d7a69b789089537" +dependencies = [ + "ext-trait", +] + +[[package]] +name = "fastlanes" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9272a674b446f53253f66582596614e41552d14968a3d1aaa590576a91c1188" +dependencies = [ + "const_for", + "core_detect", + "num-traits", + "paste", + "seq-macro", +] + [[package]] name = "fastnum" version = "0.7.5" @@ -3207,6 +3513,21 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" +[[package]] +name = "fsst-rs" +version = "0.5.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b13ac798afc0d9194eb4efefef8b9332efbd80b43f302a968cb8cb23b9d5360" +dependencies = [ + "rustc-hash", +] + +[[package]] +name = "funty" +version = "2.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6d5a32815ae3f33302d95fdcb2ce17862f8c65363dcfd29360480ba1001fc9c" + [[package]] name = "futures" version = "0.3.32" @@ -3266,6 +3587,19 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" +[[package]] +name = "futures-lite" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f78e10609fe0e0b3f4157ffab1876319b5b0db102a2c60dc4626306dc46b44ad" +dependencies = [ + "fastrand", + "futures-core", + "futures-io", + "parking", + "pin-project-lite", +] + [[package]] name = "futures-macro" version = "0.3.32" @@ -3274,7 +3608,7 @@ checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -3397,7 +3731,7 @@ checksum = "53010ccb100b96a67bc32c0175f0ed1426b31b655d562898e57325f81c023ac0" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -3461,6 +3795,10 @@ name = "hashbrown" version = "0.14.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" +dependencies = [ + "ahash", + "allocator-api2", +] [[package]] name = "hashbrown" @@ -3505,6 +3843,12 @@ version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" +[[package]] +name = "hermit-abi" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" + [[package]] name = "hex" version = "0.4.3" @@ -3648,6 +3992,15 @@ version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" +[[package]] +name = "humansize" +version = "2.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6cb51c9a029ddc91b07a787f1d86b53ccfa49b0e86688c946ebe8d3555685dd7" +dependencies = [ + "libm", +] + [[package]] name = "humantime" version = "2.4.0" @@ -3808,6 +4161,7 @@ dependencies = [ "typetag", "url", "uuid", + "vortex", "zeroize", "zstd", ] @@ -4292,7 +4646,7 @@ checksum = "e165e897f662d428f3cd3828a919dbe067c2d42bb1031eede74ef9d27ecdedd2" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -4337,7 +4691,7 @@ dependencies = [ "quote", "rustc_version", "simd_cesu8", - "syn", + "syn 2.0.118", ] [[package]] @@ -4356,7 +4710,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "38c0b942f458fe50cdac086d2f946512305e5631e720728f2a61aabcd47a6264" dependencies = [ "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -4380,6 +4734,16 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "kanal" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e3953adf0cd667798b396c2fa13552d6d9b3269d7dd1154c4c416442d1ff574" +dependencies = [ + "futures-core", + "lock_api", +] + [[package]] name = "konst" version = "0.4.3" @@ -4397,6 +4761,16 @@ version = "0.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a" +[[package]] +name = "lasso" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e14eda50a3494b3bf7b9ce51c52434a761e383d7238ce1dd5dcec2fbc13e9fb" +dependencies = [ + "dashmap", + "hashbrown 0.14.5", +] + [[package]] name = "lazy_static" version = "1.5.0" @@ -4407,8 +4781,33 @@ dependencies = [ ] [[package]] -name = "lexical-core" -version = "1.0.6" +name = "lending-iterator" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc07588c853b50689205fb5c00498aa681d89828e0ce8cbd965ebc7a5d8ae260" +dependencies = [ + "extension-traits", + "lending-iterator-proc_macros", + "macro_rules_attribute", + "never-say-never", + "nougat", + "polonius-the-crab", +] + +[[package]] +name = "lending-iterator-proc_macros" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b5445dd1c0deb1e97b8a16561d17fc686ca83e8411128fb036e9668a72d51b1d" +dependencies = [ + "proc-macro2", + "quote", + "syn 1.0.109", +] + +[[package]] +name = "lexical-core" +version = "1.0.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7d8d125a277f807e55a77304455eb7b1cb52f2b18c143b60e766c120bd64a594" dependencies = [ @@ -4475,6 +4874,16 @@ version = "0.2.186" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" +[[package]] +name = "libloading" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d7c4b02199fee7c5d21a5ae7d8cfa79a6ef5bb2fc834d6e9058e89c825efdc55" +dependencies = [ + "cfg-if 1.0.4", + "windows-link", +] + [[package]] name = "libloading" version = "0.9.0" @@ -4626,6 +5035,22 @@ dependencies = [ "twox-hash", ] +[[package]] +name = "macro_rules_attribute" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf0c9b980bf4f3a37fd7b1c066941dd1b1d0152ce6ee6e8fe8c49b9f6810d862" +dependencies = [ + "macro_rules_attribute-proc_macro", + "paste", +] + +[[package]] +name = "macro_rules_attribute-proc_macro" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "58093314a45e00c77d5c508f76e77c3396afbbc0d01506e7fae47b018bac2b1d" + [[package]] name = "marrow" version = "0.2.7" @@ -4732,6 +5157,12 @@ dependencies = [ "serde", ] +[[package]] +name = "minimal-lexical" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" + [[package]] name = "miniz_oxide" version = "0.8.9" @@ -4776,7 +5207,7 @@ dependencies = [ "cfg-if 1.0.4", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -4850,7 +5281,7 @@ checksum = "b40e46c845ac234bcba19db7ab252bc2778cbadd516a466d2f12b1580852d136" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -4870,7 +5301,7 @@ checksum = "4568f25ccbd45ab5d5603dc34318c1ec56b117531781260002151b8530a9f931" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -4885,6 +5316,12 @@ version = "0.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9252111cf132ba0929b6f8e030cac2a24b507f3a4d6db6fb2896f27b354c714b" +[[package]] +name = "never-say-never" +version = "6.6.666" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf5a574dadd7941adeaa71823ecba5e28331b8313fb2e1c6a5c7e5981ea53ad6" + [[package]] name = "nibble_vec" version = "0.1.0" @@ -4919,6 +5356,37 @@ dependencies = [ "libc", ] +[[package]] +name = "nom" +version = "7.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a" +dependencies = [ + "memchr", + "minimal-lexical", +] + +[[package]] +name = "nougat" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "97b57b9ced431322f054fc673f1d3c7fa52d80efd9df74ad2fc759f044742510" +dependencies = [ + "macro_rules_attribute", + "nougat-proc_macros", +] + +[[package]] +name = "nougat-proc_macros" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c84f77a45e99a2f9b492695d99e1c23844619caa5f3e57647cffacad773ca257" +dependencies = [ + "proc-macro2", + "quote", + "syn 1.0.109", +] + [[package]] name = "ntapi" version = "0.4.3" @@ -5027,7 +5495,7 @@ dependencies = [ "proc-macro-crate", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -5111,6 +5579,9 @@ name = "once_cell" version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +dependencies = [ + "parking_lot_core", +] [[package]] name = "once_cell_polyfill" @@ -5124,6 +5595,12 @@ version = "0.1.13" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "269bca4c2591a28585d6bf10d9ed0332b7d76900a1b02bec41bdc3a2cdcda107" +[[package]] +name = "oneshot" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cfe21416a02c693fb9f980befcb230ecc70b0b3d1cc4abf88b9675c4c1457f0c" + [[package]] name = "opaque-debug" version = "0.3.1" @@ -5486,6 +5963,18 @@ dependencies = [ "hmac 0.12.1", ] +[[package]] +name = "pco" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "553ccdc7f6999785559af4998c79712a5ab820e26b68bad9146609c19587ec82" +dependencies = [ + "better_io", + "dtype_dispatch", + "half", + "rand_xoshiro", +] + [[package]] name = "pem" version = "3.0.6" @@ -5580,7 +6069,7 @@ checksum = "c96395f0a926bc13b1c17622aaddda1ecb55d49c8f1bf9777e4d877800a43f8b" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -5595,6 +6084,17 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" +[[package]] +name = "piper" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c835479a4443ded371d6c535cbfd8d31ad92c5d23ae9770a61bc155e4992a3c1" +dependencies = [ + "atomic-waker", + "fastrand", + "futures-io", +] + [[package]] name = "pkcs1" version = "0.7.5" @@ -5645,6 +6145,26 @@ version = "0.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b4596b6d070b27117e987119b4dac604f3c58cfb0b191112e24771b2faeac1a6" +[[package]] +name = "polling" +version = "3.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d0e4f59085d47d8241c88ead0f274e8a0cb551f3625263c05eb8dd897c34218" +dependencies = [ + "cfg-if 1.0.4", + "concurrent-queue", + "hermit-abi", + "pin-project-lite", + "rustix", + "windows-sys 0.61.2", +] + +[[package]] +name = "polonius-the-crab" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2a69ee997a6282f8462abf1e0d8c38c965e968799e912b3bed8c9e8a28c2f9f" + [[package]] name = "polyval" version = "0.6.2" @@ -5719,7 +6239,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0de1b847b39c8131db0467e9df1ff60e6d0562ab8e9a16e568ad0fdb372e2f2" dependencies = [ "predicates-core", - "termtree", + "termtree 0.5.1", ] [[package]] @@ -5739,7 +6259,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "479ca8adacdd7ce8f1fb39ce9ecccbfe93a3f1344b3d0d97f20bc0196208f62b" dependencies = [ "proc-macro2", - "syn", + "syn 2.0.118", ] [[package]] @@ -5780,7 +6300,16 @@ dependencies = [ "itertools 0.14.0", "proc-macro2", "quote", - "syn", + "syn 2.0.118", +] + +[[package]] +name = "prost-types" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f94967dc7688f3054c7fac87473ffae4cc4c3904800e2d9f5b857246d8963b0a" +dependencies = [ + "prost", ] [[package]] @@ -5810,7 +6339,7 @@ checksum = "7347867d0a7e1208d93b46767be83e2b8f978c3dad35f775ac8d8847551d6fe1" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -5868,7 +6397,7 @@ dependencies = [ "proc-macro2", "pyo3-macros-backend", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -5881,7 +6410,7 @@ dependencies = [ "proc-macro2", "pyo3-build-config", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -5988,6 +6517,12 @@ version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" +[[package]] +name = "radium" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc33ff2d4973d518d823d61aa239014831e521c75da58e3df4840d3f47749d09" + [[package]] name = "radix_trie" version = "0.3.0" @@ -6094,6 +6629,15 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "rand_xoshiro" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6f97cdb2a36ed4183de61b2f824cc45c9f1037f28afe0a322e9fff4c108b5aaa" +dependencies = [ + "rand_core 0.6.4", +] + [[package]] name = "recursive" version = "0.1.1" @@ -6111,7 +6655,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "76009fbe0614077fc1a2ce255e3a1881a2e3a3527097d5dc6d8212c585e7e38b" dependencies = [ "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -6169,7 +6713,7 @@ checksum = "b7186006dcb21920990093f30e3dea63b7d6e977bf1256be20c3563a5db070da" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -6469,7 +7013,7 @@ checksum = "c0ed1a78a1b19d184b0daa629dd9a024573173ec7d485b287cb369fb3607cc1c" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -6528,7 +7072,7 @@ dependencies = [ "regex", "relative-path", "rustc_version", - "syn", + "syn 2.0.118", "unicode-ident", ] @@ -6852,7 +7396,7 @@ checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -6876,7 +7420,7 @@ checksum = "175ee3e80ae9982737ca543e96133087cbd9a485eecc3bc4de9c1a37b47ea59c" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -6929,7 +7473,7 @@ dependencies = [ "darling 0.23.0", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -7012,6 +7556,12 @@ dependencies = [ "os_str_bytes", ] +[[package]] +name = "shlex" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" + [[package]] name = "shlex" version = "2.0.1" @@ -7072,6 +7622,12 @@ version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ee5873ec9cce0195efcb7a4e9507a04cd49aec9c83d0389df45b1ef7ba2e649" +[[package]] +name = "sketches-ddsketch" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05e40b6cf54d988dc1a2223531b969c9a9e30906ad90ef64890c27b4bfbb46ea" + [[package]] name = "slab" version = "0.4.12" @@ -7087,6 +7643,23 @@ dependencies = [ "serde", ] +[[package]] +name = "smol" +version = "2.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a33bd3e260892199c3ccfc487c88b2da2265080acb316cd920da72fdfd7c599f" +dependencies = [ + "async-channel", + "async-executor", + "async-fs", + "async-io", + "async-lock", + "async-net", + "async-process", + "blocking", + "futures-lite", +] + [[package]] name = "snap" version = "1.1.1" @@ -7214,7 +7787,7 @@ checksum = "a6dd45d8fc1c79299bfbb7190e42ccbbdf6a5f52e4a6ad98d92357ea965bd289" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -7276,7 +7849,7 @@ dependencies = [ "quote", "sqlx-core", "sqlx-macros-core", - "syn", + "syn 2.0.118", ] [[package]] @@ -7297,7 +7870,7 @@ dependencies = [ "sha2 0.10.9", "sqlx-core", "sqlx-sqlite", - "syn", + "syn 2.0.118", "tokio", "url", ] @@ -7435,7 +8008,7 @@ dependencies = [ "proc-macro-crate", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -7508,7 +8081,7 @@ dependencies = [ "heck", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -7520,7 +8093,7 @@ dependencies = [ "heck", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -7545,6 +8118,17 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a7973cce6668464ea31f176d85b13c7ab3bba2cb3b77a2ed26abd7801688010a" +[[package]] +name = "syn" +version = "1.0.109" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72b64191b275b66ffe2469e8af2c1cfe3bafa67b529ead792a6d0160888b4237" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "syn" version = "2.0.118" @@ -7573,7 +8157,7 @@ checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -7617,6 +8201,12 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417" +[[package]] +name = "tap" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "55937e1799185b12863d447f42597ed69d9928686b8d88a1df17376a097d8369" + [[package]] name = "target-lexicon" version = "0.13.5" @@ -7642,6 +8232,12 @@ version = "0.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f50febec83f5ee1df3015341d8bd429f2d1cc62bcba7ea2076759d315084683" +[[package]] +name = "termtree" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d4d1330fe7f7f872cd05165130b10602d667b205fd85be09be2814b115d4ced9" + [[package]] name = "thiserror" version = "1.0.69" @@ -7668,7 +8264,7 @@ checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -7679,7 +8275,7 @@ checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -7791,7 +8387,7 @@ checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -7989,7 +8585,7 @@ checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -8076,7 +8672,7 @@ checksum = "3c36781cc0e46a83726d9879608e4cf6c2505237e263a8eb8c24502989cfdb28" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -8112,7 +8708,7 @@ checksum = "cf808357c6ed7e13ba0f3277ec8d8f21b2d501274895104263985330c726c1c5" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -8304,6 +8900,598 @@ dependencies = [ "volo", ] +[[package]] +name = "vortex" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e999910e7c7f50a1f344b47ec6ebd07ce60d856bcf3f053ab63a358104c19367" +dependencies = [ + "vortex-alp", + "vortex-array", + "vortex-btrblocks", + "vortex-buffer", + "vortex-bytebool", + "vortex-datetime-parts", + "vortex-decimal-byte-parts", + "vortex-error", + "vortex-fastlanes", + "vortex-file", + "vortex-flatbuffers", + "vortex-fsst", + "vortex-io", + "vortex-ipc", + "vortex-layout", + "vortex-mask", + "vortex-metrics", + "vortex-pco", + "vortex-proto", + "vortex-runend", + "vortex-scan", + "vortex-sequence", + "vortex-session", + "vortex-sparse", + "vortex-utils", + "vortex-zigzag", + "vortex-zstd", +] + +[[package]] +name = "vortex-alp" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a78eaf3adcedeb6a71cc28b4b837db147524a138aa72563f128aab776c572901" +dependencies = [ + "itertools 0.14.0", + "num-traits", + "prost", + "rustc-hash", + "vortex-array", + "vortex-buffer", + "vortex-error", + "vortex-fastlanes", + "vortex-mask", + "vortex-session", + "vortex-utils", +] + +[[package]] +name = "vortex-array" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eedbfa56229af2db2945e078ec505121ac8ef1179f7c9ee289420a43647144f4" +dependencies = [ + "arc-swap", + "arcref", + "arrow-arith", + "arrow-array", + "arrow-buffer", + "arrow-cast", + "arrow-data", + "arrow-ord", + "arrow-schema", + "arrow-select", + "arrow-string", + "async-lock", + "bytes", + "cfg-if 1.0.4", + "enum-iterator", + "flatbuffers", + "futures", + "half", + "humansize", + "inventory", + "itertools 0.14.0", + "jiff", + "num-traits", + "num_enum", + "parking_lot", + "paste", + "pin-project-lite", + "prost", + "rand 0.10.2", + "rustc-hash", + "simdutf8", + "smallvec", + "static_assertions", + "termtree 1.0.0", + "tracing", + "uuid", + "vortex-array-macros", + "vortex-buffer", + "vortex-compute", + "vortex-error", + "vortex-flatbuffers", + "vortex-mask", + "vortex-proto", + "vortex-session", + "vortex-utils", +] + +[[package]] +name = "vortex-array-macros" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "357e1a0737582c236a75430582c057145781bfced9bab99d5cbed93bd498f3d0" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.118", +] + +[[package]] +name = "vortex-btrblocks" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3662139b831588f59b90b526f763384dae3c868ce4192dd236e8b81fa69a8291" +dependencies = [ + "itertools 0.14.0", + "num-traits", + "pco", + "rand 0.10.2", + "vortex-alp", + "vortex-array", + "vortex-buffer", + "vortex-compressor", + "vortex-datetime-parts", + "vortex-decimal-byte-parts", + "vortex-error", + "vortex-fastlanes", + "vortex-fsst", + "vortex-pco", + "vortex-runend", + "vortex-sequence", + "vortex-sparse", + "vortex-utils", + "vortex-zigzag", + "vortex-zstd", +] + +[[package]] +name = "vortex-buffer" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "73094503b5983b0408fb1b36d6a21bf783290627cbdc994539fdff433c42966a" +dependencies = [ + "arrow-buffer", + "bitvec", + "bytes", + "itertools 0.14.0", + "simdutf8", + "vortex-error", +] + +[[package]] +name = "vortex-bytebool" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85dc654b529c47912300d8fe9e099110565e6a04602369f3741d7fe2382ca622" +dependencies = [ + "num-traits", + "vortex-array", + "vortex-buffer", + "vortex-error", + "vortex-mask", + "vortex-session", +] + +[[package]] +name = "vortex-compressor" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ed8fa7a85228c365316192f5690e0794a0367b0e97b6dadbc9123562630e2b8" +dependencies = [ + "itertools 0.14.0", + "num-traits", + "parking_lot", + "rand 0.10.2", + "rustc-hash", + "tracing", + "vortex-array", + "vortex-buffer", + "vortex-error", + "vortex-mask", + "vortex-utils", +] + +[[package]] +name = "vortex-compute" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2c5f057d9083c84e0e5c16f4965c947b8c9dcbd4d952b00c1cdf3d87f26d8b66" +dependencies = [ + "vortex-buffer", +] + +[[package]] +name = "vortex-datetime-parts" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2f9dac037c88913f5829d7559442231996cdc542ed12a692ec0d5ccfcd99151" +dependencies = [ + "num-traits", + "prost", + "vortex-array", + "vortex-buffer", + "vortex-error", + "vortex-mask", + "vortex-session", +] + +[[package]] +name = "vortex-decimal-byte-parts" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2952a8baa16f0b0ffaf51442654cf3adff0d876266543367413417af97f96689" +dependencies = [ + "num-traits", + "prost", + "vortex-array", + "vortex-buffer", + "vortex-error", + "vortex-mask", + "vortex-session", +] + +[[package]] +name = "vortex-error" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "17607fb14879f521b123c31ffaba184f3b5765e9e344ac8c07b1b0b3be5ec0e9" +dependencies = [ + "arrow-schema", + "flatbuffers", + "jiff", + "prost", + "tokio", +] + +[[package]] +name = "vortex-fastlanes" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d1595b950fd63b336f395e394ac22038641215b0eaea433ba8c76b313d52eec" +dependencies = [ + "fastlanes", + "itertools 0.14.0", + "lending-iterator", + "num-traits", + "prost", + "vortex-array", + "vortex-buffer", + "vortex-error", + "vortex-mask", + "vortex-session", +] + +[[package]] +name = "vortex-file" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8d8c9f7d11188ca817806d4f356fc277cded3c81ca4aecf5a31159b3c438cb6c" +dependencies = [ + "async-trait", + "bytes", + "flatbuffers", + "futures", + "itertools 0.14.0", + "kanal", + "moka", + "oneshot 0.2.1", + "parking_lot", + "pin-project-lite", + "tokio", + "tracing", + "url", + "vortex-alp", + "vortex-array", + "vortex-btrblocks", + "vortex-buffer", + "vortex-bytebool", + "vortex-datetime-parts", + "vortex-decimal-byte-parts", + "vortex-error", + "vortex-fastlanes", + "vortex-flatbuffers", + "vortex-fsst", + "vortex-io", + "vortex-layout", + "vortex-mask", + "vortex-metrics", + "vortex-pco", + "vortex-runend", + "vortex-scan", + "vortex-sequence", + "vortex-session", + "vortex-sparse", + "vortex-utils", + "vortex-zigzag", + "vortex-zstd", +] + +[[package]] +name = "vortex-flatbuffers" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "abde80f1c35c73ac806e4d2ef58e720bc535aa0b217be3cbe058c7d014a6ed71" +dependencies = [ + "flatbuffers", + "vortex-buffer", + "vortex-error", +] + +[[package]] +name = "vortex-fsst" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "442abb523fa145d6d26e33856ef9fa3da4a08afeb96769097e4d0c81d0e0efe3" +dependencies = [ + "fsst-rs", + "num-traits", + "prost", + "vortex-array", + "vortex-buffer", + "vortex-error", + "vortex-mask", + "vortex-session", +] + +[[package]] +name = "vortex-io" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "549b3a7b6352a7a7861df2c6f41d5c4c33987cd3c01dbd88ed364f6285582629" +dependencies = [ + "async-fs", + "async-stream", + "async-trait", + "bytes", + "custom-labels", + "futures", + "glob", + "kanal", + "oneshot 0.2.1", + "parking_lot", + "pin-project-lite", + "smol", + "tokio", + "tracing", + "vortex-array", + "vortex-buffer", + "vortex-error", + "vortex-metrics", + "vortex-session", + "vortex-utils", + "wasm-bindgen-futures", +] + +[[package]] +name = "vortex-ipc" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e64b352ba3554ea92084e3128dd845a3f25c72fc5ab427dff68f0e4d1f896c89" +dependencies = [ + "bytes", + "flatbuffers", + "futures", + "itertools 0.14.0", + "pin-project-lite", + "vortex-array", + "vortex-buffer", + "vortex-error", + "vortex-flatbuffers", + "vortex-session", +] + +[[package]] +name = "vortex-layout" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "88f4ec82f325823394430d256130a692d8f5c2154cf25fde8215b51fc963a818" +dependencies = [ + "arcref", + "arrow-array", + "arrow-schema", + "async-stream", + "async-trait", + "bit-vec", + "flatbuffers", + "futures", + "itertools 0.14.0", + "kanal", + "moka", + "once_cell", + "oneshot 0.2.1", + "parking_lot", + "paste", + "pin-project-lite", + "prost", + "rustc-hash", + "sketches-ddsketch", + "termtree 1.0.0", + "tokio", + "tracing", + "vortex-array", + "vortex-btrblocks", + "vortex-buffer", + "vortex-error", + "vortex-flatbuffers", + "vortex-io", + "vortex-mask", + "vortex-metrics", + "vortex-runend", + "vortex-scan", + "vortex-sequence", + "vortex-session", + "vortex-utils", +] + +[[package]] +name = "vortex-mask" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c91caf1d598e35e4a5c24cee0cd7d4e8c4931b30a9163ac33af8181d546ae15e" +dependencies = [ + "itertools 0.14.0", + "vortex-buffer", + "vortex-error", +] + +[[package]] +name = "vortex-metrics" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9446b43132833788c5fc0cb184912add97ffe1c41a9e9faaf7bafc9b58fdd529" +dependencies = [ + "parking_lot", + "sketches-ddsketch", +] + +[[package]] +name = "vortex-pco" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c57ea636c346087d48e25008554321a285f175d6e7036db1fefbb1998fb228df" +dependencies = [ + "pco", + "prost", + "vortex-array", + "vortex-buffer", + "vortex-error", + "vortex-mask", + "vortex-session", +] + +[[package]] +name = "vortex-proto" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "550431d87f13583e1ac7b54f16ae4450278b63ad1e7aaab56b7b229b908445a6" +dependencies = [ + "prost", + "prost-types", +] + +[[package]] +name = "vortex-runend" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "566ec7cc60f3a478db50c1b56cbc9da448b5ac920bdadc76538eb0122142de10" +dependencies = [ + "arrow-array", + "itertools 0.14.0", + "num-traits", + "prost", + "vortex-array", + "vortex-buffer", + "vortex-error", + "vortex-mask", + "vortex-session", +] + +[[package]] +name = "vortex-scan" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b5521e8ac93f143a99ccf28b9b2ada30c9d521e7fbaadfc8f5fe08787c15ae4" +dependencies = [ + "async-trait", + "futures", + "roaring", + "tracing", + "vortex-array", + "vortex-buffer", + "vortex-error", + "vortex-mask", + "vortex-session", +] + +[[package]] +name = "vortex-sequence" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e0218635ebeb62d21f56fd101b2e3803441c1bad9f3f76fcddb8cc545ea7629" +dependencies = [ + "num-traits", + "prost", + "smallvec", + "vortex-array", + "vortex-buffer", + "vortex-error", + "vortex-mask", + "vortex-proto", + "vortex-session", +] + +[[package]] +name = "vortex-session" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "66234858e92df597666ac9de927855d4cca1bac72d1aceca43eb9ffd8b7167dd" +dependencies = [ + "arc-swap", + "lasso", + "parking_lot", + "vortex-error", + "vortex-utils", +] + +[[package]] +name = "vortex-sparse" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7e1842b2365812e21a7aa0cc378dc6aa19af9f1955c2a9b79f673ef29789f26" +dependencies = [ + "itertools 0.14.0", + "num-traits", + "prost", + "vortex-array", + "vortex-buffer", + "vortex-error", + "vortex-mask", + "vortex-session", +] + +[[package]] +name = "vortex-utils" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24b6fa379006bf275b0ad5641c8b83fd52388fc4d483b83b1c1fd2870dd5d653" +dependencies = [ + "dashmap", + "hashbrown 0.17.1", + "vortex-error", +] + +[[package]] +name = "vortex-zigzag" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aadf2833802c2bf9d69d282189bf3c3e1e1aff6c7c5989f3ae8197ba528fda31" +dependencies = [ + "vortex-array", + "vortex-buffer", + "vortex-error", + "vortex-mask", + "vortex-session", + "zigzag", +] + +[[package]] +name = "vortex-zstd" +version = "0.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1264c8bb5247872a0bbbf20ef51f7177959c213a5e11f9c28b4a1c5f91879e48" +dependencies = [ + "itertools 0.14.0", + "prost", + "vortex-array", + "vortex-buffer", + "vortex-error", + "vortex-mask", + "vortex-session", + "zstd", +] + [[package]] name = "vsimd" version = "0.8.0" @@ -8410,7 +9598,7 @@ dependencies = [ "bumpalo", "proc-macro2", "quote", - "syn", + "syn 2.0.118", "wasm-bindgen-shared", ] @@ -8603,7 +9791,7 @@ checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -8614,7 +9802,7 @@ checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -8849,6 +10037,15 @@ version = "0.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4" +[[package]] +name = "wyz" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05f360fc0b24296329c78fda852a1e9ae82de9cf7b27dae4b7f62f118f77b9ed" +dependencies = [ + "tap", +] + [[package]] name = "xattr" version = "1.6.1" @@ -8988,7 +10185,7 @@ dependencies = [ "lazy_static", "libc", "more-asserts", - "oneshot", + "oneshot 0.1.13", "pin-project", "rand 0.10.2", "reqwest 0.13.4", @@ -9037,7 +10234,7 @@ checksum = "de844c262c8848816172cef550288e7dc6c7b7814b4ee56b3e1553f275f1858e" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", "synstructure", ] @@ -9058,7 +10255,7 @@ checksum = "4714fd92cf900833d49538023a9b3915155210801d1c1169eba513b2addefd71" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -9078,7 +10275,7 @@ checksum = "11532158c46691caf0f2593ea8358fed6bbf68a0315e80aae9bd41fbade684a1" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", "synstructure", ] @@ -9118,7 +10315,16 @@ checksum = "625dc425cab0dca6dc3c3319506e6593dcb08a9f387ea3b284dbd52a92c40555" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", +] + +[[package]] +name = "zigzag" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "70b40401a28d86ce16a330b863b86fd7dbee4d7c940587ab09ab8c019f9e3fdf" +dependencies = [ + "num-traits", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index f14a94af93..fe073c4736 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -152,5 +152,6 @@ url = "2.5.7" uuid = { version = "1.18", features = ["v7"] } volo = "0.10.6" volo-thrift = "0.10.8" +vortex = { version = "0.78", features = ["tokio"] } zeroize = "1.7" zstd = "0.13.3" diff --git a/bindings/python/src/data_file.rs b/bindings/python/src/data_file.rs index b0e42e7d73..2a762731c2 100644 --- a/bindings/python/src/data_file.rs +++ b/bindings/python/src/data_file.rs @@ -71,6 +71,7 @@ impl PyDataFile { DataFileFormat::Orc => "orc", DataFileFormat::Parquet => "parquet", DataFileFormat::Puffin => "puffin", + DataFileFormat::Vortex => "vortex", } } diff --git a/crates/iceberg/Cargo.toml b/crates/iceberg/Cargo.toml index 9353a31842..e419045b2c 100644 --- a/crates/iceberg/Cargo.toml +++ b/crates/iceberg/Cargo.toml @@ -80,6 +80,7 @@ typed-builder = { workspace = true } typetag = { workspace = true } url = { workspace = true } uuid = { workspace = true } +vortex = { workspace = true } zeroize = { workspace = true } zstd = { workspace = true } diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index c2013294e3..5fbd23f313 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1368,6 +1368,7 @@ pub iceberg::spec::DataFileFormat::Avro pub iceberg::spec::DataFileFormat::Orc pub iceberg::spec::DataFileFormat::Parquet pub iceberg::spec::DataFileFormat::Puffin +pub iceberg::spec::DataFileFormat::Vortex impl core::clone::Clone for iceberg::spec::DataFileFormat pub fn iceberg::spec::DataFileFormat::clone(&self) -> iceberg::spec::DataFileFormat impl core::cmp::Eq for iceberg::spec::DataFileFormat @@ -3235,18 +3236,42 @@ pub fn iceberg::writer::file_writer::ParquetWriterBuilder::fmt(&self, f: &mut co impl iceberg::writer::file_writer::FileWriterBuilder for iceberg::writer::file_writer::ParquetWriterBuilder pub type iceberg::writer::file_writer::ParquetWriterBuilder::R = iceberg::writer::file_writer::ParquetWriter pub async fn iceberg::writer::file_writer::ParquetWriterBuilder::build(&self, output_file: iceberg::io::OutputFile) -> iceberg::Result +pub struct iceberg::writer::file_writer::VortexWriter +impl iceberg::writer::CurrentFileStatus for iceberg::writer::file_writer::VortexWriter +pub fn iceberg::writer::file_writer::VortexWriter::current_file_path(&self) -> alloc::string::String +pub fn iceberg::writer::file_writer::VortexWriter::current_row_num(&self) -> usize +pub fn iceberg::writer::file_writer::VortexWriter::current_written_size(&self) -> usize +impl iceberg::writer::file_writer::FileWriter for iceberg::writer::file_writer::VortexWriter +pub async fn iceberg::writer::file_writer::VortexWriter::close(self) -> iceberg::Result> +pub async fn iceberg::writer::file_writer::VortexWriter::write(&mut self, batch: &arrow_array::record_batch::RecordBatch) -> iceberg::Result<()> +pub struct iceberg::writer::file_writer::VortexWriterBuilder +impl iceberg::writer::file_writer::VortexWriterBuilder +pub fn iceberg::writer::file_writer::VortexWriterBuilder::new(schema: iceberg::spec::SchemaRef, session: vortex_session::session::VortexSession) -> Self +impl core::clone::Clone for iceberg::writer::file_writer::VortexWriterBuilder +pub fn iceberg::writer::file_writer::VortexWriterBuilder::clone(&self) -> iceberg::writer::file_writer::VortexWriterBuilder +impl core::fmt::Debug for iceberg::writer::file_writer::VortexWriterBuilder +pub fn iceberg::writer::file_writer::VortexWriterBuilder::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl iceberg::writer::file_writer::FileWriterBuilder for iceberg::writer::file_writer::VortexWriterBuilder +pub type iceberg::writer::file_writer::VortexWriterBuilder::R = iceberg::writer::file_writer::VortexWriter +pub async fn iceberg::writer::file_writer::VortexWriterBuilder::build(&self, output_file: iceberg::io::OutputFile) -> iceberg::Result pub trait iceberg::writer::file_writer::FileWriter: core::marker::Send + iceberg::writer::CurrentFileStatus + 'static pub fn iceberg::writer::file_writer::FileWriter::close(self) -> impl core::future::future::Future> + core::marker::Send pub fn iceberg::writer::file_writer::FileWriter::write(&mut self, batch: &arrow_array::record_batch::RecordBatch) -> impl core::future::future::Future> + core::marker::Send impl iceberg::writer::file_writer::FileWriter for iceberg::writer::file_writer::ParquetWriter pub async fn iceberg::writer::file_writer::ParquetWriter::close(self) -> iceberg::Result> pub async fn iceberg::writer::file_writer::ParquetWriter::write(&mut self, batch: &arrow_array::record_batch::RecordBatch) -> iceberg::Result<()> +impl iceberg::writer::file_writer::FileWriter for iceberg::writer::file_writer::VortexWriter +pub async fn iceberg::writer::file_writer::VortexWriter::close(self) -> iceberg::Result> +pub async fn iceberg::writer::file_writer::VortexWriter::write(&mut self, batch: &arrow_array::record_batch::RecordBatch) -> iceberg::Result<()> pub trait iceberg::writer::file_writer::FileWriterBuilder: core::clone::Clone + core::marker::Send + core::marker::Sync + 'static pub type iceberg::writer::file_writer::FileWriterBuilder::R: iceberg::writer::file_writer::FileWriter pub fn iceberg::writer::file_writer::FileWriterBuilder::build(&self, output_file: iceberg::io::OutputFile) -> impl core::future::future::Future> + core::marker::Send impl iceberg::writer::file_writer::FileWriterBuilder for iceberg::writer::file_writer::ParquetWriterBuilder pub type iceberg::writer::file_writer::ParquetWriterBuilder::R = iceberg::writer::file_writer::ParquetWriter pub async fn iceberg::writer::file_writer::ParquetWriterBuilder::build(&self, output_file: iceberg::io::OutputFile) -> iceberg::Result +impl iceberg::writer::file_writer::FileWriterBuilder for iceberg::writer::file_writer::VortexWriterBuilder +pub type iceberg::writer::file_writer::VortexWriterBuilder::R = iceberg::writer::file_writer::VortexWriter +pub async fn iceberg::writer::file_writer::VortexWriterBuilder::build(&self, output_file: iceberg::io::OutputFile) -> iceberg::Result pub mod iceberg::writer::partitioning pub mod iceberg::writer::partitioning::clustered_writer pub struct iceberg::writer::partitioning::clustered_writer::ClusteredWriter where B: iceberg::writer::IcebergWriterBuilder, O: core::iter::traits::collect::IntoIterator + core::iter::traits::collect::FromIterator<::Item>, ::Item: core::clone::Clone @@ -3285,6 +3310,10 @@ impl iceberg::writer::CurrentFileStatus for iceberg::writer::file_writer::Parque pub fn iceberg::writer::file_writer::ParquetWriter::current_file_path(&self) -> alloc::string::String pub fn iceberg::writer::file_writer::ParquetWriter::current_row_num(&self) -> usize pub fn iceberg::writer::file_writer::ParquetWriter::current_written_size(&self) -> usize +impl iceberg::writer::CurrentFileStatus for iceberg::writer::file_writer::VortexWriter +pub fn iceberg::writer::file_writer::VortexWriter::current_file_path(&self) -> alloc::string::String +pub fn iceberg::writer::file_writer::VortexWriter::current_row_num(&self) -> usize +pub fn iceberg::writer::file_writer::VortexWriter::current_written_size(&self) -> usize impl iceberg::writer::CurrentFileStatus for iceberg::writer::base_writer::data_file_writer::DataFileWriter where B: iceberg::writer::file_writer::FileWriterBuilder, L: iceberg::writer::file_writer::location_generator::LocationGenerator, F: iceberg::writer::file_writer::location_generator::FileNameGenerator pub fn iceberg::writer::base_writer::data_file_writer::DataFileWriter::current_file_path(&self) -> alloc::string::String pub fn iceberg::writer::base_writer::data_file_writer::DataFileWriter::current_row_num(&self) -> usize diff --git a/crates/iceberg/src/arrow/mod.rs b/crates/iceberg/src/arrow/mod.rs index bf53633cfc..e51b2c49ac 100644 --- a/crates/iceberg/src/arrow/mod.rs +++ b/crates/iceberg/src/arrow/mod.rs @@ -29,6 +29,8 @@ pub(crate) mod delete_filter; mod int96; mod reader; +mod vortex_util; +pub(crate) use vortex_util::*; /// RecordBatch projection utilities pub mod record_batch_projector; pub(crate) mod record_batch_transformer; diff --git a/crates/iceberg/src/arrow/reader/mod.rs b/crates/iceberg/src/arrow/reader/mod.rs index 7e3c2f807f..f8715d53ac 100644 --- a/crates/iceberg/src/arrow/reader/mod.rs +++ b/crates/iceberg/src/arrow/reader/mod.rs @@ -41,6 +41,7 @@ mod positional_deletes; mod predicate_visitor; mod projection; mod row_filter; +mod vortex; pub use file_reader::ArrowFileReader; pub(crate) use options::ParquetReadOptions; use predicate_visitor::{CollectFieldIdVisitor, PredicateConverter}; diff --git a/crates/iceberg/src/arrow/reader/pipeline.rs b/crates/iceberg/src/arrow/reader/pipeline.rs index 2b7d68aa43..8fe80b48a9 100644 --- a/crates/iceberg/src/arrow/reader/pipeline.rs +++ b/crates/iceberg/src/arrow/reader/pipeline.rs @@ -39,7 +39,7 @@ use crate::error::Result; use crate::io::{FileIO, FileMetadata, FileRead}; use crate::metadata_columns::{RESERVED_FIELD_ID_FILE, is_metadata_field}; use crate::scan::{ArrowRecordBatchStream, FileScanTask, FileScanTaskStream}; -use crate::spec::Datum; +use crate::spec::{DataFileFormat, Datum}; use crate::{Error, ErrorKind}; impl ArrowReader { @@ -104,6 +104,26 @@ struct FileScanTaskReader { impl FileScanTaskReader { async fn process(self, task: FileScanTask) -> Result { + match task.data_file_format { + DataFileFormat::Parquet => self.process_parquet(task).await, + DataFileFormat::Vortex => { + super::vortex::read_vortex_task( + task, + &self.file_io, + self.batch_size, + &self.delete_file_loader, + self.scan_metrics.bytes_read_counter().clone(), + ) + .await + } + format => Err(Error::new( + ErrorKind::FeatureUnsupported, + format!("Unsupported data file format: {format}"), + )), + } + } + + async fn process_parquet(self, task: FileScanTask) -> Result { let should_load_page_index = (self.row_selection_enabled && task.predicate.is_some()) || !task.deletes.is_empty(); let mut parquet_read_options = self.parquet_read_options; diff --git a/crates/iceberg/src/arrow/reader/vortex.rs b/crates/iceberg/src/arrow/reader/vortex.rs new file mode 100644 index 0000000000..ed70ec6c9b --- /dev/null +++ b/crates/iceberg/src/arrow/reader/vortex.rs @@ -0,0 +1,1236 @@ +// 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. + +//! Read path for vortex data files, producing the same transformed arrow +//! record batch stream as the parquet read path. + +use std::sync::Arc; +use std::sync::atomic::AtomicU64; + +use arrow_array::RecordBatch; +use arrow_array::cast::AsArray; +use arrow_schema::{Field as ArrowField, Schema as ArrowSchema}; +use futures::future::BoxFuture; +use futures::{FutureExt, StreamExt}; +use parquet::arrow::PARQUET_FIELD_ID_META_KEY; +use vortex::VortexSessionDefault; +use vortex::array::VortexSessionExecute; +use vortex::array::arrow::ArrowSessionExt; +use vortex::array::buffer::BufferHandle; +use vortex::array::stream::ArrayStream; +use vortex::buffer::{Alignment, Buffer, ByteBuffer}; +use vortex::dtype::extension::Matcher; +use vortex::dtype::{DType, DecimalDType, Nullability, PType}; +use vortex::error::{VortexResult, vortex_err}; +use vortex::expr::{ + Expression, and, eq, get_item, gt, gt_eq, is_not_null, is_null, like, lit, lt, lt_eq, not, + not_eq, not_like, or, or_collect, root, select, +}; +use vortex::extension::datetime::{AnyTemporal, TimeUnit}; +use vortex::file::OpenOptionsSessionExt; +use vortex::io::VortexReadAt; +use vortex::layout::scan::split_by::SplitBy; +use vortex::scalar::{DecimalValue, Scalar}; +use vortex::scan::selection::Selection; +use vortex::session::VortexSession; + +use super::DEFAULT_RANGE_FETCH_CONCURRENCY; +use crate::arrow::caching_delete_file_loader::CachingDeleteFileLoader; +use crate::arrow::record_batch_transformer::RecordBatchTransformerBuilder; +use crate::arrow::scan_metrics::CountingFileRead; +use crate::arrow::{convert_temporal_value, to_iceberg_error}; +use crate::expr::{BoundPredicate, BoundReference, PredicateOperator}; +use crate::io::{FileIO, FileRead}; +use crate::metadata_columns::RESERVED_FIELD_ID_FILE; +use crate::scan::{ArrowRecordBatchStream, FileScanTask}; +use crate::spec::{Datum, NestedField, PrimitiveLiteral, PrimitiveType, Schema, Type}; +use crate::{Error, ErrorKind, Result}; + +/// Reads a [`FileScanTask`] pointing at a vortex data file and returns a +/// stream of arrow record batches matching the task's projected schema. +pub(super) async fn read_vortex_task( + task: FileScanTask, + file_io: &FileIO, + batch_size: Option, + delete_file_loader: &CachingDeleteFileLoader, + bytes_read_counter: Arc, +) -> Result { + // Start loading the delete files concurrently with opening the data file. + let delete_filter_rx = delete_file_loader.load_deletes(&task.deletes, Arc::clone(&task.schema)); + + // The session captures the current tokio runtime handle at construction + // time, so it is created here, inside the runtime driving the scan. + let session = VortexSession::default(); + + // Open the vortex file through iceberg's FileIO. + let input_file = file_io.new_input(&task.data_file_path)?; + let file_size = if task.file_size_in_bytes > 0 { + task.file_size_in_bytes + } else { + input_file.metadata().await?.size + }; + let reader = CountingFileRead::new(input_file.reader().await?, bytes_read_counter); + let source = Arc::new(FileReadVortexSource { + reader: Arc::new(reader), + size: file_size, + uri: Arc::from(task.data_file_path.as_str()), + }); + + let vortex_file = session + .open_options() + .with_file_size(file_size) + .open(source) + .await + .map_err(to_iceberg_error)?; + + // Vortex files do not record iceberg field ids, so columns are matched by + // name. Fields that were added to the table schema after the file was + // written are filled in by the RecordBatchTransformer below. + let file_field_names: Vec<&str> = vortex_file + .dtype() + .as_struct_fields_opt() + .ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + "Vortex data file does not contain a struct dtype", + ) + })? + .names() + .iter() + .map(|name| name.as_ref()) + .collect(); + let projected_names: Vec = task + .project_field_ids + .iter() + .filter_map(|id| task.schema.name_by_field_id(*id)) + .filter(|name| file_field_names.contains(name)) + .map(|name| name.to_string()) + .collect(); + + let delete_filter = delete_filter_rx.await.unwrap()?; + let delete_predicate = delete_filter.build_equality_delete_predicate(&task).await?; + + // In addition to the optional predicate supplied in the `FileScanTask`, + // we also have an optional predicate resulting from equality delete files. + // If both are present, we logical-AND them together to form a single filter + // predicate that gets pushed down into the vortex scan. + let final_predicate = match (&task.predicate, delete_predicate) { + (None, None) => None, + (Some(predicate), None) => Some(predicate.clone()), + (None, Some(predicate)) => Some(predicate), + (Some(filter_predicate), Some(delete_predicate)) => { + Some(filter_predicate.clone().and(delete_predicate)) + } + }; + let filter = final_predicate + .as_ref() + .map(|predicate| convert_predicate_to_vortex(predicate, &task.schema, vortex_file.dtype())) + .transpose()?; + + let mut scan_builder = vortex_file + .scan() + .map_err(to_iceberg_error)? + .with_projection(select( + projected_names + .iter() + .map(|s| s.as_str()) + .collect::>(), + root(), + )) + .with_some_filter(filter); + if let Some(batch_size) = batch_size { + scan_builder = scan_builder.with_split_by(SplitBy::RowCount(batch_size)); + } + + // Positional deletes are applied by excluding the deleted row ordinals + // from the scan. Vortex selections compose with the filter expression. + if let Some(positional_deletes) = delete_filter.get_delete_vector(&task) { + let deleted_rows: Buffer = { + let guard = positional_deletes.lock().unwrap(); + guard.iter().collect() + }; + scan_builder = scan_builder.with_selection(Selection::ExcludeByIndex(deleted_rows)); + } + + let array_stream = scan_builder.into_array_stream().map_err(to_iceberg_error)?; + + // Mirror the parquet pipeline: adapt file batches to the task schema. + let mut record_batch_transformer_builder = + RecordBatchTransformerBuilder::new(task.schema_ref(), task.project_field_ids()); + if task.project_field_ids().contains(&RESERVED_FIELD_ID_FILE) { + record_batch_transformer_builder = record_batch_transformer_builder.with_constant( + RESERVED_FIELD_ID_FILE, + Datum::string(task.data_file_path.clone()), + ); + } + if let (Some(partition_spec), Some(partition_data)) = + (task.partition_spec.clone(), task.partition.clone()) + { + record_batch_transformer_builder = + record_batch_transformer_builder.with_partition(partition_spec, partition_data)?; + } + let mut record_batch_transformer = record_batch_transformer_builder.build(); + + let arrow_field = session + .arrow() + .to_arrow_field("", array_stream.dtype()) + .map_err(to_iceberg_error)?; + let mut execution_ctx = session.create_execution_ctx(); + let task_schema = task.schema_ref(); + + let record_batch_stream = array_stream.map(move |chunk| { + let chunk = chunk.map_err(to_iceberg_error)?; + let arrow_array = session + .arrow() + .execute_arrow(chunk, Some(&arrow_field), &mut execution_ctx) + .map_err(to_iceberg_error)?; + let batch = RecordBatch::from(arrow_array.as_struct()); + let batch = attach_field_id_metadata(batch, &task_schema)?; + record_batch_transformer.process_record_batch(batch) + }); + + Ok(Box::pin(record_batch_stream) as ArrowRecordBatchStream) +} + +/// Tags every top-level column of `batch` that exists in `schema` with its +/// iceberg field id, so that the [`RecordBatchTransformer`] can match columns +/// by id like it does for parquet files. +fn attach_field_id_metadata(batch: RecordBatch, schema: &Schema) -> Result { + let fields: Vec = batch + .schema() + .fields() + .iter() + .map(|field| { + let field = field.as_ref().clone(); + match schema.field_id_by_name(field.name()) { + Some(field_id) => { + let mut metadata = field.metadata().clone(); + metadata.insert(PARQUET_FIELD_ID_META_KEY.to_string(), field_id.to_string()); + field.with_metadata(metadata) + } + None => field, + } + }) + .collect(); + let schema = Arc::new(ArrowSchema::new(fields)); + RecordBatch::try_new(schema, batch.columns().to_vec()).map_err(|err| { + Error::new( + ErrorKind::Unexpected, + "Failed to attach field id metadata to record batch", + ) + .with_source(err) + }) +} + +/// A [`VortexReadAt`] source backed by iceberg's [`FileRead`]. +struct FileReadVortexSource { + reader: Arc, + size: u64, + uri: Arc, +} + +impl VortexReadAt for FileReadVortexSource { + fn uri(&self) -> Option<&Arc> { + Some(&self.uri) + } + + fn concurrency(&self) -> usize { + DEFAULT_RANGE_FETCH_CONCURRENCY + } + + fn size(&self) -> BoxFuture<'static, VortexResult> { + futures::future::ready(Ok(self.size)).boxed() + } + + fn read_at( + &self, + offset: u64, + length: usize, + alignment: Alignment, + ) -> BoxFuture<'static, VortexResult> { + let reader = Arc::clone(&self.reader); + async move { + let bytes = reader + .read(offset..offset + length as u64) + .await + .map_err(|err| vortex_err!("Failed to read from iceberg storage: {err}"))?; + if bytes.len() != length { + return Err(vortex_err!( + "Short read from iceberg storage: expected {length} bytes, got {}", + bytes.len() + )); + } + Ok(BufferHandle::new_host( + ByteBuffer::from(bytes).aligned(alignment), + )) + } + .boxed() + } +} + +/// Converts an iceberg [`BoundPredicate`] into a vortex filter [`Expression`]. +/// +/// Vortex applies filters exactly (not just for pruning), matching the row +/// level filtering guarantees of the parquet read path. Like the parquet read +/// path, predicate leaves referencing columns that do not exist in the file +/// evaluate to "keep all rows". +fn convert_predicate_to_vortex( + predicate: &BoundPredicate, + schema: &Schema, + file_dtype: &DType, +) -> Result { + match predicate { + BoundPredicate::AlwaysTrue => Ok(lit(true)), + BoundPredicate::AlwaysFalse => Ok(lit(false)), + BoundPredicate::And(expr) => { + let [left, right] = expr.inputs(); + Ok(and( + convert_predicate_to_vortex(left, schema, file_dtype)?, + convert_predicate_to_vortex(right, schema, file_dtype)?, + )) + } + BoundPredicate::Or(expr) => { + let [left, right] = expr.inputs(); + Ok(or( + convert_predicate_to_vortex(left, schema, file_dtype)?, + convert_predicate_to_vortex(right, schema, file_dtype)?, + )) + } + BoundPredicate::Not(expr) => { + let [inner] = expr.inputs(); + Ok(not(convert_predicate_to_vortex(inner, schema, file_dtype)?)) + } + BoundPredicate::Unary(expr) => { + let Some(column) = column_expr(expr.term(), schema, file_dtype)? else { + return Ok(lit(true)); + }; + match expr.op() { + PredicateOperator::IsNull => Ok(is_null(column.expr)), + PredicateOperator::NotNull => Ok(is_not_null(column.expr)), + PredicateOperator::IsNan => nan_predicate(column.expr, expr.term().field(), true), + PredicateOperator::NotNan => nan_predicate(column.expr, expr.term().field(), false), + op => Err(unsupported_predicate(op)), + } + } + BoundPredicate::Binary(expr) => { + let Some(column) = column_expr(expr.term(), schema, file_dtype)? else { + return Ok(lit(true)); + }; + match expr.op() { + PredicateOperator::StartsWith => { + starts_with_predicate(column.expr, expr.literal(), false) + } + PredicateOperator::NotStartsWith => { + starts_with_predicate(column.expr, expr.literal(), true) + } + op => { + let literal = lit(datum_to_vortex_scalar(expr.literal(), &column.dtype)?); + match op { + PredicateOperator::LessThan => Ok(lt(column.expr, literal)), + PredicateOperator::LessThanOrEq => Ok(lt_eq(column.expr, literal)), + PredicateOperator::GreaterThan => Ok(gt(column.expr, literal)), + PredicateOperator::GreaterThanOrEq => Ok(gt_eq(column.expr, literal)), + PredicateOperator::Eq => Ok(eq(column.expr, literal)), + PredicateOperator::NotEq => Ok(not_eq(column.expr, literal)), + op => Err(unsupported_predicate(op)), + } + } + } + } + BoundPredicate::Set(expr) => { + let Some(column) = column_expr(expr.term(), schema, file_dtype)? else { + return Ok(lit(true)); + }; + let matches = expr + .literals() + .iter() + .map(|datum| { + Ok(eq( + column.expr.clone(), + lit(datum_to_vortex_scalar(datum, &column.dtype)?), + )) + }) + .collect::>>()?; + match expr.op() { + PredicateOperator::In => Ok(or_collect(matches).unwrap_or_else(|| lit(false))), + PredicateOperator::NotIn => Ok(match or_collect(matches) { + Some(any_match) => not(any_match), + None => lit(true), + }), + op => Err(unsupported_predicate(op)), + } + } + } +} + +fn unsupported_predicate(op: PredicateOperator) -> Error { + Error::new( + ErrorKind::FeatureUnsupported, + format!("Predicate operator {op} is not yet supported for vortex data files"), + ) +} + +/// A column reference resolved against the file: the vortex expression that +/// selects it, plus its dtype in the file. +struct FileColumn { + expr: Expression, + dtype: DType, +} + +/// Resolves the field referenced by a bound predicate term to a vortex column +/// expression. Nested struct fields become chains of `get_item`. +/// +/// Returns `None` when the column (or any struct along its path) does not +/// exist in the file, in which case the predicate leaf must evaluate to +/// "keep all rows", mirroring the parquet read path. +fn column_expr( + reference: &BoundReference, + schema: &Schema, + file_dtype: &DType, +) -> Result> { + // Walk the accessor positions through the table schema to build the + // column's name path. + let mut names: Vec = Vec::new(); + let mut current_struct = Some(schema.as_struct()); + let mut accessor = Some(reference.accessor()); + while let Some(acc) = accessor { + let field = current_struct + .and_then(|struct_type| struct_type.fields().get(acc.position())) + .ok_or_else(|| { + Error::new( + ErrorKind::Unexpected, + format!( + "Bound reference {} does not resolve within the table schema", + reference.field().name + ), + ) + })?; + names.push(field.name.clone()); + accessor = acc.inner(); + current_struct = match field.field_type.as_ref() { + Type::Struct(struct_type) => Some(struct_type), + _ => None, + }; + } + + // Resolve the same path in the file's dtype; vortex files are matched by + // name. + let mut dtype = file_dtype.clone(); + for name in &names { + let Some(field_dtype) = dtype + .as_struct_fields_opt() + .and_then(|fields| fields.field(name)) + else { + return Ok(None); + }; + dtype = field_dtype; + } + + let expr = names + .iter() + .fold(root(), |child, name| get_item(name.as_str(), child)); + Ok(Some(FileColumn { expr, dtype })) +} + +/// Builds a NaN check for a float column. +/// +/// Vortex compares floats using IEEE-754 total ordering, in which every NaN +/// bit pattern sorts either above `+inf` (positive NaNs) or below `-inf` +/// (negative NaNs), so NaN checks are expressed as comparisons against the +/// infinities. Null values compare to null and are excluded by the filter, +/// matching the three valued logic of the parquet read path. +fn nan_predicate(column: Expression, field: &NestedField, is_nan: bool) -> Result { + let (pos_inf, neg_inf): (Scalar, Scalar) = match field.field_type.as_ref() { + Type::Primitive(PrimitiveType::Float) => (f32::INFINITY.into(), f32::NEG_INFINITY.into()), + Type::Primitive(PrimitiveType::Double) => (f64::INFINITY.into(), f64::NEG_INFINITY.into()), + ty => { + return Err(Error::new( + ErrorKind::Unexpected, + format!( + "NaN predicate on non-float field {} of type {ty}", + field.name + ), + )); + } + }; + Ok(if is_nan { + or(gt(column.clone(), lit(pos_inf)), lt(column, lit(neg_inf))) + } else { + and( + lt_eq(column.clone(), lit(pos_inf)), + gt_eq(column, lit(neg_inf)), + ) + }) +} + +/// Builds a `LIKE` pattern matching values that start with the literal prefix, +/// escaping any SQL LIKE wildcards contained in it. +fn starts_with_predicate(column: Expression, prefix: &Datum, negated: bool) -> Result { + let PrimitiveLiteral::String(prefix) = prefix.literal() else { + return Err(Error::new( + ErrorKind::Unexpected, + "STARTS WITH predicates require a string literal", + )); + }; + let escaped = prefix + .replace('\\', "\\\\") + .replace('%', "\\%") + .replace('_', "\\_"); + let pattern = format!("{escaped}%"); + Ok(if negated { + not_like(column, lit(pattern)) + } else { + like(column, lit(pattern)) + }) +} + +fn datum_to_vortex_scalar(datum: &Datum, file_dtype: &DType) -> Result { + match (datum.data_type(), datum.literal()) { + (PrimitiveType::Boolean, PrimitiveLiteral::Boolean(v)) => Ok((*v).into()), + (PrimitiveType::Int, PrimitiveLiteral::Int(v)) => Ok((*v).into()), + (PrimitiveType::Long, PrimitiveLiteral::Long(v)) => Ok((*v).into()), + (PrimitiveType::Float, PrimitiveLiteral::Float(v)) => Ok(v.0.into()), + (PrimitiveType::Double, PrimitiveLiteral::Double(v)) => Ok(v.0.into()), + (PrimitiveType::String, PrimitiveLiteral::String(v)) => Ok(v.as_str().into()), + (PrimitiveType::Binary, PrimitiveLiteral::Binary(v)) => Ok(v.as_slice().into()), + (PrimitiveType::Decimal { precision, scale }, PrimitiveLiteral::Int128(v)) => { + let precision = u8::try_from(*precision).map_err(|err| { + Error::new( + ErrorKind::DataInvalid, + format!("Decimal precision {precision} out of range"), + ) + .with_source(err) + })?; + let scale = i8::try_from(*scale).map_err(|err| { + Error::new( + ErrorKind::DataInvalid, + format!("Decimal scale {scale} out of range"), + ) + .with_source(err) + })?; + Ok(Scalar::decimal( + DecimalValue::I128(*v), + DecimalDType::new(precision, scale), + Nullability::NonNullable, + )) + } + (PrimitiveType::Date, PrimitiveLiteral::Int(days)) => { + temporal_scalar(file_dtype, i64::from(*days), TimeUnit::Days) + } + (PrimitiveType::Time, PrimitiveLiteral::Long(micros)) => { + temporal_scalar(file_dtype, *micros, TimeUnit::Microseconds) + } + (PrimitiveType::Timestamp | PrimitiveType::Timestamptz, PrimitiveLiteral::Long(micros)) => { + temporal_scalar(file_dtype, *micros, TimeUnit::Microseconds) + } + ( + PrimitiveType::TimestampNs | PrimitiveType::TimestamptzNs, + PrimitiveLiteral::Long(nanos), + ) => temporal_scalar(file_dtype, *nanos, TimeUnit::Nanoseconds), + (ty, _) => Err(Error::new( + ErrorKind::FeatureUnsupported, + format!("Predicate literals of type {ty} are not yet supported for vortex data files"), + )), + } +} + +/// Builds a literal scalar for comparing against a temporal column. +/// +/// When the file stores the column as a vortex temporal extension type, the +/// literal is converted to the column's exact extension dtype (unit and +/// timezone), since vortex only compares extension values against matching +/// extension dtypes. Plain primitive columns are compared against the raw +/// iceberg value. +fn temporal_scalar(file_dtype: &DType, value: i64, value_unit: TimeUnit) -> Result { + match file_dtype { + DType::Extension(ext) => { + let Some(metadata) = AnyTemporal::try_match(ext) else { + return Err(Error::new( + ErrorKind::FeatureUnsupported, + format!( + "Cannot compare a temporal predicate literal to a column of type {file_dtype}" + ), + )); + }; + let converted = convert_temporal_value(value, value_unit, metadata.time_unit())?; + let nullability = ext.storage_dtype().nullability(); + let storage = match ext.storage_dtype() { + DType::Primitive(PType::I32, _) => Scalar::primitive( + i32::try_from(converted).map_err(|err| { + Error::new( + ErrorKind::DataInvalid, + format!("Temporal value {converted} overflows the column storage type"), + ) + .with_source(err) + })?, + nullability, + ), + DType::Primitive(PType::I64, _) => Scalar::primitive(converted, nullability), + dtype => { + return Err(Error::new( + ErrorKind::FeatureUnsupported, + format!("Unsupported temporal storage type {dtype}"), + )); + } + }; + Ok(Scalar::extension_ref(ext.clone(), storage)) + } + DType::Primitive(PType::I32, _) => Ok(Scalar::primitive( + i32::try_from(value).map_err(|err| { + Error::new( + ErrorKind::DataInvalid, + format!("Temporal value {value} overflows the column storage type"), + ) + .with_source(err) + })?, + Nullability::NonNullable, + )), + DType::Primitive(PType::I64, _) => Ok(Scalar::primitive(value, Nullability::NonNullable)), + dtype => Err(Error::new( + ErrorKind::FeatureUnsupported, + format!("Cannot compare a temporal predicate literal to a column of type {dtype}"), + )), + } +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use arrow_array::{ + Array, Date32Array, Decimal128Array, Float64Array, Int64Array, RecordBatch, StringArray, + StructArray, TimestampMicrosecondArray, + }; + use arrow_schema::{DataType, Field, Schema as ArrowSchema, TimeUnit}; + use futures::TryStreamExt; + use parquet::arrow::PARQUET_FIELD_ID_META_KEY; + use tempfile::TempDir; + use vortex::VortexSessionDefault; + use vortex::session::VortexSession; + + use crate::arrow::ArrowReaderBuilder; + use crate::expr::{Bind, Reference}; + use crate::io::FileIO; + use crate::runtime::Runtime; + use crate::scan::{FileScanTask, FileScanTaskDeleteFile, FileScanTaskStream}; + use crate::spec::{ + DataContentType, DataFileFormat, Datum, NestedField, PrimitiveType, Schema, SchemaRef, Type, + }; + use crate::writer::file_writer::{FileWriter, FileWriterBuilder, VortexWriterBuilder}; + + fn test_schema() -> SchemaRef { + Arc::new( + Schema::builder() + .with_schema_id(0) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(), + NestedField::optional(2, "name", Type::Primitive(PrimitiveType::String)).into(), + NestedField::optional(3, "score", Type::Primitive(PrimitiveType::Double)) + .into(), + NestedField::optional(4, "ts", Type::Primitive(PrimitiveType::Timestamp)) + .into(), + NestedField::optional(5, "date", Type::Primitive(PrimitiveType::Date)).into(), + NestedField::optional( + 6, + "info", + Type::Struct(crate::spec::StructType::new(vec![ + NestedField::optional( + 7, + "points", + Type::Primitive(PrimitiveType::Long), + ) + .into(), + ])), + ) + .into(), + NestedField::optional( + 8, + "price", + Type::Primitive(PrimitiveType::Decimal { + precision: 9, + scale: 2, + }), + ) + .into(), + ]) + .build() + .unwrap(), + ) + } + + fn test_arrow_schema() -> Arc { + let with_field_id = |field: Field, id: i32| { + field.with_metadata(std::collections::HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + id.to_string(), + )])) + }; + Arc::new(ArrowSchema::new(vec![ + with_field_id(Field::new("id", DataType::Int64, false), 1), + with_field_id(Field::new("name", DataType::Utf8, true), 2), + with_field_id(Field::new("score", DataType::Float64, true), 3), + with_field_id( + Field::new("ts", DataType::Timestamp(TimeUnit::Microsecond, None), true), + 4, + ), + with_field_id(Field::new("date", DataType::Date32, true), 5), + with_field_id( + Field::new("info", DataType::Struct(info_struct_fields()), true), + 6, + ), + with_field_id(Field::new("price", DataType::Decimal128(9, 2), true), 8), + ])) + } + + fn info_struct_fields() -> arrow_schema::Fields { + vec![Arc::new( + Field::new("points", DataType::Int64, true).with_metadata( + std::collections::HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "7".to_string(), + )]), + ), + )] + .into() + } + + fn test_batch(ids: Vec, names: Vec>) -> RecordBatch { + let scores = ids + .iter() + .map(|id| match id % 3 { + 0 => None, + 1 => Some(*id as f64 * 1.5), + _ => Some(f64::NAN), + }) + .collect::>(); + let timestamps = ids + .iter() + .map(|id| Some(id * 1_000_000)) + .collect::>(); + let dates = ids + .iter() + .map(|id| Some(*id as i32 * 10)) + .collect::>(); + let points = Int64Array::from(ids.iter().map(|id| Some(id * 100)).collect::>()); + let info = StructArray::new(info_struct_fields(), vec![Arc::new(points)], None); + let prices = Decimal128Array::from( + ids.iter() + .map(|id| Some(i128::from(*id) * 250)) + .collect::>(), + ) + .with_precision_and_scale(9, 2) + .unwrap(); + RecordBatch::try_new(test_arrow_schema(), vec![ + Arc::new(Int64Array::from(ids)), + Arc::new(StringArray::from(names)), + Arc::new(Float64Array::from(scores)), + Arc::new(TimestampMicrosecondArray::from(timestamps)), + Arc::new(Date32Array::from(dates)), + Arc::new(info), + Arc::new(prices), + ]) + .unwrap() + } + + async fn write_test_file(file_path: &str, file_io: &FileIO) -> crate::spec::DataFile { + let output = file_io.new_output(file_path).unwrap(); + let mut writer = VortexWriterBuilder::new(test_schema(), VortexSession::default()) + .build(output) + .await + .unwrap(); + writer + .write(&test_batch(vec![1, 2, 3], vec![Some("a"), Some("b"), None])) + .await + .unwrap(); + writer + .write(&test_batch(vec![4, 5], vec![Some("d"), Some("e")])) + .await + .unwrap(); + let mut builders = writer.close().await.unwrap(); + assert_eq!(builders.len(), 1); + builders + .pop() + .unwrap() + .partition_spec_id(0) + .build() + .unwrap() + } + + fn test_scan_task(file_path: &str, schema: SchemaRef, file_size: u64) -> FileScanTask { + FileScanTask::builder() + .with_file_size_in_bytes(file_size) + .with_start(0) + .with_length(file_size) + .with_data_file_path(file_path.to_string()) + .with_data_file_format(DataFileFormat::Vortex) + .with_schema(schema.clone()) + .with_project_field_ids(schema.as_struct().fields().iter().map(|f| f.id).collect()) + .with_case_sensitive(false) + .build() + } + + #[tokio::test] + async fn test_vortex_roundtrip() { + let tmp_dir = TempDir::new().unwrap(); + let file_path = format!("{}/data.vortex", tmp_dir.path().to_str().unwrap()); + let file_io = FileIO::new_with_fs(); + + let data_file = write_test_file(&file_path, &file_io).await; + assert_eq!(data_file.file_format(), DataFileFormat::Vortex); + assert_eq!(data_file.record_count(), 5); + assert_eq!(data_file.value_counts()[&1], 5); + assert_eq!(data_file.null_value_counts()[&1], 0); + assert_eq!(data_file.null_value_counts()[&2], 1); + assert_eq!(data_file.nan_value_counts()[&3], 2); + + // Min/max bounds computed from the written batches. NaN values are + // excluded from float bounds; nested fields carry no bounds. + assert_eq!(data_file.lower_bounds()[&1], Datum::long(1)); + assert_eq!(data_file.upper_bounds()[&1], Datum::long(5)); + assert_eq!(data_file.lower_bounds()[&2], Datum::string("a")); + assert_eq!(data_file.upper_bounds()[&2], Datum::string("e")); + assert_eq!(data_file.lower_bounds()[&3], Datum::double(1.5)); + assert_eq!(data_file.upper_bounds()[&3], Datum::double(6.0)); + assert_eq!( + data_file.lower_bounds()[&4], + Datum::timestamp_micros(1_000_000) + ); + assert_eq!( + data_file.upper_bounds()[&4], + Datum::timestamp_micros(5_000_000) + ); + assert_eq!(data_file.lower_bounds()[&5], Datum::date(10)); + assert_eq!(data_file.upper_bounds()[&5], Datum::date(50)); + assert_eq!( + data_file.lower_bounds()[&8].literal(), + &crate::spec::PrimitiveLiteral::Int128(250) + ); + assert_eq!( + data_file.upper_bounds()[&8].literal(), + &crate::spec::PrimitiveLiteral::Int128(1250) + ); + assert!(!data_file.lower_bounds().contains_key(&6)); + assert!(!data_file.lower_bounds().contains_key(&7)); + + let file_size = std::fs::metadata(&file_path).unwrap().len(); + assert_eq!(data_file.file_size_in_bytes(), file_size); + + let reader = ArrowReaderBuilder::new(file_io, Runtime::current()).build(); + let task = test_scan_task(&file_path, test_schema(), file_size); + let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream; + let result = reader.read(tasks).unwrap(); + let batches: Vec = result.stream().try_collect().await.unwrap(); + + let total_rows: usize = batches.iter().map(|batch| batch.num_rows()).sum(); + assert_eq!(total_rows, 5); + + let ids: Vec = batches + .iter() + .flat_map(|batch| { + batch + .column_by_name("id") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap() + .values() + .to_vec() + }) + .collect(); + assert_eq!(ids, vec![1, 2, 3, 4, 5]); + + let names: Vec> = batches + .iter() + .flat_map(|batch| { + batch + .column_by_name("name") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap() + .iter() + .map(|name| name.map(|s| s.to_string())) + .collect::>() + }) + .collect(); + assert_eq!(names, vec![ + Some("a".to_string()), + Some("b".to_string()), + None, + Some("d".to_string()), + Some("e".to_string()) + ]); + } + + #[tokio::test] + async fn test_vortex_read_with_predicate() { + let tmp_dir = TempDir::new().unwrap(); + let file_path = format!("{}/data.vortex", tmp_dir.path().to_str().unwrap()); + let file_io = FileIO::new_with_fs(); + write_test_file(&file_path, &file_io).await; + let file_size = std::fs::metadata(&file_path).unwrap().len(); + + let schema = test_schema(); + let predicate = Reference::new("id") + .greater_than(Datum::long(3)) + .bind(schema.clone(), true) + .unwrap(); + let mut task = test_scan_task(&file_path, schema, file_size); + task.predicate = Some(predicate); + + let reader = ArrowReaderBuilder::new(file_io, Runtime::current()).build(); + let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream; + let batches: Vec = reader + .read(tasks) + .unwrap() + .stream() + .try_collect() + .await + .unwrap(); + + let ids: Vec = batches + .iter() + .flat_map(|batch| { + batch + .column_by_name("id") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap() + .values() + .to_vec() + }) + .collect(); + assert_eq!(ids, vec![4, 5]); + } + + #[tokio::test] + async fn test_vortex_read_with_evolved_schema() { + let tmp_dir = TempDir::new().unwrap(); + let file_path = format!("{}/data.vortex", tmp_dir.path().to_str().unwrap()); + let file_io = FileIO::new_with_fs(); + write_test_file(&file_path, &file_io).await; + let file_size = std::fs::metadata(&file_path).unwrap().len(); + + // The table schema gained a column after the file was written; it must + // be filled with nulls when reading old files. + let evolved_schema = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(), + NestedField::optional(99, "extra", Type::Primitive(PrimitiveType::String)) + .into(), + ]) + .build() + .unwrap(), + ); + + let reader = ArrowReaderBuilder::new(file_io, Runtime::current()).build(); + let task = test_scan_task(&file_path, evolved_schema, file_size); + let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream; + let batches: Vec = reader + .read(tasks) + .unwrap() + .stream() + .try_collect() + .await + .unwrap(); + + let total_rows: usize = batches.iter().map(|batch| batch.num_rows()).sum(); + assert_eq!(total_rows, 5); + for batch in &batches { + assert_eq!(batch.num_columns(), 2); + let extra = batch.column_by_name("extra").unwrap(); + assert_eq!(extra.null_count(), extra.len()); + } + } + + const FIELD_ID_POSITIONAL_DELETE_FILE_PATH: u64 = 2147483546; + const FIELD_ID_POSITIONAL_DELETE_POS: u64 = 2147483545; + + fn write_positional_delete_file(path: &str, data_file_path: &str, positions: &[i64]) { + let schema = Arc::new(ArrowSchema::new(vec![ + Field::new("file_path", DataType::Utf8, false).with_metadata( + std::collections::HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + FIELD_ID_POSITIONAL_DELETE_FILE_PATH.to_string(), + )]), + ), + Field::new("pos", DataType::Int64, false).with_metadata( + std::collections::HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + FIELD_ID_POSITIONAL_DELETE_POS.to_string(), + )]), + ), + ])); + let batch = RecordBatch::try_new(schema.clone(), vec![ + Arc::new(StringArray::from(vec![data_file_path; positions.len()])), + Arc::new(Int64Array::from(positions.to_vec())), + ]) + .unwrap(); + let file = std::fs::File::create(path).unwrap(); + let mut writer = parquet::arrow::ArrowWriter::try_new(file, schema, None).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + } + + fn write_equality_delete_file(path: &str, ids_to_delete: &[i64]) { + let schema = Arc::new(ArrowSchema::new(vec![ + Field::new("id", DataType::Int64, false).with_metadata( + std::collections::HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "1".to_string(), + )]), + ), + ])); + let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(Int64Array::from( + ids_to_delete.to_vec(), + ))]) + .unwrap(); + let file = std::fs::File::create(path).unwrap(); + let mut writer = parquet::arrow::ArrowWriter::try_new(file, schema, None).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + } + + fn delete_file_entry(path: &str, file_type: DataContentType) -> FileScanTaskDeleteFile { + let equality_ids = (file_type == DataContentType::EqualityDeletes).then(|| vec![1]); + FileScanTaskDeleteFile::builder() + .with_file_path(path.to_string()) + .with_file_size_in_bytes(std::fs::metadata(path).unwrap().len()) + .with_file_type(file_type) + .with_partition_spec_id(0) + .with_equality_ids(equality_ids) + .build() + } + + async fn collect_ids(reader: &crate::arrow::ArrowReader, task: FileScanTask) -> Vec { + let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream; + let batches: Vec = reader + .clone() + .read(tasks) + .unwrap() + .stream() + .try_collect() + .await + .unwrap(); + batches + .iter() + .flat_map(|batch| { + batch + .column_by_name("id") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap() + .values() + .to_vec() + }) + .collect() + } + + #[tokio::test] + async fn test_vortex_read_with_positional_deletes() { + let tmp_dir = TempDir::new().unwrap(); + let file_path = format!("{}/data.vortex", tmp_dir.path().to_str().unwrap()); + let delete_path = format!("{}/pos-del.parquet", tmp_dir.path().to_str().unwrap()); + let file_io = FileIO::new_with_fs(); + write_test_file(&file_path, &file_io).await; + let file_size = std::fs::metadata(&file_path).unwrap().len(); + + // Delete row ordinals 0 and 3 (ids 1 and 4). + write_positional_delete_file(&delete_path, &file_path, &[0, 3]); + + let mut task = test_scan_task(&file_path, test_schema(), file_size); + task.deletes = vec![delete_file_entry( + &delete_path, + DataContentType::PositionDeletes, + )]; + + let reader = ArrowReaderBuilder::new(file_io, Runtime::current()).build(); + assert_eq!(collect_ids(&reader, task).await, vec![2, 3, 5]); + } + + #[tokio::test] + async fn test_vortex_read_with_equality_deletes() { + let tmp_dir = TempDir::new().unwrap(); + let file_path = format!("{}/data.vortex", tmp_dir.path().to_str().unwrap()); + let delete_path = format!("{}/eq-del.parquet", tmp_dir.path().to_str().unwrap()); + let file_io = FileIO::new_with_fs(); + write_test_file(&file_path, &file_io).await; + let file_size = std::fs::metadata(&file_path).unwrap().len(); + + write_equality_delete_file(&delete_path, &[2, 5]); + + let mut task = test_scan_task(&file_path, test_schema(), file_size); + task.deletes = vec![delete_file_entry( + &delete_path, + DataContentType::EqualityDeletes, + )]; + + let reader = ArrowReaderBuilder::new(file_io, Runtime::current()).build(); + assert_eq!(collect_ids(&reader, task).await, vec![1, 3, 4]); + } + + #[tokio::test] + async fn test_vortex_read_with_deletes_and_predicate() { + let tmp_dir = TempDir::new().unwrap(); + let file_path = format!("{}/data.vortex", tmp_dir.path().to_str().unwrap()); + let pos_delete_path = format!("{}/pos-del.parquet", tmp_dir.path().to_str().unwrap()); + let eq_delete_path = format!("{}/eq-del.parquet", tmp_dir.path().to_str().unwrap()); + let file_io = FileIO::new_with_fs(); + write_test_file(&file_path, &file_io).await; + let file_size = std::fs::metadata(&file_path).unwrap().len(); + + // Positional delete of ordinal 4 (id 5), equality delete of id 3, and + // a predicate id > 1: expect [2, 4]. + write_positional_delete_file(&pos_delete_path, &file_path, &[4]); + write_equality_delete_file(&eq_delete_path, &[3]); + + let schema = test_schema(); + let mut task = test_scan_task(&file_path, schema.clone(), file_size); + task.deletes = vec![ + delete_file_entry(&pos_delete_path, DataContentType::PositionDeletes), + delete_file_entry(&eq_delete_path, DataContentType::EqualityDeletes), + ]; + task.predicate = Some( + Reference::new("id") + .greater_than(Datum::long(1)) + .bind(schema, true) + .unwrap(), + ); + + let reader = ArrowReaderBuilder::new(file_io, Runtime::current()).build(); + assert_eq!(collect_ids(&reader, task).await, vec![2, 4]); + } + + async fn ids_matching(predicate: crate::expr::Predicate) -> Vec { + let tmp_dir = TempDir::new().unwrap(); + let file_path = format!("{}/data.vortex", tmp_dir.path().to_str().unwrap()); + let file_io = FileIO::new_with_fs(); + write_test_file(&file_path, &file_io).await; + let file_size = std::fs::metadata(&file_path).unwrap().len(); + + let schema = test_schema(); + let mut task = test_scan_task(&file_path, schema.clone(), file_size); + task.predicate = Some(predicate.bind(schema, true).unwrap()); + + let reader = ArrowReaderBuilder::new(file_io, Runtime::current()).build(); + collect_ids(&reader, task).await + } + + #[tokio::test] + async fn test_vortex_read_with_timestamp_predicate() { + // ts = id seconds; ts >= 3s selects ids 3, 4, 5. + let ids = ids_matching( + Reference::new("ts").greater_than_or_equal_to(Datum::timestamp_micros(3_000_000)), + ) + .await; + assert_eq!(ids, vec![3, 4, 5]); + } + + #[tokio::test] + async fn test_vortex_read_with_date_predicate() { + // date = id * 10 days; date > 20 days selects ids 3, 4, 5. + let ids = ids_matching(Reference::new("date").greater_than(Datum::date(20))).await; + assert_eq!(ids, vec![3, 4, 5]); + } + + #[tokio::test] + async fn test_vortex_read_with_decimal_predicate() { + // price = id * 2.50; price > 5.00 selects ids 3, 4, 5. + let ids = ids_matching( + Reference::new("price").greater_than(Datum::decimal_from_str("5.00").unwrap()), + ) + .await; + assert_eq!(ids, vec![3, 4, 5]); + } + + #[tokio::test] + async fn test_vortex_read_with_in_predicate() { + let ids = ids_matching(Reference::new("id").is_in([ + Datum::long(2), + Datum::long(4), + Datum::long(7), + ])) + .await; + assert_eq!(ids, vec![2, 4]); + } + + #[tokio::test] + async fn test_vortex_read_with_starts_with_predicate() { + // names are [a, b, null, d, e]. + let ids = ids_matching(Reference::new("name").starts_with(Datum::string("d"))).await; + assert_eq!(ids, vec![4]); + + // Rows with a null name are excluded by NOT STARTS WITH. + let ids = ids_matching(Reference::new("name").not_starts_with(Datum::string("d"))).await; + assert_eq!(ids, vec![1, 2, 5]); + } + + #[tokio::test] + async fn test_vortex_read_with_nan_predicates() { + // scores are [1.5, NaN, null, 6.0, NaN]. + let ids = ids_matching(Reference::new("score").is_nan()).await; + assert_eq!(ids, vec![2, 5]); + + // Rows with a null score are excluded by NOT NAN. + let ids = ids_matching(Reference::new("score").is_not_nan()).await; + assert_eq!(ids, vec![1, 4]); + } + + #[tokio::test] + async fn test_vortex_read_with_nested_field_predicate() { + // info.points = id * 100; points > 300 selects ids 4, 5. + let ids = ids_matching(Reference::new("info.points").greater_than(Datum::long(300))).await; + assert_eq!(ids, vec![4, 5]); + } + + #[tokio::test] + async fn test_vortex_read_with_predicate_on_missing_column() { + let tmp_dir = TempDir::new().unwrap(); + let file_path = format!("{}/data.vortex", tmp_dir.path().to_str().unwrap()); + let file_io = FileIO::new_with_fs(); + write_test_file(&file_path, &file_io).await; + let file_size = std::fs::metadata(&file_path).unwrap().len(); + + // "extra" was added to the table schema after the file was written. + // Following the parquet read path, predicate leaves on columns missing + // from the file keep all rows, so only the id predicate takes effect. + let evolved_schema = Arc::new( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(), + NestedField::optional(99, "extra", Type::Primitive(PrimitiveType::String)) + .into(), + ]) + .build() + .unwrap(), + ); + let predicate = Reference::new("extra") + .equal_to(Datum::string("x")) + .and(Reference::new("id").greater_than(Datum::long(3))); + + let mut task = test_scan_task(&file_path, evolved_schema.clone(), file_size); + task.predicate = Some(predicate.bind(evolved_schema, true).unwrap()); + + let reader = ArrowReaderBuilder::new(file_io, Runtime::current()).build(); + assert_eq!(collect_ids(&reader, task).await, vec![4, 5]); + } +} diff --git a/crates/iceberg/src/arrow/vortex_util.rs b/crates/iceberg/src/arrow/vortex_util.rs new file mode 100644 index 0000000000..37c73ccda0 --- /dev/null +++ b/crates/iceberg/src/arrow/vortex_util.rs @@ -0,0 +1,56 @@ +// 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. + +//! Shared helpers for the vortex file format integration. + +use vortex::error::VortexError; +use vortex::extension::datetime::TimeUnit; + +use crate::{Error, ErrorKind, Result}; + +/// Converts a [`VortexError`] into an iceberg [`Error`]. +pub(crate) fn to_iceberg_error(err: VortexError) -> Error { + Error::new(ErrorKind::Unexpected, "Vortex error").with_source(err) +} + +/// Losslessly converts a temporal value between units; conversions that would +/// lose precision (e.g. microseconds to seconds) are rejected. +pub(crate) fn convert_temporal_value(value: i64, from: TimeUnit, to: TimeUnit) -> Result { + fn nanos_per(unit: TimeUnit) -> i64 { + match unit { + TimeUnit::Nanoseconds => 1, + TimeUnit::Microseconds => 1_000, + TimeUnit::Milliseconds => 1_000_000, + TimeUnit::Seconds => 1_000_000_000, + TimeUnit::Days => 86_400_000_000_000, + } + } + + let (from_nanos, to_nanos) = (nanos_per(from), nanos_per(to)); + if from_nanos % to_nanos != 0 { + return Err(Error::new( + ErrorKind::FeatureUnsupported, + format!("Lossy temporal unit conversion from {from} to {to}"), + )); + } + value.checked_mul(from_nanos / to_nanos).ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!("Temporal value {value} overflows when converted from {from} to {to}"), + ) + }) +} diff --git a/crates/iceberg/src/expr/accessor.rs b/crates/iceberg/src/expr/accessor.rs index c4623def99..8f94bc020f 100644 --- a/crates/iceberg/src/expr/accessor.rs +++ b/crates/iceberg/src/expr/accessor.rs @@ -52,6 +52,10 @@ impl StructAccessor { self.position } + pub(crate) fn inner(&self) -> Option<&StructAccessor> { + self.inner.as_deref() + } + pub(crate) fn r#type(&self) -> &PrimitiveType { &self.r#type } diff --git a/crates/iceberg/src/spec/manifest/data_file.rs b/crates/iceberg/src/spec/manifest/data_file.rs index 77bd046f8a..108e566cbe 100644 --- a/crates/iceberg/src/spec/manifest/data_file.rs +++ b/crates/iceberg/src/spec/manifest/data_file.rs @@ -383,6 +383,8 @@ pub enum DataFileFormat { Parquet, /// Puffin file format: Puffin, + /// Vortex file format: + Vortex, } impl FromStr for DataFileFormat { @@ -394,6 +396,7 @@ impl FromStr for DataFileFormat { "orc" => Ok(Self::Orc), "parquet" => Ok(Self::Parquet), "puffin" => Ok(Self::Puffin), + "vortex" => Ok(Self::Vortex), _ => Err(Error::new( ErrorKind::DataInvalid, format!("Unsupported data file format: {s}"), @@ -409,6 +412,7 @@ impl std::fmt::Display for DataFileFormat { DataFileFormat::Orc => write!(f, "orc"), DataFileFormat::Parquet => write!(f, "parquet"), DataFileFormat::Puffin => write!(f, "puffin"), + DataFileFormat::Vortex => write!(f, "vortex"), } } } diff --git a/crates/iceberg/src/writer/file_writer/mod.rs b/crates/iceberg/src/writer/file_writer/mod.rs index 101919f5b3..0eaca931db 100644 --- a/crates/iceberg/src/writer/file_writer/mod.rs +++ b/crates/iceberg/src/writer/file_writer/mod.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -//! This module contains the writer for data file format supported by iceberg: parquet, orc. +//! This module contains the writer for data file format supported by iceberg: parquet, orc, vortex. use arrow_array::RecordBatch; use futures::Future; @@ -26,6 +26,8 @@ use crate::spec::DataFileBuilder; mod parquet_writer; pub use parquet_writer::{ParquetWriter, ParquetWriterBuilder}; +mod vortex_writer; +pub use vortex_writer::{VortexWriter, VortexWriterBuilder}; use crate::io::OutputFile; diff --git a/crates/iceberg/src/writer/file_writer/vortex_writer.rs b/crates/iceberg/src/writer/file_writer/vortex_writer.rs new file mode 100644 index 0000000000..0fc3e2a14f --- /dev/null +++ b/crates/iceberg/src/writer/file_writer/vortex_writer.rs @@ -0,0 +1,427 @@ +// 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. + +//! The module contains the file writer for the vortex file format. + +use std::collections::HashMap; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; + +use arrow_array::RecordBatch; +use arrow_schema::SchemaRef as ArrowSchemaRef; +use bytes::Bytes; +use futures::channel::mpsc; +use futures::{SinkExt, StreamExt}; +use vortex::array::ArrayRef; +use vortex::array::arrow::FromArrowArray; +use vortex::array::stats::TypedStatsSetRef; +use vortex::array::stream::ArrayStreamAdapter; +use vortex::dtype::arrow::FromArrowType; +use vortex::dtype::extension::Matcher; +use vortex::dtype::{DType, PType}; +use vortex::expr::stats::{Stat, StatsProvider, StatsProviderExt}; +use vortex::extension::datetime::{AnyTemporal, TimeUnit}; +use vortex::file::{WriteOptionsSessionExt, WriteStrategyBuilder, WriteSummary}; +use vortex::io::{IoBuf, VortexWrite}; +use vortex::layout::LayoutStrategy; +use vortex::scalar::Scalar; +use vortex::session::VortexSession; + +use super::{FileWriter, FileWriterBuilder}; +use crate::arrow::{convert_temporal_value, to_iceberg_error}; +use crate::io::{FileWrite, OutputFile}; +use crate::spec::{ + DataContentType, DataFileBuilder, DataFileFormat, Datum, PrimitiveLiteral, PrimitiveType, + SchemaRef, Struct, Type, +}; +use crate::{Error, ErrorKind, Result}; + +/// VortexWriterBuilder is used to build a [`VortexWriter`]. +#[derive(Clone, Debug)] +pub struct VortexWriterBuilder { + schema: SchemaRef, + session: VortexSession, +} + +impl VortexWriterBuilder { + /// Create a new `VortexWriterBuilder`. + /// + /// The session is shared by all writers the builder builds. It captures + /// the current tokio runtime handle at construction time, so it must be + /// created from within the runtime that will drive the writes. + pub fn new(schema: SchemaRef, session: VortexSession) -> Self { + Self { schema, session } + } +} + +impl FileWriterBuilder for VortexWriterBuilder { + type R = VortexWriter; + + async fn build(&self, output_file: OutputFile) -> Result { + Ok(VortexWriter { + schema: self.schema.clone(), + session: self.session.clone(), + output_file, + inner: None, + current_row_num: 0, + }) + } +} + +/// `VortexWriter` writes arrow data into vortex files on storage. +/// +/// Record batches are streamed into a vortex file writer running as a +/// background task, so data is compressed and flushed to storage as it +/// arrives instead of being buffered until close. +pub struct VortexWriter { + schema: SchemaRef, + session: VortexSession, + output_file: OutputFile, + inner: Option, + current_row_num: usize, +} + +/// The state of an in-progress vortex file write. +/// +/// Vortex's push-based writer is not `Send`, so instead of holding it in +/// place the batches are sent through a bounded channel to a spawned task +/// that owns the write end-to-end and returns the [`WriteSummary`] when +/// finished. The channel is bounded so the task applies backpressure instead +/// of buffering batches. +struct StreamingWriter { + arrow_schema: ArrowSchemaRef, + batches: mpsc::Sender, + task: tokio::task::JoinHandle>, + bytes_written: Arc, + strategy: Arc, +} + +impl StreamingWriter { + /// Awaits the background task after a failure to surface its error. + async fn task_error(self) -> Error { + match self.task.await { + Ok(Ok(_)) => Error::new( + ErrorKind::Unexpected, + "Vortex write task stopped accepting batches but reported success", + ), + Ok(Err(err)) => err, + Err(err) => { + Error::new(ErrorKind::Unexpected, "Vortex write task panicked").with_source(err) + } + } + } +} + +impl VortexWriter { + async fn start(&self, arrow_schema: ArrowSchemaRef) -> Result { + let dtype = DType::from_arrow(arrow_schema.clone()); + let file = self.output_file.writer().await?; + let bytes_written = Arc::new(AtomicU64::new(0)); + let strategy = WriteStrategyBuilder::default().build(); + + let (batches, receiver) = mpsc::channel::(1); + let stream = ArrayStreamAdapter::new(dtype, receiver.map(Ok)); + let write_options = self + .session + .write_options() + .with_strategy(Arc::clone(&strategy)); + + let mut sink = FileWriteSink { + file, + bytes_written: Arc::clone(&bytes_written), + }; + let task = tokio::spawn(async move { + let summary = write_options + .write(&mut sink, stream) + .await + .map_err(to_iceberg_error)?; + // The vortex writer flushes but does not close its sink; finalize + // the iceberg output file explicitly. + sink.file.close().await?; + Ok(summary) + }); + + Ok(StreamingWriter { + arrow_schema, + batches, + task, + bytes_written, + strategy, + }) + } + + fn data_file_builder( + &self, + summary: &WriteSummary, + arrow_schema: &ArrowSchemaRef, + ) -> Result { + let record_count = summary.row_count(); + + // Take value/null/NaN counts and min/max bounds for top-level + // primitive fields from the statistics collected by the vortex writer. + // Nested fields are left unset; metrics evaluators treat missing + // entries as "rows might match". + let mut value_counts: HashMap = HashMap::new(); + let mut null_value_counts: HashMap = HashMap::new(); + let mut nan_value_counts: HashMap = HashMap::new(); + let mut lower_bounds: HashMap = HashMap::new(); + let mut upper_bounds: HashMap = HashMap::new(); + let file_stats = summary.footer().statistics(); + for (index, field) in arrow_schema.fields().iter().enumerate() { + let Some(iceberg_field) = self.schema.field_by_name(field.name()) else { + continue; + }; + let Type::Primitive(primitive_type) = iceberg_field.field_type.as_ref() else { + continue; + }; + // Top-level columns hold exactly one value per row. + value_counts.insert(iceberg_field.id, record_count); + let Some(stats) = file_stats.filter(|stats| index < stats.stats_sets().len()) else { + continue; + }; + let (stats_set, field_dtype) = stats.get(index); + let typed_stats = stats_set.as_typed_ref(field_dtype); + if let Some(null_count) = typed_stats.get_as::(Stat::NullCount).as_exact() { + null_value_counts.insert(iceberg_field.id, null_count); + } + if matches!(primitive_type, PrimitiveType::Float | PrimitiveType::Double) + && let Some(nan_count) = typed_stats.get_as::(Stat::NaNCount).as_exact() + { + nan_value_counts.insert(iceberg_field.id, nan_count); + } + if let Some(min) = stat_bound(&typed_stats, Stat::Min, primitive_type) { + lower_bounds.insert(iceberg_field.id, min); + } + if let Some(max) = stat_bound(&typed_stats, Stat::Max, primitive_type) { + upper_bounds.insert(iceberg_field.id, max); + } + } + + let mut builder = DataFileBuilder::default(); + builder + .content(DataContentType::Data) + .file_path(self.output_file.location().to_string()) + .file_format(DataFileFormat::Vortex) + .partition(Struct::empty()) + .record_count(record_count) + .file_size_in_bytes(summary.size()) + .value_counts(value_counts) + .null_value_counts(null_value_counts) + .nan_value_counts(nan_value_counts) + .lower_bounds(lower_bounds) + .upper_bounds(upper_bounds); + // Vortex files can be split at arbitrary row offsets, so no physical + // split offsets are recorded. + + Ok(builder) + } +} + +/// Converts a min/max statistic into an iceberg bound [`Datum`]. +/// +/// Truncated variable-length statistics remain sound bounds: the min is +/// prefix-truncated (still a lower bound) and the max is upper-adjusted or +/// absent, so both exact and inexact values are used. +fn stat_bound( + stats: &TypedStatsSetRef, + stat: Stat, + primitive_type: &PrimitiveType, +) -> Option { + let value = stats.get(stat).into_inner()?; + vortex_scalar_to_datum(value, primitive_type) +} + +/// Converts a vortex [`Scalar`] into an iceberg [`Datum`] of the given +/// primitive type. +/// +/// Returns `None` for null scalars, NaN float values (excluded from bounds by +/// the iceberg spec; NaN counts are tracked separately), and types whose +/// bounds are not computed (`Uuid`, `Fixed`). +fn vortex_scalar_to_datum(scalar: Scalar, primitive_type: &PrimitiveType) -> Option { + match primitive_type { + PrimitiveType::Boolean => Some(Datum::bool(scalar.as_bool_opt()?.value()?)), + PrimitiveType::Int => Some(Datum::int(scalar.as_primitive_opt()?.typed_value::()?)), + PrimitiveType::Long => Some(Datum::long( + scalar.as_primitive_opt()?.typed_value::()?, + )), + PrimitiveType::Float => { + let value = scalar.as_primitive_opt()?.typed_value::()?; + (!value.is_nan()).then(|| Datum::float(value)) + } + PrimitiveType::Double => { + let value = scalar.as_primitive_opt()?.typed_value::()?; + (!value.is_nan()).then(|| Datum::double(value)) + } + PrimitiveType::Date => { + let days = temporal_value(scalar, TimeUnit::Days)?; + Some(Datum::date(i32::try_from(days).ok()?)) + } + PrimitiveType::Time => { + Datum::time_micros(temporal_value(scalar, TimeUnit::Microseconds)?).ok() + } + PrimitiveType::Timestamp => Some(Datum::timestamp_micros(temporal_value( + scalar, + TimeUnit::Microseconds, + )?)), + PrimitiveType::Timestamptz => Some(Datum::timestamptz_micros(temporal_value( + scalar, + TimeUnit::Microseconds, + )?)), + PrimitiveType::TimestampNs => Some(Datum::timestamp_nanos(temporal_value( + scalar, + TimeUnit::Nanoseconds, + )?)), + PrimitiveType::TimestamptzNs => Some(Datum::timestamptz_nanos(temporal_value( + scalar, + TimeUnit::Nanoseconds, + )?)), + PrimitiveType::String => Some(Datum::string(scalar.as_utf8_opt()?.value()?.as_str())), + PrimitiveType::Binary => Some(Datum::binary( + scalar.as_binary_opt()?.value()?.as_slice().iter().copied(), + )), + PrimitiveType::Decimal { .. } => { + let value = scalar.as_decimal_opt()?.decimal_value()?.cast::()?; + Some(Datum::new( + primitive_type.clone(), + PrimitiveLiteral::Int128(value), + )) + } + // Uuid and Fixed bounds are not computed. + _ => None, + } +} + +/// Extracts a temporal scalar's storage value converted to the given unit. +/// +/// Vortex stores temporal columns as extension types whose metadata carries +/// the time unit; plain integer columns are assumed to already be in the +/// requested unit. +fn temporal_value(scalar: Scalar, unit: TimeUnit) -> Option { + match scalar.dtype() { + DType::Extension(ext) => { + let metadata = AnyTemporal::try_match(ext)?; + let storage = scalar.as_extension_opt()?.to_storage_scalar(); + let storage = storage.as_primitive_opt()?; + let value = match storage.ptype() { + PType::I32 => i64::from(storage.typed_value::()?), + PType::I64 => storage.typed_value::()?, + _ => return None, + }; + convert_temporal_value(value, metadata.time_unit(), unit).ok() + } + DType::Primitive(PType::I32, _) => scalar + .as_primitive_opt()? + .typed_value::() + .map(i64::from), + DType::Primitive(PType::I64, _) => scalar.as_primitive_opt()?.typed_value::(), + _ => None, + } +} + +/// Adapts an iceberg [`FileWrite`] into a [`VortexWrite`] sink, counting the +/// bytes flushed to storage. +struct FileWriteSink { + file: Box, + bytes_written: Arc, +} + +impl VortexWrite for FileWriteSink { + async fn write_all(&mut self, buffer: B) -> std::io::Result { + self.file + .write(Bytes::copy_from_slice(buffer.as_slice())) + .await + .map_err(std::io::Error::other)?; + self.bytes_written + .fetch_add(buffer.as_slice().len() as u64, Ordering::Relaxed); + Ok(buffer) + } + + async fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + + async fn shutdown(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +impl FileWriter for VortexWriter { + async fn write(&mut self, batch: &RecordBatch) -> Result<()> { + // Skip empty batch + if batch.num_rows() == 0 { + return Ok(()); + } + + if self.inner.is_none() { + self.inner = Some(self.start(batch.schema()).await?); + } + let inner = self.inner.as_mut().expect("writer started above"); + + let array = ArrayRef::from_arrow(batch.clone(), false).map_err(to_iceberg_error)?; + if inner.batches.send(array).await.is_err() { + // The background task exited early; surface its error. + let inner = self.inner.take().expect("writer started above"); + return Err(inner.task_error().await); + } + self.current_row_num += batch.num_rows(); + + Ok(()) + } + + async fn close(mut self) -> Result> { + let Some(inner) = self.inner.take() else { + return Ok(vec![]); + }; + + let StreamingWriter { + arrow_schema, + batches, + task, + .. + } = inner; + // Close the input channel to signal end-of-stream to the write task. + drop(batches); + let summary = task.await.map_err(|err| { + Error::new(ErrorKind::Unexpected, "Vortex write task panicked").with_source(err) + })??; + + Ok(vec![self.data_file_builder(&summary, &arrow_schema)?]) + } +} + +impl super::super::CurrentFileStatus for VortexWriter { + fn current_file_path(&self) -> String { + self.output_file.location().to_string() + } + + fn current_row_num(&self) -> usize { + self.current_row_num + } + + fn current_written_size(&self) -> usize { + // Bytes already flushed to storage plus bytes buffered by the layout + // strategy. This underestimates the final file size by the footer, + // which is only serialized at close time. + self.inner + .as_ref() + .map(|inner| { + (inner.bytes_written.load(Ordering::Relaxed) + inner.strategy.buffered_bytes()) + as usize + }) + .unwrap_or(0) + } +}