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 878c33d6 Unify global index construction behind a common table builder
(#1091)
878c33d6 is described below
commit 878c33d6f56f9c3a040152774595f35548cb9bbf
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Oct 10 16:12:25 2026 +0800
Unify global index construction behind a common table builder (#1091)
---
bindings/c/src/tests.rs | 7 +-
.../python/python/pypaimon_rust/datafusion.pyi | 15 +-
bindings/python/src/context.rs | 2 +-
bindings/python/src/index_build.rs | 50 ++-
bindings/python/src/table.rs | 6 +-
crates/integrations/datafusion/src/procedures.rs | 82 +----
crates/integrations/datafusion/tests/procedures.rs | 4 +-
crates/paimon/examples/ivfpq_build_benchmark.rs | 3 +-
crates/paimon/src/table/data_evolution_reader.rs | 2 +-
crates/paimon/src/table/de_vector_read/tests.rs | 3 +-
crates/paimon/src/table/de_vector_scan/tests.rs | 3 +-
.../src/table/full_text_index_build_builder.rs | 58 ++--
.../table/full_text_index_build_builder/tests.rs | 24 +-
.../paimon/src/table/global_index_build_builder.rs | 348 +++++++++++++++++++++
.../paimon/src/table/global_index_build_common.rs | 23 ++
.../paimon/src/table/global_index_drop_builder.rs | 27 +-
crates/paimon/src/table/global_index_types.rs | 72 ++---
.../paimon/src/table/lumina_index_build_builder.rs | 63 ++--
.../src/table/lumina_index_build_builder/tests.rs | 15 +-
.../src/table/lumina_index_build_builder/writer.rs | 23 +-
crates/paimon/src/table/mod.rs | 38 +--
.../src/table/sorted_global_index_build_builder.rs | 39 +--
.../preparation_tests.rs | 20 +-
.../sorted_global_index_build_builder/tests.rs | 88 +++---
.../paimon/src/table/vector_search_test_utils.rs | 5 +-
.../paimon/src/table/vindex_index_build_builder.rs | 67 ++--
.../src/table/vindex_index_build_builder/tests.rs | 39 ++-
.../src/table/vindex_index_build_builder/timing.rs | 6 +-
.../src/table/vindex_index_build_builder/writer.rs | 24 +-
crates/paimon/src/table/write_builder.rs | 2 +-
crates/paimon/tests/scan_planning_parity_test.rs | 2 +-
31 files changed, 699 insertions(+), 461 deletions(-)
diff --git a/bindings/c/src/tests.rs b/bindings/c/src/tests.rs
index b8dbd785..6eb98979 100644
--- a/bindings/c/src/tests.rs
+++ b/bindings/c/src/tests.rs
@@ -3123,7 +3123,7 @@ fn test_two_commits_same_builder() {
// * A primary-key vector table backed by a real vindex IVF-flat ANN segment
// built in-process (bucket-local ANN search, residual filter supported).
// * A data-evolution (append) vector table whose global index is produced by
-// the public `new_vindex_index_build_builder(...).execute()` path.
+// the public
`new_global_index_build_builder().with_index_type(...).execute()` path.
//
// Both fixtures live entirely on the in-memory FileIO, so no temp dirs or
// on-disk schema files are needed: the written data file keeps `schema_id ==
0`,
@@ -3482,7 +3482,7 @@ fn append_vector_batch(ids: Vec<i32>, vectors: Vec<[f32;
2]>) -> RecordBatch {
}
/// Build a data-evolution vector table: write vectors via the public write
path,
-/// then build the global vindex index via `new_vindex_index_build_builder`.
+/// then build the global vindex index via `new_global_index_build_builder`.
fn build_append_vector_table(path: &str) -> Table {
let file_io = memory_file_io();
setup_table_dirs(&file_io, path);
@@ -3524,7 +3524,8 @@ fn build_append_vector_table(path: &str) -> Table {
write_builder.new_commit().commit(messages).await.unwrap();
let built = table
- .new_vindex_index_build_builder(INDEX_TYPE)
+ .new_global_index_build_builder()
+ .with_index_type(INDEX_TYPE)
.with_index_column("embedding")
.execute()
.await
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index 0974cea2..e38d5e5e 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -223,7 +223,7 @@ class Table:
def new_stream_scan(self) -> StreamTableScan: ...
def new_vector_search_builder(self) -> "VectorSearchBuilder": ...
def new_batch_vector_search_builder(self) -> "BatchVectorSearchBuilder":
...
- def new_sorted_global_index_build_builder(self) ->
"SortedGlobalIndexBuildBuilder": ...
+ def new_global_index_build_builder(self) -> "GlobalIndexBuildBuilder": ...
def new_batch_write_builder(self) -> "BatchWriteBuilder": ...
def new_postpone_fixed_bucket_write_builder(self) ->
"PostponeFixedBucketWriteBuilder": ...
def new_stream_write_builder(self) -> "StreamWriteBuilder": ...
@@ -260,11 +260,14 @@ class Table:
"""
...
-class SortedGlobalIndexBuildBuilder:
- def with_index_column(self, column: str) ->
"SortedGlobalIndexBuildBuilder": ...
- def with_index_type(self, index_type: str) ->
"SortedGlobalIndexBuildBuilder": ...
- def with_options(self, options: Dict[str, str]) ->
"SortedGlobalIndexBuildBuilder": ...
- def with_partition_filter(self, predicate: Dict[str, Any]) ->
"SortedGlobalIndexBuildBuilder": ...
+class GlobalIndexBuildBuilder:
+ @staticmethod
+ def supports_index_type(index_type: str) -> bool: ...
+ def with_index_columns(self, columns: List[str]) ->
"GlobalIndexBuildBuilder": ...
+ def with_index_column(self, column: str) -> "GlobalIndexBuildBuilder": ...
+ def with_index_type(self, index_type: str) -> "GlobalIndexBuildBuilder":
...
+ def with_options(self, options: Dict[str, str]) ->
"GlobalIndexBuildBuilder": ...
+ def with_partition_filter(self, predicate: Dict[str, Any]) ->
"GlobalIndexBuildBuilder": ...
def build(self) -> List["CommitMessage"]:
"""Build unpublished index files; the caller decides whether to commit
or abort."""
...
diff --git a/bindings/python/src/context.rs b/bindings/python/src/context.rs
index 37a1f03e..f0baad1b 100644
--- a/bindings/python/src/context.rs
+++ b/bindings/python/src/context.rs
@@ -726,7 +726,7 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_,
PyModule>) -> PyResult<()>
let this = PyModule::new(py, "datafusion")?;
this.add_class::<PaimonCatalog>()?;
this.add_class::<crate::table::PyTable>()?;
- this.add_class::<crate::index_build::PySortedGlobalIndexBuildBuilder>()?;
+ this.add_class::<crate::index_build::PyGlobalIndexBuildBuilder>()?;
this.add_class::<crate::vector_search::PyVectorSearchBuilder>()?;
this.add_class::<crate::vector_search::PyBatchVectorSearchBuilder>()?;
this.add_class::<crate::vector_search::PyVectorScan>()?;
diff --git a/bindings/python/src/index_build.rs
b/bindings/python/src/index_build.rs
index 2b216153..d81e7252 100644
--- a/bindings/python/src/index_build.rs
+++ b/bindings/python/src/index_build.rs
@@ -15,13 +15,13 @@
// specific language governing permissions and limitations
// under the License.
-//! Parameter and commit-message wrappers for the core sorted-index builder.
+//! Parameter and commit-message wrappers for the core global-index builder.
use std::collections::HashMap;
use std::sync::Arc;
use paimon::spec::Predicate;
-use paimon::table::{SortedGlobalIndexBuildBuilder, Table};
+use paimon::table::{GlobalIndexBuildBuilder, Table};
use paimon_datafusion::runtime::runtime;
use pyo3::prelude::*;
use pyo3::types::PyDict;
@@ -30,33 +30,41 @@ use crate::error::to_py_err;
use crate::predicate::dict_to_table_predicate;
use crate::write::PyCommitMessage;
-#[pyclass(
- name = "SortedGlobalIndexBuildBuilder",
- module = "pypaimon_rust.datafusion"
-)]
-pub struct PySortedGlobalIndexBuildBuilder {
+enum PyIndexColumns {
+ Column(String),
+ Columns(Vec<String>),
+}
+
+#[pyclass(name = "GlobalIndexBuildBuilder", module =
"pypaimon_rust.datafusion")]
+pub struct PyGlobalIndexBuildBuilder {
table: Arc<Table>,
- column: Option<String>,
+ columns: Option<PyIndexColumns>,
index_type: String,
options: HashMap<String, String>,
partition_filters: Vec<Predicate>,
}
-impl PySortedGlobalIndexBuildBuilder {
+impl PyGlobalIndexBuildBuilder {
pub(crate) fn new(table: Arc<Table>) -> Self {
Self {
table,
- column: None,
+ columns: None,
index_type: "btree".into(),
options: HashMap::new(),
partition_filters: vec![],
}
}
- fn builder(&self) -> paimon::Result<SortedGlobalIndexBuildBuilder<'_>> {
- let mut builder = self.table.new_sorted_global_index_build_builder();
- if let Some(column) = &self.column {
- builder.with_index_column(column);
+ fn builder(&self) -> paimon::Result<GlobalIndexBuildBuilder<'_>> {
+ let mut builder = self.table.new_global_index_build_builder();
+ match &self.columns {
+ Some(PyIndexColumns::Column(column)) => {
+ builder.with_index_column(column);
+ }
+ Some(PyIndexColumns::Columns(columns)) => {
+
builder.with_index_columns(&columns.iter().map(String::as_str).collect::<Vec<_>>());
+ }
+ None => {}
}
builder
.with_index_type(&self.index_type)
@@ -69,9 +77,19 @@ impl PySortedGlobalIndexBuildBuilder {
}
#[pymethods]
-impl PySortedGlobalIndexBuildBuilder {
+impl PyGlobalIndexBuildBuilder {
+ #[staticmethod]
+ fn supports_index_type(index_type: &str) -> bool {
+ GlobalIndexBuildBuilder::supports_index_type(index_type)
+ }
+
+ fn with_index_columns(mut slf: PyRefMut<'_, Self>, columns: Vec<String>)
-> PyRefMut<'_, Self> {
+ slf.columns = Some(PyIndexColumns::Columns(columns));
+ slf
+ }
+
fn with_index_column(mut slf: PyRefMut<'_, Self>, column: String) ->
PyRefMut<'_, Self> {
- slf.column = Some(column);
+ slf.columns = Some(PyIndexColumns::Column(column));
slf
}
diff --git a/bindings/python/src/table.rs b/bindings/python/src/table.rs
index a2771e41..363c4a07 100644
--- a/bindings/python/src/table.rs
+++ b/bindings/python/src/table.rs
@@ -206,10 +206,8 @@ impl PyTable {
crate::vector_search::PyBatchVectorSearchBuilder::new(Arc::clone(&self.inner))
}
- fn new_sorted_global_index_build_builder(
- &self,
- ) -> crate::index_build::PySortedGlobalIndexBuildBuilder {
-
crate::index_build::PySortedGlobalIndexBuildBuilder::new(Arc::clone(&self.inner))
+ fn new_global_index_build_builder(&self) ->
crate::index_build::PyGlobalIndexBuildBuilder {
+
crate::index_build::PyGlobalIndexBuildBuilder::new(Arc::clone(&self.inner))
}
fn new_batch_write_builder(&self) -> PyBatchWriteBuilder {
diff --git a/crates/integrations/datafusion/src/procedures.rs
b/crates/integrations/datafusion/src/procedures.rs
index 18416eec..b908da7d 100644
--- a/crates/integrations/datafusion/src/procedures.rs
+++ b/crates/integrations/datafusion/src/procedures.rs
@@ -64,21 +64,18 @@ use paimon::api::{
PermissionColumns, PermissionResource, PolicyType, ResourceType, RowFilter,
};
use paimon::catalog::{Catalog, Identifier, RESTCatalog};
-use paimon::lumina::LUMINA_IDENTIFIER;
+use paimon::lumina::{is_lumina_index_type, LUMINA_IDENTIFIER};
use paimon::spec::Snapshot;
use paimon::table::{
- normalize_global_index_type_for_drop, BranchManager, SnapshotManager,
Table, TagManager,
+ normalize_global_index_type, BranchManager, SnapshotManager, Table,
TagManager,
SUPPORTED_GLOBAL_INDEX_TYPES_FOR_DROP,
};
-use paimon::vindex::is_vindex_index_type;
use crate::error::to_datafusion_error;
/// Default `index_type` for the global index procedures when the argument is
/// omitted, matching Java's `CreateGlobalIndexProcedure`.
const DEFAULT_GLOBAL_INDEX_TYPE: &str = "btree";
-/// Java `NativeFullTextGlobalIndexerFactory.IDENTIFIER`.
-const FULL_TEXT_GLOBAL_INDEX_TYPE: &str = "full-text";
/// Resolve a snapshot by id: try live snapshot file first, then fall back to
tag metadata.
async fn resolve_snapshot_by_id(
@@ -716,13 +713,18 @@ async fn proc_create_lumina_index(
) -> DFResult<DataFrame> {
let table = get_table(catalog, catalog_name, args).await?;
let index_column = require_arg(args, "index_column")?;
- let mut builder = table.new_lumina_index_build_builder();
+ let mut builder = table.new_global_index_build_builder();
builder.with_index_column(index_column);
let index_type = normalize_index_type(
args.get("index_type")
.map(String::as_str)
.unwrap_or(LUMINA_IDENTIFIER),
);
+ if !is_lumina_index_type(&index_type) {
+ return Err(DataFusionError::NotImplemented(format!(
+ "Unsupported Lumina index type: {index_type}"
+ )));
+ }
builder.with_index_type(&index_type);
if let Some(options) = args.get("options") {
builder.with_options(parse_key_value_options(options)?);
@@ -743,45 +745,14 @@ async fn proc_create_global_index(
.get("index_type")
.map(String::as_str)
.unwrap_or(DEFAULT_GLOBAL_INDEX_TYPE);
- let index_type = normalize_index_type(index_type_arg);
- let index_type = index_type.as_str();
- if is_scalar_global_index_type(index_type) {
- let mut builder = table.new_sorted_global_index_build_builder();
- builder.with_index_column(index_column);
- builder.with_index_type(index_type);
- if let Some(options) = args.get("options") {
- builder.with_options(parse_key_value_options(options)?);
- }
- builder.execute().await.map_err(to_datafusion_error)?;
- } else if is_vindex_index_type(index_type) {
- let mut builder = table.new_vindex_index_build_builder(index_type);
- builder.with_index_column(index_column);
- if let Some(options) = args.get("options") {
- builder.with_options(parse_key_value_options(options)?);
- }
- builder.execute().await.map_err(to_datafusion_error)?;
- } else if index_type == FULL_TEXT_GLOBAL_INDEX_TYPE {
- #[cfg(feature = "fulltext")]
- {
- let mut builder = table.new_full_text_index_build_builder();
- builder.with_index_column(index_column);
- if let Some(options) = args.get("options") {
- builder.with_options(parse_key_value_options(options)?);
- }
- builder.execute().await.map_err(to_datafusion_error)?;
- }
- #[cfg(not(feature = "fulltext"))]
- return Err(DataFusionError::NotImplemented(
- "create_global_index with index_type => 'full-text' requires the
'fulltext' feature"
- .to_string(),
- ));
- } else {
- // Echo the raw argument, not the normalized one, so a typo stays
visible.
- return Err(DataFusionError::NotImplemented(format!(
- "create_global_index only supports index_type => 'btree',
'bitmap', 'multivalue', 'fm', 'full-text', \
- or vindex types ('ivf-flat', 'ivf-pq', 'ivf-sq', 'ivf-rq',
'diskann'), got '{index_type_arg}'"
- )));
+ let mut builder = table.new_global_index_build_builder();
+ builder
+ .with_index_column(index_column)
+ .with_index_type(index_type_arg);
+ if let Some(options) = args.get("options") {
+ builder.with_options(parse_key_value_options(options)?);
}
+ builder.execute().await.map_err(to_datafusion_error)?;
ok_result(ctx)
}
@@ -799,7 +770,7 @@ async fn proc_drop_global_index(
.unwrap_or(DEFAULT_GLOBAL_INDEX_TYPE);
let index_type = normalize_index_type(index_type_arg);
let index_type = index_type.as_str();
- if normalize_global_index_type_for_drop(index_type).is_none() {
+ if normalize_global_index_type(index_type).is_none() {
// Echo the raw argument, not the normalized one, so a typo stays
visible.
return Err(DataFusionError::NotImplemented(format!(
"unsupported global index type '{index_type_arg}'; supported:
{SUPPORTED_GLOBAL_INDEX_TYPES_FOR_DROP}"
@@ -829,17 +800,10 @@ async fn proc_drop_global_index(
ok_result(ctx)
}
-/// Precondition: `index_type` is already canonical (see
`normalize_index_type`).
-fn is_scalar_global_index_type(index_type: &str) -> bool {
- matches!(index_type, "btree" | "bitmap" | "multivalue" | "fm")
-}
-
/// Canonicalize a procedure's `index_type` argument: trim, then lowercase.
/// Mirrors `indexType.toLowerCase(Locale.ROOT).trim()` in Java's Flink and
Spark
/// `CreateGlobalIndexProcedure` / `DropGlobalIndexProcedure`. Normalizing at
this
-/// boundary keeps the core builders' exact matching intact -- they are the
analog
-/// of Java's `GlobalIndexer`, which likewise receives an already-canonical
value
-/// and persists it into index metadata.
+/// boundary and the core builder both accept procedure-style identifiers.
fn normalize_index_type(index_type: &str) -> String {
index_type.trim().to_ascii_lowercase()
}
@@ -1442,18 +1406,6 @@ mod tests {
assert_eq!(normalize_index_type(" Full-Text "), "full-text");
}
- #[test]
- fn test_scalar_global_index_type_predicate() {
- assert!(is_scalar_global_index_type("btree"));
- assert!(is_scalar_global_index_type("bitmap"));
- assert!(is_scalar_global_index_type("multivalue"));
- assert!(is_scalar_global_index_type("fm"));
- assert!(!is_scalar_global_index_type("ivf-flat"));
- assert!(!is_scalar_global_index_type("lumina"));
- // The predicate requires a canonical input; callers normalize first.
- assert!(!is_scalar_global_index_type("BTREE"));
- }
-
#[test]
fn test_reject_unknown_args_covers_tag_and_index_procedures() {
// A typo'd optional arg on a tag/index procedure used to be silently
dropped:
diff --git a/crates/integrations/datafusion/tests/procedures.rs
b/crates/integrations/datafusion/tests/procedures.rs
index d89c4547..abc06c63 100644
--- a/crates/integrations/datafusion/tests/procedures.rs
+++ b/crates/integrations/datafusion/tests/procedures.rs
@@ -544,7 +544,7 @@ async fn
test_create_global_index_rejects_unsupported_index_types() {
index_type => '{index_type}'\
)"
),
- "only supports index_type => 'btree', 'bitmap', 'multivalue',
'fm', 'full-text', or vindex types",
+ "Unsupported global index type",
)
.await;
}
@@ -675,7 +675,7 @@ async fn
test_global_index_procedures_echo_raw_unsupported_type() {
assert_sql_error(
&sql_context,
"CALL sys.create_global_index(table => 'test_db.btree_echo',
index_column => 'id', index_type => 'Ivf-Hnsw-Flat')",
- "got 'Ivf-Hnsw-Flat'",
+ "Unsupported global index type: 'Ivf-Hnsw-Flat'",
)
.await;
assert_sql_error(
diff --git a/crates/paimon/examples/ivfpq_build_benchmark.rs
b/crates/paimon/examples/ivfpq_build_benchmark.rs
index cbfe1f3f..84a7844e 100644
--- a/crates/paimon/examples/ivfpq_build_benchmark.rs
+++ b/crates/paimon/examples/ivfpq_build_benchmark.rs
@@ -62,7 +62,8 @@ async fn main() -> Result<(), Box<dyn Error>> {
]);
let started = Instant::now();
let built_shards = table
- .new_vindex_index_build_builder("ivf-pq")
+ .new_global_index_build_builder()
+ .with_index_type("ivf-pq")
.with_index_column(&column)
.with_options(options.clone())
.execute()
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index b1dbbddb..a29c9241 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -7360,7 +7360,7 @@ mod tests {
.await
.unwrap();
table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("id")
.execute()
.await
diff --git a/crates/paimon/src/table/de_vector_read/tests.rs
b/crates/paimon/src/table/de_vector_read/tests.rs
index ed64c2cb..0d76e309 100644
--- a/crates/paimon/src/table/de_vector_read/tests.rs
+++ b/crates/paimon/src/table/de_vector_read/tests.rs
@@ -1516,7 +1516,8 @@ async fn
filtered_rerank_rejects_metric_from_a_skipped_shard() {
.await
.unwrap();
let count = table
- .new_vindex_index_build_builder(IVF_FLAT_IDENTIFIER)
+ .new_global_index_build_builder()
+ .with_index_type(IVF_FLAT_IDENTIFIER)
.with_index_column("embedding")
.with_options(HashMap::from([(
"ivf-flat.distance.metric".to_string(),
diff --git a/crates/paimon/src/table/de_vector_scan/tests.rs
b/crates/paimon/src/table/de_vector_scan/tests.rs
index aa1b14f9..310a442f 100644
--- a/crates/paimon/src/table/de_vector_scan/tests.rs
+++ b/crates/paimon/src/table/de_vector_scan/tests.rs
@@ -506,7 +506,8 @@ async fn
scalar_coverage_gaps_follow_scalar_and_vector_modes() {
.unwrap();
if vector_covers_tail {
table
-
.new_vindex_index_build_builder(crate::vindex::IVF_FLAT_IDENTIFIER)
+ .new_global_index_build_builder()
+ .with_index_type(crate::vindex::IVF_FLAT_IDENTIFIER)
.with_index_column("embedding")
.with_options(HashMap::from([(
"ivf-flat.nlist".to_string(),
diff --git a/crates/paimon/src/table/full_text_index_build_builder.rs
b/crates/paimon/src/table/full_text_index_build_builder.rs
index 1ae932bd..0ee0924f 100644
--- a/crates/paimon/src/table/full_text_index_build_builder.rs
+++ b/crates/paimon/src/table/full_text_index_build_builder.rs
@@ -21,12 +21,11 @@
//! `NativeFullTextGlobalIndexWriter`: the table's row-id space is cut into
//! `global-index.row-count-per-shard` shards, each shard's text column is fed
to
//! the shared `paimon-ftindex-core` writer with shard-relative row ids, and
every
-//! non-empty shard becomes one `full-text` index file committed in a single
-//! snapshot. Java writes the same archive through the same native core, so the
+//! non-empty shard becomes one unpublished `full-text` index file. Java
writes the same archive through the same native core, so the
//! files are readable by both `FullTextSearchBuilder` and Java.
use crate::spec::{
- CoreOptions, DataField, DataType, GlobalIndexMeta, IndexFileMeta,
ROW_ID_FIELD_NAME,
+ CoreOptions, DataField, DataType, GlobalIndexMeta, IndexFileMeta,
Predicate, ROW_ID_FIELD_NAME,
};
use crate::table::global_index_build_common::vector::{
find_index_field, plan_vector_index_shards, VectorIndexShard,
@@ -45,15 +44,15 @@ use paimon_ftindex_core::io::PosWriter;
use paimon_ftindex_core::{FullTextIndexConfig, FullTextIndexWriter};
use std::collections::{BTreeMap, HashMap};
-const INDEX_DIR: &str = "index";
/// Java `NativeFullTextIndexOptions.FULL_TEXT_PREFIX`: only these options
reach
/// the native writer, with the prefix removed.
const FULL_TEXT_OPTION_PREFIX: &str = "full-text.";
-pub struct FullTextIndexBuildBuilder<'a> {
+pub(super) struct FullTextIndexBuildBuilder<'a> {
table: &'a Table,
index_column: Option<String>,
options: HashMap<String, String>,
+ pub(super) partition_filter: Option<Predicate>,
}
impl<'a> FullTextIndexBuildBuilder<'a> {
@@ -62,6 +61,7 @@ impl<'a> FullTextIndexBuildBuilder<'a> {
table,
index_column: None,
options: HashMap::new(),
+ partition_filter: None,
}
}
@@ -77,10 +77,7 @@ impl<'a> FullTextIndexBuildBuilder<'a> {
self
}
- /// Build full-text index files for every row range of the latest snapshot
- /// that is not yet covered by a full-text index on the column. Returns the
- /// number of index files committed.
- pub async fn execute(&self) -> Result<usize> {
+ pub(super) async fn prepare(&self) -> Result<(Option<i64>,
Vec<CommitMessage>)> {
// Building the index scans the table's rows.
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
@@ -115,17 +112,15 @@ impl<'a> FullTextIndexBuildBuilder<'a> {
// The table's manager resolves a REST-managed table's latest snapshot
// through the catalog, like reads and commit validation do.
let snapshot_manager = self.table.snapshot_manager();
- let snapshot = snapshot_manager
- .get_latest_snapshot()
- .await?
- .ok_or_else(|| Error::DataInvalid {
- message: "Cannot build full-text index without a
snapshot".to_string(),
- source: None,
- })?;
+ let Some(snapshot) = snapshot_manager.get_latest_snapshot().await?
else {
+ return Ok((None, vec![]));
+ };
- let manifest_entries = self
- .table
- .new_read_builder()
+ let mut read_builder = self.table.new_read_builder();
+ if let Some(filter) = &self.partition_filter {
+ read_builder.with_filter(filter.clone());
+ }
+ let manifest_entries = read_builder
.new_scan()
.with_scan_all_files()
.plan_manifest_entries(&snapshot)
@@ -150,7 +145,7 @@ impl<'a> FullTextIndexBuildBuilder<'a> {
"full-text",
)?;
if shards.is_empty() {
- return Ok(0);
+ return Ok((Some(snapshot.id()), vec![]));
}
validate_existing_index_overlap(
@@ -201,14 +196,10 @@ impl<'a> FullTextIndexBuildBuilder<'a> {
messages.push(message);
}
if messages.is_empty() {
- return Ok(0);
+ return Ok((Some(snapshot.id()), vec![]));
}
- let file_count = messages.len();
- commit
- .commit_if_latest_snapshot(messages, snapshot.id())
- .await?;
- Ok(file_count)
+ Ok((Some(snapshot.id()), messages))
}
async fn build_shard(
@@ -272,16 +263,13 @@ impl<'a> FullTextIndexBuildBuilder<'a> {
})
.await?;
- let table_location = self.table.location().trim_end_matches('/');
let file_name = format!(
"{FULL_TEXT_GLOBAL_INDEX_TYPE}-global-index-{}.index",
uuid::Uuid::new_v4()
);
- let index_path = format!("{table_location}/{INDEX_DIR}/{file_name}");
- self.table
- .file_io()
- .mkdirs(&format!("{table_location}/{INDEX_DIR}/"))
- .await?;
+ let (index_path, external_path) =
+
super::global_index_build_common::prepare_index_file_path(self.table,
&file_name)
+ .await?;
let upload: Result<i64> = async {
copy_local_file_to_output(
local_index.path(),
@@ -312,13 +300,15 @@ impl<'a> FullTextIndexBuildBuilder<'a> {
file_size,
row_count: cursor.row_count,
deletion_vectors_ranges: None,
- external_path: None,
+ external_path,
global_index_meta: Some(GlobalIndexMeta {
row_range_start: shard.row_range_start,
row_range_end: shard.row_range_end,
index_field_id,
extra_field_ids: None,
- source_meta: None,
+ source_meta: Some(
+
crate::spec::DataEvolutionIndexSourceMeta::new(shard.snapshot_id)?.serialize(),
+ ),
index_meta: Some(index_meta.to_vec()),
}),
}))
diff --git a/crates/paimon/src/table/full_text_index_build_builder/tests.rs
b/crates/paimon/src/table/full_text_index_build_builder/tests.rs
index f59ccfc3..3dc16531 100644
--- a/crates/paimon/src/table/full_text_index_build_builder/tests.rs
+++ b/crates/paimon/src/table/full_text_index_build_builder/tests.rs
@@ -169,7 +169,8 @@ async fn
test_execute_builds_shards_readable_by_full_text_search() {
assert!(search(&table, "paimon").await.is_empty());
let file_count = table
- .new_full_text_index_build_builder()
+ .new_global_index_build_builder()
+ .with_index_type("full-text")
.with_index_column("name")
.execute()
.await
@@ -224,7 +225,8 @@ async fn test_execute_only_indexes_new_rows() {
write_rows(&table, vec![1, 2], vec![Some("paimon one"),
Some("other")]).await;
let builder = || {
- let mut builder = table.new_full_text_index_build_builder();
+ let mut builder = table.new_global_index_build_builder();
+ builder.with_index_type("full-text");
builder.with_index_column("name");
builder
};
@@ -259,7 +261,8 @@ async fn test_execute_indexes_all_null_shard() {
// Java's writer counts NULL rows, so an all-NULL shard still yields an
// (empty) index file that marks the range as indexed.
let file_count = table
- .new_full_text_index_build_builder()
+ .new_global_index_build_builder()
+ .with_index_type("full-text")
.with_index_column("name")
.execute()
.await
@@ -275,7 +278,8 @@ async fn
test_execute_passes_prefixed_options_to_native_writer() {
write_rows(&table, vec![1], vec![Some("Running Paimon")]).await;
let file_count = table
- .new_full_text_index_build_builder()
+ .new_global_index_build_builder()
+ .with_index_type("full-text")
.with_index_column("name")
.with_options(HashMap::from([
("full-text.stem".to_string(), "false".to_string()),
@@ -309,7 +313,8 @@ async fn
test_execute_rejects_invalid_native_options_before_reading() {
setup_dirs(&table).await;
let err = table
- .new_full_text_index_build_builder()
+ .new_global_index_build_builder()
+ .with_index_type("full-text")
.with_index_column("name")
.with_options(HashMap::from([(
"full-text.tokenizer".to_string(),
@@ -328,7 +333,8 @@ async fn
test_execute_rejects_invalid_native_options_before_reading() {
async fn test_execute_rejects_unsupported_tables_and_columns() {
let expect_err = |table: Table, column: &'static str| async move {
table
- .new_full_text_index_build_builder()
+ .new_global_index_build_builder()
+ .with_index_type("full-text")
.with_index_column(column)
.execute()
.await
@@ -394,7 +400,8 @@ async fn
test_execute_rejects_unsupported_tables_and_columns() {
async fn test_execute_requires_index_column() {
let table = test_table("memory:/ft_no_column", table_options("10"));
let err = table
- .new_full_text_index_build_builder()
+ .new_global_index_build_builder()
+ .with_index_type("full-text")
.execute()
.await
.unwrap_err();
@@ -611,7 +618,8 @@ async fn
test_build_search_and_drop_through_the_catalog_snapshot() {
);
let file_count = table
- .new_full_text_index_build_builder()
+ .new_global_index_build_builder()
+ .with_index_type("full-text")
.with_index_column("name")
.execute()
.await
diff --git a/crates/paimon/src/table/global_index_build_builder.rs
b/crates/paimon/src/table/global_index_build_builder.rs
new file mode 100644
index 00000000..428de822
--- /dev/null
+++ b/crates/paimon/src/table/global_index_build_builder.rs
@@ -0,0 +1,348 @@
+// 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 std::collections::HashMap;
+use std::time::Instant;
+
+#[cfg(feature = "fulltext")]
+use super::full_text_index_build_builder::FullTextIndexBuildBuilder;
+use super::global_index_build_common::{resolve_index_fields, IndexColumns};
+use super::global_index_types::{
+ normalize_global_index_type, normalize_queryable_global_index_type,
BTREE_GLOBAL_INDEX_TYPE,
+ FULL_TEXT_GLOBAL_INDEX_TYPE,
+};
+use super::lumina_index_build_builder::LuminaIndexBuildBuilder;
+use super::sorted_global_index_build_builder::SortedGlobalIndexBuildBuilder;
+use super::vindex_index_build_builder::timing::{
+ vector_index_build_timing_enabled, VectorIndexBuildTiming,
+};
+use super::vindex_index_build_builder::VindexIndexBuildBuilder;
+use super::{CommitMessage, Table, TableCommit};
+use crate::lumina::LUMINA_IDENTIFIER;
+use crate::spec::{CoreOptions, Predicate};
+use crate::{Error, Result};
+
+/// Build global indexes by type, like Java's `GlobalIndexer.create`.
+/// Sorting and native writer selection are internal implementation details.
+pub struct GlobalIndexBuildBuilder<'a> {
+ table: &'a Table,
+ columns: Option<IndexColumns>,
+ index_type: String,
+ options: HashMap<String, String>,
+ partition_filter: Option<Predicate>,
+}
+
+struct PreparedBuild {
+ snapshot_id: Option<i64>,
+ messages: Vec<CommitMessage>,
+ vector_timings: Vec<VectorIndexBuildTiming>,
+}
+
+impl From<(Option<i64>, Vec<CommitMessage>)> for PreparedBuild {
+ fn from((snapshot_id, messages): (Option<i64>, Vec<CommitMessage>)) ->
Self {
+ Self {
+ snapshot_id,
+ messages,
+ vector_timings: vec![],
+ }
+ }
+}
+
+impl<'a> GlobalIndexBuildBuilder<'a> {
+ pub(crate) fn new(table: &'a Table) -> Self {
+ Self {
+ table,
+ columns: None,
+ index_type: BTREE_GLOBAL_INDEX_TYPE.into(),
+ options: HashMap::new(),
+ partition_filter: None,
+ }
+ }
+
+ /// Whether this build includes a writer for the requested index type.
+ pub fn supports_index_type(index_type: &str) -> bool {
+
normalize_global_index_type(index_type.trim()).is_some_and(|index_type| {
+ index_type != FULL_TEXT_GLOBAL_INDEX_TYPE || cfg!(feature =
"fulltext")
+ })
+ }
+
+ pub fn with_index_column(&mut self, column: &str) -> &mut Self {
+ self.columns = Some(IndexColumns::Column(column.into()));
+ self
+ }
+
+ /// Select columns in physical tuple-key order; only BTree supports
multiple columns.
+ pub fn with_index_columns(&mut self, columns: &[&str]) -> &mut Self {
+ self.columns = Some(IndexColumns::Columns(
+ columns.iter().map(|column| (*column).into()).collect(),
+ ));
+ self
+ }
+
+ pub fn with_index_type(&mut self, index_type: &str) -> &mut Self {
+ self.index_type = index_type.into();
+ self
+ }
+
+ pub fn with_options(&mut self, options: HashMap<String, String>) -> &mut
Self {
+ self.options = options;
+ self
+ }
+
+ /// Restrict the build to partitions. Repeated calls combine filters with
AND.
+ pub fn with_partition_filter(&mut self, filter: Predicate) -> Result<&mut
Self> {
+ super::partition_filter::validate_partition_filter(self.table,
&filter)?;
+ self.partition_filter = Some(match self.partition_filter.take() {
+ Some(previous) => Predicate::and(vec![previous, filter]),
+ None => filter,
+ });
+ Ok(self)
+ }
+
+ /// Prepare files without publishing a snapshot. The caller owns the
returned messages.
+ /// No snapshot or no uncovered rows produces an empty list.
+ pub async fn build(&self) -> Result<Vec<CommitMessage>> {
+ Ok(self.prepare().await?.messages)
+ }
+
+ /// Prepare and commit, rejecting changes to the snapshot used for the
build.
+ pub async fn execute(&self) -> Result<usize> {
+ let prepared = self.prepare().await?;
+ if prepared.messages.is_empty() {
+ return Ok(0);
+ }
+ let count = prepared
+ .messages
+ .iter()
+ .map(|message| message.new_index_files.len())
+ .sum();
+ let commit_start =
vector_index_build_timing_enabled().then(Instant::now);
+ TableCommit::new(
+ self.table.clone(),
+ format!("global-index-create-{}", uuid::Uuid::new_v4()),
+ )
+ .commit_if_latest_snapshot(
+ prepared.messages,
+ prepared
+ .snapshot_id
+ .expect("nonempty index build has a snapshot"),
+ )
+ .await?;
+ if let Some(start) = commit_start {
+ let elapsed = start.elapsed();
+ for timing in prepared.vector_timings {
+ timing.log(self.index_type.trim(), elapsed);
+ }
+ }
+ // Once submitted, messages may have been committed even if an error
is returned.
+ // Never abort their files on a failed or uncertain commit.
+ Ok(count)
+ }
+
+ async fn prepare(&self) -> Result<PreparedBuild> {
+
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
+ self.table.ensure_not_branch_reference_for_write()?;
+ let index_type =
normalize_global_index_type(self.index_type.trim()).ok_or_else(|| {
+ Error::Unsupported {
+ message: format!("Unsupported global index type: '{}'",
self.index_type),
+ }
+ })?;
+ let columns = self.columns.as_ref().ok_or_else(|| Error::DataInvalid {
+ message: "Global index column is required".into(),
+ source: None,
+ })?;
+ let fields = resolve_index_fields(self.table, columns, index_type)?;
+ let columns = fields.iter().map(|field|
field.name()).collect::<Vec<_>>();
+ if normalize_queryable_global_index_type(index_type).is_some() {
+ let mut builder = SortedGlobalIndexBuildBuilder::new(self.table);
+ builder
+ .with_index_columns(&columns)
+ .with_index_type(index_type)
+ .with_options(self.options.clone());
+ if let Some(filter) = &self.partition_filter {
+ builder.with_partition_filter(filter.clone())?;
+ }
+ return Ok(builder.prepare().await?.into());
+ }
+ if index_type == FULL_TEXT_GLOBAL_INDEX_TYPE {
+ #[cfg(feature = "fulltext")]
+ {
+ let mut builder = FullTextIndexBuildBuilder::new(self.table);
+ builder
+ .with_index_column(columns[0])
+ .with_options(self.options.clone());
+ builder.partition_filter = self.partition_filter.clone();
+ return Ok(builder.prepare().await?.into());
+ }
+ #[cfg(not(feature = "fulltext"))]
+ return Err(Error::Unsupported {
+ message: "Full-text global index build requires the 'fulltext'
feature".into(),
+ });
+ }
+ if index_type == LUMINA_IDENTIFIER {
+ let mut builder = LuminaIndexBuildBuilder::new(self.table);
+ builder
+ .with_index_column(columns[0])
+ .with_options(self.options.clone());
+ builder.partition_filter = self.partition_filter.clone();
+ return Ok(builder.prepare().await?.into());
+ }
+ let mut builder = VindexIndexBuildBuilder::new(self.table, index_type);
+ builder
+ .with_index_column(columns[0])
+ .with_options(self.options.clone());
+ builder.partition_filter = self.partition_filter.clone();
+ let (snapshot_id, messages, vector_timings) = builder.prepare().await?;
+ Ok(PreparedBuild {
+ snapshot_id,
+ messages,
+ vector_timings,
+ })
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::catalog::Identifier;
+ use crate::io::FileIOBuilder;
+ use crate::spec::{ArrayType, DataType, FloatType, IntType, Schema,
TableSchema, VarCharType};
+
+ fn empty_table() -> Table {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("text", DataType::VarChar(VarCharType::string_type()))
+ .column(
+ "vector",
+
DataType::Array(ArrayType::new(DataType::Float(FloatType::new()))),
+ )
+ .options(HashMap::from([
+ ("row-tracking.enabled".into(), "true".into()),
+ ("data-evolution.enabled".into(), "true".into()),
+ ("global-index.enabled".into(), "true".into()),
+ ]))
+ .build()
+ .unwrap();
+ Table::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("default", "empty"),
+ "memory:/empty_global_index".into(),
+ TableSchema::new(0, &schema),
+ None,
+ )
+ }
+
+ #[tokio::test]
+ async fn all_index_families_share_empty_build_and_execute_contract() {
+ let table = empty_table();
+ for (kind, column) in [
+ ("btree", "id"),
+ ("bitmap", "id"),
+ ("multivalue", "vector"),
+ ("fm", "text"),
+ ("full-text", "text"),
+ ("lumina", "vector"),
+ ("lumina-vector-ann", "vector"),
+ ("ivf-flat", "vector"),
+ ("ivf-pq", "vector"),
+ ("ivf-sq", "vector"),
+ ("ivf-rq", "vector"),
+ ("diskann", "vector"),
+ ] {
+ if !GlobalIndexBuildBuilder::supports_index_type(kind) {
+ continue;
+ }
+ let mut options = HashMap::from([(
+ format!(
+ "{}.dimension",
+ if kind == "lumina-vector-ann" {
+ "lumina"
+ } else {
+ kind
+ }
+ ),
+ "2".into(),
+ )]);
+ if kind == "ivf-pq" {
+ options.insert("ivf-pq.pq.m".into(), "1".into());
+ }
+ options.insert("global-index.row-count-per-shard".into(),
"4".into());
+ let mut builder = table.new_global_index_build_builder();
+ builder
+ .with_index_column(column)
+ .with_index_type(&format!(" {} ", kind.to_uppercase()))
+ .with_options(options);
+ assert!(builder.build().await.unwrap().is_empty(), "{kind}");
+ assert_eq!(builder.execute().await.unwrap(), 0, "{kind}");
+ }
+ assert!(table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .is_none());
+ }
+
+ #[tokio::test]
+ async fn column_validation_precedes_any_index_io() {
+ let table = empty_table();
+ for kind in ["bitmap", "fm", "full-text", "ivf-flat", "lumina"] {
+ let mut builder = table.new_global_index_build_builder();
+ builder
+ .with_index_columns(&["id", "text"])
+ .with_index_type(kind);
+ assert!(
+ matches!(builder.build().await, Err(Error::Unsupported {
message }) if message.contains("multiple index columns"))
+ );
+ }
+ for columns in [vec![], vec![""], vec!["id", "id"]] {
+ assert!(table
+ .new_global_index_build_builder()
+ .with_index_columns(&columns)
+ .build()
+ .await
+ .is_err());
+ }
+ assert!(matches!(
+ table
+ .new_global_index_build_builder()
+ .with_index_column("missing")
+ .build()
+ .await,
+ Err(Error::ColumnNotExist { .. })
+ ));
+ assert!(matches!(
+ table
+ .new_global_index_build_builder()
+ .with_index_type("unknown")
+ .build()
+ .await,
+ Err(Error::Unsupported { .. })
+ ));
+ assert!(!GlobalIndexBuildBuilder::supports_index_type("unknown"));
+ }
+
+ #[cfg(not(feature = "fulltext"))]
+ #[tokio::test]
+ async fn full_text_without_feature_is_explicitly_unsupported() {
+ assert!(!GlobalIndexBuildBuilder::supports_index_type("full-text"));
+ assert!(
+
matches!(empty_table().new_global_index_build_builder().with_index_column("text")
+ .with_index_type("full-text").build().await,
Err(Error::Unsupported { message }) if message.contains("fulltext"))
+ );
+ }
+}
diff --git a/crates/paimon/src/table/global_index_build_common.rs
b/crates/paimon/src/table/global_index_build_common.rs
index aab1784f..77775b18 100644
--- a/crates/paimon/src/table/global_index_build_common.rs
+++ b/crates/paimon/src/table/global_index_build_common.rs
@@ -235,3 +235,26 @@ pub(crate) async fn validate_existing_index_overlap(
}
Ok(())
}
+
+/// Resolve and create the table's global index directory, including external
placement.
+pub(crate) async fn prepare_index_file_path(
+ table: &Table,
+ file_name: &str,
+) -> Result<(String, Option<String>)> {
+ let external_path = super::external_path::new_index_external_path(
+ table.schema().options(),
+ false,
+ "",
+ file_name,
+ )?;
+ let path = external_path.clone().unwrap_or_else(|| {
+ format!(
+ "{}/index/{file_name}",
+ table.location().trim_end_matches('/')
+ )
+ });
+ if let Some((directory, _)) = path.rsplit_once('/') {
+ table.file_io().mkdirs(&format!("{directory}/")).await?;
+ }
+ Ok((path, external_path))
+}
diff --git a/crates/paimon/src/table/global_index_drop_builder.rs
b/crates/paimon/src/table/global_index_drop_builder.rs
index 155a708e..7b2f811b 100644
--- a/crates/paimon/src/table/global_index_drop_builder.rs
+++ b/crates/paimon/src/table/global_index_drop_builder.rs
@@ -17,8 +17,7 @@
use super::global_index_build_common::IndexColumns;
use super::global_index_types::{
- normalize_global_index_type_for_drop, BTREE_GLOBAL_INDEX_TYPE,
- SUPPORTED_GLOBAL_INDEX_TYPES_FOR_DROP,
+ normalize_global_index_type, BTREE_GLOBAL_INDEX_TYPE,
SUPPORTED_GLOBAL_INDEX_TYPES_FOR_DROP,
};
use crate::spec::{FileKind, IndexFileMeta, IndexManifest};
use crate::table::{CommitMessage, Table, TableCommit};
@@ -73,13 +72,11 @@ impl<'a> GlobalIndexDropBuilder<'a> {
self.table.ensure_not_branch_reference_for_write()?;
let index_type =
-
normalize_global_index_type_for_drop(&self.index_type).ok_or_else(|| {
- Error::Unsupported {
- message: format!(
- "unsupported global index type '{}'; supported: {}",
- self.index_type, SUPPORTED_GLOBAL_INDEX_TYPES_FOR_DROP
- ),
- }
+ normalize_global_index_type(&self.index_type).ok_or_else(||
Error::Unsupported {
+ message: format!(
+ "unsupported global index type '{}'; supported: {}",
+ self.index_type, SUPPORTED_GLOBAL_INDEX_TYPES_FOR_DROP
+ ),
})?;
let index_column = self
.index_column
@@ -121,9 +118,7 @@ impl<'a> GlobalIndexDropBuilder<'a> {
if entry.kind != FileKind::Add {
continue;
}
- if
normalize_global_index_type_for_drop(&entry.index_file.index_type)
- != Some(index_type)
- {
+ if normalize_global_index_type(&entry.index_file.index_type) !=
Some(index_type) {
continue;
}
let Some(global_meta) =
entry.index_file.global_index_meta.as_ref() else {
@@ -218,7 +213,7 @@ mod tests {
}
refused(
table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.execute()
.await
.unwrap_err(),
@@ -226,7 +221,8 @@ mod tests {
);
refused(
table
- .new_vindex_index_build_builder("vector")
+ .new_global_index_build_builder()
+ .with_index_type("vector")
.execute()
.await
.unwrap_err(),
@@ -234,7 +230,8 @@ mod tests {
);
refused(
table
- .new_lumina_index_build_builder()
+ .new_global_index_build_builder()
+ .with_index_type("lumina")
.execute()
.await
.unwrap_err(),
diff --git a/crates/paimon/src/table/global_index_types.rs
b/crates/paimon/src/table/global_index_types.rs
index a7a03841..a51a54cc 100644
--- a/crates/paimon/src/table/global_index_types.rs
+++ b/crates/paimon/src/table/global_index_types.rs
@@ -55,14 +55,14 @@ pub const SUPPORTED_GLOBAL_INDEX_TYPES_FOR_DROP: &str =
"btree, bitmap, multivalue, fm, full-text, lumina, lumina-vector-ann,
ivf-flat, ivf-pq, ivf-sq, ivf-rq, diskann";
/// Canonicalize any supported global index type to a stable `&'static str`, or
-/// `None` if unsupported. Case-insensitive. Order: sorted -> full-text ->
lumina
-/// -> vindex.
+/// `None` if unsupported. Case-insensitive and shared by build and drop.
+/// Order: scalar -> full-text -> lumina -> vindex.
///
/// Both lumina aliases (`lumina`, `lumina-vector-ann`) canonicalize to
/// `"lumina"`; each vindex type keeps its own identity so dropping one vindex
/// type never matches another on the same column. Callers should compare the
/// canonical form of BOTH the request and each stored entry's `index_type`.
-pub fn normalize_global_index_type_for_drop(index_type: &str) ->
Option<&'static str> {
+pub fn normalize_global_index_type(index_type: &str) -> Option<&'static str> {
if let Some(queryable) = normalize_queryable_global_index_type(index_type)
{
return Some(queryable);
}
@@ -100,77 +100,47 @@ mod tests {
#[test]
fn sorted_types_canonicalize_case_insensitively() {
- assert_eq!(normalize_global_index_type_for_drop("BTREE"),
Some("btree"));
+ assert_eq!(normalize_global_index_type("BTREE"), Some("btree"));
+ assert_eq!(normalize_global_index_type("Bitmap"), Some("bitmap"));
assert_eq!(
- normalize_global_index_type_for_drop("Bitmap"),
- Some("bitmap")
- );
- assert_eq!(
- normalize_global_index_type_for_drop("MultiValue"),
+ normalize_global_index_type("MultiValue"),
Some("multivalue")
);
- assert_eq!(normalize_global_index_type_for_drop("FM"), Some("fm"));
+ assert_eq!(normalize_global_index_type("FM"), Some("fm"));
}
#[test]
fn full_text_canonicalizes_case_insensitively() {
- assert_eq!(
- normalize_global_index_type_for_drop("full-text"),
- Some("full-text")
- );
- assert_eq!(
- normalize_global_index_type_for_drop("Full-Text"),
- Some("full-text")
- );
+ assert_eq!(normalize_global_index_type("full-text"),
Some("full-text"));
+ assert_eq!(normalize_global_index_type("Full-Text"),
Some("full-text"));
// Full-text is dropped by its own identity, never as a queryable
scalar index.
assert_eq!(normalize_queryable_global_index_type("full-text"), None);
}
#[test]
fn lumina_aliases_canonicalize_to_lumina() {
+ assert_eq!(normalize_global_index_type("lumina"), Some("lumina"));
assert_eq!(
- normalize_global_index_type_for_drop("lumina"),
- Some("lumina")
- );
- assert_eq!(
- normalize_global_index_type_for_drop("lumina-vector-ann"),
- Some("lumina")
- );
- assert_eq!(
- normalize_global_index_type_for_drop("LUMINA"),
+ normalize_global_index_type("lumina-vector-ann"),
Some("lumina")
);
+ assert_eq!(normalize_global_index_type("LUMINA"), Some("lumina"));
}
#[test]
fn vindex_types_keep_distinct_identity() {
- assert_eq!(
- normalize_global_index_type_for_drop("ivf-flat"),
- Some("ivf-flat")
- );
- assert_eq!(
- normalize_global_index_type_for_drop("IVF-PQ"),
- Some("ivf-pq")
- );
- assert_eq!(
- normalize_global_index_type_for_drop("IVF-SQ"),
- Some("ivf-sq")
- );
- assert_eq!(
- normalize_global_index_type_for_drop("IVF-RQ"),
- Some("ivf-rq")
- );
- assert_eq!(
- normalize_global_index_type_for_drop("DiskANN"),
- Some("diskann")
- );
+ assert_eq!(normalize_global_index_type("ivf-flat"), Some("ivf-flat"));
+ assert_eq!(normalize_global_index_type("IVF-PQ"), Some("ivf-pq"));
+ assert_eq!(normalize_global_index_type("IVF-SQ"), Some("ivf-sq"));
+ assert_eq!(normalize_global_index_type("IVF-RQ"), Some("ivf-rq"));
+ assert_eq!(normalize_global_index_type("DiskANN"), Some("diskann"));
}
#[test]
fn unsupported_types_return_none() {
- assert_eq!(normalize_global_index_type_for_drop("hash"), None);
- assert_eq!(normalize_global_index_type_for_drop("ivf-hnsw-flat"),
None);
- assert_eq!(normalize_global_index_type_for_drop("ivf-hnsw-sq"), None);
- assert_eq!(normalize_global_index_type_for_drop(""), None);
+ assert_eq!(normalize_global_index_type("hash"), None);
+ assert_eq!(normalize_global_index_type("ivf-hnsw-flat"), None);
+ assert_eq!(normalize_global_index_type("ivf-hnsw-sq"), None);
+ assert_eq!(normalize_global_index_type(""), None);
}
}
diff --git a/crates/paimon/src/table/lumina_index_build_builder.rs
b/crates/paimon/src/table/lumina_index_build_builder.rs
index e50a7311..b96f43d9 100644
--- a/crates/paimon/src/table/lumina_index_build_builder.rs
+++ b/crates/paimon/src/table/lumina_index_build_builder.rs
@@ -28,19 +28,17 @@ use validation::{
};
use writer::abort_on_build_error;
-use crate::lumina::{
- is_lumina_index_type, LuminaIndexMeta, LuminaVectorIndexOptions,
LUMINA_IDENTIFIER,
-};
-use crate::spec::CoreOptions;
-use crate::table::{CommitMessage, RowRange, SnapshotManager, Table,
TableCommit};
+use crate::lumina::{LuminaIndexMeta, LuminaVectorIndexOptions,
LUMINA_IDENTIFIER};
+use crate::spec::{CoreOptions, Predicate};
+use crate::table::{CommitMessage, RowRange, Table, TableCommit};
use crate::{Error, Result};
use std::collections::HashMap;
-pub struct LuminaIndexBuildBuilder<'a> {
+pub(super) struct LuminaIndexBuildBuilder<'a> {
table: &'a Table,
index_column: Option<String>,
- index_type: String,
options: HashMap<String, String>,
+ pub(super) partition_filter: Option<Predicate>,
}
impl<'a> LuminaIndexBuildBuilder<'a> {
@@ -48,8 +46,8 @@ impl<'a> LuminaIndexBuildBuilder<'a> {
Self {
table,
index_column: None,
- index_type: LUMINA_IDENTIFIER.to_string(),
options: HashMap::new(),
+ partition_filter: None,
}
}
@@ -58,29 +56,17 @@ impl<'a> LuminaIndexBuildBuilder<'a> {
self
}
- pub fn with_index_type(&mut self, index_type: &str) -> &mut Self {
- self.index_type = index_type.to_string();
- self
- }
-
pub fn with_options(&mut self, options: HashMap<String, String>) -> &mut
Self {
self.options = options;
self
}
- pub async fn execute(&self) -> Result<usize> {
+ pub(super) async fn prepare(&self) -> Result<(Option<i64>,
Vec<CommitMessage>)> {
// Building the index scans the table's rows.
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
self.table.ensure_not_branch_reference_for_write()?;
- if !is_lumina_index_type(&self.index_type) {
- return Err(Error::DataInvalid {
- message: format!("Unsupported Lumina index type: {}",
self.index_type),
- source: None,
- });
- }
-
let index_column = self
.index_column
.as_deref()
@@ -89,7 +75,9 @@ impl<'a> LuminaIndexBuildBuilder<'a> {
source: None,
})?;
- let core_options = CoreOptions::new(self.table.schema().options());
+ let mut merged_options = self.table.schema().options().clone();
+ merged_options.extend(self.options.clone());
+ let core_options = CoreOptions::new(&merged_options);
validate_table_options(self.table, &core_options)?;
let rows_per_shard = core_options.global_index_row_count_per_shard()?;
@@ -103,21 +91,16 @@ impl<'a> LuminaIndexBuildBuilder<'a> {
let dimension = lumina_options.dimension;
let index_meta =
LuminaIndexMeta::new(lumina_options.to_lumina_options()).serialize()?;
- let snapshot_manager = SnapshotManager::new(
- self.table.file_io().clone(),
- self.table.location().to_string(),
- );
- let snapshot = snapshot_manager
- .get_latest_snapshot()
- .await?
- .ok_or_else(|| Error::DataInvalid {
- message: "Cannot build Lumina index without a
snapshot".to_string(),
- source: None,
- })?;
+ let snapshot_manager = self.table.snapshot_manager();
+ let Some(snapshot) = snapshot_manager.get_latest_snapshot().await?
else {
+ return Ok((None, vec![]));
+ };
- let manifest_entries = self
- .table
- .new_read_builder()
+ let mut read_builder = self.table.new_read_builder();
+ if let Some(filter) = &self.partition_filter {
+ read_builder.with_filter(filter.clone());
+ }
+ let manifest_entries = read_builder
.new_scan()
.with_scan_all_files()
.plan_manifest_entries(&snapshot)
@@ -141,7 +124,7 @@ impl<'a> LuminaIndexBuildBuilder<'a> {
&indexed,
)?;
if shards.is_empty() {
- return Ok(0);
+ return Ok((Some(snapshot.id()), vec![]));
}
crate::table::global_index_build_common::validate_existing_index_overlap(
@@ -186,11 +169,7 @@ impl<'a> LuminaIndexBuildBuilder<'a> {
messages.push(message);
}
- commit
- .commit_if_latest_snapshot(messages, snapshot.id())
- .await?;
-
- Ok(shard_count)
+ Ok((Some(snapshot.id()), messages))
}
}
diff --git a/crates/paimon/src/table/lumina_index_build_builder/tests.rs
b/crates/paimon/src/table/lumina_index_build_builder/tests.rs
index b4e59878..43e5a93a 100644
--- a/crates/paimon/src/table/lumina_index_build_builder/tests.rs
+++ b/crates/paimon/src/table/lumina_index_build_builder/tests.rs
@@ -393,7 +393,8 @@ async fn test_execute_rejects_primary_key_table() {
);
let err = table
- .new_lumina_index_build_builder()
+ .new_global_index_build_builder()
+ .with_index_type("lumina")
.with_index_column("embedding")
.execute()
.await
@@ -411,7 +412,8 @@ async fn test_execute_rejects_deletion_vectors_table() {
let table = test_table(options);
let err = table
- .new_lumina_index_build_builder()
+ .new_global_index_build_builder()
+ .with_index_type("lumina")
.with_index_column("embedding")
.execute()
.await
@@ -867,7 +869,8 @@ async fn test_execute_writes_lumina_index_manifest() {
.unwrap();
let shard_count = table
- .new_lumina_index_build_builder()
+ .new_global_index_build_builder()
+ .with_index_type("lumina")
.with_index_column("embedding")
.execute()
.await
@@ -1059,7 +1062,8 @@ async fn lumina_second_build_without_new_data_is_noop() {
assert!(!names_before.is_empty());
let built = table
- .new_lumina_index_build_builder()
+ .new_global_index_build_builder()
+ .with_index_type("lumina")
.with_index_column("embedding")
.execute()
.await
@@ -1167,7 +1171,8 @@ async fn lumina_incremental_build_indexes_only_new_rows()
{
// error. Without the native Lumina library the gap build surfaces a
// library-load error instead; with it present it succeeds.
let result = table
- .new_lumina_index_build_builder()
+ .new_global_index_build_builder()
+ .with_index_type("lumina")
.with_index_column("embedding")
.execute()
.await;
diff --git a/crates/paimon/src/table/lumina_index_build_builder/writer.rs
b/crates/paimon/src/table/lumina_index_build_builder/writer.rs
index df661f26..8ad84596 100644
--- a/crates/paimon/src/table/lumina_index_build_builder/writer.rs
+++ b/crates/paimon/src/table/lumina_index_build_builder/writer.rs
@@ -28,8 +28,6 @@ use crate::table::{CommitMessage, TableCommit};
use crate::{Error, Result};
use std::path::PathBuf;
-const INDEX_DIR: &str = "index";
-
impl LuminaIndexBuildBuilder<'_> {
pub(super) async fn build_index_file(
&self,
@@ -61,18 +59,11 @@ impl LuminaIndexBuildBuilder<'_> {
builder.dump(&temp_path_str)?;
let file_name = format!("lumina-global-index-{}.index",
uuid::Uuid::new_v4());
- self.table
- .file_io()
- .mkdirs(&format!(
- "{}/{INDEX_DIR}/",
- self.table.location().trim_end_matches('/')
- ))
+ let (index_path, external_path) =
+ crate::table::global_index_build_common::prepare_index_file_path(
+ self.table, &file_name,
+ )
.await?;
- let index_path = format!(
- "{}/{INDEX_DIR}/{}",
- self.table.location().trim_end_matches('/'),
- file_name
- );
let write_result: Result<i64> = async {
copy_local_file_to_output(&temp_path,
self.table.file_io().new_output(&index_path)?)
.await?;
@@ -97,13 +88,15 @@ impl LuminaIndexBuildBuilder<'_> {
file_size,
row_count,
deletion_vectors_ranges: None,
- external_path: None,
+ external_path,
global_index_meta: Some(GlobalIndexMeta {
row_range_start: shard.row_range_start,
row_range_end: shard.row_range_end,
index_field_id,
extra_field_ids: None,
- source_meta: None,
+ source_meta: Some(
+
crate::spec::DataEvolutionIndexSourceMeta::new(shard.snapshot_id)?.serialize(),
+ ),
index_meta: Some(index_meta),
}),
})
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 5c577881..3877bd6f 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -74,6 +74,7 @@ mod full_text_index_adapter;
mod full_text_index_build_builder;
#[cfg(feature = "fulltext")]
mod full_text_search_builder;
+mod global_index_build_builder;
pub(crate) mod global_index_build_common;
mod global_index_drop_builder;
pub(crate) mod global_index_scanner;
@@ -188,21 +189,17 @@ pub use format_partition::{
pub use format_partition_stats::FormatTablePartitionStatsCollector;
pub use format_partition_truncate::FormatTableTruncator;
#[cfg(feature = "fulltext")]
-pub use full_text_index_build_builder::FullTextIndexBuildBuilder;
-#[cfg(feature = "fulltext")]
pub use full_text_search_builder::FullTextSearchBuilder;
use futures::stream::BoxStream;
+pub use global_index_build_builder::GlobalIndexBuildBuilder;
pub use global_index_drop_builder::GlobalIndexDropBuilder;
-pub use global_index_types::{
- normalize_global_index_type_for_drop,
SUPPORTED_GLOBAL_INDEX_TYPES_FOR_DROP,
-};
+pub use global_index_types::{normalize_global_index_type,
SUPPORTED_GLOBAL_INDEX_TYPES_FOR_DROP};
pub use hybrid_search_builder::{
HybridSearchBuilder, HybridSearchRanker, HybridSearchRoute,
HybridSearchRouteKind,
};
pub use incremental_scan::{
IncrementalPlan, IncrementalScan, IncrementalScanMode, IncrementalSplit,
};
-pub use lumina_index_build_builder::LuminaIndexBuildBuilder;
pub use object_table::{ObjectEntry, ObjectTable};
pub use partition_row_count::PartitionRowCount;
pub use partition_stat::PartitionStat;
@@ -218,9 +215,6 @@ pub use scan_trace::ScanTrace;
pub use schema_manager::SchemaManager;
pub use snapshot_commit::{RESTSnapshotCommit, RenamingSnapshotCommit,
SnapshotCommit};
pub use snapshot_manager::SnapshotManager;
-pub use sorted_global_index_build_builder::{
- BTreeGlobalIndexBuildBuilder, SortedGlobalIndexBuildBuilder,
-};
pub use source::{
merge_row_ranges, DataSplit, DataSplitBuilder, DeletionFile,
PartitionBucket, Plan, RowRange,
};
@@ -239,7 +233,6 @@ pub use update_assignment::UpdateAssignment;
pub use vector_read::{BatchVectorRead, VectorRead};
pub use vector_scan::{VectorScan, VectorScanPlan};
pub use vector_search_builder::VectorSearchBuilder;
-pub use vindex_index_build_builder::VindexIndexBuildBuilder;
pub use write_builder::WriteBuilder;
use crate::catalog::{validate_branch_name, Identifier, DEFAULT_MAIN_BRANCH};
@@ -555,34 +548,15 @@ impl Table {
BatchVectorSearchBuilder::new(self)
}
- /// Create a builder for `full-text` global index files.
- ///
- /// Reference:
[NativeFullTextGlobalIndexWriter](https://github.com/apache/paimon/blob/master/paimon-full-text/src/main/java/org/apache/paimon/fulltext/index/NativeFullTextGlobalIndexWriter.java)
- #[cfg(feature = "fulltext")]
- pub fn new_full_text_index_build_builder(&self) ->
FullTextIndexBuildBuilder<'_> {
- FullTextIndexBuildBuilder::new(self)
- }
-
- pub fn new_lumina_index_build_builder(&self) ->
LuminaIndexBuildBuilder<'_> {
- LuminaIndexBuildBuilder::new(self)
- }
-
- pub fn new_sorted_global_index_build_builder(&self) ->
SortedGlobalIndexBuildBuilder<'_> {
- SortedGlobalIndexBuildBuilder::new(self)
- }
-
- pub fn new_btree_global_index_build_builder(&self) ->
BTreeGlobalIndexBuildBuilder<'_> {
- self.new_sorted_global_index_build_builder()
+ /// Build global index files selected by index type, without publishing
them.
+ pub fn new_global_index_build_builder(&self) ->
GlobalIndexBuildBuilder<'_> {
+ GlobalIndexBuildBuilder::new(self)
}
pub fn new_global_index_drop_builder(&self) -> GlobalIndexDropBuilder<'_> {
GlobalIndexDropBuilder::new(self)
}
- pub fn new_vindex_index_build_builder(&self, index_type: &str) ->
VindexIndexBuildBuilder<'_> {
- VindexIndexBuildBuilder::new(self, index_type)
- }
-
/// Create a write builder for write/commit.
///
/// Reference: [pypaimon
FileStoreTable.new_write_builder](https://github.com/apache/paimon/blob/master/paimon-python/pypaimon/table/file_store_table.py).
diff --git a/crates/paimon/src/table/sorted_global_index_build_builder.rs
b/crates/paimon/src/table/sorted_global_index_build_builder.rs
index 7b8e08e0..1c116235 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder.rs
@@ -63,7 +63,7 @@ fn make_index_key_codec(index_type: &str, data_type:
&DataType) -> (KeyComparato
}
}
-pub struct SortedGlobalIndexBuildBuilder<'a> {
+pub(super) struct SortedGlobalIndexBuildBuilder<'a> {
table: &'a Table,
index_column: Option<IndexColumns>,
index_type: String,
@@ -71,10 +71,6 @@ pub struct SortedGlobalIndexBuildBuilder<'a> {
partition_filter: Option<Predicate>,
}
-/// Backward-compatible name retained for callers that used the original
-/// BTree-only builder API before it also supported bitmap and multivalue.
-pub type BTreeGlobalIndexBuildBuilder<'a> = SortedGlobalIndexBuildBuilder<'a>;
-
impl<'a> SortedGlobalIndexBuildBuilder<'a> {
pub(crate) fn new(table: &'a Table) -> Self {
Self {
@@ -86,11 +82,6 @@ impl<'a> SortedGlobalIndexBuildBuilder<'a> {
}
}
- pub fn with_index_column(&mut self, column: &str) -> &mut Self {
- self.index_column = Some(IndexColumns::Column(column.to_string()));
- self
- }
-
/// Select BTree columns in physical tuple-key order.
pub fn with_index_columns(&mut self, columns: &[&str]) -> &mut Self {
self.index_column = Some(IndexColumns::Columns(
@@ -120,32 +111,6 @@ impl<'a> SortedGlobalIndexBuildBuilder<'a> {
Ok(self)
}
- /// Build index files without publishing a snapshot, like Java's index
builders.
- /// The caller owns the returned messages and decides whether to commit or
abort.
- /// A build with no snapshot or uncovered rows returns an empty list.
- pub async fn build(&self) -> Result<Vec<CommitMessage>> {
- self.prepare().await.map(|(_, messages)| messages)
- }
-
- /// Build and commit, rejecting a concurrent change to the planned
snapshot.
- pub async fn execute(&self) -> Result<usize> {
- let (snapshot_id, messages) = self.prepare().await?;
- if messages.is_empty() {
- return Ok(0);
- }
- let count = messages
- .iter()
- .map(|message| message.new_index_files.len())
- .sum();
- self.new_commit()
- .commit_if_latest_snapshot(
- messages,
- snapshot_id.expect("nonempty build has a snapshot"),
- )
- .await?;
- Ok(count)
- }
-
fn new_commit(&self) -> TableCommit {
TableCommit::new(
self.table.clone(),
@@ -157,7 +122,7 @@ impl<'a> SortedGlobalIndexBuildBuilder<'a> {
)
}
- async fn prepare(&self) -> Result<(Option<i64>, Vec<CommitMessage>)> {
+ pub(super) async fn prepare(&self) -> Result<(Option<i64>,
Vec<CommitMessage>)> {
// Building the index scans the table's rows.
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
diff --git
a/crates/paimon/src/table/sorted_global_index_build_builder/preparation_tests.rs
b/crates/paimon/src/table/sorted_global_index_build_builder/preparation_tests.rs
index b23a7a49..c50ebc2e 100644
---
a/crates/paimon/src/table/sorted_global_index_build_builder/preparation_tests.rs
+++
b/crates/paimon/src/table/sorted_global_index_build_builder/preparation_tests.rs
@@ -144,7 +144,7 @@ async fn
staged_indexes_reject_target_column_changes_and_preserve_files() {
vec![None; 2],
)
.await;
- let mut builder = table.new_sorted_global_index_build_builder();
+ let mut builder = table.new_global_index_build_builder();
builder.with_index_column("name").with_index_type(kind);
let mut messages = builder.build().await.unwrap();
let file = messages[0].new_index_files[0].clone();
@@ -202,7 +202,7 @@ async fn
staged_indexes_reject_deleted_partial_columns_and_removed_source_ranges
let updates = update_name(&table, "changed").await;
let commit = TableCommit::new(table.clone(), "update".into());
commit.commit(updates.clone()).await.unwrap();
- let mut builder = table.new_sorted_global_index_build_builder();
+ let mut builder = table.new_global_index_build_builder();
builder.with_index_column("name").with_index_type(kind);
let messages = builder.build().await.unwrap();
if remove_all {
@@ -242,7 +242,7 @@ async fn
staged_index_survives_append_and_unrelated_column_update() {
vec![None; 2],
)
.await;
- let mut builder = table.new_sorted_global_index_build_builder();
+ let mut builder = table.new_global_index_build_builder();
builder.with_index_column("name");
let messages = builder.build().await.unwrap();
assert_eq!(
@@ -297,7 +297,7 @@ async fn
equivalent_empty_partition_encodings_share_source_checks() {
for changed in [false, true] {
let table = table(false, false);
append(&table, vec![1], vec![Some("a")], vec![None]).await;
- let mut builder = table.new_sorted_global_index_build_builder();
+ let mut builder = table.new_global_index_build_builder();
builder.with_index_column("name");
let mut messages = builder.build().await.unwrap();
for message in &mut messages {
@@ -343,7 +343,7 @@ async fn
prepared_files_are_invisible_until_explicit_commit() {
.await
.unwrap()
.unwrap();
- let mut builder = table.new_sorted_global_index_build_builder();
+ let mut builder = table.new_global_index_build_builder();
builder.with_index_column("name").with_index_type(kind);
let messages = builder.build().await.unwrap();
assert_eq!(messages.len(), 2);
@@ -410,7 +410,7 @@ async fn
caller_can_abort_prepared_files_without_changing_snapshot() {
for kind in ["btree", "bitmap"] {
let table = table(false, true);
append(&table, vec![1, 2], vec![Some("a"), None], vec![None; 2]).await;
- let mut builder = table.new_sorted_global_index_build_builder();
+ let mut builder = table.new_global_index_build_builder();
builder.with_index_column("name").with_index_type(kind);
let messages = builder.build().await.unwrap();
let path = messages[0].new_index_files[0]
@@ -474,7 +474,7 @@ async fn
partition_builds_only_cover_selected_partitions_and_new_rows() {
.await;
let predicates = PredicateBuilder::new(table.schema().fields());
let selected = predicates.equal("pt", Datum::Int(1)).unwrap();
- let mut builder = table.new_sorted_global_index_build_builder();
+ let mut builder = table.new_global_index_build_builder();
builder
.with_index_column("id")
.with_index_type(kind)
@@ -503,7 +503,7 @@ async fn
partition_builds_only_cover_selected_partitions_and_new_rows() {
assert_eq!(builder.execute().await.unwrap(), 1);
assert!(builder.build().await.unwrap().is_empty());
- let mut all = table.new_sorted_global_index_build_builder();
+ let mut all = table.new_global_index_build_builder();
all.with_index_column("id").with_index_type(kind);
let remaining = all.build().await.unwrap();
assert_eq!(
@@ -526,7 +526,7 @@ async fn
partition_builds_only_cover_selected_partitions_and_new_rows() {
#[tokio::test]
async fn empty_and_disjoint_partition_builds_publish_nothing() {
let table = table(true, false);
- let mut builder = table.new_sorted_global_index_build_builder();
+ let mut builder = table.new_global_index_build_builder();
builder.with_index_column("id");
assert!(builder.build().await.unwrap().is_empty());
assert_eq!(builder.execute().await.unwrap(), 0);
@@ -562,7 +562,7 @@ async fn
empty_and_disjoint_partition_builds_publish_nothing() {
.is_null("pt")
.unwrap();
assert!(unpartitioned
- .new_sorted_global_index_build_builder()
+ .new_global_index_build_builder()
.with_partition_filter(filter)
.is_err());
}
diff --git a/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
b/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
index 61122928..8f9008a5 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder/tests.rs
@@ -720,7 +720,7 @@ async fn assert_btree_build_version(version: u32) {
.unwrap();
let shard_count = table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("name")
.execute()
.await
@@ -932,7 +932,7 @@ async fn column_group_btree_table(rows: i32, version: u32)
-> Table {
.any(|file| file.write_cols.as_deref() == Some(&["payload".into()])));
for column in ["id", "name"] {
table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column(column)
.execute()
.await
@@ -1081,7 +1081,7 @@ async fn test_execute_writes_and_queries_fm_index() {
assert_eq!(
table
- .new_sorted_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("name")
.with_index_type(FM_GLOBAL_INDEX_TYPE)
.execute()
@@ -1192,7 +1192,7 @@ async fn
test_failed_fm_build_removes_partial_index_file() {
options.insert("fm-index.partition-row-count".to_string(),
"1".to_string());
options.insert("fm-index.compression".to_string(), "none".to_string());
let error = table
- .new_sorted_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("name")
.with_index_type(FM_GLOBAL_INDEX_TYPE)
.with_options(options)
@@ -1244,7 +1244,7 @@ async fn test_global_index_prunes_during_manifest_read() {
}
table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("name")
.execute()
.await
@@ -1297,7 +1297,7 @@ async fn
test_scalar_full_search_includes_unindexed_rows() {
.await
.unwrap();
table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("name")
.execute()
.await
@@ -1365,7 +1365,7 @@ async fn
test_empty_global_index_ranges_skip_legacy_manifests() {
.await
.unwrap();
table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("name")
.execute()
.await
@@ -1441,7 +1441,7 @@ async fn
test_detail_mode_defers_manifest_pruning_for_unindexed_ranges() {
.await
.unwrap();
table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("name")
.execute()
.await
@@ -1504,7 +1504,7 @@ async fn
test_execute_writes_bitmap_index_manifest_and_java_file() {
.unwrap();
let shard_count = table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("name")
.with_index_type(BITMAP_GLOBAL_INDEX_TYPE)
.execute()
@@ -1606,7 +1606,7 @@ async fn
test_execute_multivalue_index_and_array_queries_end_to_end() {
.unwrap();
let shard_count = table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("items")
.with_index_type(MULTIVALUE_GLOBAL_INDEX_TYPE)
.execute()
@@ -1778,7 +1778,7 @@ async fn
test_bitmap_floating_candidates_preserve_residual_results() {
for column in ["f", "d"] {
let shard_count = table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column(column)
.with_index_type(BITMAP_GLOBAL_INDEX_TYPE)
.execute()
@@ -1931,7 +1931,7 @@ async fn bitmap_second_build_without_new_data_is_noop() {
.unwrap();
let first_built = table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("name")
.with_index_type(BITMAP_GLOBAL_INDEX_TYPE)
.execute()
@@ -1946,7 +1946,7 @@ async fn bitmap_second_build_without_new_data_is_noop() {
assert!(!files_after_first.is_empty());
let built = table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("name")
.with_index_type(BITMAP_GLOBAL_INDEX_TYPE)
.execute()
@@ -1992,7 +1992,7 @@ async fn
bitmap_build_after_btree_on_same_field_still_indexes() {
.unwrap();
let btree_built = table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("name")
.execute()
.await
@@ -2000,7 +2000,7 @@ async fn
bitmap_build_after_btree_on_same_field_still_indexes() {
assert!(btree_built > 0);
let bitmap_built = table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("name")
.with_index_type(BITMAP_GLOBAL_INDEX_TYPE)
.execute()
@@ -2050,7 +2050,7 @@ async fn bitmap_incremental_build_indexes_only_new_rows()
{
.unwrap();
let first_built = table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("name")
.with_index_type(BITMAP_GLOBAL_INDEX_TYPE)
.execute()
@@ -2076,7 +2076,7 @@ async fn bitmap_incremental_build_indexes_only_new_rows()
{
.unwrap();
let second_built = table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("name")
.with_index_type(BITMAP_GLOBAL_INDEX_TYPE)
.execute()
@@ -2215,7 +2215,7 @@ async fn second_build_without_new_data_is_noop() {
.unwrap();
let first_built = table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("name")
.execute()
.await
@@ -2226,7 +2226,7 @@ async fn second_build_without_new_data_is_noop() {
assert!(!files_after_first.is_empty());
let built = table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("name")
.execute()
.await
@@ -2269,7 +2269,7 @@ async fn incremental_build_indexes_only_new_rows() {
.unwrap();
let first_built = table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("name")
.execute()
.await
@@ -2296,7 +2296,7 @@ async fn incremental_build_indexes_only_new_rows() {
.unwrap();
let second_built = table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("name")
.execute()
.await
@@ -2356,7 +2356,7 @@ async fn first_build_indexes_full_coverage() {
.unwrap();
let built = table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("name")
.execute()
.await
@@ -2401,7 +2401,7 @@ async fn
incremental_build_splits_gap_across_records_per_range_grid() {
.unwrap();
let first_built = table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("name")
.execute()
.await
@@ -2457,7 +2457,7 @@ async fn
incremental_build_splits_gap_across_records_per_range_grid() {
);
let second_built = table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("name")
.execute()
.await
@@ -2578,7 +2578,7 @@ async fn
incremental_build_splits_gap_around_mid_coverage_indexed_hole() {
// Build: gap = coverage minus the hole = [0, hole_start-1] and
// [hole_end+1, last_row]; two shards since the grid does not split here.
let built = table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("name")
.execute()
.await
@@ -2673,7 +2673,7 @@ async fn
test_composite_btree_build_query_coverage_and_drop() {
.unwrap();
assert_eq!(
table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_columns(&["name", "id"])
.execute()
.await
@@ -2682,7 +2682,7 @@ async fn
test_composite_btree_build_query_coverage_and_drop() {
);
assert_eq!(
table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column(" name , id ")
.execute()
.await
@@ -2800,7 +2800,7 @@ async fn
test_composite_btree_build_query_coverage_and_drop() {
assert_eq!(scan_ids(&table, point).await, vec![1, 1]);
assert_eq!(
table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_columns(&["name", "id"])
.execute()
.await
@@ -2810,7 +2810,7 @@ async fn
test_composite_btree_build_query_coverage_and_drop() {
// Independent scalar and reversed tuple definitions coexist.
assert!(
table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("name")
.execute()
.await
@@ -2819,7 +2819,7 @@ async fn
test_composite_btree_build_query_coverage_and_drop() {
);
assert!(
table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_columns(&["id", "name"])
.execute()
.await
@@ -2872,7 +2872,7 @@ async fn test_composite_btree_column_validation() {
let table = test_table(table_options("10"));
for columns in ["", "name,", "name,name", "name,missing"] {
assert!(table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column(columns)
.execute()
.await
@@ -2887,7 +2887,7 @@ async fn test_composite_btree_column_validation() {
for index_type in ["bitmap", "multivalue", "fm"] {
assert!(matches!(
table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_columns(&["name", "id"])
.with_index_type(index_type)
.execute()
@@ -2920,7 +2920,7 @@ async fn
test_composite_btree_all_null_tuple_and_residual_filter() {
.await
.unwrap();
table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_columns(&["name", "id"])
.execute()
.await
@@ -3059,7 +3059,7 @@ async fn
test_composite_btree_literal_precision_preserves_scan_results() {
}
for columns in [["d", "id"], ["ts", "id"]] {
table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_columns(&columns)
.execute()
.await
@@ -3099,7 +3099,7 @@ async fn
test_composite_btree_literal_precision_preserves_scan_results() {
for index_type in ["btree", "bitmap"] {
for column in ["d", "ts"] {
table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column(column)
.with_index_type(index_type)
.execute()
@@ -3162,14 +3162,14 @@ async fn
test_composite_btree_intersects_independent_indexes() {
.await
.unwrap();
table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_type("bitmap")
.with_index_column("flag")
.execute()
.await
.unwrap();
table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_columns(&["name", "id"])
.execute()
.await
@@ -3204,7 +3204,7 @@ async fn
test_composite_btree_intersects_independent_indexes() {
.await
.unwrap();
table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_columns(&["flag", "id"])
.execute()
.await
@@ -3250,12 +3250,12 @@ async fn
test_composite_btree_combines_different_index_coverage() {
.await
.unwrap();
table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_columns(&["name", "id"])
.execute()
.await
.unwrap();
- let mut independent = table.new_btree_global_index_build_builder();
+ let mut independent = table.new_global_index_build_builder();
if independent_type == "bitmap" {
independent
.with_index_type("bitmap")
@@ -3404,7 +3404,7 @@ async fn
test_composite_btree_preserves_exact_column_names() {
.unwrap();
assert_eq!(
table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_columns(&["a,b", " c "])
.execute()
.await
@@ -3413,7 +3413,7 @@ async fn
test_composite_btree_preserves_exact_column_names() {
);
assert_eq!(
table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("a,b")
.execute()
.await
@@ -3422,7 +3422,7 @@ async fn
test_composite_btree_preserves_exact_column_names() {
);
assert_eq!(
table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column(" c ")
.execute()
.await
@@ -3472,7 +3472,7 @@ async fn
test_composite_btree_preserves_exact_column_names() {
1
);
assert!(table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_columns(&[])
.execute()
.await
diff --git a/crates/paimon/src/table/vector_search_test_utils.rs
b/crates/paimon/src/table/vector_search_test_utils.rs
index 65c41c5e..257a1897 100644
--- a/crates/paimon/src/table/vector_search_test_utils.rs
+++ b/crates/paimon/src/table/vector_search_test_utils.rs
@@ -187,14 +187,15 @@ pub(super) async fn de_vector_table() -> Table {
.unwrap();
let built = table
- .new_vindex_index_build_builder(IVF_FLAT_IDENTIFIER)
+ .new_global_index_build_builder()
+ .with_index_type(IVF_FLAT_IDENTIFIER)
.with_index_column("embedding")
.execute()
.await
.unwrap();
assert!(built > 0, "DE fixture must build a global vector index");
let built = table
- .new_sorted_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("id")
.with_index_type("btree")
.execute()
diff --git a/crates/paimon/src/table/vindex_index_build_builder.rs
b/crates/paimon/src/table/vindex_index_build_builder.rs
index e63f3658..ca049b19 100644
--- a/crates/paimon/src/table/vindex_index_build_builder.rs
+++ b/crates/paimon/src/table/vindex_index_build_builder.rs
@@ -18,26 +18,25 @@
mod extraction;
mod pipeline;
mod planning;
-mod timing;
+pub(super) mod timing;
mod validation;
mod writer;
use planning::plan_vindex_shards;
-use timing::vector_index_build_timing_enabled;
use validation::{checked_i32, find_index_field, validate_table_options,
validate_vector_field};
-use crate::spec::CoreOptions;
-use crate::table::{CommitMessage, RowRange, SnapshotManager, Table,
TableCommit};
+use crate::spec::{CoreOptions, Predicate};
+use crate::table::{CommitMessage, RowRange, Table, TableCommit};
use crate::vindex::{is_vindex_index_type, VindexVectorIndexOptions};
use crate::{Error, Result};
use std::collections::HashMap;
-use std::time::Instant;
-pub struct VindexIndexBuildBuilder<'a> {
+pub(super) struct VindexIndexBuildBuilder<'a> {
table: &'a Table,
index_column: Option<String>,
index_type: String,
options: HashMap<String, String>,
+ pub(super) partition_filter: Option<Predicate>,
}
impl<'a> VindexIndexBuildBuilder<'a> {
@@ -47,6 +46,7 @@ impl<'a> VindexIndexBuildBuilder<'a> {
index_column: None,
index_type: index_type.to_string(),
options: HashMap::new(),
+ partition_filter: None,
}
}
@@ -60,7 +60,13 @@ impl<'a> VindexIndexBuildBuilder<'a> {
self
}
- pub async fn execute(&self) -> Result<usize> {
+ pub(super) async fn prepare(
+ &self,
+ ) -> Result<(
+ Option<i64>,
+ Vec<CommitMessage>,
+ Vec<timing::VectorIndexBuildTiming>,
+ )> {
// Building the index scans the table's rows.
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
@@ -81,15 +87,20 @@ impl<'a> VindexIndexBuildBuilder<'a> {
source: None,
})?;
- let core_options = CoreOptions::new(self.table.schema().options());
+ let mut merged_options = self.table.schema().options().clone();
+ merged_options.extend(self.options.clone());
+ let core_options = CoreOptions::new(&merged_options);
validate_table_options(self.table, &core_options)?;
let rows_per_shard = core_options.global_index_row_count_per_shard()?;
let index_field = find_index_field(self.table, index_column)?;
validate_vector_field(index_field)?;
+ // The builder consumes shard sizing; it is not a native writer option.
+ let mut writer_options = self.options.clone();
+ writer_options.remove("global-index.row-count-per-shard");
let vindex_options = VindexVectorIndexOptions::new(
self.table.schema().options(),
- &self.options,
+ &writer_options,
&self.index_type,
index_field,
)?;
@@ -103,21 +114,16 @@ impl<'a> VindexIndexBuildBuilder<'a> {
source: Some(Box::new(e)),
})?;
- let snapshot_manager = SnapshotManager::new(
- self.table.file_io().clone(),
- self.table.location().to_string(),
- );
- let snapshot = snapshot_manager
- .get_latest_snapshot()
- .await?
- .ok_or_else(|| Error::DataInvalid {
- message: "Cannot build vindex index without a
snapshot".to_string(),
- source: None,
- })?;
+ let snapshot_manager = self.table.snapshot_manager();
+ let Some(snapshot) = snapshot_manager.get_latest_snapshot().await?
else {
+ return Ok((None, vec![], vec![]));
+ };
- let manifest_entries = self
- .table
- .new_read_builder()
+ let mut read_builder = self.table.new_read_builder();
+ if let Some(filter) = &self.partition_filter {
+ read_builder.with_filter(filter.clone());
+ }
+ let manifest_entries = read_builder
.new_scan()
.with_scan_all_files()
.plan_manifest_entries(&snapshot)
@@ -141,7 +147,7 @@ impl<'a> VindexIndexBuildBuilder<'a> {
&indexed,
)?;
if shards.is_empty() {
- return Ok(0);
+ return Ok((Some(snapshot.id()), vec![], vec![]));
}
crate::table::global_index_build_common::validate_existing_index_overlap(
@@ -196,18 +202,7 @@ impl<'a> VindexIndexBuildBuilder<'a> {
}
}
- let commit_start =
vector_index_build_timing_enabled().then(Instant::now);
- commit
- .commit_if_latest_snapshot(messages, snapshot.id())
- .await?;
- if let Some(commit_start) = commit_start {
- let commit = commit_start.elapsed();
- for timing in timings {
- timing.log(&self.index_type, commit);
- }
- }
-
- Ok(shard_count)
+ Ok((Some(snapshot.id()), messages, timings))
}
}
diff --git a/crates/paimon/src/table/vindex_index_build_builder/tests.rs
b/crates/paimon/src/table/vindex_index_build_builder/tests.rs
index fb5c1499..97fad54d 100644
--- a/crates/paimon/src/table/vindex_index_build_builder/tests.rs
+++ b/crates/paimon/src/table/vindex_index_build_builder/tests.rs
@@ -594,7 +594,8 @@ async fn vindex_second_build_without_new_data_is_noop() {
assert!(!names_before.is_empty());
let built = table
- .new_vindex_index_build_builder(IVF_FLAT_IDENTIFIER)
+ .new_global_index_build_builder()
+ .with_index_type(IVF_FLAT_IDENTIFIER)
.with_index_column("embedding")
.execute()
.await
@@ -634,7 +635,8 @@ async fn vindex_incremental_build_indexes_only_new_rows() {
)
.await;
let first_built = table
- .new_vindex_index_build_builder(IVF_FLAT_IDENTIFIER)
+ .new_global_index_build_builder()
+ .with_index_type(IVF_FLAT_IDENTIFIER)
.with_index_column("embedding")
.with_options(HashMap::from([(
"ivf-flat.train.sample-ratio".to_string(),
@@ -667,7 +669,8 @@ async fn vindex_incremental_build_indexes_only_new_rows() {
// End-to-end: build #2 must SUCCEED and index the appended rows.
let second_built = table
- .new_vindex_index_build_builder(IVF_FLAT_IDENTIFIER)
+ .new_global_index_build_builder()
+ .with_index_type(IVF_FLAT_IDENTIFIER)
.with_index_column("embedding")
.with_options(HashMap::from([(
"vindex.build.granule.enabled".to_string(),
@@ -747,7 +750,8 @@ async fn
vindex_small_training_sample_preserves_tail_cluster_recall() {
.await;
assert_eq!(
table
- .new_vindex_index_build_builder(crate::vindex::IVF_SQ_IDENTIFIER)
+ .new_global_index_build_builder()
+ .with_index_type(crate::vindex::IVF_SQ_IDENTIFIER)
.with_index_column("embedding")
.execute()
.await
@@ -823,7 +827,8 @@ async fn
vindex_granule_training_sees_file_periodic_cluster() {
assert_eq!(
table
- .new_vindex_index_build_builder(crate::vindex::IVF_SQ_IDENTIFIER)
+ .new_global_index_build_builder()
+ .with_index_type(crate::vindex::IVF_SQ_IDENTIFIER)
.with_index_column("embedding")
.execute()
.await
@@ -967,7 +972,8 @@ async fn
vindex_granule_training_sees_one_oversized_row_group() {
assert_eq!(physical_granules.len(), 1);
assert_eq!(physical_granules[0].row_count, LARGE_ROWS as i64);
- let mut builder =
table.new_vindex_index_build_builder(crate::vindex::IVF_SQ_IDENTIFIER);
+ let mut builder =
+ super::VindexIndexBuildBuilder::new(&table,
crate::vindex::IVF_SQ_IDENTIFIER);
builder
.with_index_column("embedding")
.with_options(HashMap::from([(
@@ -996,7 +1002,20 @@ async fn
vindex_granule_training_sees_one_oversized_row_group() {
assert!(plan.rest.is_empty(), "snapshot {}", snapshot.id());
assert_eq!(plan.first, vec![RowRange::new(0, total_rows as i64 -
1)]);
}
- assert_eq!(builder.execute().await.unwrap(), 1);
+ assert_eq!(
+ table
+ .new_global_index_build_builder()
+ .with_index_column("embedding")
+ .with_index_type(crate::vindex::IVF_SQ_IDENTIFIER)
+ .with_options(HashMap::from([(
+ "vindex.build.granule.enabled".to_string(),
+ granule_enabled.to_string()
+ )]))
+ .execute()
+ .await
+ .unwrap(),
+ 1
+ );
let result = table
.new_vector_search_builder()
.with_vector_column("embedding")
@@ -1033,7 +1052,8 @@ async fn
vindex_build_cleans_written_shards_when_later_shard_fails() {
.await;
let error = table
- .new_vindex_index_build_builder(IVF_FLAT_IDENTIFIER)
+ .new_global_index_build_builder()
+ .with_index_type(IVF_FLAT_IDENTIFIER)
.with_index_column("embedding")
.execute()
.await
@@ -1099,7 +1119,8 @@ async fn
vindex_build_coexists_with_different_index_type_on_same_field() {
// A native-build failure over the tiny synthetic dataset is tolerated;
// only the overlap error is forbidden.
let result = table
- .new_vindex_index_build_builder(IVF_FLAT_IDENTIFIER)
+ .new_global_index_build_builder()
+ .with_index_type(IVF_FLAT_IDENTIFIER)
.with_index_column("embedding")
.execute()
.await;
diff --git a/crates/paimon/src/table/vindex_index_build_builder/timing.rs
b/crates/paimon/src/table/vindex_index_build_builder/timing.rs
index ae43568f..87b3f417 100644
--- a/crates/paimon/src/table/vindex_index_build_builder/timing.rs
+++ b/crates/paimon/src/table/vindex_index_build_builder/timing.rs
@@ -20,14 +20,14 @@ use std::time::Duration;
const VECTOR_INDEX_BUILD_TIMING_ENV: &str =
"PAIMON_LOG_VECTOR_INDEX_BUILD_TIMING";
-pub(super) fn vector_index_build_timing_enabled() -> bool {
+pub(in crate::table) fn vector_index_build_timing_enabled() -> bool {
static ENABLED: OnceLock<bool> = OnceLock::new();
*ENABLED.get_or_init(|| {
std::env::var_os(VECTOR_INDEX_BUILD_TIMING_ENV).is_some_and(|value|
value == "1")
})
}
-pub(super) struct VectorIndexBuildTiming {
+pub(in crate::table) struct VectorIndexBuildTiming {
pub(super) total_without_commit: Duration,
pub(super) source_batch_wait: Duration,
pub(super) oss_read: Duration,
@@ -59,7 +59,7 @@ pub(super) struct VectorIndexBuildTiming {
}
impl VectorIndexBuildTiming {
- pub(super) fn log(self, index_type: &str, commit: Duration) {
+ pub(in crate::table) fn log(self, index_type: &str, commit: Duration) {
let total = self.total_without_commit.saturating_add(commit);
let accounted = self
.source_batch_wait
diff --git a/crates/paimon/src/table/vindex_index_build_builder/writer.rs
b/crates/paimon/src/table/vindex_index_build_builder/writer.rs
index 8c7fc126..bc4c77d2 100644
--- a/crates/paimon/src/table/vindex_index_build_builder/writer.rs
+++ b/crates/paimon/src/table/vindex_index_build_builder/writer.rs
@@ -39,7 +39,6 @@ use std::time::{Duration, Instant};
use tokio::io::AsyncWriteExt;
use tokio_util::io::SyncIoBridge;
-const INDEX_DIR: &str = "index";
const VECTOR_BUFFER_BYTES: usize = 8 * 1024 * 1024;
pub(super) struct BuiltIndexFile {
@@ -426,23 +425,16 @@ impl<'a> VindexIndexBuildBuilder<'a> {
index_meta: Vec<u8>,
row_count: i64,
) -> Result<IndexFileMeta> {
- self.table
- .file_io()
- .mkdirs(&format!(
- "{}/{INDEX_DIR}/",
- self.table.location().trim_end_matches('/')
- ))
- .await?;
let file_name = format!(
"vector-{}-global-index-{}.index",
self.index_type,
uuid::Uuid::new_v4()
);
- let index_path = format!(
- "{}/{INDEX_DIR}/{}",
- self.table.location().trim_end_matches('/'),
- file_name
- );
+ let (index_path, external_path) =
+ crate::table::global_index_build_common::prepare_index_file_path(
+ self.table, &file_name,
+ )
+ .await?;
let write_result = async {
let async_writer = self
.table
@@ -484,13 +476,15 @@ impl<'a> VindexIndexBuildBuilder<'a> {
)?,
row_count,
deletion_vectors_ranges: None,
- external_path: None,
+ external_path,
global_index_meta: Some(GlobalIndexMeta {
row_range_start: shard.row_range_start,
row_range_end: shard.row_range_end,
index_field_id,
extra_field_ids: None,
- source_meta: None,
+ source_meta: Some(
+
crate::spec::DataEvolutionIndexSourceMeta::new(shard.snapshot_id)?.serialize(),
+ ),
index_meta: Some(index_meta),
}),
})
diff --git a/crates/paimon/src/table/write_builder.rs
b/crates/paimon/src/table/write_builder.rs
index bbde338f..1211996e 100644
--- a/crates/paimon/src/table/write_builder.rs
+++ b/crates/paimon/src/table/write_builder.rs
@@ -536,7 +536,7 @@ mod tests {
.unwrap();
let index_err = table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("value")
.execute()
.await
diff --git a/crates/paimon/tests/scan_planning_parity_test.rs
b/crates/paimon/tests/scan_planning_parity_test.rs
index f08134db..c5fb07d7 100644
--- a/crates/paimon/tests/scan_planning_parity_test.rs
+++ b/crates/paimon/tests/scan_planning_parity_test.rs
@@ -190,7 +190,7 @@ async fn
global_index_ranges_intersect_positions_after_candidate_assignment() {
append_ids(&table, 3, 6).await;
assert!(
table
- .new_btree_global_index_build_builder()
+ .new_global_index_build_builder()
.with_index_column("id")
.with_index_type(index_type)
.execute()
