This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 4c70d287 feat(fulltext): read global indexes via paimon-ftindex-core
(#571)
4c70d287 is described below
commit 4c70d2871e0a738c1a75554211b0243173804aeb
Author: shyjsarah <[email protected]>
AuthorDate: Thu Jul 23 14:06:41 2026 +0800
feat(fulltext): read global indexes via paimon-ftindex-core (#571)
---
Cargo.lock | 340 ++-------
DEPENDENCIES.rust.tsv | 22 -
bindings/python/DEPENDENCIES.rust.tsv | 23 -
crates/integrations/datafusion/Cargo.toml | 2 +
.../integrations/datafusion/DEPENDENCIES.rust.tsv | 22 -
.../integrations/datafusion/tests/read_tables.rs | 158 +++-
crates/paimon/Cargo.toml | 7 +-
crates/paimon/DEPENDENCIES.rust.tsv | 24 -
.../{tantivy/full_text_search.rs => full_text.rs} | 87 ++-
crates/paimon/src/lib.rs | 4 +-
crates/paimon/src/table/data_file_reader.rs | 2 +-
crates/paimon/src/table/full_text_index_adapter.rs | 411 +++++++++++
.../paimon/src/table/full_text_search_builder.rs | 806 +++++++++++++++++++--
crates/paimon/src/table/mod.rs | 2 +
crates/paimon/src/tantivy/directory.rs | 399 ----------
crates/paimon/src/tantivy/mod.rs | 23 -
crates/paimon/src/tantivy/reader.rs | 198 -----
crates/paimon/src/tantivy/writer.rs | 242 -------
18 files changed, 1399 insertions(+), 1373 deletions(-)
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 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
[email protected] X X
X
@@ -214,7 +213,6 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected] X
[email protected]
X
@@ -278,12 +276,10 @@ [email protected] X
[email protected] X
[email protected] X
X
[email protected] X
X
[email protected] X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
@@ -322,11 +318,9 @@ [email protected] X
X
[email protected]
X
[email protected]
X
[email protected] X X
X
[email protected] X X
X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
@@ -338,7 +332,6 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
[email protected]
X
[email protected]
X
[email protected]
X X
[email protected] X
X
@@ -365,7 +358,6 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X X
X
[email protected] X X
X
[email protected] X
X
@@ -393,7 +385,6 @@ [email protected] X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected] X
[email protected] X
@@ -469,7 +460,6 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
@@ -498,10 +488,8 @@ [email protected] X
X
[email protected] X
X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X X
X
[email protected] X X
X
[email protected] X
X X
[email protected] X
X X
@@ -545,7 +533,6 @@ [email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
[email protected] X
[email protected]
X
[email protected] X
X
@@ -579,23 +566,15 @@ [email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected] X
@@ -703,7 +682,6 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] 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 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X X
X
[email protected] X
X
@@ -170,7 +169,6 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected] X
X
[email protected] X
X
@@ -197,7 +195,6 @@ [email protected] X
X
[email protected] X
X
[email protected] X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
@@ -230,11 +227,9 @@ [email protected] X
[email protected] X
[email protected] X
X
[email protected] X
X
[email protected] X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected]
X
@@ -267,11 +262,9 @@ [email protected] X
X
[email protected]
X
[email protected]
X
[email protected] X X
X
[email protected] X X
X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
@@ -280,7 +273,6 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
[email protected]
X
[email protected]
X
[email protected]
X X
[email protected] X
X
@@ -303,7 +295,6 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
@@ -327,7 +318,6 @@ [email protected] X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected] X
[email protected] X
@@ -391,7 +381,6 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
@@ -419,10 +408,8 @@ [email protected] X
X
[email protected] X
X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X X
X
[email protected] X X
X
[email protected] X
X X
[email protected] X
X X
@@ -463,7 +450,6 @@ [email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
[email protected] X
[email protected]
X
[email protected] X
X
@@ -493,23 +479,15 @@ [email protected] X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected] X
[email protected] X
X
@@ -584,7 +562,6 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] 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 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
[email protected] X X
X
@@ -183,7 +182,6 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected] X
[email protected]
X
@@ -242,11 +240,9 @@ [email protected] X
[email protected] X
[email protected] X
[email protected] X
X
[email protected] X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
@@ -284,11 +280,9 @@ [email protected] X
X
[email protected] X
X
[email protected]
X
[email protected] X X
X
[email protected] X X
X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
@@ -299,7 +293,6 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
[email protected]
X
[email protected]
X
[email protected]
X X
[email protected] X
X
@@ -325,7 +318,6 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X X
X
[email protected] X X
X
[email protected] X
X
@@ -345,7 +337,6 @@ [email protected] X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected] X
[email protected] X
@@ -403,7 +394,6 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
@@ -424,10 +414,8 @@ [email protected] X
X
[email protected] X
X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X X
X
[email protected] X X
X
[email protected] X
X X
[email protected] X
X X
@@ -466,7 +454,6 @@ [email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
[email protected] X
[email protected]
X
[email protected] X
X
@@ -494,23 +481,15 @@ [email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected] X
X
@@ -611,7 +590,6 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] 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<FileSystemCatalog>,
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<FileSystemCatalog>,
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<i32> = 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<i32> = 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 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
[email protected] X X
X
@@ -148,13 +147,11 @@ [email protected] X
X
[email protected] X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected] X
[email protected]
X
@@ -179,7 +176,6 @@ [email protected] X
X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
@@ -217,11 +213,9 @@ [email protected] X
[email protected] X
[email protected] X
X
[email protected] X
X
[email protected] X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
@@ -257,11 +251,9 @@ [email protected]
X
[email protected]
X
[email protected]
X
[email protected] X X
X
[email protected] X X
X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
@@ -272,7 +264,6 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
[email protected]
X
[email protected]
X
[email protected]
X X
[email protected] X
X
@@ -299,7 +290,6 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X X
X
[email protected] X X
X
[email protected] X
X
@@ -324,7 +314,6 @@ [email protected] X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected] X
[email protected] X
@@ -384,7 +373,6 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
@@ -411,10 +399,8 @@ [email protected] X
X
[email protected] X
X
[email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X X
X
[email protected] X X
X
[email protected] X
X X
[email protected] X
X X
@@ -456,7 +442,6 @@ [email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
[email protected] X
[email protected]
X
[email protected] X
X
@@ -484,23 +469,15 @@ [email protected]
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected]
X
[email protected] X
X
@@ -602,7 +579,6 @@ [email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] X
X
[email protected] 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<roaring::RoaringTreemap>,
}
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::<std::collections::HashSet<_>>();
+ 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<Self> {
+ 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<crate::table::RowRange> {
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<dyn FileRead>,
+ file_name: String,
+ query: String,
+ limit: usize,
+ include_row_ids: Option<RoaringTreemap>,
+) -> crate::Result<FullTextSearchResult> {
+ 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<RoaringTreemap>,
+) -> crate::Result<FullTextSearchResult> {
+ 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<R: SeekRead + 'static>(
+ input: R,
+ file_name: &str,
+ query: &str,
+ limit: usize,
+ include_row_ids: Option<RoaringTreemap>,
+) -> crate::Result<FullTextSearchResult> {
+ 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<dyn FileRead>,
+ handle: tokio::runtime::Handle,
+ permits: Arc<tokio::sync::Semaphore>,
+}
+
+impl AsyncFileSeekRead {
+ fn new(reader: Box<dyn FileRead>, 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::<io::Result<Vec<_>>>()?;
+
+ 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::<Vec<_>>()
+ .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<File>,
+}
+
+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<Mutex<Vec<Range<u64>>>>,
+ }
+
+ #[async_trait::async_trait]
+ impl FileRead for BytesFileRead {
+ async fn read(&self, range: Range<u64>) -> crate::Result<Bytes> {
+ 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<u8> {
+ 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<u64>) -> crate::Result<Bytes> {
+ 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<AtomicUsize>,
+ maximum: Arc<AtomicUsize>,
+ }
+
+ #[async_trait::async_trait]
+ impl FileRead for ConcurrentFileRead {
+ async fn read(&self, range: Range<u64>) -> crate::Result<Bytes> {
+ 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::<Vec<_>>();
+ 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<String>,
query_text: Option<String>,
limit: Option<usize>,
+ include_row_ids: Option<RoaringTreemap>,
}
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<Vec<RowRange>> {
@@ -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::<Vec<_>>()
+ .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<RoaringTreemap>,
+}
+
+fn plan_full_text_index_searches(
+ entries: Vec<&IndexManifestEntry>,
+ include_row_ids: Option<&RoaringTreemap>,
+) -> crate::Result<Vec<FullTextIndexSearchPlan>> {
+ 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::<Vec<_>>();
+ let mut active = Vec::<usize>::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<Vec<RowRange>> {
@@ -305,15 +403,22 @@ async fn read_raw_full_text_search(
table: &Table,
search: &FullTextSearch,
raw_ranges: &[RowRange],
+ table_options: &HashMap<String, String>,
) -> crate::Result<SearchResult> {
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<Vec<(i64, String)>> {
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::<StringArray>() {
+ 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::<LargeStringArray>() {
+ 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<Option<(i64, String)>> {
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<String, String>,
+) -> crate::Result<FullTextIndexConfig> {
+ let native_options: HashMap<String, String> = 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<RoaringTreemap> {
+ 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<String> {
+ 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::<serde_json::Value>(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<SearchResult> {
+ 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::<serde_json::Value>(&normalized).unwrap(),
+ serde_json::json!({"match": {"query": "123"}})
+ );
+
+ let normalized = normalize_query_text("\"hello\"", "body").unwrap();
+ assert_eq!(
+ serde_json::from_str::<serde_json::Value>(&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<_>>(),
+ vec!["first", "second", "overlap"]
+ );
+ let decoded = plans
+ .iter()
+ .map(|plan| {
+ plan.local_filter
+ .as_ref()
+ .unwrap()
+ .iter()
+ .collect::<Vec<_>>()
+ })
+ .collect::<Vec<_>>();
+ 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<HashMap<PathBuf, FileMeta>>,
- reader: Arc<dyn FileRead>,
- /// In-memory storage for atomic_write (used by Tantivy for meta.json).
- atomic_data: Arc<Mutex<HashMap<PathBuf, Vec<u8>>>>,
-}
-
-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<Self> {
- let reader: Arc<dyn FileRead> = 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<dyn FileRead>, range: Range<u64>) ->
io::Result<Bytes> {
- 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::<Vec<_>>())
- .finish()
- }
-}
-
-/// A `FileHandle` for a single file within the archive.
-#[derive(Clone)]
-struct ArchiveFileHandle {
- reader: Arc<dyn FileRead>,
- 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<usize>) -> io::Result<OwnedBytes> {
- 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<Arc<dyn FileHandle>,
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<bool, OpenReadError> {
- Ok(self.files.contains_key(path) ||
self.atomic_data.lock().unwrap().contains_key(path))
- }
-
- fn atomic_read(&self, path: &Path) -> Result<Vec<u8>, 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<WritePtr, OpenWriteError> {
- Ok(io::BufWriter::new(Box::new(
- VecTerminatingWrite(Vec::new()),
- )))
- }
-
- fn sync_directory(&self) -> io::Result<()> {
- Ok(())
- }
-
- fn acquire_lock(&self, _lock: &Lock) -> Result<DirectoryLock, LockError> {
- Ok(DirectoryLock::from(Box::new(())))
- }
-
- fn watch(&self, _watch_callback: WatchCallback) ->
tantivy::Result<WatchHandle> {
- Ok(WatchHandle::empty())
- }
-}
-
-/// Dummy writer for lock file support.
-struct VecTerminatingWrite(Vec<u8>);
-
-impl io::Write for VecTerminatingWrite {
- fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
- 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<Self> {
- 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<Self> {
- 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<SearchResult> {
- 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<Self> {
- 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<bool> {
- 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());
- }
-}
