diff --git a/Cargo.lock b/Cargo.lock index 126dfeb1..ffcb3c7a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -512,7 +512,7 @@ dependencies = [ "futures-lite", "parking", "polling", - "rustix 1.1.4", + "rustix", "slab", "windows-sys 0.61.2", ] @@ -554,7 +554,7 @@ dependencies = [ "cfg-if", "event-listener", "futures-lite", - "rustix 1.1.4", + "rustix", ] [[package]] @@ -569,7 +569,7 @@ dependencies = [ "cfg-if", "futures-core", "futures-io", - "rustix 1.1.4", + "rustix", "signal-hook-registry", "slab", "windows-sys 0.61.2", @@ -770,7 +770,7 @@ dependencies = [ "proc-macro2", "quote", "regex", - "rustc-hash 2.1.3", + "rustc-hash", "shlex 1.3.0", "syn 2.0.119", ] @@ -2346,12 +2346,6 @@ dependencies = [ "windows-sys 0.60.2", ] -[[package]] -name = "downcast-rs" -version = "1.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "75b325c5dbd37f80359721ad39aca5a29fb04c89279657cffdda8736d0c0b9d2" - [[package]] name = "downcast-rs" version = "2.0.2" @@ -2604,23 +2598,13 @@ dependencies = [ "percent-encoding", ] -[[package]] -name = "fs4" -version = "0.8.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f7e180ac76c23b45e767bd7ae9579bc0bb458618c4bc71835926e098e61d15f8" -dependencies = [ - "rustix 0.38.44", - "windows-sys 0.52.0", -] - [[package]] name = "fs4" version = "0.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8640e34b88f7652208ce9e88b1a37a2ae95227d84abec377ccd3c5cfeb141ed4" dependencies = [ - "rustix 1.1.4", + "rustix", "windows-sys 0.59.0", ] @@ -2636,7 +2620,7 @@ version = "0.5.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b13ac798afc0d9194eb4efefef8b9332efbd80b43f302a968cb8cb23b9d5360" dependencies = [ - "rustc-hash 2.1.3", + "rustc-hash", ] [[package]] @@ -2882,8 +2866,6 @@ version = "0.15.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" dependencies = [ - "allocator-api2", - "equivalent", "foldhash 0.1.5", ] @@ -3338,18 +3320,6 @@ dependencies = [ "generic-array", ] -[[package]] -name = "instant" -version = "0.1.13" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e0242819d153cba4b4b05a5a8f2a7e9bbf97b6055b2a002b395c96b5ff3c0222" -dependencies = [ - "cfg-if", - "js-sys", - "wasm-bindgen", - "web-sys", -] - [[package]] name = "integer-encoding" version = "3.0.4" @@ -3377,15 +3347,6 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" -[[package]] -name = "itertools" -version = "0.12.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ba291022dbbd398a455acf126c1e341954079855bc60dfdda641363bd6922569" -dependencies = [ - "either", -] - [[package]] name = "itertools" version = "0.13.0" @@ -3431,7 +3392,7 @@ dependencies = [ "jieba-macros", "phf 0.13.1", "regex", - "rustc-hash 2.1.3", + "rustc-hash", ] [[package]] @@ -3770,12 +3731,6 @@ dependencies = [ "redox_syscall 0.9.0", ] -[[package]] -name = "linux-raw-sys" -version = "0.4.15" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d26c52dbd32dccf2d10cac7725f8eae5296885fb5703b261f7d0a0739ec807ab" - [[package]] name = "linux-raw-sys" version = "0.12.1" @@ -3803,15 +3758,6 @@ version = "0.4.33" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" -[[package]] -name = "lru" -version = "0.12.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "234cf4f4a04dc1f57e24b96cc0cd600cf2af460d4161ac5ecdd0af8e1f3b2a38" -dependencies = [ - "hashbrown 0.15.5", -] - [[package]] name = "lru" version = "0.16.4" @@ -3916,16 +3862,6 @@ dependencies = [ "slab", ] -[[package]] -name = "measure_time" -version = "0.8.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dbefd235b0aadd181626f281e1d684e116972988c14c264e42069d5e8a5775cc" -dependencies = [ - "instant", - "log", -] - [[package]] name = "measure_time" version = "0.9.0" @@ -4195,16 +4131,6 @@ dependencies = [ "libm", ] -[[package]] -name = "num_cpus" -version = "1.17.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "91df4bbde75afed763b708b7eee1e8e7651e02d97f6d5dd763e89367e957b23b" -dependencies = [ - "hermit-abi", - "libc", -] - [[package]] name = "num_enum" version = "0.7.6" @@ -4578,15 +4504,6 @@ dependencies = [ "hashbrown 0.14.5", ] -[[package]] -name = "ownedbytes" -version = "0.7.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c3a059efb063b8f425b948e042e6b9bd85edfe60e913630ed727b23e2dfcc558" -dependencies = [ - "stable_deref_trait", -] - [[package]] name = "ownedbytes" version = "0.9.0" @@ -4655,7 +4572,6 @@ dependencies = [ "sha2 0.10.9", "snafu 0.9.1", "snap", - "tantivy 0.22.1", "tempfile", "tokio", "tokio-util", @@ -4690,6 +4606,7 @@ dependencies = [ "arrow-schema", "arrow-select", "async-trait", + "bytes", "chrono", "constant_time_eq", "datafusion", @@ -4698,6 +4615,7 @@ dependencies = [ "lexical-write-float", "log", "paimon", + "paimon-ftindex-core", "parquet", "serde", "serde_json", @@ -4717,8 +4635,8 @@ dependencies = [ "roaring", "serde", "serde_json", - "tantivy 0.26.1", - "tantivy-common 0.11.0", + "tantivy", + "tantivy-common", "tantivy-fst", "tantivy-jieba", "tempfile", @@ -5080,7 +4998,7 @@ dependencies = [ "concurrent-queue", "hermit-abi", "pin-project-lite", - "rustix 1.1.4", + "rustix", "windows-sys 0.61.2", ] @@ -5356,7 +5274,7 @@ dependencies = [ "pin-project-lite", "quinn-proto", "quinn-udp", - "rustc-hash 2.1.3", + "rustc-hash", "rustls", "socket2", "thiserror 2.0.19", @@ -5378,7 +5296,7 @@ dependencies = [ "rand 0.10.2", "rand_pcg", "ring", - "rustc-hash 2.1.3", + "rustc-hash", "rustls", "rustls-pki-types", "slab", @@ -5505,16 +5423,6 @@ version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" -[[package]] -name = "rand_distr" -version = "0.4.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "32cb0b9bc82b0a0876c2dd994a7e7a2683d3e7390ca40e6886785ef0c7e3ee31" -dependencies = [ - "num-traits", - "rand 0.8.7", -] - [[package]] name = "rand_pcg" version = "0.10.2" @@ -5952,12 +5860,6 @@ dependencies = [ "serde_derive", ] -[[package]] -name = "rustc-hash" -version = "1.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "08d43f7aa6b08d49f382cde6a7982047c3426db949b1424bc4b7ec9ae12c6ce2" - [[package]] name = "rustc-hash" version = "2.1.3" @@ -5973,19 +5875,6 @@ dependencies = [ "semver", ] -[[package]] -name = "rustix" -version = "0.38.44" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fdb5bc1ae2baa591800df16c9ca78619bf65c0488b41b96ccec5d11220d8c154" -dependencies = [ - "bitflags 2.13.1", - "errno", - "libc", - "linux-raw-sys 0.4.15", - "windows-sys 0.59.0", -] - [[package]] name = "rustix" version = "1.1.4" @@ -5995,7 +5884,7 @@ dependencies = [ "bitflags 2.13.1", "errno", "libc", - "linux-raw-sys 0.12.1", + "linux-raw-sys", "windows-sys 0.61.2", ] @@ -6440,15 +6329,6 @@ version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ee5873ec9cce0195efcb7a4e9507a04cd49aec9c83d0389df45b1ef7ba2e649" -[[package]] -name = "sketches-ddsketch" -version = "0.2.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "85636c14b73d81f541e525f585c0a2109e6744e1565b5c1668e31c70c10ed65c" -dependencies = [ - "serde", -] - [[package]] name = "sketches-ddsketch" version = "0.4.0" @@ -6758,57 +6638,6 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417" -[[package]] -name = "tantivy" -version = "0.22.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "96599ea6fccd844fc833fed21d2eecac2e6a7c1afd9e044057391d78b1feb141" -dependencies = [ - "aho-corasick", - "arc-swap", - "base64", - "bitpacking", - "byteorder", - "census", - "crc32fast", - "crossbeam-channel", - "downcast-rs 1.2.1", - "fastdivide", - "fnv", - "fs4 0.8.4", - "htmlescape", - "itertools 0.12.1", - "levenshtein_automata", - "log", - "lru 0.12.5", - "lz4_flex 0.11.6", - "measure_time 0.8.3", - "memmap2", - "num_cpus", - "once_cell", - "oneshot 0.1.13", - "rayon", - "regex", - "rust-stemmers", - "rustc-hash 1.1.0", - "serde", - "serde_json", - "sketches-ddsketch 0.2.2", - "smallvec", - "tantivy-bitpacker 0.6.0", - "tantivy-columnar 0.3.0", - "tantivy-common 0.7.0", - "tantivy-fst", - "tantivy-query-grammar 0.22.0", - "tantivy-stacker 0.3.0", - "tantivy-tokenizer-api 0.3.0", - "tempfile", - "thiserror 1.0.69", - "time", - "uuid", - "winapi", -] - [[package]] name = "tantivy" version = "0.26.1" @@ -6825,35 +6654,35 @@ dependencies = [ "crc32fast", "crossbeam-channel", "datasketches", - "downcast-rs 2.0.2", + "downcast-rs", "fastdivide", "fnv", - "fs4 0.13.1", + "fs4", "htmlescape", "itertools 0.14.0", "levenshtein_automata", "log", - "lru 0.16.4", + "lru", "lz4_flex 0.13.1", - "measure_time 0.9.0", + "measure_time", "memmap2", "once_cell", "oneshot 0.1.13", "rayon", "regex", "rust-stemmers", - "rustc-hash 2.1.3", + "rustc-hash", "serde", "serde_json", - "sketches-ddsketch 0.4.0", + "sketches-ddsketch", "smallvec", - "tantivy-bitpacker 0.10.0", - "tantivy-columnar 0.7.0", - "tantivy-common 0.11.0", + "tantivy-bitpacker", + "tantivy-columnar", + "tantivy-common", "tantivy-fst", - "tantivy-query-grammar 0.26.0", - "tantivy-stacker 0.7.0", - "tantivy-tokenizer-api 0.7.0", + "tantivy-query-grammar", + "tantivy-stacker", + "tantivy-tokenizer-api", "tempfile", "thiserror 2.0.19", "time", @@ -6862,15 +6691,6 @@ dependencies = [ "winapi", ] -[[package]] -name = "tantivy-bitpacker" -version = "0.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "284899c2325d6832203ac6ff5891b297fc5239c3dc754c5bc1977855b23c10df" -dependencies = [ - "bitpacking", -] - [[package]] name = "tantivy-bitpacker" version = "0.10.0" @@ -6880,49 +6700,20 @@ dependencies = [ "bitpacking", ] -[[package]] -name = "tantivy-columnar" -version = "0.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "12722224ffbe346c7fec3275c699e508fd0d4710e629e933d5736ec524a1f44e" -dependencies = [ - "downcast-rs 1.2.1", - "fastdivide", - "itertools 0.12.1", - "serde", - "tantivy-bitpacker 0.6.0", - "tantivy-common 0.7.0", - "tantivy-sstable 0.3.0", - "tantivy-stacker 0.3.0", -] - [[package]] name = "tantivy-columnar" version = "0.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c57166f5bcfd478f370ab8445afb4678dce44801fa5ce5c451aaf8595583c5dc" dependencies = [ - "downcast-rs 2.0.2", + "downcast-rs", "fastdivide", "itertools 0.14.0", "serde", - "tantivy-bitpacker 0.10.0", - "tantivy-common 0.11.0", - "tantivy-sstable 0.7.0", - "tantivy-stacker 0.7.0", -] - -[[package]] -name = "tantivy-common" -version = "0.7.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8019e3cabcfd20a1380b491e13ff42f57bb38bf97c3d5fa5c07e50816e0621f4" -dependencies = [ - "async-trait", - "byteorder", - "ownedbytes 0.7.0", - "serde", - "time", + "tantivy-bitpacker", + "tantivy-common", + "tantivy-sstable", + "tantivy-stacker", ] [[package]] @@ -6933,7 +6724,7 @@ checksum = "bbf10915aa75da3c3b0d58b58853d2e889efbaf32d4982a4c3715dde6bba23e5" dependencies = [ "async-trait", "byteorder", - "ownedbytes 0.9.0", + "ownedbytes", "serde", "time", ] @@ -6957,16 +6748,7 @@ checksum = "3392170e86f1c387170aba7d171a466ffdc98a8b55b006e19ac64b123a7b690a" dependencies = [ "jieba-rs", "lazy_static", - "tantivy-tokenizer-api 0.7.0", -] - -[[package]] -name = "tantivy-query-grammar" -version = "0.22.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "847434d4af57b32e309f4ab1b4f1707a6c566656264caa427ff4285c4d9d0b82" -dependencies = [ - "nom", + "tantivy-tokenizer-api", ] [[package]] @@ -6982,18 +6764,6 @@ dependencies = [ "serde_json", ] -[[package]] -name = "tantivy-sstable" -version = "0.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c69578242e8e9fc989119f522ba5b49a38ac20f576fc778035b96cc94f41f98e" -dependencies = [ - "tantivy-bitpacker 0.6.0", - "tantivy-common 0.7.0", - "tantivy-fst", - "zstd", -] - [[package]] name = "tantivy-sstable" version = "0.7.0" @@ -7002,23 +6772,12 @@ checksum = "8a2cfc3ac5164cbadc28965ffb145a8f47582a60ae5897859ad8d4316596c606" dependencies = [ "futures-util", "itertools 0.14.0", - "tantivy-bitpacker 0.10.0", - "tantivy-common 0.11.0", + "tantivy-bitpacker", + "tantivy-common", "tantivy-fst", "zstd", ] -[[package]] -name = "tantivy-stacker" -version = "0.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c56d6ff5591fc332739b3ce7035b57995a3ce29a93ffd6012660e0949c956ea8" -dependencies = [ - "murmurhash32", - "rand_distr", - "tantivy-common 0.7.0", -] - [[package]] name = "tantivy-stacker" version = "0.7.0" @@ -7026,16 +6785,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6cbb051742da9d53ca9e8fff43a9b10e319338b24e2c0e15d0372df19ffeb951" dependencies = [ "murmurhash32", - "tantivy-common 0.11.0", -] - -[[package]] -name = "tantivy-tokenizer-api" -version = "0.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2a0dcade25819a89cfe6f17d932c9cedff11989936bf6dd4f336d50392053b04" -dependencies = [ - "serde", + "tantivy-common", ] [[package]] @@ -7079,7 +6829,7 @@ dependencies = [ "fastrand", "getrandom 0.4.3", "once_cell", - "rustix 1.1.4", + "rustix", "windows-sys 0.61.2", ] @@ -7581,7 +7331,7 @@ dependencies = [ "itertools 0.14.0", "num-traits", "prost 0.14.4", - "rustc-hash 2.1.3", + "rustc-hash", "vortex-array", "vortex-buffer", "vortex-error", @@ -7626,7 +7376,7 @@ dependencies = [ "pin-project-lite", "prost 0.14.4", "rand 0.10.2", - "rustc-hash 2.1.3", + "rustc-hash", "simdutf8", "smallvec", "static_assertions", @@ -7720,7 +7470,7 @@ dependencies = [ "num-traits", "parking_lot", "rand 0.10.2", - "rustc-hash 2.1.3", + "rustc-hash", "tracing", "vortex-array", "vortex-buffer", @@ -7930,8 +7680,8 @@ dependencies = [ "paste", "pin-project-lite", "prost 0.14.4", - "rustc-hash 2.1.3", - "sketches-ddsketch 0.4.0", + "rustc-hash", + "sketches-ddsketch", "termtree", "tokio", "tracing", @@ -7968,7 +7718,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6e079c9fe032af05b2e38bf91e63d2bdf36bd01e3920bc4a352648b2b21e9f70" dependencies = [ "parking_lot", - "sketches-ddsketch 0.4.0", + "sketches-ddsketch", ] [[package]] @@ -8578,7 +8328,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32e45ad4206f6d2479085147f02bc2ef834ac85886624a23575ae137c8aa8156" dependencies = [ "libc", - "rustix 1.1.4", + "rustix", ] [[package]] diff --git a/DEPENDENCIES.rust.tsv b/DEPENDENCIES.rust.tsv index 166d7ce6..cb427ff2 100644 --- a/DEPENDENCIES.rust.tsv +++ b/DEPENDENCIES.rust.tsv @@ -183,7 +183,6 @@ digest@0.11.3 X X displaydoc@0.2.6 X X dlv-list@0.5.2 X X dns-lookup@3.0.1 X X -downcast-rs@1.2.1 X X downcast-rs@2.0.2 X X dtype_dispatch@0.2.1 X dunce@1.0.5 X X X @@ -214,7 +213,6 @@ foreign-types@0.3.2 X X foreign-types-shared@0.1.1 X X form_urlencoded@1.2.2 X X fs4@0.13.1 X X -fs4@0.8.4 X X fs_extra@1.3.0 X fsst-rs@0.5.11 X funty@2.0.0 X @@ -278,12 +276,10 @@ include-flate-codegen@0.3.4 X include-flate-compress@0.3.4 X indexmap@2.14.0 X X inout@0.1.4 X X -instant@0.1.13 X integer-encoding@3.0.4 X inventory@0.3.24 X X ipnet@2.12.0 X X is_terminal_polyfill@1.70.2 X X -itertools@0.12.1 X X itertools@0.13.0 X X itertools@0.14.0 X X itoa@1.0.18 X X @@ -322,11 +318,9 @@ liblzma-sys@0.4.7 X X libm@0.2.16 X libredox@0.1.18 X linux-raw-sys@0.12.1 X X X -linux-raw-sys@0.4.15 X X X litemap@0.8.2 X lock_api@0.4.14 X X log@0.4.33 X X -lru@0.12.5 X lru@0.16.4 X lz4_flex@0.11.6 X lz4_flex@0.13.1 X @@ -338,7 +332,6 @@ matrixmultiply@0.3.11 X X md-5@0.10.6 X X md-5@0.11.0 X X mea@0.6.4 X -measure_time@0.8.3 X measure_time@0.9.0 X memchr@2.8.3 X X memmap2@0.9.11 X X @@ -365,7 +358,6 @@ num-integer@0.1.46 X X num-iter@0.1.46 X X num-rational@0.4.2 X X num-traits@0.2.19 X X -num_cpus@1.17.0 X X num_enum@0.7.6 X X X num_enum_derive@0.7.6 X X X object@0.37.3 X X @@ -393,7 +385,6 @@ orc-rust@0.8.0 X ordered-float@2.10.1 X ordered-float@5.3.0 X ordered-multimap@0.7.3 X -ownedbytes@0.7.0 X ownedbytes@0.9.0 X paimon@0.3.0 X paimon-c@0.3.0 X @@ -469,7 +460,6 @@ rand_chacha@0.9.0 X X rand_core@0.10.1 X X rand_core@0.6.4 X X rand_core@0.9.5 X X -rand_distr@0.4.3 X X rand_xoshiro@0.6.0 X X rawpointer@0.2.1 X X rayon@1.12.0 X X @@ -498,10 +488,8 @@ roxmltree@0.21.1 X X rsa@0.9.10 X X rust-ini@0.21.3 X rust-stemmers@1.2.0 X X -rustc-hash@1.1.0 X X rustc-hash@2.1.3 X X rustc_version@0.4.1 X X -rustix@0.38.44 X X X rustix@1.1.4 X X X rustls@0.23.42 X X X rustls-native-certs@0.8.4 X X X @@ -545,7 +533,6 @@ simd-adler32@0.3.10 X simd_cesu8@1.2.0 X X simdutf8@0.1.5 X X siphasher@1.0.3 X X -sketches-ddsketch@0.2.2 X sketches-ddsketch@0.4.0 X slab@0.4.12 X smallvec@1.15.2 X X @@ -579,23 +566,15 @@ synstructure@0.13.2 X system-configuration@0.7.0 X X system-configuration-sys@0.6.0 X X tagptr@0.2.0 X X -tantivy@0.22.1 X tantivy@0.26.1 X tantivy-bitpacker@0.10.0 X -tantivy-bitpacker@0.6.0 X -tantivy-columnar@0.3.0 X tantivy-columnar@0.7.0 X tantivy-common@0.11.0 X -tantivy-common@0.7.0 X tantivy-fst@0.5.0 X X tantivy-jieba@0.20.0 X -tantivy-query-grammar@0.22.0 X tantivy-query-grammar@0.26.0 X -tantivy-sstable@0.3.0 X tantivy-sstable@0.7.0 X -tantivy-stacker@0.3.0 X tantivy-stacker@0.7.0 X -tantivy-tokenizer-api@0.3.0 X tantivy-tokenizer-api@0.7.0 X tap@1.0.1 X target-lexicon@0.13.5 X @@ -703,7 +682,6 @@ windows-link@0.2.1 X X windows-registry@0.6.1 X X windows-result@0.4.1 X X windows-strings@0.5.1 X X -windows-sys@0.52.0 X X windows-sys@0.59.0 X X windows-sys@0.60.2 X X windows-sys@0.61.2 X X diff --git a/bindings/python/DEPENDENCIES.rust.tsv b/bindings/python/DEPENDENCIES.rust.tsv index 041777ad..f2749c96 100644 --- a/bindings/python/DEPENDENCIES.rust.tsv +++ b/bindings/python/DEPENDENCIES.rust.tsv @@ -148,7 +148,6 @@ digest@0.11.3 X X displaydoc@0.2.6 X X dlv-list@0.5.2 X X dns-lookup@3.0.1 X X -downcast-rs@1.2.1 X X downcast-rs@2.0.2 X X dunce@1.0.5 X X X either@1.16.0 X X @@ -170,7 +169,6 @@ foreign-types@0.3.2 X X foreign-types-shared@0.1.1 X X form_urlencoded@1.2.2 X X fs4@0.13.1 X X -fs4@0.8.4 X X fs_extra@1.3.0 X futures@0.3.33 X X futures-channel@0.3.33 X X @@ -197,7 +195,6 @@ hashbrown@0.16.1 X X hashbrown@0.17.1 X X hdfs-native@0.13.5 X heck@0.5.0 X X -hermit-abi@0.5.2 X X hex@0.4.3 X X hmac@0.12.1 X X hmac@0.13.0 X X @@ -230,11 +227,9 @@ include-flate-codegen@0.3.4 X include-flate-compress@0.3.4 X indexmap@2.14.0 X X inout@0.1.4 X X -instant@0.1.13 X integer-encoding@3.0.4 X inventory@0.3.24 X X ipnet@2.12.0 X X -itertools@0.12.1 X X itertools@0.14.0 X X itoa@1.0.18 X X jieba-macros@0.10.3 X @@ -267,11 +262,9 @@ liblzma-sys@0.4.7 X X libm@0.2.16 X libredox@0.1.18 X linux-raw-sys@0.12.1 X X X -linux-raw-sys@0.4.15 X X X litemap@0.8.2 X lock_api@0.4.14 X X log@0.4.33 X X -lru@0.12.5 X lru@0.16.4 X lz4_flex@0.11.6 X lz4_flex@0.13.1 X @@ -280,7 +273,6 @@ matrixmultiply@0.3.11 X X md-5@0.10.6 X X md-5@0.11.0 X X mea@0.6.4 X -measure_time@0.8.3 X measure_time@0.9.0 X memchr@2.8.3 X X memmap2@0.9.11 X X @@ -303,7 +295,6 @@ num-integer@0.1.46 X X num-iter@0.1.46 X X num-rational@0.4.2 X X num-traits@0.2.19 X X -num_cpus@1.17.0 X X object@0.37.3 X X object_store@0.13.2 X X once_cell@1.21.4 X X @@ -327,7 +318,6 @@ orc-rust@0.8.0 X ordered-float@2.10.1 X ordered-float@5.3.0 X ordered-multimap@0.7.3 X -ownedbytes@0.7.0 X ownedbytes@0.9.0 X paimon@0.3.0 X paimon-datafusion@0.3.0 X @@ -391,7 +381,6 @@ rand_chacha@0.3.1 X X rand_chacha@0.9.0 X X rand_core@0.6.4 X X rand_core@0.9.5 X X -rand_distr@0.4.3 X X rawpointer@0.2.1 X X rayon@1.12.0 X X rayon-core@1.13.0 X X @@ -419,10 +408,8 @@ roxmltree@0.21.1 X X rsa@0.9.10 X X rust-ini@0.21.3 X rust-stemmers@1.2.0 X X -rustc-hash@1.1.0 X X rustc-hash@2.1.3 X X rustc_version@0.4.1 X X -rustix@0.38.44 X X X rustix@1.1.4 X X X rustls@0.23.42 X X X rustls-native-certs@0.8.4 X X X @@ -463,7 +450,6 @@ simd-adler32@0.3.10 X simd_cesu8@1.2.0 X X simdutf8@0.1.5 X X siphasher@1.0.3 X X -sketches-ddsketch@0.2.2 X sketches-ddsketch@0.4.0 X slab@0.4.12 X smallvec@1.15.2 X X @@ -493,23 +479,15 @@ sync_wrapper@1.0.2 X synstructure@0.13.2 X system-configuration@0.7.0 X X system-configuration-sys@0.6.0 X X -tantivy@0.22.1 X tantivy@0.26.1 X tantivy-bitpacker@0.10.0 X -tantivy-bitpacker@0.6.0 X -tantivy-columnar@0.3.0 X tantivy-columnar@0.7.0 X tantivy-common@0.11.0 X -tantivy-common@0.7.0 X tantivy-fst@0.5.0 X X tantivy-jieba@0.20.0 X -tantivy-query-grammar@0.22.0 X tantivy-query-grammar@0.26.0 X -tantivy-sstable@0.3.0 X tantivy-sstable@0.7.0 X -tantivy-stacker@0.3.0 X tantivy-stacker@0.7.0 X -tantivy-tokenizer-api@0.3.0 X tantivy-tokenizer-api@0.7.0 X target-lexicon@0.13.5 X tempfile@3.27.0 X X @@ -584,7 +562,6 @@ windows-link@0.2.1 X X windows-registry@0.6.1 X X windows-result@0.4.1 X X windows-strings@0.5.1 X X -windows-sys@0.52.0 X X windows-sys@0.59.0 X X windows-sys@0.60.2 X X windows-sys@0.61.2 X X diff --git a/crates/integrations/datafusion/Cargo.toml b/crates/integrations/datafusion/Cargo.toml index 37fb1947..17acaea0 100644 --- a/crates/integrations/datafusion/Cargo.toml +++ b/crates/integrations/datafusion/Cargo.toml @@ -50,7 +50,9 @@ uuid = { version = "1", features = ["v4"] } [dev-dependencies] arrow-array = { workspace = true } arrow-schema = { workspace = true } +bytes = "1.7.1" flate2 = "1" +paimon-ftindex-core = "0.1.0" parquet = { workspace = true } tar = "0.4" tempfile = "3" diff --git a/crates/integrations/datafusion/DEPENDENCIES.rust.tsv b/crates/integrations/datafusion/DEPENDENCIES.rust.tsv index f230597d..9724882f 100644 --- a/crates/integrations/datafusion/DEPENDENCIES.rust.tsv +++ b/crates/integrations/datafusion/DEPENDENCIES.rust.tsv @@ -152,7 +152,6 @@ digest@0.10.7 X X digest@0.11.3 X X displaydoc@0.2.6 X X dlv-list@0.5.2 X X -downcast-rs@1.2.1 X X downcast-rs@2.0.2 X X dtype_dispatch@0.2.1 X dunce@1.0.5 X X X @@ -183,7 +182,6 @@ foreign-types@0.3.2 X X foreign-types-shared@0.1.1 X X form_urlencoded@1.2.2 X X fs4@0.13.1 X X -fs4@0.8.4 X X fs_extra@1.3.0 X fsst-rs@0.5.11 X funty@2.0.0 X @@ -242,11 +240,9 @@ include-flate@0.3.4 X include-flate-codegen@0.3.4 X include-flate-compress@0.3.4 X indexmap@2.14.0 X X -instant@0.1.13 X integer-encoding@3.0.4 X inventory@0.3.24 X X ipnet@2.12.0 X X -itertools@0.12.1 X X itertools@0.13.0 X X itertools@0.14.0 X X itoa@1.0.18 X X @@ -284,11 +280,9 @@ liblzma@0.4.7 X X liblzma-sys@0.4.7 X X libm@0.2.16 X linux-raw-sys@0.12.1 X X X -linux-raw-sys@0.4.15 X X X litemap@0.8.2 X lock_api@0.4.14 X X log@0.4.33 X X -lru@0.12.5 X lru@0.16.4 X lz4_flex@0.11.6 X lz4_flex@0.13.1 X @@ -299,7 +293,6 @@ matrixmultiply@0.3.11 X X md-5@0.10.6 X X md-5@0.11.0 X X mea@0.6.4 X -measure_time@0.8.3 X measure_time@0.9.0 X memchr@2.8.3 X X memmap2@0.9.11 X X @@ -325,7 +318,6 @@ num-integer@0.1.46 X X num-iter@0.1.46 X X num-rational@0.4.2 X X num-traits@0.2.19 X X -num_cpus@1.17.0 X X num_enum@0.7.6 X X X num_enum_derive@0.7.6 X X X object@0.37.3 X X @@ -345,7 +337,6 @@ orc-rust@0.8.0 X ordered-float@2.10.1 X ordered-float@5.3.0 X ordered-multimap@0.7.3 X -ownedbytes@0.7.0 X ownedbytes@0.9.0 X paimon@0.3.0 X paimon-datafusion@0.3.0 X @@ -403,7 +394,6 @@ rand_chacha@0.9.0 X X rand_core@0.10.1 X X rand_core@0.6.4 X X rand_core@0.9.5 X X -rand_distr@0.4.3 X X rand_xoshiro@0.6.0 X X rawpointer@0.2.1 X X rayon@1.12.0 X X @@ -424,10 +414,8 @@ rle-decode-fast@1.0.3 X X roaring@0.11.4 X X rust-ini@0.21.3 X rust-stemmers@1.2.0 X X -rustc-hash@1.1.0 X X rustc-hash@2.1.3 X X rustc_version@0.4.1 X X -rustix@0.38.44 X X X rustix@1.1.4 X X X rustls@0.23.42 X X X rustls-native-certs@0.8.4 X X X @@ -466,7 +454,6 @@ simd-adler32@0.3.10 X simd_cesu8@1.2.0 X X simdutf8@0.1.5 X X siphasher@1.0.3 X X -sketches-ddsketch@0.2.2 X sketches-ddsketch@0.4.0 X slab@0.4.12 X smallvec@1.15.2 X X @@ -494,23 +481,15 @@ synstructure@0.13.2 X system-configuration@0.7.0 X X system-configuration-sys@0.6.0 X X tagptr@0.2.0 X X -tantivy@0.22.1 X tantivy@0.26.1 X tantivy-bitpacker@0.10.0 X -tantivy-bitpacker@0.6.0 X -tantivy-columnar@0.3.0 X tantivy-columnar@0.7.0 X tantivy-common@0.11.0 X -tantivy-common@0.7.0 X tantivy-fst@0.5.0 X X tantivy-jieba@0.20.0 X -tantivy-query-grammar@0.22.0 X tantivy-query-grammar@0.26.0 X -tantivy-sstable@0.3.0 X tantivy-sstable@0.7.0 X -tantivy-stacker@0.3.0 X tantivy-stacker@0.7.0 X -tantivy-tokenizer-api@0.3.0 X tantivy-tokenizer-api@0.7.0 X tap@1.0.1 X tempfile@3.27.0 X X @@ -611,7 +590,6 @@ windows-link@0.2.1 X X windows-registry@0.6.1 X X windows-result@0.4.1 X X windows-strings@0.5.1 X X -windows-sys@0.52.0 X X windows-sys@0.59.0 X X windows-sys@0.61.2 X X windows-targets@0.52.6 X X diff --git a/crates/integrations/datafusion/tests/read_tables.rs b/crates/integrations/datafusion/tests/read_tables.rs index 87655c25..b4637b5b 100644 --- a/crates/integrations/datafusion/tests/read_tables.rs +++ b/crates/integrations/datafusion/tests/read_tables.rs @@ -1461,34 +1461,21 @@ async fn test_case_insensitive_column_not_supported_via_sql() { mod fulltext_tests { use std::sync::Arc; + use bytes::Bytes; use datafusion::arrow::array::Int32Array; use paimon::catalog::Identifier; + use paimon::spec::{GlobalIndexMeta, IndexFileMeta}; use paimon::table::BranchManager; - use paimon::{Catalog, CatalogOptions, FileSystemCatalog, Options}; + use paimon::{Catalog, CatalogOptions, CommitMessage, FileSystemCatalog, Options, TableCommit}; use paimon_datafusion::{register_full_text_search, SQLContext}; use super::common::string_value; + use paimon_ftindex_core::io::PosWriter; + use paimon_ftindex_core::{FullTextIndexConfig, FullTextIndexWriter}; - /// Extract the bundled tar.gz into a temp dir and return (tempdir, warehouse_path). - fn extract_test_warehouse() -> (tempfile::TempDir, String) { - let archive_path = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) - .join("testdata/test_tantivy_fulltext.tar.gz"); - let file = std::fs::File::open(&archive_path) - .unwrap_or_else(|e| panic!("Failed to open {}: {e}", archive_path.display())); - let decoder = flate2::read::GzDecoder::new(file); - let mut archive = tar::Archive::new(decoder); - + async fn create_fulltext_context() -> (SQLContext, Arc, tempfile::TempDir) { let tmp = tempfile::tempdir().expect("Failed to create temp dir"); - let db_dir = tmp.path().join("default.db"); - std::fs::create_dir_all(&db_dir).unwrap(); - archive.unpack(&db_dir).unwrap(); - let warehouse = format!("file://{}", tmp.path().display()); - (tmp, warehouse) - } - - async fn create_fulltext_context() -> (SQLContext, Arc, tempfile::TempDir) { - let (tmp, warehouse) = extract_test_warehouse(); let mut options = Options::new(); options.set(CatalogOptions::WAREHOUSE, warehouse); let catalog = Arc::new(FileSystemCatalog::new(options).expect("Failed to create catalog")); @@ -1498,9 +1485,113 @@ mod fulltext_tests { .await .expect("Failed to register catalog"); register_full_text_search(ctx.ctx(), catalog.clone(), "default"); + run_fulltext_sql( + &ctx, + "CREATE TABLE paimon.default.test_fulltext ( + id INT, + content STRING + ) WITH ( + 'row-tracking.enabled' = 'true', + 'global-index.search-mode' = 'full' + )", + ) + .await; + for insert in [ + "INSERT INTO paimon.default.test_fulltext VALUES (0, 'alphaft lakehouse storage')", + "INSERT INTO paimon.default.test_fulltext VALUES (1, 'betafft gammaft full-text search')", + "INSERT INTO paimon.default.test_fulltext VALUES (2, 'alphaft supports indexed queries')", + "INSERT INTO paimon.default.test_fulltext VALUES (3, 'gammaft over raw fallback rows')", + "INSERT INTO paimon.default.test_fulltext VALUES (4, 'alphaft tables can be queried')", + ] { + run_fulltext_sql(&ctx, insert).await; + } + build_fulltext_index(catalog.as_ref()).await; (ctx, catalog, tmp) } + async fn build_fulltext_index(catalog: &dyn Catalog) { + let table = catalog + .get_table(&Identifier::new("default", "test_fulltext")) + .await + .expect("table should exist"); + let content_field_id = table + .schema() + .fields() + .iter() + .find(|field| field.name() == "content") + .expect("content field") + .id(); + + let mut writer = + FullTextIndexWriter::new(FullTextIndexConfig::new().with_text_fields(["content"])) + .expect("full-text writer"); + for (row_id, content) in [ + (0, "alphaft lakehouse storage"), + (1, "betafft gammaft full-text search"), + (2, "alphaft supports indexed queries"), + (3, "gammaft over raw fallback rows"), + (4, "alphaft tables can be queried"), + ] { + writer + .add_document_fields(row_id, [("content", content)]) + .expect("add full-text document"); + } + let mut index_bytes = Vec::new(); + writer + .write(&mut PosWriter::new(&mut index_bytes)) + .expect("write full-text index"); + + let index_file_name = "ft-datafusion-fulltext.index"; + table + .file_io() + .mkdirs(&format!( + "{}/index/", + table.location().trim_end_matches('/') + )) + .await + .expect("create index dir"); + table + .file_io() + .new_output(&format!( + "{}/index/{index_file_name}", + table.location().trim_end_matches('/') + )) + .expect("index output") + .write(Bytes::from(index_bytes.clone())) + .await + .expect("write index file"); + + let mut message = CommitMessage::new(Vec::new(), 0, Vec::new()); + message.new_index_files = vec![IndexFileMeta { + index_type: "full-text".to_string(), + file_name: index_file_name.to_string(), + file_size: i64::try_from(index_bytes.len()).unwrap(), + row_count: 5, + deletion_vectors_ranges: None, + global_index_meta: Some(GlobalIndexMeta { + row_range_start: 0, + row_range_end: 4, + index_field_id: content_field_id, + extra_field_ids: None, + index_meta: None, + source_meta: None, + }), + }]; + TableCommit::new(table, "test-user".to_string()) + .commit(vec![message]) + .await + .expect("commit full-text index manifest"); + } + + async fn run_fulltext_sql(ctx: &SQLContext, sql: &str) { + ctx.sql(sql) + .await + .unwrap_or_else(|e| panic!("Failed to plan `{sql}`: {e}")) + .collect() + .await + .unwrap_or_else(|e| panic!("Failed to execute `{sql}`: {e}")); + } + fn extract_id_content_rows( batches: &[datafusion::arrow::record_batch::RecordBatch], ) -> Vec<(i32, String)> { @@ -1524,12 +1615,12 @@ mod fulltext_tests { rows } - /// Search for 'paimon' — rows 0, 2, 4 mention "paimon". + /// Search for 'alphaft' — rows 0, 2, 4 contain the token. #[tokio::test] async fn test_full_text_search_paimon() { let (ctx, _catalog, _tmp) = create_fulltext_context().await; let batches = ctx - .sql("SELECT id, content FROM full_text_search('paimon.default.test_tantivy_fulltext', 'content', 'paimon', 10)") + .sql("SELECT id, content FROM full_text_search('paimon.default.test_fulltext', 'content', 'alphaft', 10)") .await .expect("SQL should parse") .collect() @@ -1541,14 +1632,14 @@ mod fulltext_tests { assert_eq!( ids, vec![0, 2, 4], - "Searching 'paimon' should match rows 0, 2, 4" + "Searching 'alphaft' should match rows 0, 2, 4" ); } #[tokio::test] async fn test_full_text_search_branch() { let (ctx, catalog, _tmp) = create_fulltext_context().await; - let identifier = Identifier::new("default", "test_tantivy_fulltext"); + let identifier = Identifier::new("default", "test_fulltext"); let table = catalog.get_table(&identifier).await.expect("load table"); let snapshot_manager = table.snapshot_manager(); let snapshot = snapshot_manager @@ -1572,7 +1663,7 @@ mod fulltext_tests { .expect("delete main snapshots"); let batches = ctx - .sql("SELECT id, content FROM full_text_search('paimon.default.test_tantivy_fulltext$branch_b1', 'content', 'paimon', 10)") + .sql("SELECT id, content FROM full_text_search('paimon.default.test_fulltext$branch_b1', 'content', 'alphaft', 10)") .await .expect("SQL should parse") .collect() @@ -1584,12 +1675,12 @@ mod fulltext_tests { assert_eq!(ids, vec![0, 2, 4]); } - /// Search for 'tantivy' — only row 1. + /// Search for 'betafft' — only row 1. #[tokio::test] - async fn test_full_text_search_tantivy() { + async fn test_full_text_search_betafft() { let (ctx, _catalog, _tmp) = create_fulltext_context().await; let batches = ctx - .sql("SELECT id, content FROM full_text_search('paimon.default.test_tantivy_fulltext', 'content', 'tantivy', 10)") + .sql("SELECT id, content FROM full_text_search('paimon.default.test_fulltext', 'content', 'betafft', 10)") .await .expect("SQL should parse") .collect() @@ -1598,15 +1689,15 @@ mod fulltext_tests { let rows = extract_id_content_rows(&batches); let ids: Vec = rows.iter().map(|(id, _)| *id).collect(); - assert_eq!(ids, vec![1], "Searching 'tantivy' should match row 1"); + assert_eq!(ids, vec![1], "Searching 'betafft' should match row 1"); } - /// Search for 'search' — rows 1, 3 mention "full-text search". + /// Search for 'gammaft' — rows 1, 3 contain the token. #[tokio::test] async fn test_full_text_search_search() { let (ctx, _catalog, _tmp) = create_fulltext_context().await; let batches = ctx - .sql("SELECT id, content FROM full_text_search('paimon.default.test_tantivy_fulltext', 'content', 'search', 10)") + .sql("SELECT id, content FROM full_text_search('paimon.default.test_fulltext', 'content', 'gammaft', 10)") .await .expect("SQL should parse") .collect() @@ -1615,8 +1706,11 @@ mod fulltext_tests { let rows = extract_id_content_rows(&batches); let ids: Vec = rows.iter().map(|(id, _)| *id).collect(); - assert!(ids.contains(&1), "Searching 'search' should match row 1"); - assert!(ids.contains(&3), "Searching 'search' should match row 3"); + assert_eq!( + ids, + vec![1, 3], + "Searching 'gammaft' should match rows 1, 3" + ); } } diff --git a/crates/paimon/Cargo.toml b/crates/paimon/Cargo.toml index a2042c5a..043f95f8 100644 --- a/crates/paimon/Cargo.toml +++ b/crates/paimon/Cargo.toml @@ -42,7 +42,7 @@ storage-all = [ "storage-gcs", "storage-hdfs", ] -fulltext = ["tantivy", "tempfile", "dep:paimon-ftindex-core"] +fulltext = ["dep:paimon-ftindex-core", "dep:tempfile"] vortex = ["dep:vortex"] storage-memory = ["opendal/services-memory"] @@ -112,10 +112,9 @@ md-5 = "0.10" regex = "1" uuid = { version = "1", features = ["v4"] } urlencoding = "2.1" -tantivy = { version = "0.22", optional = true } -tempfile = { version = "3", optional = true } -paimon-ftindex-core = { version = "0.1.0", optional = true } paimon-mosaic-core = "0.2.0" +paimon-ftindex-core = { version = "0.1.0", optional = true } +tempfile = { version = "3", optional = true } paimon-vindex-core = "0.2.0" vortex = { version = "0.75.0", features = ["tokio"], optional = true } libloading = "0.9" diff --git a/crates/paimon/DEPENDENCIES.rust.tsv b/crates/paimon/DEPENDENCIES.rust.tsv index b032f87b..d78da445 100644 --- a/crates/paimon/DEPENDENCIES.rust.tsv +++ b/crates/paimon/DEPENDENCIES.rust.tsv @@ -124,7 +124,6 @@ digest@0.11.3 X X displaydoc@0.2.6 X X dlv-list@0.5.2 X X dns-lookup@3.0.1 X X -downcast-rs@1.2.1 X X downcast-rs@2.0.2 X X dtype_dispatch@0.2.1 X dunce@1.0.5 X X X @@ -148,13 +147,11 @@ find-msvc-tools@0.1.9 X X flatbuffers@25.12.19 X flate2@1.1.9 X X fnv@1.0.7 X X -foldhash@0.1.5 X foldhash@0.2.0 X foreign-types@0.3.2 X X foreign-types-shared@0.1.1 X X form_urlencoded@1.2.2 X X fs4@0.13.1 X X -fs4@0.8.4 X X fs_extra@1.3.0 X fsst-rs@0.5.11 X funty@2.0.0 X @@ -179,7 +176,6 @@ glob@0.3.3 X X h2@0.4.15 X half@2.7.1 X X hashbrown@0.14.5 X X -hashbrown@0.15.5 X X hashbrown@0.16.1 X X hashbrown@0.17.1 X X hdfs-native@0.13.5 X @@ -217,11 +213,9 @@ include-flate-codegen@0.3.4 X include-flate-compress@0.3.4 X indexmap@2.14.0 X X inout@0.1.4 X X -instant@0.1.13 X integer-encoding@3.0.4 X inventory@0.3.24 X X ipnet@2.12.0 X X -itertools@0.12.1 X X itertools@0.13.0 X X itertools@0.14.0 X X itoa@1.0.18 X X @@ -257,11 +251,9 @@ libloading@0.9.0 X libm@0.2.16 X libredox@0.1.18 X linux-raw-sys@0.12.1 X X X -linux-raw-sys@0.4.15 X X X litemap@0.8.2 X lock_api@0.4.14 X X log@0.4.33 X X -lru@0.12.5 X lru@0.16.4 X lz4_flex@0.11.6 X lz4_flex@0.13.1 X @@ -272,7 +264,6 @@ matrixmultiply@0.3.11 X X md-5@0.10.6 X X md-5@0.11.0 X X mea@0.6.4 X -measure_time@0.8.3 X measure_time@0.9.0 X memchr@2.8.3 X X memmap2@0.9.11 X X @@ -299,7 +290,6 @@ num-integer@0.1.46 X X num-iter@0.1.46 X X num-rational@0.4.2 X X num-traits@0.2.19 X X -num_cpus@1.17.0 X X num_enum@0.7.6 X X X num_enum_derive@0.7.6 X X X once_cell@1.21.4 X X @@ -324,7 +314,6 @@ orc-rust@0.8.0 X ordered-float@2.10.1 X ordered-float@5.3.0 X ordered-multimap@0.7.3 X -ownedbytes@0.7.0 X ownedbytes@0.9.0 X paimon@0.3.0 X paimon-ftindex-core@0.1.0 X @@ -384,7 +373,6 @@ rand_chacha@0.9.0 X X rand_core@0.10.1 X X rand_core@0.6.4 X X rand_core@0.9.5 X X -rand_distr@0.4.3 X X rand_xoshiro@0.6.0 X X rawpointer@0.2.1 X X rayon@1.12.0 X X @@ -411,10 +399,8 @@ roxmltree@0.21.1 X X rsa@0.9.10 X X rust-ini@0.21.3 X rust-stemmers@1.2.0 X X -rustc-hash@1.1.0 X X rustc-hash@2.1.3 X X rustc_version@0.4.1 X X -rustix@0.38.44 X X X rustix@1.1.4 X X X rustls@0.23.42 X X X rustls-native-certs@0.8.4 X X X @@ -456,7 +442,6 @@ simd-adler32@0.3.10 X simd_cesu8@1.2.0 X X simdutf8@0.1.5 X X siphasher@1.0.3 X X -sketches-ddsketch@0.2.2 X sketches-ddsketch@0.4.0 X slab@0.4.12 X smallvec@1.15.2 X X @@ -484,23 +469,15 @@ synstructure@0.13.2 X system-configuration@0.7.0 X X system-configuration-sys@0.6.0 X X tagptr@0.2.0 X X -tantivy@0.22.1 X tantivy@0.26.1 X tantivy-bitpacker@0.10.0 X -tantivy-bitpacker@0.6.0 X -tantivy-columnar@0.3.0 X tantivy-columnar@0.7.0 X tantivy-common@0.11.0 X -tantivy-common@0.7.0 X tantivy-fst@0.5.0 X X tantivy-jieba@0.20.0 X -tantivy-query-grammar@0.22.0 X tantivy-query-grammar@0.26.0 X -tantivy-sstable@0.3.0 X tantivy-sstable@0.7.0 X -tantivy-stacker@0.3.0 X tantivy-stacker@0.7.0 X -tantivy-tokenizer-api@0.3.0 X tantivy-tokenizer-api@0.7.0 X tap@1.0.1 X tempfile@3.27.0 X X @@ -602,7 +579,6 @@ windows-link@0.2.1 X X windows-registry@0.6.1 X X windows-result@0.4.1 X X windows-strings@0.5.1 X X -windows-sys@0.52.0 X X windows-sys@0.59.0 X X windows-sys@0.60.2 X X windows-sys@0.61.2 X X diff --git a/crates/paimon/src/tantivy/full_text_search.rs b/crates/paimon/src/full_text.rs similarity index 69% rename from crates/paimon/src/tantivy/full_text_search.rs rename to crates/paimon/src/full_text.rs index 2f57ef8a..0c0b5f71 100644 --- a/crates/paimon/src/tantivy/full_text_search.rs +++ b/crates/paimon/src/full_text.rs @@ -15,18 +15,13 @@ // specific language governing permissions and limitations // under the License. -//! Full-text search types for global index. -//! -//! Reference: [org.apache.paimon.predicate.FullTextSearch](https://github.com/apache/paimon/blob/master/paimon-common/src/main/java/org/apache/paimon/predicate/FullTextSearch.java) - /// Full-text search predicate. -/// -/// Reference: `org.apache.paimon.predicate.FullTextSearch` #[derive(Debug, Clone)] pub struct FullTextSearch { pub query_text: String, pub field_name: String, pub limit: usize, + pub include_row_ids: Option, } impl FullTextSearch { @@ -50,8 +45,14 @@ impl FullTextSearch { query_text, field_name, limit, + include_row_ids: None, }) } + + pub fn with_include_row_ids(mut self, include_row_ids: roaring::RoaringTreemap) -> Self { + self.include_row_ids = Some(include_row_ids); + self + } } /// Search result containing parallel arrays of row IDs and scores. @@ -108,8 +109,18 @@ impl SearchResult { pub fn or(&self, other: &SearchResult) -> Self { let mut row_ids = self.row_ids.clone(); let mut scores = self.scores.clone(); - row_ids.extend_from_slice(&other.row_ids); - scores.extend_from_slice(&other.scores); + row_ids.reserve(other.row_ids.len()); + scores.reserve(other.scores.len()); + let mut seen = row_ids + .iter() + .copied() + .collect::>(); + for (&row_id, &score) in other.row_ids.iter().zip(&other.scores) { + if seen.insert(row_id) { + row_ids.push(row_id); + scores.push(score); + } + } Self { row_ids, scores } } @@ -155,6 +166,31 @@ impl SearchResult { Ok(Self { row_ids, scores }) } + pub(crate) fn without_row_ranges( + &self, + ranges: &[crate::table::RowRange], + ) -> crate::Result { + if ranges.is_empty() { + return Ok(self.clone()); + } + let index = crate::table::global_index_scanner::RowRangeIndex::create(ranges.to_vec()); + let mut row_ids = Vec::with_capacity(self.row_ids.len()); + let mut scores = Vec::with_capacity(self.scores.len()); + for (&row_id, &score) in self.row_ids.iter().zip(&self.scores) { + let row_id_i64 = i64::try_from(row_id).map_err(|_| crate::Error::DataInvalid { + message: format!( + "Full-text search row id {row_id} exceeds i64::MAX and cannot be checked against row ranges" + ), + source: None, + })?; + if !index.intersects(row_id_i64, row_id_i64) { + row_ids.push(row_id); + scores.push(score); + } + } + Ok(Self { row_ids, scores }) + } + /// Convert to sorted, merged row ranges. pub fn to_row_ranges(&self) -> Vec { if self.row_ids.is_empty() { @@ -191,6 +227,18 @@ mod tests { assert_eq!(fts.query_text, "hello"); assert_eq!(fts.limit, 10); assert_eq!(fts.field_name, "text"); + assert!(fts.include_row_ids.is_none()); + } + + #[test] + fn test_full_text_search_clone_preserves_include_row_ids() { + let mut include_row_ids = roaring::RoaringTreemap::new(); + include_row_ids.insert(42); + let search = FullTextSearch::new("hello".into(), 10, "text".into()) + .unwrap() + .with_include_row_ids(include_row_ids.clone()); + + assert_eq!(search.clone().include_row_ids, Some(include_row_ids)); } #[test] @@ -213,4 +261,27 @@ mod tests { assert_eq!(filtered.row_ids, vec![1, 4]); assert_eq!(filtered.scores, vec![0.1, 0.2]); } + + #[test] + fn test_search_result_or_preserves_left_score_and_order() { + let left = SearchResult::new(vec![1, 2], vec![0.4, 0.5]); + let right = SearchResult::new(vec![2, 3], vec![0.8, 0.6]); + + let merged = left.or(&right); + + assert_eq!(merged.row_ids, vec![1, 2, 3]); + assert_eq!(merged.scores, vec![0.4, 0.5, 0.6]); + } + + #[test] + fn test_search_result_without_row_ranges() { + let result = SearchResult::new(vec![1, 2, 3, 4], vec![0.1, 0.9, 0.8, 0.2]); + + let filtered = result + .without_row_ranges(&[crate::table::RowRange::new(2, 3)]) + .unwrap(); + + assert_eq!(filtered.row_ids, vec![1, 4]); + assert_eq!(filtered.scores, vec![0.1, 0.2]); + } } diff --git a/crates/paimon/src/lib.rs b/crates/paimon/src/lib.rs index 85ad6b51..e372a40b 100644 --- a/crates/paimon/src/lib.rs +++ b/crates/paimon/src/lib.rs @@ -32,13 +32,13 @@ mod deletion_vector; pub mod file_index; #[cfg(feature = "fulltext")] pub mod ftindex; +#[cfg(feature = "fulltext")] +pub mod full_text; pub mod io; pub mod lumina; mod predicate_stats; pub mod spec; pub mod table; -#[cfg(feature = "fulltext")] -pub mod tantivy; pub mod variant; pub mod vector_search; pub mod vindex; diff --git a/crates/paimon/src/table/data_file_reader.rs b/crates/paimon/src/table/data_file_reader.rs index 7c4861f7..b94016d1 100644 --- a/crates/paimon/src/table/data_file_reader.rs +++ b/crates/paimon/src/table/data_file_reader.rs @@ -170,7 +170,7 @@ impl DataFileReader { file_meta, data_fields, dv, - None, + split.row_ranges().map(|ranges| ranges.to_vec()), )?; while let Some(batch) = stream.next().await { yield batch?; diff --git a/crates/paimon/src/table/full_text_index_adapter.rs b/crates/paimon/src/table/full_text_index_adapter.rs new file mode 100644 index 00000000..680d1c3d --- /dev/null +++ b/crates/paimon/src/table/full_text_index_adapter.rs @@ -0,0 +1,411 @@ +// 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. + +use crate::ftindex::reader::FullTextArchiveReader; +use crate::io::FileRead; +use bytes::Bytes; +use futures::StreamExt; +use paimon_ftindex_core::io::{ReadRequest, SeekRead}; +use paimon_ftindex_core::FullTextSearchResult; +use roaring::RoaringTreemap; +use std::fs::File; +use std::io::{self, Read, Seek, SeekFrom}; +use std::sync::{mpsc, Arc, Mutex}; + +const FULL_TEXT_RANGE_READ_CONCURRENCY: usize = 8; + +pub(crate) async fn search_full_text_index( + reader: Box, + file_name: String, + query: String, + limit: usize, + include_row_ids: Option, +) -> crate::Result { + let handle = + tokio::runtime::Handle::try_current().map_err(|error| crate::Error::UnexpectedError { + message: "Full-text index reader requires a Tokio runtime".to_string(), + source: Some(Box::new(error)), + })?; + let task_file_name = file_name.clone(); + tokio::task::spawn_blocking(move || { + search_native_index( + AsyncFileSeekRead::new(reader, handle), + &task_file_name, + &query, + limit, + include_row_ids, + ) + }) + .await + .map_err(|error| crate::Error::UnexpectedError { + message: format!("Full-text index task for '{file_name}' failed: {error}"), + source: None, + })? +} + +pub(crate) async fn search_full_text_file( + file: File, + file_name: String, + query: String, + limit: usize, + include_row_ids: Option, +) -> crate::Result { + let task_file_name = file_name.clone(); + tokio::task::spawn_blocking(move || { + search_native_index( + StdFileSeekRead::new(file), + &task_file_name, + &query, + limit, + include_row_ids, + ) + }) + .await + .map_err(|error| crate::Error::UnexpectedError { + message: format!("Full-text index task for '{file_name}' failed: {error}"), + source: None, + })? +} + +fn search_native_index( + input: R, + file_name: &str, + query: &str, + limit: usize, + include_row_ids: Option, +) -> crate::Result { + let reader = FullTextArchiveReader::from_seek_read(input) + .map_err(|error| native_error(file_name, error))?; + let hits = match include_row_ids { + Some(include) => reader.search_with_include(query, limit, &include), + None => reader.search(query, limit), + } + .map_err(|error| native_error(file_name, error))?; + Ok(FullTextSearchResult { + row_ids: hits.row_ids, + scores: hits.scores, + }) +} + +fn native_error(file_name: &str, error: impl std::fmt::Display) -> crate::Error { + crate::Error::UnexpectedError { + message: format!("Failed to read full-text index '{file_name}': {error}"), + source: None, + } +} + +struct AsyncFileSeekRead { + reader: Arc, + handle: tokio::runtime::Handle, + permits: Arc, +} + +impl AsyncFileSeekRead { + fn new(reader: Box, handle: tokio::runtime::Handle) -> Self { + Self { + reader: Arc::from(reader), + handle, + permits: Arc::new(tokio::sync::Semaphore::new( + FULL_TEXT_RANGE_READ_CONCURRENCY, + )), + } + } +} + +impl SeekRead for AsyncFileSeekRead { + fn pread(&self, requests: &mut [ReadRequest<'_>]) -> io::Result<()> { + if requests.is_empty() { + return Ok(()); + } + let ranges = requests + .iter() + .map(|request| { + let length = u64::try_from(request.buf.len()).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidInput, + "full-text read length overflow", + ) + })?; + let end = request.pos.checked_add(length).ok_or_else(|| { + io::Error::new(io::ErrorKind::InvalidInput, "full-text read range overflow") + })?; + Ok((request.pos, end)) + }) + .collect::>>()?; + + let reader = Arc::clone(&self.reader); + let permits = Arc::clone(&self.permits); + let (tx, rx) = mpsc::sync_channel(1); + self.handle.spawn(async move { + let results = futures::stream::iter(ranges.into_iter().map(|(start, end)| { + let reader = Arc::clone(&reader); + let permits = Arc::clone(&permits); + async move { + if start == end { + return Ok(Bytes::new()); + } + let _permit = permits + .acquire_owned() + .await + .map_err(|_| io::Error::other("full-text range reader closed"))?; + reader + .read(start..end) + .await + .map_err(|error| io::Error::other(error.to_string())) + } + })) + .buffered(FULL_TEXT_RANGE_READ_CONCURRENCY) + .collect::>() + .await; + let _ = tx.send(results); + }); + + let results = rx + .recv() + .map_err(|_| io::Error::other("full-text async read task was cancelled"))?; + for (request, result) in requests.iter_mut().zip(results) { + let bytes = result?; + if bytes.len() != request.buf.len() { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + format!( + "full-text read expected {} bytes, got {}", + request.buf.len(), + bytes.len() + ), + )); + } + request.buf.copy_from_slice(&bytes); + } + Ok(()) + } +} + +struct StdFileSeekRead { + file: Mutex, +} + +impl StdFileSeekRead { + fn new(file: File) -> Self { + Self { + file: Mutex::new(file), + } + } +} + +impl SeekRead for StdFileSeekRead { + fn pread(&self, requests: &mut [ReadRequest<'_>]) -> io::Result<()> { + let mut file = self + .file + .lock() + .map_err(|_| io::Error::other("full-text temporary file lock poisoned"))?; + for request in requests { + file.seek(SeekFrom::Start(request.pos))?; + file.read_exact(request.buf)?; + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use paimon_ftindex_core::io::PosWriter; + use paimon_ftindex_core::{FullTextIndexConfig, FullTextIndexWriter}; + use std::ops::Range; + use std::sync::atomic::{AtomicUsize, Ordering}; + + struct BytesFileRead { + data: Bytes, + ranges: Arc>>>, + } + + #[async_trait::async_trait] + impl FileRead for BytesFileRead { + async fn read(&self, range: Range) -> crate::Result { + self.ranges.lock().unwrap().push(range.clone()); + tokio::task::yield_now().await; + let start = usize::try_from(range.start).unwrap(); + let end = usize::try_from(range.end).unwrap(); + Ok(self.data.slice(start..end)) + } + } + + fn test_index_bytes() -> Vec { + let mut writer = FullTextIndexWriter::new(FullTextIndexConfig::new()).unwrap(); + writer.add_document(7, "apache paimon").unwrap(); + writer.add_document(9, "unrelated").unwrap(); + let mut bytes = Vec::new(); + writer.write(&mut PosWriter::new(&mut bytes)).unwrap(); + bytes + } + + #[test] + fn test_range_backed_search_works_on_current_thread_runtime() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + let data = Bytes::from(test_index_bytes()); + let ranges = Arc::new(Mutex::new(Vec::new())); + let reader = BytesFileRead { + data, + ranges: Arc::clone(&ranges), + }; + + let result = search_full_text_index( + Box::new(reader), + "test.index".to_string(), + r#"{"match":{"query":"paimon"}}"#.to_string(), + 10, + None, + ) + .await + .unwrap(); + + assert_eq!(result.row_ids, vec![7]); + assert!(!ranges.lock().unwrap().is_empty()); + }); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_range_backed_search_works_on_multi_thread_runtime() { + let data = Bytes::from(test_index_bytes()); + let ranges = Arc::new(Mutex::new(Vec::new())); + let result = search_full_text_index( + Box::new(BytesFileRead { + data, + ranges: Arc::clone(&ranges), + }), + "test.index".to_string(), + r#"{"match":{"query":"paimon"}}"#.to_string(), + 10, + None, + ) + .await + .unwrap(); + + assert_eq!(result.row_ids, vec![7]); + assert!(!ranges.lock().unwrap().is_empty()); + } + + #[test] + fn test_async_seek_reader_preserves_multi_range_order() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + let data = Bytes::from_static(b"0123456789"); + let reader = AsyncFileSeekRead::new( + Box::new(BytesFileRead { + data, + ranges: Arc::new(Mutex::new(Vec::new())), + }), + tokio::runtime::Handle::current(), + ); + let result = tokio::task::spawn_blocking(move || { + let mut first = [0u8; 3]; + let mut second = [0u8; 2]; + reader.pread(&mut [ + ReadRequest::new(4, &mut first), + ReadRequest::new(1, &mut second), + ])?; + Ok::<_, io::Error>((first, second)) + }) + .await + .unwrap() + .unwrap(); + + assert_eq!(&result.0, b"456"); + assert_eq!(&result.1, b"12"); + }); + } + + struct ShortFileRead; + + #[async_trait::async_trait] + impl FileRead for ShortFileRead { + async fn read(&self, range: Range) -> crate::Result { + let requested = usize::try_from(range.end - range.start).unwrap(); + Ok(Bytes::from(vec![0; requested.saturating_sub(1)])) + } + } + + #[tokio::test] + async fn test_async_seek_reader_rejects_short_reads() { + let reader = + AsyncFileSeekRead::new(Box::new(ShortFileRead), tokio::runtime::Handle::current()); + let error = tokio::task::spawn_blocking(move || { + let mut buffer = [0u8; 4]; + reader.pread(&mut [ReadRequest::new(0, &mut buffer)]) + }) + .await + .unwrap() + .unwrap_err(); + + assert_eq!(error.kind(), io::ErrorKind::UnexpectedEof); + } + + struct ConcurrentFileRead { + active: Arc, + maximum: Arc, + } + + #[async_trait::async_trait] + impl FileRead for ConcurrentFileRead { + async fn read(&self, range: Range) -> crate::Result { + let active = self.active.fetch_add(1, Ordering::SeqCst) + 1; + self.maximum.fetch_max(active, Ordering::SeqCst); + tokio::task::yield_now().await; + self.active.fetch_sub(1, Ordering::SeqCst); + Ok(Bytes::from(vec![ + 0; + usize::try_from(range.end - range.start) + .unwrap() + ])) + } + } + + #[tokio::test] + async fn test_async_seek_reader_bounds_range_concurrency() { + let active = Arc::new(AtomicUsize::new(0)); + let maximum = Arc::new(AtomicUsize::new(0)); + let reader = AsyncFileSeekRead::new( + Box::new(ConcurrentFileRead { + active: Arc::clone(&active), + maximum: Arc::clone(&maximum), + }), + tokio::runtime::Handle::current(), + ); + tokio::task::spawn_blocking(move || { + let mut buffers = [[0u8; 1]; FULL_TEXT_RANGE_READ_CONCURRENCY * 2]; + let mut requests = buffers + .iter_mut() + .enumerate() + .map(|(position, buffer)| ReadRequest::new(position as u64, buffer)) + .collect::>(); + reader.pread(&mut requests) + }) + .await + .unwrap() + .unwrap(); + + assert!(maximum.load(Ordering::SeqCst) <= FULL_TEXT_RANGE_READ_CONCURRENCY); + } +} diff --git a/crates/paimon/src/table/full_text_search_builder.rs b/crates/paimon/src/table/full_text_search_builder.rs index 30e1d358..005d65e6 100644 --- a/crates/paimon/src/table/full_text_search_builder.rs +++ b/crates/paimon/src/table/full_text_search_builder.rs @@ -19,26 +19,29 @@ //! //! Reference: [FullTextSearchBuilderImpl.java](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilderImpl.java) -use crate::io::{FileIO, FileIOBuilder}; +use crate::full_text::{FullTextSearch, SearchResult}; +use crate::io::FileIO; use crate::spec::{ CoreOptions, DataField, FileKind, GlobalIndexSearchMode, IndexFileMeta, IndexManifest, IndexManifestEntry, ROW_ID_FIELD_NAME, }; +use crate::table::full_text_index_adapter::{search_full_text_file, search_full_text_index}; use crate::table::global_index_scanner::{ deleted_row_ranges_for_data_evolution_dvs, search_limit_with_deleted_rows, unindexed_ranges_for_global_index_entries, RowRangeIndex, }; use crate::table::{find_field_id_by_name, merge_row_ranges, RowRange, Table}; -use crate::tantivy::full_text_search::{FullTextSearch, SearchResult}; -use crate::tantivy::reader::TantivyFullTextReader; -use crate::tantivy::writer::TantivyFullTextWriter; use arrow_array::{Array, Int64Array, LargeStringArray, RecordBatch, StringArray}; -use futures::TryStreamExt; +use futures::{StreamExt, TryStreamExt}; +use paimon_ftindex_core::io::PosWriter; +use paimon_ftindex_core::{FullTextIndexConfig, FullTextIndexWriter}; +use roaring::RoaringTreemap; +use serde_json::json; use std::collections::{HashMap, HashSet}; -use uuid::Uuid; const INDEX_DIR: &str = "index"; -const TANTIVY_FULLTEXT_INDEX_TYPE: &str = "tantivy-fulltext"; +const FULL_TEXT_INDEX_TYPE: &str = "full-text"; +const FULL_TEXT_INDEX_SEARCH_CONCURRENCY: usize = 8; /// Builder for executing full-text search on a Paimon table. /// @@ -58,6 +61,7 @@ pub struct FullTextSearchBuilder<'a> { text_column: Option, query_text: Option, limit: Option, + include_row_ids: Option, } impl<'a> FullTextSearchBuilder<'a> { @@ -67,6 +71,7 @@ impl<'a> FullTextSearchBuilder<'a> { text_column: None, query_text: None, limit: None, + include_row_ids: None, } } @@ -82,6 +87,12 @@ impl<'a> FullTextSearchBuilder<'a> { self } + /// Set candidate row IDs to include in the search. + pub fn with_include_row_ids(&mut self, include_row_ids: RoaringTreemap) -> &mut Self { + self.include_row_ids = Some(include_row_ids); + self + } + /// Set the top-k limit for results. pub fn with_limit(&mut self, limit: usize) -> &mut Self { self.limit = Some(limit); @@ -91,7 +102,7 @@ impl<'a> FullTextSearchBuilder<'a> { /// Execute the full-text search and return row ranges. /// /// This reads the latest snapshot, loads the index manifest, and evaluates - /// the search against Tantivy indexes. + /// the search against full-text indexes. /// /// Reference: `FullTextSearchBuilder.executeLocal()` pub async fn execute(&self) -> crate::Result> { @@ -117,7 +128,14 @@ impl<'a> FullTextSearchBuilder<'a> { message: "Limit must be set via with_limit()".to_string(), })?; - let search = FullTextSearch::new(query_text.to_string(), limit, text_column.to_string())?; + let mut search = FullTextSearch::new( + normalize_query_text(query_text, text_column)?, + limit, + text_column.to_string(), + )?; + if let Some(include_row_ids) = &self.include_row_ids { + search = search.with_include_row_ids(include_row_ids.clone()); + } let snapshot_manager = self.table.snapshot_manager(); @@ -150,7 +168,7 @@ impl<'a> FullTextSearchBuilder<'a> { } } -/// Evaluate a full-text search query against Tantivy indexes found in the index manifest. +/// Evaluate a full-text search query against full-text indexes found in the index manifest. struct FullTextSearchEvaluation<'a> { table: Option<&'a Table>, file_io: &'a FileIO, @@ -174,20 +192,21 @@ async fn evaluate_full_text_search( None => return Ok(SearchResult::empty()), }; - // Collect tantivy fulltext entries for the target field. + // Collect full-text entries for the target field. let fulltext_entries: Vec<_> = index_entries .iter() .filter(|e| { e.kind == FileKind::Add - && e.index_file.index_type == TANTIVY_FULLTEXT_INDEX_TYPE + && e.index_file.index_type == FULL_TEXT_INDEX_TYPE && e.index_file .global_index_meta .as_ref() .is_some_and(|m| m.index_field_id == field_id) }) .collect(); + let has_fulltext_entries = !fulltext_entries.is_empty(); - if fulltext_entries.is_empty() && search_mode == GlobalIndexSearchMode::Fast { + if !has_fulltext_entries && search_mode == GlobalIndexSearchMode::Fast { return Ok(SearchResult::empty()); } @@ -206,13 +225,16 @@ async fn evaluate_full_text_search( let mut merged = SearchResult::empty(); if !fulltext_entries.is_empty() { - let futures: Vec<_> = fulltext_entries - .into_iter() - .map(|entry| { + let plans = + plan_full_text_index_searches(fulltext_entries, search.include_row_ids.as_ref())?; + let results = futures::stream::iter(plans) + .map(|plan| { + let entry = plan.entry; let global_meta = entry.index_file.global_index_meta.as_ref().unwrap(); let path = format!("{table_path}/{INDEX_DIR}/{}", entry.index_file.file_name); let file_name = entry.index_file.file_name.clone(); let query_text = search.query_text.clone(); + let local_filter = plan.local_filter; let row_range_start = global_meta.row_range_start; let row_range_end = global_meta.row_range_end; let limit = search_limit_with_deleted_rows( @@ -221,31 +243,24 @@ async fn evaluate_full_text_search( row_range_end, deleted_row_index.as_ref(), ); - let input = evaluation.file_io.new_input(&path); async move { - let input = input?; - let reader = TantivyFullTextReader::from_input_file(&input) - .await - .map_err(|e| crate::Error::UnexpectedError { - message: format!( - "Failed to open Tantivy full-text index '{}': {}", - file_name, e - ), - source: None, - })?; - let result = reader.search(&query_text, limit)?; - Ok::<_, crate::Error>(result.offset(row_range_start)) + let input = evaluation.file_io.new_input(&path)?; + let reader = Box::new(input.reader().await?); + let result = + search_full_text_index(reader, file_name, query_text, limit, local_filter) + .await?; + Ok::<_, crate::Error>(search_result_from_core(result)?.offset(row_range_start)) } }) - .collect(); - - let results = futures::future::try_join_all(futures).await?; + .buffered(FULL_TEXT_INDEX_SEARCH_CONCURRENCY) + .try_collect::>() + .await?; for r in &results { merged = merged.or(r); } } - if search_mode != GlobalIndexSearchMode::Fast { + if search_mode != GlobalIndexSearchMode::Fast && has_fulltext_entries { let detail_ranges = if search_mode == GlobalIndexSearchMode::Detail { let table = evaluation.table.ok_or_else(|| crate::Error::DataInvalid { message: "Full-text raw search in detail mode requires table context".to_string(), @@ -262,14 +277,17 @@ async fn evaluate_full_text_search( search_mode, evaluation.next_row_id, &detail_ranges, - is_tantivy_fulltext_index_file, + is_full_text_index_file, ); if !raw_ranges.is_empty() { let table = evaluation.table.ok_or_else(|| crate::Error::DataInvalid { message: "Full-text raw search requires table context".to_string(), source: None, })?; - let raw_result = read_raw_full_text_search(table, search, &raw_ranges).await?; + let raw_result = + read_raw_full_text_search(table, search, &raw_ranges, evaluation.table_options) + .await?; + merged = merged.without_row_ranges(&raw_ranges)?; merged = merged.or(&raw_result); } } @@ -279,8 +297,88 @@ async fn evaluate_full_text_search( .top_k(search.limit)) } -fn is_tantivy_fulltext_index_file(index_file: &IndexFileMeta) -> bool { - index_file.index_type == TANTIVY_FULLTEXT_INDEX_TYPE +struct FullTextIndexSearchPlan { + entry: IndexManifestEntry, + local_filter: Option, +} + +fn plan_full_text_index_searches( + entries: Vec<&IndexManifestEntry>, + include_row_ids: Option<&RoaringTreemap>, +) -> crate::Result> { + let mut ranges = Vec::with_capacity(entries.len()); + for (index, entry) in entries.iter().enumerate() { + let meta = entry.index_file.global_index_meta.as_ref().ok_or_else(|| { + crate::Error::DataInvalid { + message: format!( + "Full-text index '{}' is missing global index metadata", + entry.index_file.file_name + ), + source: None, + } + })?; + if meta.row_range_start < 0 || meta.row_range_end < meta.row_range_start { + return Err(crate::Error::DataInvalid { + message: format!( + "Invalid full-text row range [{}, {}] for '{}'", + meta.row_range_start, meta.row_range_end, entry.index_file.file_name + ), + source: None, + }); + } + ranges.push(( + meta.row_range_start as u64, + meta.row_range_end as u64, + index, + )); + } + ranges.sort_unstable_by_key(|(start, _, _)| *start); + + let Some(include_row_ids) = include_row_ids else { + return Ok(entries + .into_iter() + .map(|entry| FullTextIndexSearchPlan { + entry: entry.clone(), + local_filter: None, + }) + .collect()); + }; + + let mut local_filters = (0..entries.len()) + .map(|_| RoaringTreemap::new()) + .collect::>(); + let mut active = Vec::::new(); + let mut next_range = 0usize; + for row_id in include_row_ids.iter() { + while next_range < ranges.len() && ranges[next_range].0 <= row_id { + active.push(next_range); + next_range += 1; + } + active.retain(|range_index| ranges[*range_index].1 >= row_id); + for range_index in &active { + let (start, _, original_index) = ranges[*range_index]; + local_filters[original_index].insert(row_id - start); + } + if next_range == ranges.len() && active.is_empty() { + break; + } + } + + let mut plans = Vec::with_capacity(entries.len()); + for (entry, local_filter) in entries.into_iter().zip(local_filters) { + if local_filter.is_empty() { + continue; + } + plans.push(FullTextIndexSearchPlan { + entry: entry.clone(), + local_filter: Some(local_filter), + }); + } + Ok(plans) +} + +fn is_full_text_index_file(index_file: &IndexFileMeta) -> bool { + index_file.index_type == FULL_TEXT_INDEX_TYPE } async fn detail_data_ranges_for_table(table: &Table) -> crate::Result> { @@ -305,15 +403,22 @@ async fn read_raw_full_text_search( table: &Table, search: &FullTextSearch, raw_ranges: &[RowRange], + table_options: &HashMap, ) -> crate::Result { if raw_ranges.is_empty() { return Ok(SearchResult::empty()); } + let raw_ranges = merge_row_ranges(raw_ranges.to_vec()); + let row_range_start = raw_ranges.first().map(RowRange::from).unwrap_or(0); + let row_range_end = raw_ranges + .last() + .map(RowRange::to) + .unwrap_or(row_range_start); let mut read_builder = table.new_read_builder(); read_builder .with_projection(&[search.field_name.as_str(), ROW_ID_FIELD_NAME])? - .with_row_ranges(raw_ranges.to_vec()); + .with_row_ranges(raw_ranges); let plan = read_builder.new_scan().plan().await?; if plan.splits().is_empty() { return Ok(SearchResult::empty()); @@ -321,31 +426,85 @@ async fn read_raw_full_text_search( let read = read_builder.new_read()?; let mut stream = read.to_arrow(plan.splits())?; - let mut writer = TantivyFullTextWriter::new()?; + let config = raw_full_text_config(table_options)?; + let mut writer = tokio::task::spawn_blocking(move || { + FullTextIndexWriter::new(config).map_err(full_text_error) + }) + .await + .map_err(raw_index_task_error)??; + let mut document_count = 0usize; while let Some(batch) = stream.try_next().await? { - add_raw_full_text_batch(&batch, search, &mut writer)?; + let documents = raw_full_text_batch_documents(&batch, search, row_range_start)?; + document_count += documents.len(); + writer = tokio::task::spawn_blocking(move || { + for (row_id, text) in documents { + writer.add_document(row_id, text).map_err(full_text_error)?; + } + Ok::<_, crate::Error>(writer) + }) + .await + .map_err(raw_index_task_error)??; } - let memory_io = FileIOBuilder::new("memory").build()?; - let output = memory_io.new_output(&format!("/raw-fulltext-{}.archive", Uuid::new_v4()))?; - if !writer.finish(&output).await? { + if document_count == 0 { return Ok(SearchResult::empty()); } - let input = output.to_input_file(); - let reader = TantivyFullTextReader::from_input_file(&input) - .await - .map_err(|e| crate::Error::UnexpectedError { - message: format!("Failed to open raw Tantivy full-text index: {e}"), + let temporary_index = tokio::task::spawn_blocking(move || { + let mut temporary_index = + tempfile::NamedTempFile::new().map_err(|error| crate::Error::UnexpectedError { + message: format!("Failed to create temporary full-text index: {error}"), + source: None, + })?; + writer + .write(&mut PosWriter::new(temporary_index.as_file_mut())) + .map_err(full_text_error)?; + temporary_index.as_file_mut().sync_all().map_err(|error| { + crate::Error::UnexpectedError { + message: format!("Failed to flush temporary full-text index: {error}"), + source: None, + } + })?; + Ok::<_, crate::Error>(temporary_index) + }) + .await + .map_err(raw_index_task_error)??; + let index_file = temporary_index + .reopen() + .map_err(|error| crate::Error::UnexpectedError { + message: format!("Failed to reopen temporary full-text index: {error}"), source: None, })?; - reader.search(&search.query_text, search.limit) + let local_filter = match search.include_row_ids.as_ref() { + Some(include_row_ids) => Some(local_filter( + include_row_ids, + row_range_start, + row_range_end, + )?), + None => None, + }; + let result = search_full_text_file( + index_file, + "raw-full-text.index".to_string(), + search.query_text.clone(), + search.limit, + local_filter, + ) + .await?; + search_result_from_core(result).map(|result| result.offset(row_range_start)) +} + +fn raw_index_task_error(error: tokio::task::JoinError) -> crate::Error { + crate::Error::UnexpectedError { + message: format!("Full-text raw index task failed: {error}"), + source: None, + } } -fn add_raw_full_text_batch( +fn raw_full_text_batch_documents( batch: &RecordBatch, search: &FullTextSearch, - writer: &mut TantivyFullTextWriter, -) -> crate::Result<()> { + row_range_start: i64, +) -> crate::Result> { let text_index = batch .schema() @@ -376,17 +535,33 @@ fn add_raw_full_text_batch( let column = batch.column(text_index); if let Some(strings) = column.as_any().downcast_ref::() { + let mut documents = Vec::with_capacity(batch.num_rows()); for row in 0..batch.num_rows() { - add_raw_full_text_row(row_ids, row, get_string_value(strings, row), writer)?; + if let Some(document) = raw_full_text_document( + row_ids, + row, + get_string_value(strings, row), + row_range_start, + )? { + documents.push(document); + } } - return Ok(()); + return Ok(documents); } if let Some(strings) = column.as_any().downcast_ref::() { + let mut documents = Vec::with_capacity(batch.num_rows()); for row in 0..batch.num_rows() { - add_raw_full_text_row(row_ids, row, get_large_string_value(strings, row), writer)?; + if let Some(document) = raw_full_text_document( + row_ids, + row, + get_large_string_value(strings, row), + row_range_start, + )? { + documents.push(document); + } } - return Ok(()); + return Ok(documents); } Err(crate::Error::DataInvalid { @@ -411,41 +586,153 @@ fn get_large_string_value(strings: &LargeStringArray, row: usize) -> Option<&str } } -fn add_raw_full_text_row( +fn raw_full_text_document( row_ids: &Int64Array, row: usize, text: Option<&str>, - writer: &mut TantivyFullTextWriter, -) -> crate::Result<()> { + row_range_start: i64, +) -> crate::Result> { if row_ids.is_null(row) { return Err(crate::Error::DataInvalid { message: "Full-text raw search found null _ROW_ID".to_string(), source: None, }); } - let row_id = u64::try_from(row_ids.value(row)).map_err(|_| crate::Error::DataInvalid { - message: format!( - "Negative _ROW_ID {} cannot be used for global index search", - row_ids.value(row) - ), + let row_id = row_ids.value(row); + if row_id < 0 { + return Err(crate::Error::DataInvalid { + message: format!("Negative _ROW_ID {row_id} cannot be used for global index search"), + source: None, + }); + } + match text { + Some(text) => { + let local_row_id = + row_id + .checked_sub(row_range_start) + .ok_or_else(|| crate::Error::DataInvalid { + message: format!( + "Raw full-text row id {row_id} is before row range start {row_range_start}" + ), + source: None, + })?; + Ok(Some((local_row_id, text.to_string()))) + } + None => Ok(None), + } +} + +fn raw_full_text_config( + table_options: &HashMap, +) -> crate::Result { + let native_options: HashMap = table_options + .iter() + .filter_map(|(key, value)| { + key.strip_prefix("full-text.") + .map(|native_key| (native_key.to_string(), value.clone())) + }) + .collect(); + FullTextIndexConfig::from_options(&native_options).map_err(full_text_error) +} + +fn local_filter( + global: &RoaringTreemap, + row_range_start: i64, + row_range_end: i64, +) -> crate::Result { + if row_range_start < 0 || row_range_end < row_range_start { + return Err(crate::Error::DataInvalid { + message: format!("Invalid full-text row range [{row_range_start}, {row_range_end}]"), + source: None, + }); + } + + let start = u64::try_from(row_range_start).map_err(|_| crate::Error::DataInvalid { + message: format!("Invalid negative row range start {row_range_start}"), + source: None, + })?; + let end = u64::try_from(row_range_end).map_err(|_| crate::Error::DataInvalid { + message: format!("Invalid negative row range end {row_range_end}"), source: None, })?; - writer.add_document(row_id, text) + let mut local = RoaringTreemap::new(); + for row_id in global + .iter() + .filter(|row_id| *row_id >= start && *row_id <= end) + { + local.insert(row_id - start); + } + Ok(local) +} + +fn normalize_query_text(query_text: &str, _field_name: &str) -> crate::Result { + if is_full_text_query_dsl(query_text) { + return Ok(query_text.to_string()); + } + serde_json::to_string(&json!({ + "match": { + "query": query_text, + } + })) + .map_err(|e| crate::Error::ConfigInvalid { + message: format!("Failed to build full-text query JSON: {e}"), + }) +} + +fn is_full_text_query_dsl(query_text: &str) -> bool { + let Ok(serde_json::Value::Object(object)) = + serde_json::from_str::(query_text) + else { + return false; + }; + object.keys().any(|key| { + matches!( + key.as_str(), + "match" | "multi_match" | "match_phrase" | "phrase" | "boolean" | "boost" + ) + }) +} + +fn search_result_from_core( + result: paimon_ftindex_core::FullTextSearchResult, +) -> crate::Result { + let mut row_ids = Vec::with_capacity(result.row_ids.len()); + for row_id in result.row_ids { + row_ids.push( + u64::try_from(row_id).map_err(|_| crate::Error::DataInvalid { + message: format!("Full-text index returned negative row id {row_id}"), + source: None, + })?, + ); + } + Ok(SearchResult::new(row_ids, result.scores)) +} + +fn full_text_error(error: impl std::fmt::Display) -> crate::Error { + crate::Error::UnexpectedError { + message: format!("Full-text index error: {error}"), + source: None, + } } #[cfg(test)] mod tests { use super::*; use crate::catalog::Identifier; - use crate::spec::{DataType, IntType, Schema, TableSchema, VarCharType}; + use crate::io::FileIOBuilder; + use crate::spec::{DataType, GlobalIndexMeta, IntType, Schema, TableSchema, VarCharType}; use crate::table::table_write::TableWrite; use crate::table::TableCommit; use arrow_array::StringArray; use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema as ArrowSchema}; + use bytes::Bytes; + use paimon_ftindex_core::io::SliceReader; + use paimon_ftindex_core::FullTextIndexReader; + use roaring::RoaringTreemap; use std::sync::Arc; #[tokio::test] - async fn test_evaluate_full_mode_without_fulltext_entries_uses_raw_path() { + async fn test_evaluate_full_mode_without_fulltext_entries_returns_empty() { let file_io = FileIOBuilder::new("memory").build().unwrap(); let fields = vec![DataField::new( 1, @@ -455,7 +742,7 @@ mod tests { let search = FullTextSearch::new("hello".to_string(), 10, "body".to_string()).unwrap(); let options = HashMap::from([("global-index.search-mode".to_string(), "full".to_string())]); - let err = evaluate_full_text_search( + let result = evaluate_full_text_search( FullTextSearchEvaluation { table: None, file_io: &file_io, @@ -468,16 +755,12 @@ mod tests { &search, ) .await - .unwrap_err(); - assert!( - err.to_string() - .contains("Full-text raw search requires table context"), - "unexpected error: {err}" - ); + .unwrap(); + assert!(result.is_empty()); } #[tokio::test] - async fn test_execute_full_mode_without_index_manifest_searches_raw_rows() { + async fn test_execute_full_mode_without_index_manifest_returns_empty() { let file_io = FileIOBuilder::new("memory").build().unwrap(); let table_path = "memory:/full_text_raw_no_manifest"; setup_dirs(&file_io, table_path).await; @@ -485,7 +768,13 @@ mod tests { let mut table_write = TableWrite::new(&table, "test-user".to_string()).unwrap(); table_write - .write_arrow_batch(&text_batch(vec!["hello world", "goodbye"])) + .write_arrow_batch(&text_batch(vec![ + "alphaft lakehouse", + "betafft gammaft", + "alphaft indexed", + "gammaft fallback", + "alphaft queried", + ])) .await .unwrap(); let messages = table_write.prepare_commit().await.unwrap(); @@ -497,11 +786,372 @@ mod tests { let mut builder = table.new_full_text_search_builder(); builder .with_text_column("body") - .with_query_text("hello") + .with_query_text("alphaft") .with_limit(10); let row_ranges = builder.execute().await.unwrap(); - assert_eq!(row_ranges, vec![RowRange::new(0, 0)]); + assert!(row_ranges.is_empty()); + } + + #[tokio::test] + async fn test_raw_full_text_search_applies_include_row_ids() { + let file_io = FileIOBuilder::new("memory").build().unwrap(); + let table_path = "memory:/full_text_raw_with_filter"; + setup_dirs(&file_io, table_path).await; + let table = full_text_raw_table(&file_io, table_path); + + let mut table_write = TableWrite::new(&table, "test-user".to_string()).unwrap(); + table_write + .write_arrow_batch(&text_batch(vec!["hello world", "hello filtered"])) + .await + .unwrap(); + let messages = table_write.prepare_commit().await.unwrap(); + TableCommit::new(table.clone(), "test-user".to_string()) + .commit(messages) + .await + .unwrap(); + + let mut filter = RoaringTreemap::new(); + filter.insert(1); + + let search = FullTextSearch::new( + r#"{"match":{"column":"text","query":"hello"}}"#.to_string(), + 10, + "body".to_string(), + ) + .unwrap() + .with_include_row_ids(filter); + let row_ranges = read_raw_full_text_search( + &table, + &search, + &[RowRange::new(0, 1)], + table.schema().options(), + ) + .await + .unwrap() + .to_row_ranges(); + + assert_eq!(row_ranges, vec![RowRange::new(1, 1)]); + } + + #[tokio::test] + async fn test_evaluate_reads_new_full_text_index_file() { + let file_io = FileIOBuilder::new("memory").build().unwrap(); + let table_path = "memory:/full_text_indexed"; + setup_dirs(&file_io, table_path).await; + file_io + .mkdirs(&format!("{table_path}/index/")) + .await + .unwrap(); + + let mut writer = + FullTextIndexWriter::new(FullTextIndexConfig::new().with_text_fields(["body"])) + .unwrap(); + writer + .add_document_fields(0, [("body", "hello world")]) + .unwrap(); + writer + .add_document_fields(1, [("body", "goodbye")]) + .unwrap(); + let mut index_bytes = Vec::new(); + writer.write(&mut PosWriter::new(&mut index_bytes)).unwrap(); + file_io + .new_output(&format!("{table_path}/index/ft-0")) + .unwrap() + .write(Bytes::from(index_bytes.clone())) + .await + .unwrap(); + + let fields = vec![DataField::new( + 1, + "body".to_string(), + DataType::VarChar(VarCharType::string_type()), + )]; + let search = FullTextSearch::new( + normalize_query_text("hello", "body").unwrap(), + 10, + "body".to_string(), + ) + .unwrap(); + let entry = IndexManifestEntry { + kind: FileKind::Add, + partition: Vec::new(), + bucket: 0, + index_file: IndexFileMeta { + index_type: FULL_TEXT_INDEX_TYPE.to_string(), + file_name: "ft-0".to_string(), + file_size: i64::try_from(index_bytes.len()).unwrap(), + row_count: 2, + deletion_vectors_ranges: None, + global_index_meta: Some(GlobalIndexMeta { + row_range_start: 100, + row_range_end: 101, + index_field_id: 1, + extra_field_ids: None, + index_meta: None, + source_meta: None, + }), + }, + version: 1, + }; + let result = evaluate_full_text_search( + FullTextSearchEvaluation { + table: None, + file_io: &file_io, + table_path, + table_options: &HashMap::new(), + schema_fields: &fields, + next_row_id: Some(102), + }, + &[entry], + &search, + ) + .await + .unwrap(); + + assert_eq!(result.row_ids, vec![100]); + } + + #[tokio::test] + async fn test_indexed_full_text_search_offsets_include_row_ids() { + let file_io = FileIOBuilder::new("memory").build().unwrap(); + let table_path = "memory:/full_text_indexed_filter"; + setup_dirs(&file_io, table_path).await; + file_io + .mkdirs(&format!("{table_path}/index/")) + .await + .unwrap(); + + let mut writer = + FullTextIndexWriter::new(FullTextIndexConfig::new().with_text_fields(["body"])) + .unwrap(); + writer + .add_document_fields(0, [("body", "hello world")]) + .unwrap(); + writer + .add_document_fields(1, [("body", "hello world")]) + .unwrap(); + let mut index_bytes = Vec::new(); + writer.write(&mut PosWriter::new(&mut index_bytes)).unwrap(); + file_io + .new_output(&format!("{table_path}/index/ft-filter")) + .unwrap() + .write(Bytes::from(index_bytes.clone())) + .await + .unwrap(); + + let fields = vec![DataField::new( + 1, + "body".to_string(), + DataType::VarChar(VarCharType::string_type()), + )]; + let mut filter = RoaringTreemap::new(); + filter.insert(101); + let search = FullTextSearch::new( + normalize_query_text("hello", "body").unwrap(), + 10, + "body".to_string(), + ) + .unwrap() + .with_include_row_ids(filter); + let entry = IndexManifestEntry { + kind: FileKind::Add, + partition: Vec::new(), + bucket: 0, + index_file: IndexFileMeta { + index_type: FULL_TEXT_INDEX_TYPE.to_string(), + file_name: "ft-filter".to_string(), + file_size: i64::try_from(index_bytes.len()).unwrap(), + row_count: 2, + deletion_vectors_ranges: None, + global_index_meta: Some(GlobalIndexMeta { + row_range_start: 100, + row_range_end: 101, + index_field_id: 1, + extra_field_ids: None, + index_meta: None, + source_meta: None, + }), + }, + version: 1, + }; + let missing_empty_filter_entry = full_text_index_entry("missing-filtered-shard", 200, 201); + + let result = evaluate_full_text_search( + FullTextSearchEvaluation { + table: None, + file_io: &file_io, + table_path, + table_options: &HashMap::new(), + schema_fields: &fields, + next_row_id: Some(102), + }, + &[entry, missing_empty_filter_entry], + &search, + ) + .await + .unwrap(); + + assert_eq!(result.row_ids, vec![101]); + } + + #[test] + fn test_normalize_query_text_only_passes_query_dsl_objects() { + let normalized = normalize_query_text("123", "body").unwrap(); + assert_eq!( + serde_json::from_str::(&normalized).unwrap(), + serde_json::json!({"match": {"query": "123"}}) + ); + + let normalized = normalize_query_text("\"hello\"", "body").unwrap(); + assert_eq!( + serde_json::from_str::(&normalized).unwrap(), + serde_json::json!({"match": {"query": "\"hello\""}}) + ); + + let dsl = r#"{"match":{"query":"hello"}}"#; + assert_eq!(normalize_query_text(dsl, "body").unwrap(), dsl); + } + + #[test] + fn test_raw_full_text_config_preserves_native_text_fields() { + let options = HashMap::from([( + "full-text.text-fields".to_string(), + "native_text,secondary".to_string(), + )]); + + let config = raw_full_text_config(&options).unwrap(); + + assert_eq!(config.text_fields, vec!["native_text", "secondary"]); + assert_eq!(config.default_text_field(), "native_text"); + } + + #[test] + fn test_raw_batches_keep_one_bm25_corpus_and_configured_field() { + let options = HashMap::from([( + "full-text.text-field".to_string(), + "native_text".to_string(), + )]); + let config = raw_full_text_config(&options).unwrap(); + let search = FullTextSearch::new( + r#"{"match":{"column":"native_text","query":"rare common"}}"#.to_string(), + 10, + "body".to_string(), + ) + .unwrap(); + let batches = [ + raw_text_batch(&[10, 11], &["rare common", "common common common common"]), + raw_text_batch(&[20, 21], &["rare", "unrelated"]), + ]; + + let mut batched_writer = FullTextIndexWriter::new(config.clone()).unwrap(); + let mut all_documents = Vec::new(); + for batch in &batches { + let documents = raw_full_text_batch_documents(batch, &search, 10).unwrap(); + for (row_id, text) in &documents { + batched_writer.add_document(*row_id, text).unwrap(); + } + all_documents.extend(documents); + } + let mut reference_writer = FullTextIndexWriter::new(config).unwrap(); + for (row_id, text) in all_documents { + reference_writer.add_document(row_id, text).unwrap(); + } + + let query = search.query_text.as_str(); + let batched_result = write_and_search_test_index(&mut batched_writer, query); + let reference_result = write_and_search_test_index(&mut reference_writer, query); + + assert!(!batched_result.row_ids.is_empty()); + assert_eq!(batched_result.row_ids, reference_result.row_ids); + assert_eq!(batched_result.scores, reference_result.scores); + } + + fn raw_text_batch(row_ids: &[i64], texts: &[&str]) -> RecordBatch { + let schema = Arc::new(ArrowSchema::new(vec![ + ArrowField::new("body", ArrowDataType::Utf8, true), + ArrowField::new(ROW_ID_FIELD_NAME, ArrowDataType::Int64, false), + ])); + RecordBatch::try_new( + schema, + vec![ + Arc::new(StringArray::from(texts.to_vec())), + Arc::new(Int64Array::from(row_ids.to_vec())), + ], + ) + .unwrap() + } + + fn write_and_search_test_index( + writer: &mut FullTextIndexWriter, + query: &str, + ) -> paimon_ftindex_core::FullTextSearchResult { + let mut bytes = Vec::new(); + writer.write(&mut PosWriter::new(&mut bytes)).unwrap(); + FullTextIndexReader::open(SliceReader::new(bytes)) + .unwrap() + .search(query, 10) + .unwrap() + } + + #[test] + fn test_plan_full_text_index_searches_partitions_once_and_preserves_order() { + let entries = [ + full_text_index_entry("first", 100, 109), + full_text_index_entry("second", 0, 9), + full_text_index_entry("overlap", 104, 106), + full_text_index_entry("empty", 200, 209), + ]; + let mut include_row_ids = RoaringTreemap::new(); + include_row_ids.insert(1); + include_row_ids.insert(105); + include_row_ids.insert(999); + + let plans = plan_full_text_index_searches(entries.iter().collect(), Some(&include_row_ids)) + .unwrap(); + + assert_eq!( + plans + .iter() + .map(|plan| plan.entry.index_file.file_name.as_str()) + .collect::>(), + vec!["first", "second", "overlap"] + ); + let decoded = plans + .iter() + .map(|plan| { + plan.local_filter + .as_ref() + .unwrap() + .iter() + .collect::>() + }) + .collect::>(); + assert_eq!(decoded, vec![vec![5], vec![1], vec![1]]); + } + + fn full_text_index_entry(name: &str, start: i64, end: i64) -> IndexManifestEntry { + IndexManifestEntry { + kind: FileKind::Add, + partition: Vec::new(), + bucket: 0, + index_file: IndexFileMeta { + index_type: FULL_TEXT_INDEX_TYPE.to_string(), + file_name: name.to_string(), + file_size: 0, + row_count: i32::try_from(end - start + 1).unwrap(), + deletion_vectors_ranges: None, + global_index_meta: Some(GlobalIndexMeta { + row_range_start: start, + row_range_end: end, + index_field_id: 1, + extra_field_ids: None, + index_meta: None, + source_meta: None, + }), + }, + version: 1, + } } async fn setup_dirs(file_io: &FileIO, table_path: &str) { diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs index 5c34918d..b7d48a63 100644 --- a/crates/paimon/src/table/mod.rs +++ b/crates/paimon/src/table/mod.rs @@ -43,6 +43,8 @@ mod format_table_read; mod format_table_scan; mod format_write_builder; #[cfg(feature = "fulltext")] +mod full_text_index_adapter; +#[cfg(feature = "fulltext")] mod full_text_search_builder; pub(crate) mod global_index_build_common; mod global_index_drop_builder; diff --git a/crates/paimon/src/tantivy/directory.rs b/crates/paimon/src/tantivy/directory.rs deleted file mode 100644 index 22440a60..00000000 --- a/crates/paimon/src/tantivy/directory.rs +++ /dev/null @@ -1,399 +0,0 @@ -// 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. - -//! Archive-backed Tantivy Directory implementation. -//! -//! Reads a Tantivy index packed into a single archive file. The archive format -//! (Big-Endian, compatible with Java Paimon): -//! -//! ```text -//! [fileCount: 4 bytes BE] -//! for each file: -//! [nameLen: 4 bytes BE] -//! [name: nameLen bytes UTF-8] -//! [dataLen: 8 bytes BE] -//! [data: dataLen bytes] -//! ``` -//! -//! Reference: `org.apache.paimon.tantivy.index.TantivyFullTextGlobalIndexWriter.packIndex()` -//! Reference: `org.apache.paimon.tantivy.index.TantivyFullTextGlobalIndexReader.parseArchiveHeader()` - -use bytes::Bytes; -use std::collections::HashMap; -use std::fmt; -use std::io; -use std::ops::Range; -use std::path::{Path, PathBuf}; -use std::sync::{Arc, Mutex}; -use tantivy::directory::error::{DeleteError, LockError, OpenReadError, OpenWriteError}; -use tantivy::directory::Directory; -use tantivy::directory::{ - AntiCallToken, DirectoryLock, FileHandle, Lock, OwnedBytes, TerminatingWrite, WatchCallback, - WatchHandle, WritePtr, -}; -use tantivy::HasLen; - -use crate::io::FileRead; - -/// Metadata for a single file within the archive. -#[derive(Clone, Debug)] -struct FileMeta { - /// Absolute byte offset of the file data within the archive. - offset: u64, - length: usize, -} - -/// A read-only Tantivy `Directory` backed by an archive file. -/// -/// Only the archive header (file names, offsets, lengths) is parsed eagerly. -/// Actual file data is read on demand via `FileRead`. -#[derive(Clone)] -pub struct ArchiveDirectory { - files: Arc>, - reader: Arc, - /// In-memory storage for atomic_write (used by Tantivy for meta.json). - atomic_data: Arc>>>, -} - -impl ArchiveDirectory { - /// Create an `ArchiveDirectory` from an async `FileRead`. - /// - /// Only the archive header (file names, offsets, lengths) is read eagerly. - /// Actual file data is read on demand when Tantivy requests it. - pub async fn from_reader(reader: impl FileRead, file_size: u64) -> io::Result { - let reader: Arc = Arc::new(reader); - - if file_size < 4 { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - "Archive too small to contain file count", - )); - } - - // Read file count (4 bytes). - let buf = reader - .read(0..4) - .await - .map_err(|e| io::Error::other(e.to_string()))?; - let file_count = i32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]); - if file_count < 0 { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - format!("Negative file count in archive: {file_count}"), - )); - } - let file_count = file_count as usize; - - let mut pos: u64 = 4; - let mut files = HashMap::with_capacity(file_count); - - for _ in 0..file_count { - // Read name_len (4 bytes). - let buf = reader - .read(pos..pos + 4) - .await - .map_err(|e| io::Error::other(e.to_string()))?; - let name_len = i32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]); - if name_len < 0 { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - format!("Negative name length in archive: {name_len}"), - )); - } - let name_len = name_len as u64; - pos += 4; - - // Read name + data_len together in a single IO call. - let meta_buf = reader - .read(pos..pos + name_len + 8) - .await - .map_err(|e| io::Error::other(e.to_string()))?; - - let name = String::from_utf8(meta_buf[..name_len as usize].to_vec()).map_err(|e| { - io::Error::new( - io::ErrorKind::InvalidData, - format!("Invalid UTF-8 in file name: {}", e), - ) - })?; - - let dl = name_len as usize; - let data_len = i64::from_be_bytes([ - meta_buf[dl], - meta_buf[dl + 1], - meta_buf[dl + 2], - meta_buf[dl + 3], - meta_buf[dl + 4], - meta_buf[dl + 5], - meta_buf[dl + 6], - meta_buf[dl + 7], - ]); - if data_len < 0 { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - format!("Negative data length in archive: {data_len}"), - )); - } - let data_len = data_len as u64; - pos += name_len + 8; - - let data_offset = pos; - files.insert( - PathBuf::from(&name), - FileMeta { - offset: data_offset, - length: data_len as usize, - }, - ); - - // Skip past file data — do NOT read it. - pos += data_len; - } - - Ok(Self { - files: Arc::new(files), - reader, - atomic_data: Arc::new(Mutex::new(HashMap::new())), - }) - } -} - -/// Bridge sync Tantivy `FileHandle::read_bytes` to async `FileRead::read`. -/// -/// Uses `block_in_place` on multi-threaded tokio runtimes (no thread spawn -/// overhead). Falls back to a scoped thread for current-thread runtimes. -fn block_on_read(reader: &Arc, range: Range) -> io::Result { - let handle = tokio::runtime::Handle::current(); - let do_read = || { - handle - .block_on(reader.read(range.clone())) - .map_err(|e| io::Error::other(e.to_string())) - }; - - match handle.runtime_flavor() { - tokio::runtime::RuntimeFlavor::MultiThread => tokio::task::block_in_place(do_read), - _ => { - // Current-thread runtime: block_in_place is not available, - // fall back to a scoped thread. - let reader = Arc::clone(reader); - std::thread::scope(|s| { - s.spawn(move || { - handle - .block_on(reader.read(range)) - .map_err(|e| io::Error::other(e.to_string())) - }) - .join() - .map_err(|_| io::Error::other("reader thread panicked"))? - }) - } - } -} - -impl fmt::Debug for ArchiveDirectory { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("ArchiveDirectory") - .field("files", &self.files.keys().collect::>()) - .finish() - } -} - -/// A `FileHandle` for a single file within the archive. -#[derive(Clone)] -struct ArchiveFileHandle { - reader: Arc, - file_offset: u64, - file_length: usize, -} - -impl fmt::Debug for ArchiveFileHandle { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("ArchiveFileHandle") - .field("offset", &self.file_offset) - .field("length", &self.file_length) - .finish() - } -} - -impl HasLen for ArchiveFileHandle { - fn len(&self) -> usize { - self.file_length - } -} - -impl FileHandle for ArchiveFileHandle { - fn read_bytes(&self, range: Range) -> io::Result { - if range.end > self.file_length { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - format!( - "Read range {:?} exceeds file length {}", - range, self.file_length - ), - )); - } - - let abs_start = self.file_offset + range.start as u64; - let abs_end = self.file_offset + range.end as u64; - let data = block_on_read(&self.reader, abs_start..abs_end)?; - Ok(OwnedBytes::new(data.to_vec())) - } -} - -impl Directory for ArchiveDirectory { - fn get_file_handle(&self, path: &Path) -> Result, OpenReadError> { - let meta = self - .files - .get(path) - .ok_or_else(|| OpenReadError::FileDoesNotExist(path.to_path_buf()))?; - - Ok(Arc::new(ArchiveFileHandle { - reader: self.reader.clone(), - file_offset: meta.offset, - file_length: meta.length, - })) - } - - fn exists(&self, path: &Path) -> Result { - Ok(self.files.contains_key(path) || self.atomic_data.lock().unwrap().contains_key(path)) - } - - fn atomic_read(&self, path: &Path) -> Result, OpenReadError> { - if let Some(data) = self.atomic_data.lock().unwrap().get(path) { - return Ok(data.clone()); - } - let meta = self - .files - .get(path) - .ok_or_else(|| OpenReadError::FileDoesNotExist(path.to_path_buf()))?; - - let data = block_on_read(&self.reader, meta.offset..meta.offset + meta.length as u64) - .map_err(|e| OpenReadError::wrap_io_error(e, path.to_path_buf()))?; - Ok(data.to_vec()) - } - - fn atomic_write(&self, path: &Path, data: &[u8]) -> io::Result<()> { - self.atomic_data - .lock() - .unwrap() - .insert(path.to_path_buf(), data.to_vec()); - Ok(()) - } - - fn delete(&self, _path: &Path) -> Result<(), DeleteError> { - Ok(()) - } - - fn open_write(&self, _path: &Path) -> Result { - Ok(io::BufWriter::new(Box::new( - VecTerminatingWrite(Vec::new()), - ))) - } - - fn sync_directory(&self) -> io::Result<()> { - Ok(()) - } - - fn acquire_lock(&self, _lock: &Lock) -> Result { - Ok(DirectoryLock::from(Box::new(()))) - } - - fn watch(&self, _watch_callback: WatchCallback) -> tantivy::Result { - Ok(WatchHandle::empty()) - } -} - -/// Dummy writer for lock file support. -struct VecTerminatingWrite(Vec); - -impl io::Write for VecTerminatingWrite { - fn write(&mut self, buf: &[u8]) -> io::Result { - self.0.extend_from_slice(buf); - Ok(buf.len()) - } - - fn flush(&mut self) -> io::Result<()> { - Ok(()) - } -} - -impl TerminatingWrite for VecTerminatingWrite { - fn terminate_ref(&mut self, _token: AntiCallToken) -> io::Result<()> { - Ok(()) - } -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::io::FileIOBuilder; - use crate::tantivy::writer::TantivyFullTextWriter; - - async fn make_test_dir() -> ArchiveDirectory { - let file_io = FileIOBuilder::new("memory").build().unwrap(); - - let mut writer = TantivyFullTextWriter::new().unwrap(); - writer.add_document(0, Some("hello")).unwrap(); - writer.add_document(1, Some("world")).unwrap(); - let output = file_io.new_output("/test_archive.bin").unwrap(); - writer.finish(&output).await.unwrap(); - - let input = output.to_input_file(); - let metadata = input.metadata().await.unwrap(); - let reader = input.reader().await.unwrap(); - ArchiveDirectory::from_reader(reader, metadata.size) - .await - .unwrap() - } - - #[tokio::test] - async fn test_parse_archive() { - let dir = make_test_dir().await; - // Tantivy index files should be present. - assert!(!dir.files.is_empty()); - } - - #[tokio::test] - async fn test_read_file_from_archive() { - let dir = make_test_dir().await; - // Find a non-empty file (some tantivy index files can be 0 bytes). - let non_empty_path = dir - .files - .iter() - .find(|(_, meta)| meta.length > 0) - .map(|(p, _)| p.clone()) - .expect("archive should contain at least one non-empty file"); - let handle = dir.get_file_handle(&non_empty_path).unwrap(); - assert!(handle.len() > 0); - let data = handle.read_bytes(0..handle.len()).unwrap(); - assert_eq!(data.len(), handle.len()); - } - - #[tokio::test] - async fn test_atomic_read_write() { - let dir = make_test_dir().await; - - // atomic_write + atomic_read - dir.atomic_write(Path::new("meta.json"), b"{}").unwrap(); - let data = dir.atomic_read(Path::new("meta.json")).unwrap(); - assert_eq!(&data, b"{}"); - } - - #[tokio::test] - async fn test_file_not_found() { - let dir = make_test_dir().await; - assert!(dir.get_file_handle(Path::new("missing.txt")).is_err()); - } -} diff --git a/crates/paimon/src/tantivy/mod.rs b/crates/paimon/src/tantivy/mod.rs deleted file mode 100644 index 09229374..00000000 --- a/crates/paimon/src/tantivy/mod.rs +++ /dev/null @@ -1,23 +0,0 @@ -// 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. - -//! Tantivy-based full-text search for global index. - -pub(crate) mod directory; -pub mod full_text_search; -pub mod reader; -pub mod writer; diff --git a/crates/paimon/src/tantivy/reader.rs b/crates/paimon/src/tantivy/reader.rs deleted file mode 100644 index 17488506..00000000 --- a/crates/paimon/src/tantivy/reader.rs +++ /dev/null @@ -1,198 +0,0 @@ -// 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. - -//! Tantivy full-text index reader. -//! -//! Reads a Tantivy index from an archive (packed by `TantivyFullTextWriter`) -//! and performs full-text search queries. -//! -//! Reference: `org.apache.paimon.tantivy.index.TantivyFullTextGlobalIndexReader` - -use crate::io::{FileRead, InputFile}; -use crate::tantivy::directory::ArchiveDirectory; -use crate::tantivy::full_text_search::SearchResult; -use tantivy::collector::TopDocs; -use tantivy::query::QueryParser; -use tantivy::{Index, IndexReader, ReloadPolicy}; - -/// Reader for a Tantivy full-text index stored in archive format. -pub struct TantivyFullTextReader { - reader: IndexReader, - index: Index, -} - -impl TantivyFullTextReader { - /// Open a reader from an `InputFile` (on-demand reading, no full load). - pub async fn from_input_file(input: &InputFile) -> crate::Result { - let metadata = input.metadata().await?; - let reader = input.reader().await?; - Self::from_reader(reader, metadata.size).await - } - - /// Open a reader from an async `FileRead` and file size. - pub async fn from_reader(reader: impl FileRead, file_size: u64) -> crate::Result { - let directory = ArchiveDirectory::from_reader(reader, file_size) - .await - .map_err(|e| crate::Error::UnexpectedError { - message: format!("Failed to parse Tantivy archive: {}", e), - source: None, - })?; - - let index = Index::open(directory).map_err(|e| crate::Error::UnexpectedError { - message: format!("Failed to open Tantivy index from archive: {}", e), - source: None, - })?; - - let reader = index - .reader_builder() - .reload_policy(ReloadPolicy::Manual) - .try_into() - .map_err(|e| crate::Error::UnexpectedError { - message: format!("Failed to create Tantivy reader: {}", e), - source: None, - })?; - - Ok(Self { reader, index }) - } - - /// Search the index and return top-N results ranked by score. - pub fn search(&self, query_text: &str, limit: usize) -> crate::Result { - let schema = self.index.schema(); - let text_field = schema - .get_field("text") - .map_err(|_| crate::Error::UnexpectedError { - message: "Tantivy schema missing 'text' field".to_string(), - source: None, - })?; - - let searcher = self.reader.searcher(); - let query_parser = QueryParser::for_index(&self.index, vec![text_field]); - let query = - query_parser - .parse_query(query_text) - .map_err(|e| crate::Error::UnexpectedError { - message: format!("Failed to parse query '{}': {}", query_text, e), - source: None, - })?; - - let top_docs = searcher - .search(&query, &TopDocs::with_limit(limit)) - .map_err(|e| crate::Error::UnexpectedError { - message: format!("Tantivy search failed: {}", e), - source: None, - })?; - - let mut row_ids = Vec::with_capacity(top_docs.len()); - let mut scores = Vec::with_capacity(top_docs.len()); - - for (score, doc_address) in &top_docs { - let segment_reader = searcher.segment_reader(doc_address.segment_ord); - let fast_fields = segment_reader.fast_fields().u64("row_id").map_err(|e| { - crate::Error::UnexpectedError { - message: format!("Failed to get row_id fast field: {}", e), - source: None, - } - })?; - let row_id = fast_fields.first(doc_address.doc_id).ok_or_else(|| { - crate::Error::UnexpectedError { - message: format!( - "Missing row_id for doc_id {} in segment {}", - doc_address.doc_id, doc_address.segment_ord - ), - source: None, - } - })?; - row_ids.push(row_id); - scores.push(*score); - } - - Ok(SearchResult::new(row_ids, scores)) - } -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::io::FileIOBuilder; - use crate::tantivy::writer::TantivyFullTextWriter; - - async fn create_test_reader() -> TantivyFullTextReader { - let file_io = FileIOBuilder::new("memory").build().unwrap(); - let mut writer = TantivyFullTextWriter::new().unwrap(); - writer - .add_document(0, Some("the quick brown fox jumps over the lazy dog")) - .unwrap(); - writer - .add_document(1, Some("rust programming language")) - .unwrap(); - writer - .add_document(2, Some("apache paimon data lake")) - .unwrap(); - writer - .add_document(3, Some("full text search with tantivy")) - .unwrap(); - writer - .add_document(4, Some("the fox is quick and brown")) - .unwrap(); - let output = file_io.new_output("/test_reader_index.archive").unwrap(); - writer.finish(&output).await.unwrap(); - TantivyFullTextReader::from_input_file(&output.to_input_file()) - .await - .unwrap() - } - - #[tokio::test] - async fn test_search_basic() { - let reader = create_test_reader().await; - let result = reader.search("fox", 10).unwrap(); - assert_eq!(result.len(), 2); - assert!(result.row_ids.contains(&0)); - assert!(result.row_ids.contains(&4)); - } - - #[tokio::test] - async fn test_search_limit() { - let reader = create_test_reader().await; - let result = reader.search("fox", 1).unwrap(); - assert_eq!(result.len(), 1); - } - - #[tokio::test] - async fn test_search_no_match() { - let reader = create_test_reader().await; - let result = reader.search("nonexistent", 10).unwrap(); - assert!(result.is_empty()); - } - - #[tokio::test] - async fn test_search_scored() { - let reader = create_test_reader().await; - let result = reader.search("tantivy", 10).unwrap(); - assert_eq!(result.len(), 1); - assert!(result.row_ids.contains(&3)); - assert!(result.scores[0] > 0.0); - } - - #[tokio::test] - async fn test_search_with_offset() { - let reader = create_test_reader().await; - let result = reader.search("fox", 10).unwrap(); - let offset_result = result.offset(1000); - assert!(offset_result.row_ids.contains(&1000)); - assert!(offset_result.row_ids.contains(&1004)); - } -} diff --git a/crates/paimon/src/tantivy/writer.rs b/crates/paimon/src/tantivy/writer.rs deleted file mode 100644 index a12ed300..00000000 --- a/crates/paimon/src/tantivy/writer.rs +++ /dev/null @@ -1,242 +0,0 @@ -// 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. - -//! Tantivy full-text index writer. -//! -//! Writes documents (rowId + text) to a local Tantivy index, then packs the -//! index directory into a single archive file (Big-Endian, Java-compatible) -//! and writes it to an `OutputFile`. -//! -//! Reference: `org.apache.paimon.tantivy.index.TantivyFullTextGlobalIndexWriter` - -use bytes::Bytes; -use std::io::Read; - -use crate::io::OutputFile; -use tantivy::schema::{Field, NumericOptions, Schema, TEXT}; -use tantivy::{Index, IndexWriter, TantivyDocument}; - -/// Builds the fixed schema: `row_id` (u64 fast+stored+indexed) + `text` (full-text). -fn build_schema() -> (Schema, Field, Field) { - let mut builder = Schema::builder(); - let row_id_field = builder.add_u64_field( - "row_id", - NumericOptions::default() - .set_stored() - .set_indexed() - .set_fast(), - ); - let text_field = builder.add_text_field("text", TEXT); - (builder.build(), row_id_field, text_field) -} - -/// Writer for creating a Tantivy full-text index and packing it into an archive. -pub struct TantivyFullTextWriter { - writer: IndexWriter, - row_id_field: Field, - text_field: Field, - temp_dir: tempfile::TempDir, - row_count: u64, -} - -impl TantivyFullTextWriter { - /// Create a new writer. The index is built in a temporary directory. - pub fn new() -> crate::Result { - let temp_dir = tempfile::tempdir().map_err(|e| crate::Error::UnexpectedError { - message: format!("Failed to create temp directory for Tantivy index: {}", e), - source: None, - })?; - - let (schema, row_id_field, text_field) = build_schema(); - let index = Index::create_in_dir(temp_dir.path(), schema).map_err(|e| { - crate::Error::UnexpectedError { - message: format!("Failed to create Tantivy index: {}", e), - source: None, - } - })?; - let writer = index - .writer(50_000_000) - .map_err(|e| crate::Error::UnexpectedError { - message: format!("Failed to create Tantivy writer: {}", e), - source: None, - })?; - - Ok(Self { - writer, - row_id_field, - text_field, - temp_dir, - row_count: 0, - }) - } - - /// Add a document with the given row ID and text content. - /// If text is None, the row ID is still incremented (null value). - pub fn add_document(&mut self, row_id: u64, text: Option<&str>) -> crate::Result<()> { - if let Some(text) = text { - let mut doc = TantivyDocument::new(); - doc.add_u64(self.row_id_field, row_id); - doc.add_text(self.text_field, text); - self.writer - .add_document(doc) - .map_err(|e| crate::Error::UnexpectedError { - message: format!("Failed to add document: {}", e), - source: None, - })?; - } - self.row_count += 1; - Ok(()) - } - - /// Commit, pack the index into an archive, and write it to the given `OutputFile`. - /// - /// Returns `false` if no documents were written (nothing is written to the file). - /// - /// Reference: `TantivyFullTextGlobalIndexWriter.packIndex()` - pub async fn finish(mut self, output: &OutputFile) -> crate::Result { - if self.row_count == 0 { - return Ok(false); - } - - self.writer - .commit() - .map_err(|e| crate::Error::UnexpectedError { - message: format!("Failed to commit Tantivy index: {}", e), - source: None, - })?; - - // Drop the writer to release file locks before packing. - drop(self.writer); - - // Stream the archive directly to the OutputFile. - let mut file_writer = output.writer().await?; - - // Collect file entries from the temp directory. - let mut entries: Vec<(String, std::path::PathBuf)> = Vec::new(); - for entry in - std::fs::read_dir(self.temp_dir.path()).map_err(|e| crate::Error::UnexpectedError { - message: format!("Failed to read Tantivy index directory: {}", e), - source: None, - })? - { - let entry = entry.map_err(|e| crate::Error::UnexpectedError { - message: format!("Failed to read directory entry: {}", e), - source: None, - })?; - if entry.file_type().map(|t| t.is_file()).unwrap_or(false) { - let name = entry.file_name().to_string_lossy().to_string(); - entries.push((name, entry.path())); - } - } - - // Write file count (4 bytes BE). - file_writer - .write(Bytes::from((entries.len() as i32).to_be_bytes().to_vec())) - .await?; - - // Write each file: name_len + name + data_len + data (chunked). - const CHUNK_SIZE: usize = 4 * 1024 * 1024; // 4 MiB - for (name, path) in &entries { - let name_bytes = name.as_bytes(); - file_writer - .write(Bytes::from( - (name_bytes.len() as i32).to_be_bytes().to_vec(), - )) - .await?; - file_writer.write(Bytes::from(name_bytes.to_vec())).await?; - - let file_len = std::fs::metadata(path) - .map_err(|e| crate::Error::UnexpectedError { - message: format!("Failed to stat index file '{}': {}", name, e), - source: None, - })? - .len(); - file_writer - .write(Bytes::from((file_len as i64).to_be_bytes().to_vec())) - .await?; - - let mut file = - std::fs::File::open(path).map_err(|e| crate::Error::UnexpectedError { - message: format!("Failed to open index file '{}': {}", name, e), - source: None, - })?; - let mut buf = vec![0u8; CHUNK_SIZE]; - loop { - let n = file - .read(&mut buf) - .map_err(|e| crate::Error::UnexpectedError { - message: format!("Failed to read index file '{}': {}", name, e), - source: None, - })?; - if n == 0 { - break; - } - file_writer.write(Bytes::copy_from_slice(&buf[..n])).await?; - } - } - - file_writer.close().await?; - Ok(true) - } - - /// Number of rows processed (including nulls). - pub fn row_count(&self) -> u64 { - self.row_count - } -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::io::FileIOBuilder; - use crate::tantivy::reader::TantivyFullTextReader; - - #[tokio::test] - async fn test_write_and_read_roundtrip() { - let file_io = FileIOBuilder::new("memory").build().unwrap(); - - let mut writer = TantivyFullTextWriter::new().unwrap(); - writer.add_document(0, Some("hello world")).unwrap(); - writer.add_document(1, Some("foo bar baz")).unwrap(); - writer.add_document(2, None).unwrap(); - writer.add_document(3, Some("hello again")).unwrap(); - - let output = file_io.new_output("/test_index.archive").unwrap(); - let written = writer.finish(&output).await.unwrap(); - assert!(written); - - let input = output.to_input_file(); - let reader = TantivyFullTextReader::from_input_file(&input) - .await - .unwrap(); - let result = reader.search("hello", 10).unwrap(); - assert_eq!(result.len(), 2); - assert!(result.row_ids.contains(&0)); - assert!(result.row_ids.contains(&3)); - } - - #[tokio::test] - async fn test_empty_writer() { - let file_io = FileIOBuilder::new("memory").build().unwrap(); - let output = file_io.new_output("/empty_index.archive").unwrap(); - - let writer = TantivyFullTextWriter::new().unwrap(); - let written = writer.finish(&output).await.unwrap(); - assert!(!written); - assert!(!output.exists().await.unwrap()); - } -}