Skip to content

Commit 38e92cf

Browse files
shyjsarahshyjsarah
committed
feat(fulltext): read global indexes via paimon-ftindex-core
1 parent 9e83dea commit 38e92cf

13 files changed

Lines changed: 1363 additions & 986 deletions

File tree

‎crates/integrations/datafusion/Cargo.toml‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,9 @@ uuid = { version = "1", features = ["v4"] }
4949
[dev-dependencies]
5050
arrow-array = { workspace = true }
5151
arrow-schema = { workspace = true }
52+
bytes = "1.7.1"
5253
flate2 = "1"
54+
paimon-ftindex-core = { git = "https://github.com/apache/paimon-full-text.git", tag = "v0.1.0-rc5" }
5355
parquet = { workspace = true }
5456
tar = "0.4"
5557
tempfile = "3"

‎crates/integrations/datafusion/tests/read_tables.rs‎

Lines changed: 126 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -1449,34 +1449,21 @@ async fn test_case_insensitive_column_not_supported_via_sql() {
14491449
mod fulltext_tests {
14501450
use std::sync::Arc;
14511451

1452+
use bytes::Bytes;
14521453
use datafusion::arrow::array::Int32Array;
14531454
use paimon::catalog::Identifier;
1455+
use paimon::spec::{GlobalIndexMeta, IndexFileMeta};
14541456
use paimon::table::BranchManager;
1455-
use paimon::{Catalog, CatalogOptions, FileSystemCatalog, Options};
1457+
use paimon::{Catalog, CatalogOptions, CommitMessage, FileSystemCatalog, Options, TableCommit};
14561458
use paimon_datafusion::{register_full_text_search, SQLContext};
14571459

14581460
use super::common::string_value;
1461+
use paimon_ftindex_core::io::PosWriter;
1462+
use paimon_ftindex_core::{FullTextIndexConfig, FullTextIndexWriter};
14591463

1460-
/// Extract the bundled tar.gz into a temp dir and return (tempdir, warehouse_path).
1461-
fn extract_test_warehouse() -> (tempfile::TempDir, String) {
1462-
let archive_path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
1463-
.join("testdata/test_tantivy_fulltext.tar.gz");
1464-
let file = std::fs::File::open(&archive_path)
1465-
.unwrap_or_else(|e| panic!("Failed to open {}: {e}", archive_path.display()));
1466-
let decoder = flate2::read::GzDecoder::new(file);
1467-
let mut archive = tar::Archive::new(decoder);
1468-
1464+
async fn create_fulltext_context() -> (SQLContext, Arc<FileSystemCatalog>, tempfile::TempDir) {
14691465
let tmp = tempfile::tempdir().expect("Failed to create temp dir");
1470-
let db_dir = tmp.path().join("default.db");
1471-
std::fs::create_dir_all(&db_dir).unwrap();
1472-
archive.unpack(&db_dir).unwrap();
1473-
14741466
let warehouse = format!("file://{}", tmp.path().display());
1475-
(tmp, warehouse)
1476-
}
1477-
1478-
async fn create_fulltext_context() -> (SQLContext, Arc<FileSystemCatalog>, tempfile::TempDir) {
1479-
let (tmp, warehouse) = extract_test_warehouse();
14801467
let mut options = Options::new();
14811468
options.set(CatalogOptions::WAREHOUSE, warehouse);
14821469
let catalog = Arc::new(FileSystemCatalog::new(options).expect("Failed to create catalog"));
@@ -1486,9 +1473,113 @@ mod fulltext_tests {
14861473
.await
14871474
.expect("Failed to register catalog");
14881475
register_full_text_search(ctx.ctx(), catalog.clone(), "default");
1476+
run_fulltext_sql(
1477+
&ctx,
1478+
"CREATE TABLE paimon.default.test_fulltext (
1479+
id INT,
1480+
content STRING
1481+
) WITH (
1482+
'row-tracking.enabled' = 'true',
1483+
'global-index.search-mode' = 'full'
1484+
)",
1485+
)
1486+
.await;
1487+
for insert in [
1488+
"INSERT INTO paimon.default.test_fulltext VALUES (0, 'alphaft lakehouse storage')",
1489+
"INSERT INTO paimon.default.test_fulltext VALUES (1, 'betafft gammaft full-text search')",
1490+
"INSERT INTO paimon.default.test_fulltext VALUES (2, 'alphaft supports indexed queries')",
1491+
"INSERT INTO paimon.default.test_fulltext VALUES (3, 'gammaft over raw fallback rows')",
1492+
"INSERT INTO paimon.default.test_fulltext VALUES (4, 'alphaft tables can be queried')",
1493+
] {
1494+
run_fulltext_sql(&ctx, insert).await;
1495+
}
1496+
build_fulltext_index(catalog.as_ref()).await;
14891497
(ctx, catalog, tmp)
14901498
}
14911499

1500+
async fn build_fulltext_index(catalog: &dyn Catalog) {
1501+
let table = catalog
1502+
.get_table(&Identifier::new("default", "test_fulltext"))
1503+
.await
1504+
.expect("table should exist");
1505+
let content_field_id = table
1506+
.schema()
1507+
.fields()
1508+
.iter()
1509+
.find(|field| field.name() == "content")
1510+
.expect("content field")
1511+
.id();
1512+
1513+
let mut writer =
1514+
FullTextIndexWriter::new(FullTextIndexConfig::new().with_text_fields(["content"]))
1515+
.expect("full-text writer");
1516+
for (row_id, content) in [
1517+
(0, "alphaft lakehouse storage"),
1518+
(1, "betafft gammaft full-text search"),
1519+
(2, "alphaft supports indexed queries"),
1520+
(3, "gammaft over raw fallback rows"),
1521+
(4, "alphaft tables can be queried"),
1522+
] {
1523+
writer
1524+
.add_document_fields(row_id, [("content", content)])
1525+
.expect("add full-text document");
1526+
}
1527+
let mut index_bytes = Vec::new();
1528+
writer
1529+
.write(&mut PosWriter::new(&mut index_bytes))
1530+
.expect("write full-text index");
1531+
1532+
let index_file_name = "ft-datafusion-fulltext.index";
1533+
table
1534+
.file_io()
1535+
.mkdirs(&format!(
1536+
"{}/index/",
1537+
table.location().trim_end_matches('/')
1538+
))
1539+
.await
1540+
.expect("create index dir");
1541+
table
1542+
.file_io()
1543+
.new_output(&format!(
1544+
"{}/index/{index_file_name}",
1545+
table.location().trim_end_matches('/')
1546+
))
1547+
.expect("index output")
1548+
.write(Bytes::from(index_bytes.clone()))
1549+
.await
1550+
.expect("write index file");
1551+
1552+
let mut message = CommitMessage::new(Vec::new(), 0, Vec::new());
1553+
message.new_index_files = vec![IndexFileMeta {
1554+
index_type: "full-text".to_string(),
1555+
file_name: index_file_name.to_string(),
1556+
file_size: index_bytes.len() as i32,
1557+
row_count: 5,
1558+
deletion_vectors_ranges: None,
1559+
global_index_meta: Some(GlobalIndexMeta {
1560+
row_range_start: 0,
1561+
row_range_end: 4,
1562+
index_field_id: content_field_id,
1563+
extra_field_ids: None,
1564+
index_meta: None,
1565+
source_meta: None,
1566+
}),
1567+
}];
1568+
TableCommit::new(table, "test-user".to_string())
1569+
.commit(vec![message])
1570+
.await
1571+
.expect("commit full-text index manifest");
1572+
}
1573+
1574+
async fn run_fulltext_sql(ctx: &SQLContext, sql: &str) {
1575+
ctx.sql(sql)
1576+
.await
1577+
.unwrap_or_else(|e| panic!("Failed to plan `{sql}`: {e}"))
1578+
.collect()
1579+
.await
1580+
.unwrap_or_else(|e| panic!("Failed to execute `{sql}`: {e}"));
1581+
}
1582+
14921583
fn extract_id_content_rows(
14931584
batches: &[datafusion::arrow::record_batch::RecordBatch],
14941585
) -> Vec<(i32, String)> {
@@ -1512,12 +1603,12 @@ mod fulltext_tests {
15121603
rows
15131604
}
15141605

1515-
/// Search for 'paimon' — rows 0, 2, 4 mention "paimon".
1606+
/// Search for 'alphaft' — rows 0, 2, 4 contain the token.
15161607
#[tokio::test]
15171608
async fn test_full_text_search_paimon() {
15181609
let (ctx, _catalog, _tmp) = create_fulltext_context().await;
15191610
let batches = ctx
1520-
.sql("SELECT id, content FROM full_text_search('paimon.default.test_tantivy_fulltext', 'content', 'paimon', 10)")
1611+
.sql("SELECT id, content FROM full_text_search('paimon.default.test_fulltext', 'content', 'alphaft', 10)")
15211612
.await
15221613
.expect("SQL should parse")
15231614
.collect()
@@ -1529,14 +1620,14 @@ mod fulltext_tests {
15291620
assert_eq!(
15301621
ids,
15311622
vec![0, 2, 4],
1532-
"Searching 'paimon' should match rows 0, 2, 4"
1623+
"Searching 'alphaft' should match rows 0, 2, 4"
15331624
);
15341625
}
15351626

15361627
#[tokio::test]
15371628
async fn test_full_text_search_branch() {
15381629
let (ctx, catalog, _tmp) = create_fulltext_context().await;
1539-
let identifier = Identifier::new("default", "test_tantivy_fulltext");
1630+
let identifier = Identifier::new("default", "test_fulltext");
15401631
let table = catalog.get_table(&identifier).await.expect("load table");
15411632
let snapshot_manager = table.snapshot_manager();
15421633
let snapshot = snapshot_manager
@@ -1560,7 +1651,7 @@ mod fulltext_tests {
15601651
.expect("delete main snapshots");
15611652

15621653
let batches = ctx
1563-
.sql("SELECT id, content FROM full_text_search('paimon.default.test_tantivy_fulltext$branch_b1', 'content', 'paimon', 10)")
1654+
.sql("SELECT id, content FROM full_text_search('paimon.default.test_fulltext$branch_b1', 'content', 'alphaft', 10)")
15641655
.await
15651656
.expect("SQL should parse")
15661657
.collect()
@@ -1572,12 +1663,12 @@ mod fulltext_tests {
15721663
assert_eq!(ids, vec![0, 2, 4]);
15731664
}
15741665

1575-
/// Search for 'tantivy' — only row 1.
1666+
/// Search for 'betafft' — only row 1.
15761667
#[tokio::test]
1577-
async fn test_full_text_search_tantivy() {
1668+
async fn test_full_text_search_betafft() {
15781669
let (ctx, _catalog, _tmp) = create_fulltext_context().await;
15791670
let batches = ctx
1580-
.sql("SELECT id, content FROM full_text_search('paimon.default.test_tantivy_fulltext', 'content', 'tantivy', 10)")
1671+
.sql("SELECT id, content FROM full_text_search('paimon.default.test_fulltext', 'content', 'betafft', 10)")
15811672
.await
15821673
.expect("SQL should parse")
15831674
.collect()
@@ -1586,15 +1677,15 @@ mod fulltext_tests {
15861677

15871678
let rows = extract_id_content_rows(&batches);
15881679
let ids: Vec<i32> = rows.iter().map(|(id, _)| *id).collect();
1589-
assert_eq!(ids, vec![1], "Searching 'tantivy' should match row 1");
1680+
assert_eq!(ids, vec![1], "Searching 'betafft' should match row 1");
15901681
}
15911682

1592-
/// Search for 'search' — rows 1, 3 mention "full-text search".
1683+
/// Search for 'gammaft' — rows 1, 3 contain the token.
15931684
#[tokio::test]
15941685
async fn test_full_text_search_search() {
15951686
let (ctx, _catalog, _tmp) = create_fulltext_context().await;
15961687
let batches = ctx
1597-
.sql("SELECT id, content FROM full_text_search('paimon.default.test_tantivy_fulltext', 'content', 'search', 10)")
1688+
.sql("SELECT id, content FROM full_text_search('paimon.default.test_fulltext', 'content', 'gammaft', 10)")
15981689
.await
15991690
.expect("SQL should parse")
16001691
.collect()
@@ -1603,8 +1694,11 @@ mod fulltext_tests {
16031694

16041695
let rows = extract_id_content_rows(&batches);
16051696
let ids: Vec<i32> = rows.iter().map(|(id, _)| *id).collect();
1606-
assert!(ids.contains(&1), "Searching 'search' should match row 1");
1607-
assert!(ids.contains(&3), "Searching 'search' should match row 3");
1697+
assert_eq!(
1698+
ids,
1699+
vec![1, 3],
1700+
"Searching 'gammaft' should match rows 1, 3"
1701+
);
16081702
}
16091703
}
16101704

‎crates/paimon/Cargo.toml‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -41,7 +41,7 @@ storage-all = [
4141
"storage-gcs",
4242
"storage-hdfs",
4343
]
44-
fulltext = ["tantivy", "tempfile"]
44+
fulltext = ["dep:paimon-ftindex-core", "dep:tempfile"]
4545
vortex = ["dep:vortex"]
4646

4747
storage-memory = ["opendal/services-memory"]
@@ -102,9 +102,9 @@ md-5 = "0.10"
102102
regex = "1"
103103
uuid = { version = "1", features = ["v4"] }
104104
urlencoding = "2.1"
105-
tantivy = { version = "0.22", optional = true }
106-
tempfile = { version = "3", optional = true }
107105
paimon-mosaic-core = "0.2.0"
106+
paimon-ftindex-core = { git = "https://github.com/apache/paimon-full-text.git", tag = "v0.1.0-rc5", optional = true }
107+
tempfile = { version = "3", optional = true }
108108
paimon-vindex-core = "0.2.0"
109109
vortex = { version = "0.75.0", features = ["tokio"], optional = true }
110110
libloading = "0.9"

0 commit comments

Comments
 (0)