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 bf1dcdd Upgrade vindex and add create index support (#476)
bf1dcdd is described below
commit bf1dcddd265c33186f1815b89e910554170315dc
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Jul 7 21:44:49 2026 +0800
Upgrade vindex and add create index support (#476)
---
bindings/c/DEPENDENCIES.rust.tsv | 2 +-
bindings/python/DEPENDENCIES.rust.tsv | 2 +-
crates/integration_tests/DEPENDENCIES.rust.tsv | 2 +-
.../integrations/datafusion/DEPENDENCIES.rust.tsv | 2 +-
crates/integrations/datafusion/src/procedures.rs | 33 +-
.../integrations/datafusion/tests/read_tables.rs | 106 +++
crates/paimon/Cargo.toml | 2 +-
crates/paimon/DEPENDENCIES.rust.tsv | 2 +-
crates/paimon/src/table/mod.rs | 6 +
.../paimon/src/table/vindex_index_build_builder.rs | 995 +++++++++++++++++++++
crates/paimon/src/vindex/mod.rs | 518 +++++++++++
docs/src/sql.md | 179 +++-
12 files changed, 1824 insertions(+), 25 deletions(-)
diff --git a/bindings/c/DEPENDENCIES.rust.tsv b/bindings/c/DEPENDENCIES.rust.tsv
index 0d136bc..ba66630 100644
--- a/bindings/c/DEPENDENCIES.rust.tsv
+++ b/bindings/c/DEPENDENCIES.rust.tsv
@@ -173,7 +173,7 @@ [email protected] X
[email protected]
X
[email protected] X
[email protected] X
[email protected] X
[email protected] X
[email protected] X
[email protected] X
X
[email protected] X
X
diff --git a/bindings/python/DEPENDENCIES.rust.tsv
b/bindings/python/DEPENDENCIES.rust.tsv
index 91ac670..c8bc735 100644
--- a/bindings/python/DEPENDENCIES.rust.tsv
+++ b/bindings/python/DEPENDENCIES.rust.tsv
@@ -285,7 +285,7 @@ [email protected]
X
[email protected]
X
[email protected] X
[email protected] X
[email protected] X
[email protected] X
[email protected] X
X
[email protected] X
X
[email protected] X
diff --git a/crates/integration_tests/DEPENDENCIES.rust.tsv
b/crates/integration_tests/DEPENDENCIES.rust.tsv
index 0e65d0e..5933f8b 100644
--- a/crates/integration_tests/DEPENDENCIES.rust.tsv
+++ b/crates/integration_tests/DEPENDENCIES.rust.tsv
@@ -173,7 +173,7 @@ [email protected] X
[email protected]
X
[email protected] X
[email protected] X
[email protected] X
[email protected] X
[email protected] X
[email protected] X
X
[email protected] X
X
diff --git a/crates/integrations/datafusion/DEPENDENCIES.rust.tsv
b/crates/integrations/datafusion/DEPENDENCIES.rust.tsv
index b6dd6fe..7809a72 100644
--- a/crates/integrations/datafusion/DEPENDENCIES.rust.tsv
+++ b/crates/integrations/datafusion/DEPENDENCIES.rust.tsv
@@ -259,7 +259,7 @@ [email protected]
X
[email protected]
X
[email protected] X
[email protected] X
[email protected] X
[email protected] X
[email protected] X
X
[email protected] X
X
[email protected] X
diff --git a/crates/integrations/datafusion/src/procedures.rs
b/crates/integrations/datafusion/src/procedures.rs
index 6587983..0778839 100644
--- a/crates/integrations/datafusion/src/procedures.rs
+++ b/crates/integrations/datafusion/src/procedures.rs
@@ -24,6 +24,7 @@
//! - `CALL sys.rollback_to_timestamp(table => '...', timestamp => ...)`
//! - `CALL sys.create_tag_from_timestamp(table => '...', tag => '...',
timestamp => ...)`
//! - `CALL sys.create_global_index(table => '...', index_column => '...',
index_type => 'btree')`
+//! - `CALL sys.create_global_index(table => '...', index_column => '...',
index_type => 'ivf-pq')`
//! - `CALL sys.drop_global_index(table => '...', index_column => '...',
index_type => 'btree')`
//! - `CALL sys.create_lumina_index(table => '...', index_column => '...')`
@@ -42,6 +43,7 @@ use datafusion::sql::sqlparser::ast::{
use paimon::catalog::{Catalog, Identifier};
use paimon::spec::Snapshot;
use paimon::table::{SnapshotManager, Table, TagManager};
+use paimon::vindex::is_vindex_index_type;
use crate::error::to_datafusion_error;
@@ -543,20 +545,29 @@ async fn proc_create_global_index(
.get("index_type")
.map(String::as_str)
.unwrap_or("btree");
- if !index_type.eq_ignore_ascii_case("btree") {
+ if index_type.eq_ignore_ascii_case("btree") {
+ if args.contains_key("options") {
+ return Err(DataFusionError::NotImplemented(
+ "create_global_index options are not supported for btree
yet".to_string(),
+ ));
+ }
+
+ let mut builder = table.new_btree_global_index_build_builder();
+ builder.with_index_column(index_column);
+ 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 {
return Err(DataFusionError::NotImplemented(format!(
- "create_global_index only supports index_type => 'btree', got
'{index_type}'"
+ "create_global_index only supports index_type => 'btree' or vindex
types \
+ ('ivf-flat', 'ivf-pq', 'ivf-hnsw-flat', 'ivf-hnsw-sq'), got
'{index_type}'"
)));
}
- if args.contains_key("options") {
- return Err(DataFusionError::NotImplemented(
- "create_global_index options are not supported for btree
yet".to_string(),
- ));
- }
-
- let mut builder = table.new_btree_global_index_build_builder();
- builder.with_index_column(index_column);
- builder.execute().await.map_err(to_datafusion_error)?;
ok_result(ctx)
}
diff --git a/crates/integrations/datafusion/tests/read_tables.rs
b/crates/integrations/datafusion/tests/read_tables.rs
index 718a1cd..f2b8894 100644
--- a/crates/integrations/datafusion/tests/read_tables.rs
+++ b/crates/integrations/datafusion/tests/read_tables.rs
@@ -1501,6 +1501,7 @@ mod vector_search_tests {
ctx.register_catalog("paimon", catalog.clone())
.await
.expect("Failed to register catalog");
+ register_vector_search(ctx.ctx(), catalog.clone(), "default");
(ctx, catalog, tmp)
}
@@ -1527,6 +1528,27 @@ mod vector_search_tests {
.expect("Failed to build table schema")
}
+ fn build_vindex_table_schema() -> Schema {
+ let mut options = std::collections::HashMap::new();
+ options.insert("row-tracking.enabled".to_string(), "true".to_string());
+ options.insert("data-evolution.enabled".to_string(),
"true".to_string());
+ options.insert("global-index.enabled".to_string(), "true".to_string());
+ options.insert(
+ "global-index.row-count-per-shard".to_string(),
+ "3".to_string(),
+ );
+
+ Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column(
+ "embedding",
+
DataType::Array(ArrayType::new(DataType::Float(FloatType::new()))),
+ )
+ .options(options)
+ .build()
+ .expect("Failed to build table schema")
+ }
+
fn build_vector_batch(ids: Vec<i32>, vectors: Vec<Vec<f32>>) ->
RecordBatch {
let element_field = Arc::new(ArrowField::new("element",
ArrowDataType::Float32, true));
let mut vector_builder =
@@ -1823,6 +1845,90 @@ mod vector_search_tests {
"one same-direction neighbor should be returned, got {ids:?}"
);
}
+
+ #[tokio::test]
+ async fn test_vindex_build_then_vector_search_query() {
+ let (ctx, catalog, _tmp) = create_empty_vector_search_context().await;
+ let identifier = Identifier::new("default", "vindex_build_query_e2e");
+ catalog
+ .create_table(&identifier, build_vindex_table_schema(), false)
+ .await
+ .expect("Failed to create table");
+ let table = catalog
+ .get_table(&identifier)
+ .await
+ .expect("Failed to load table");
+
+ let write_builder = table
+ .new_write_builder()
+ .with_commit_user("test-user")
+ .expect("Failed to configure write builder");
+ let mut table_write = write_builder
+ .new_write()
+ .expect("Failed to create table write");
+ table_write
+ .write_arrow_batch(&build_vector_batch(
+ vec![0, 1, 2, 3, 4, 5],
+ vec![
+ vec![1.0, 0.0],
+ vec![0.9, 0.1],
+ vec![0.0, 1.0],
+ vec![-1.0, 0.0],
+ vec![0.0, -1.0],
+ vec![0.7, 0.3],
+ ],
+ ))
+ .await
+ .expect("Failed to write vector batch");
+ let messages = table_write
+ .prepare_commit()
+ .await
+ .expect("Failed to prepare commit");
+ write_builder
+ .new_commit()
+ .commit(messages)
+ .await
+ .expect("Failed to commit vector data");
+
+ ctx.sql(
+ "CALL sys.create_global_index( \
+ table => 'default.vindex_build_query_e2e', \
+ index_column => 'embedding', \
+ index_type => 'ivf-flat', \
+ options =>
'ivf-flat.dimension=2,ivf-flat.nlist=1,ivf-flat.distance.metric=l2')",
+ )
+ .await
+ .expect("vindex index build SQL should parse")
+ .collect()
+ .await
+ .expect("vindex index build SQL should execute");
+
+ let index_batches = ctx
+ .sql("SELECT index_type, row_count, row_range_start,
row_range_end, index_field_name FROM
paimon.default.`vindex_build_query_e2e$table_indexes` WHERE index_type =
'ivf-flat'")
+ .await
+ .expect("index metadata SQL should parse")
+ .collect()
+ .await
+ .expect("index metadata query should execute");
+ let index_rows = extract_index_rows(&index_batches);
+ assert_eq!(
+ index_rows,
+ vec![
+ ("ivf-flat".to_string(), 3, 0, 2, "embedding".to_string()),
+ ("ivf-flat".to_string(), 3, 3, 5, "embedding".to_string()),
+ ]
+ );
+
+ let search_batches = ctx
+ .sql("SELECT id FROM
vector_search('paimon.default.vindex_build_query_e2e', 'embedding', '[1.0,
0.0]', 2)")
+ .await
+ .expect("vector_search SQL should parse")
+ .collect()
+ .await
+ .expect("vector_search query should execute");
+ let ids = extract_ids(&search_batches);
+ assert_eq!(ids, vec![0, 1]);
+ }
}
// ======================= Hybrid Search Tests =======================
diff --git a/crates/paimon/Cargo.toml b/crates/paimon/Cargo.toml
index 7755a8e..72fbfe9 100644
--- a/crates/paimon/Cargo.toml
+++ b/crates/paimon/Cargo.toml
@@ -105,7 +105,7 @@ urlencoding = "2.1"
tantivy = { version = "0.22", optional = true }
tempfile = { version = "3", optional = true }
paimon-mosaic-core = { version = "0.1.0", optional = true }
-paimon-vindex-core = "0.1.0"
+paimon-vindex-core = "0.2.0"
vortex = { version = "0.75.0", features = ["tokio"], optional = true }
libloading = "0.9"
# Keep CI on the dependency set that passed before unicode-segmentation 1.13.3.
diff --git a/crates/paimon/DEPENDENCIES.rust.tsv
b/crates/paimon/DEPENDENCIES.rust.tsv
index 9e06a05..7762797 100644
--- a/crates/paimon/DEPENDENCIES.rust.tsv
+++ b/crates/paimon/DEPENDENCIES.rust.tsv
@@ -220,7 +220,7 @@ [email protected] X
[email protected]
X
[email protected]
X
[email protected] X
[email protected] X
[email protected] X
[email protected] X
[email protected] X
X
[email protected] X
X
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index edc3719..9e6b623 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -66,6 +66,7 @@ pub(crate) mod table_write;
mod tag_manager;
pub(crate) mod time_travel;
mod vector_search_builder;
+mod vindex_index_build_builder;
mod write_builder;
use crate::Result;
@@ -99,6 +100,7 @@ pub use table_update::TableUpdate;
pub use table_write::TableWrite;
pub use tag_manager::TagManager;
pub use vector_search_builder::{BatchVectorSearchBuilder, VectorSearchBuilder};
+pub use vindex_index_build_builder::VindexIndexBuildBuilder;
pub use write_builder::WriteBuilder;
use crate::catalog::Identifier;
@@ -218,6 +220,10 @@ impl Table {
BTreeGlobalIndexDropBuilder::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/vindex_index_build_builder.rs
b/crates/paimon/src/table/vindex_index_build_builder.rs
new file mode 100644
index 0000000..7985411
--- /dev/null
+++ b/crates/paimon/src/table/vindex_index_build_builder.rs
@@ -0,0 +1,995 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::spec::{
+ bucket_dir_name, BinaryRow, CoreOptions, DataField, DataFileMeta,
DataType, FileKind,
+ GlobalIndexMeta, IndexFileMeta, IndexManifest, ROW_ID_FIELD_NAME,
+};
+use crate::table::{
+ CommitMessage, DataSplitBuilder, RowRange, SnapshotManager, Table,
TableCommit,
+};
+use crate::vindex::{is_vindex_index_type, VindexVectorIndexOptions};
+use crate::{Error, Result};
+use arrow_array::{Array, FixedSizeListArray, Float32Array, Int64Array,
ListArray, RecordBatch};
+use bytes::Bytes;
+use futures::TryStreamExt;
+use paimon_vindex_core::index::{VectorIndexConfig, VectorIndexTrainer,
VectorIndexWriter};
+use paimon_vindex_core::io::PosWriter;
+use std::collections::HashMap;
+
+const INDEX_DIR: &str = "index";
+
+pub struct VindexIndexBuildBuilder<'a> {
+ table: &'a Table,
+ index_column: Option<String>,
+ index_type: String,
+ options: HashMap<String, String>,
+}
+
+impl<'a> VindexIndexBuildBuilder<'a> {
+ pub(crate) fn new(table: &'a Table, index_type: &str) -> Self {
+ Self {
+ table,
+ index_column: None,
+ index_type: index_type.to_string(),
+ options: HashMap::new(),
+ }
+ }
+
+ pub fn with_index_column(&mut self, column: &str) -> &mut Self {
+ self.index_column = Some(column.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> {
+ if !is_vindex_index_type(&self.index_type) {
+ return Err(Error::DataInvalid {
+ message: format!("Unsupported vindex index type: {}",
self.index_type),
+ source: None,
+ });
+ }
+
+ let index_column = self
+ .index_column
+ .as_deref()
+ .ok_or_else(|| Error::DataInvalid {
+ message: "vindex index column is required".to_string(),
+ source: None,
+ })?;
+
+ let core_options = CoreOptions::new(self.table.schema().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)?;
+ let vindex_options = VindexVectorIndexOptions::new(
+ self.table.schema().options(),
+ &self.options,
+ &self.index_type,
+ index_field,
+ )?;
+ let dimension = checked_i32(
+ vindex_options.dimension() as u64,
+ "vindex dimension is too large for Rust builder",
+ )?;
+ let index_meta =
+ serde_json::to_vec(&vindex_options.native_options).map_err(|e|
Error::DataInvalid {
+ message: format!("Failed to serialize vindex options metadata:
{e}"),
+ 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 manifest_entries = self
+ .table
+ .new_read_builder()
+ .new_scan()
+ .with_scan_all_files()
+ .plan_manifest_entries(&snapshot)
+ .await?;
+ let shards = plan_vindex_shards(
+ self.table.location(),
+ self.table.schema().partition_keys(),
+ self.table.schema().fields(),
+ &core_options,
+ snapshot.id(),
+ manifest_entries,
+ rows_per_shard,
+ )?;
+ if shards.is_empty() {
+ return Ok(0);
+ }
+
+ validate_existing_index_overlap(
+ self.table,
+ snapshot.index_manifest(),
+ index_field.id(),
+ &shards,
+ )
+ .await?;
+
+ let shard_count = shards.len();
+ let mut messages = Vec::with_capacity(shard_count);
+ for shard in shards {
+ let vectors = extract_vectors(self.table, &shard, index_column,
dimension).await?;
+ let index_file = self
+ .build_index_file(
+ &shard,
+ &vectors,
+ dimension,
+ index_field.id(),
+ vindex_options.config.clone(),
+ index_meta.clone(),
+ )
+ .await?;
+ let mut message =
CommitMessage::new(shard.partition_bytes.clone(), 0, vec![]);
+ message.new_index_files = vec![index_file];
+ messages.push(message);
+ }
+
+ TableCommit::new(
+ self.table.clone(),
+ format!(
+ "global-index-{}-create-{}",
+ self.index_type,
+ uuid::Uuid::new_v4()
+ ),
+ )
+ .commit_if_latest_snapshot(messages, snapshot.id())
+ .await?;
+
+ Ok(shard_count)
+ }
+
+ async fn build_index_file(
+ &self,
+ shard: &VindexIndexShard,
+ vectors: &[f32],
+ dimension: i32,
+ index_field_id: i32,
+ config: VectorIndexConfig,
+ index_meta: Vec<u8>,
+ ) -> Result<IndexFileMeta> {
+ let row_count = checked_row_count(shard.row_range_start,
shard.row_range_end)?;
+ validate_vector_buffer(vectors, row_count, dimension)?;
+ let row_count_usize = usize::try_from(row_count).map_err(|e|
Error::DataInvalid {
+ message: format!("Invalid vindex row count: {row_count}"),
+ source: Some(Box::new(e)),
+ })?;
+ let ids = (0..i64::from(row_count)).collect::<Vec<_>>();
+
+ let training =
+ VectorIndexTrainer::train(config, vectors,
row_count_usize).map_err(|e| {
+ Error::DataInvalid {
+ message: format!("Failed to train vindex index: {e}"),
+ source: Some(Box::new(e)),
+ }
+ })?;
+ let mut writer = VectorIndexWriter::new(training);
+ writer
+ .add_vectors(&ids, vectors, row_count_usize)
+ .map_err(|e| Error::DataInvalid {
+ message: format!("Failed to add vectors to vindex index: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ let mut bytes = Vec::new();
+ {
+ let mut output = PosWriter::new(&mut bytes);
+ writer.write(&mut output).map_err(|e| Error::DataInvalid {
+ message: format!("Failed to serialize vindex index: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ }
+
+ 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
+ );
+ self.table
+ .file_io()
+ .new_output(&index_path)?
+ .write(Bytes::from(bytes))
+ .await?;
+
+ let status = self.table.file_io().get_status(&index_path).await?;
+ Ok(IndexFileMeta {
+ index_type: self.index_type.clone(),
+ file_name,
+ file_size: checked_i32(
+ status.size,
+ "Index file is too large for Rust IndexFileMeta",
+ )?,
+ row_count,
+ deletion_vectors_ranges: None,
+ 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,
+ index_meta: Some(index_meta),
+ }),
+ })
+ }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct VindexIndexShard {
+ pub partition: BinaryRow,
+ pub partition_bytes: Vec<u8>,
+ pub files: Vec<DataFileMeta>,
+ pub row_range_start: i64,
+ pub row_range_end: i64,
+ snapshot_id: i64,
+ source_bucket: i32,
+ total_buckets: i32,
+ bucket_path: String,
+}
+
+fn validate_table_options(table: &Table, core_options: &CoreOptions) ->
Result<()> {
+ if !core_options.row_tracking_enabled() {
+ return Err(Error::DataInvalid {
+ message: "vindex index build requires 'row-tracking.enabled' =
'true'".to_string(),
+ source: None,
+ });
+ }
+ if !core_options.data_evolution_enabled() {
+ return Err(Error::DataInvalid {
+ message: "vindex index build requires 'data-evolution.enabled' =
'true'".to_string(),
+ source: None,
+ });
+ }
+ if !core_options.global_index_enabled() {
+ return Err(Error::DataInvalid {
+ message: "vindex index build requires 'global-index.enabled' =
'true'".to_string(),
+ source: None,
+ });
+ }
+ if !table.schema().primary_keys().is_empty() {
+ return Err(Error::Unsupported {
+ message: "vindex index build does not support primary-key
tables".to_string(),
+ });
+ }
+ if core_options.deletion_vectors_enabled() {
+ return Err(Error::Unsupported {
+ message:
+ "vindex index build does not support tables with
deletion-vectors.enabled=true"
+ .to_string(),
+ });
+ }
+ Ok(())
+}
+
+fn find_index_field<'a>(table: &'a Table, column: &str) -> Result<&'a
DataField> {
+ table
+ .schema()
+ .fields()
+ .iter()
+ .find(|field| field.name() == column)
+ .ok_or_else(|| Error::ColumnNotExist {
+ full_name: table.identifier().full_name(),
+ column: column.to_string(),
+ })
+}
+
+fn validate_vector_field(field: &DataField) -> Result<()> {
+ let is_array_float = matches!(
+ field.data_type(),
+ DataType::Array(array) if matches!(array.element_type(),
DataType::Float(_))
+ );
+ let is_vector_float = matches!(
+ field.data_type(),
+ DataType::Vector(vector) if matches!(vector.element_type(),
DataType::Float(_))
+ );
+ if !is_array_float && !is_vector_float {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "vindex index requires ARRAY<FLOAT> or VECTOR<FLOAT> column,
got {:?} for column '{}'",
+ field.data_type(),
+ field.name()
+ ),
+ source: None,
+ });
+ }
+ Ok(())
+}
+
+fn plan_vindex_shards(
+ table_location: &str,
+ partition_keys: &[String],
+ schema_fields: &[DataField],
+ core_options: &CoreOptions,
+ snapshot_id: i64,
+ entries: Vec<crate::spec::ManifestEntry>,
+ rows_per_shard: i64,
+) -> Result<Vec<VindexIndexShard>> {
+ if rows_per_shard <= 0 {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Option 'global-index.row-count-per-shard' must be greater
than 0, got: {rows_per_shard}"
+ ),
+ source: None,
+ });
+ }
+
+ let mut by_partition_bucket: HashMap<(Vec<u8>, i32, i32),
Vec<DataFileMeta>> = HashMap::new();
+ for entry in entries {
+ if *entry.kind() != FileKind::Add {
+ continue;
+ }
+ if entry.file().first_row_id.is_none() {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Data file '{}' is missing first_row_id; cannot build a
complete vindex index",
+ entry.file().file_name
+ ),
+ source: None,
+ });
+ }
+ let (partition, bucket, total_buckets, file) = entry.into_parts();
+ by_partition_bucket
+ .entry((partition, bucket, total_buckets))
+ .or_default()
+ .push(file);
+ }
+
+ let mut result = Vec::new();
+ for ((partition_bytes, source_bucket, total_buckets), files) in
by_partition_bucket {
+ let partition = if partition_keys.is_empty() {
+ BinaryRow::new(0)
+ } else {
+ BinaryRow::from_serialized_bytes(&partition_bytes)?
+ };
+ let bucket_path = bucket_path(
+ table_location,
+ partition_keys,
+ schema_fields,
+ core_options,
+ &partition,
+ source_bucket,
+ )?;
+ let mut files_by_shard: HashMap<i64, Vec<DataFileMeta>> =
HashMap::new();
+ for file in files {
+ let (file_start, file_end) = file.row_id_range().ok_or_else(||
Error::DataInvalid {
+ message: format!(
+ "Data file '{}' is missing first_row_id; cannot build a
complete vindex index",
+ file.file_name
+ ),
+ source: None,
+ })?;
+ let start_shard = file_start / rows_per_shard;
+ let end_shard = file_end / rows_per_shard;
+ for shard_id in start_shard..=end_shard {
+ files_by_shard
+ .entry(shard_id * rows_per_shard)
+ .or_default()
+ .push(file.clone());
+ }
+ }
+
+ let mut shard_starts =
files_by_shard.keys().copied().collect::<Vec<_>>();
+ shard_starts.sort_unstable();
+ for shard_start in shard_starts {
+ let shard_end = shard_start + rows_per_shard - 1;
+ let mut shard_files =
files_by_shard.remove(&shard_start).unwrap_or_default();
+ shard_files.sort_by_key(|file| file.first_row_id);
+ let groups = group_contiguous_files(shard_files)?;
+ for group in groups {
+ let group_start = group
+ .first()
+ .and_then(|file| file.first_row_id)
+ .expect("planned groups are non-empty and row-id
assigned");
+ let group_end = group
+ .iter()
+ .map(|file| file.row_id_range().unwrap().1)
+ .max()
+ .unwrap();
+ let row_range_start = group_start.max(shard_start);
+ let row_range_end = group_end.min(shard_end);
+ result.push(VindexIndexShard {
+ partition: partition.clone(),
+ partition_bytes: partition_bytes.clone(),
+ files: group,
+ row_range_start,
+ row_range_end,
+ snapshot_id,
+ source_bucket,
+ total_buckets,
+ bucket_path: bucket_path.clone(),
+ });
+ }
+ }
+ }
+ result.sort_by(|a, b| {
+ a.partition
+ .to_serialized_bytes()
+ .cmp(&b.partition.to_serialized_bytes())
+ .then(a.source_bucket.cmp(&b.source_bucket))
+ .then(a.row_range_start.cmp(&b.row_range_start))
+ });
+ Ok(result)
+}
+
+fn group_contiguous_files(mut files: Vec<DataFileMeta>) ->
Result<Vec<Vec<DataFileMeta>>> {
+ if files.is_empty() {
+ return Ok(Vec::new());
+ }
+ files.sort_by_key(|file| file.first_row_id);
+ let mut groups = Vec::new();
+ let mut current = Vec::new();
+ let mut current_end = None;
+ for file in files {
+ let (file_start, file_end) = file.row_id_range().ok_or_else(||
Error::DataInvalid {
+ message: format!(
+ "Data file '{}' is missing first_row_id; cannot build a
complete vindex index",
+ file.file_name
+ ),
+ source: None,
+ })?;
+ match current_end {
+ None => {
+ current.push(file);
+ current_end = Some(file_end);
+ }
+ Some(end) if file_start <= end + 1 => {
+ current.push(file);
+ current_end = Some(end.max(file_end));
+ }
+ Some(_) => {
+ groups.push(std::mem::take(&mut current));
+ current.push(file);
+ current_end = Some(file_end);
+ }
+ }
+ }
+ if !current.is_empty() {
+ groups.push(current);
+ }
+ Ok(groups)
+}
+
+fn bucket_path(
+ table_location: &str,
+ partition_keys: &[String],
+ schema_fields: &[DataField],
+ core_options: &CoreOptions,
+ partition: &BinaryRow,
+ bucket: i32,
+) -> Result<String> {
+ let base = table_location.trim_end_matches('/');
+ if partition_keys.is_empty() {
+ return Ok(format!("{base}/{}", bucket_dir_name(bucket)));
+ }
+ let computer = crate::spec::PartitionComputer::new(
+ partition_keys,
+ schema_fields,
+ core_options.partition_default_name(),
+ core_options.legacy_partition_name(),
+ )?;
+ Ok(format!(
+ "{base}/{}{}",
+ computer.generate_partition_path(partition)?,
+ bucket_dir_name(bucket)
+ ))
+}
+
+async fn validate_existing_index_overlap(
+ table: &Table,
+ index_manifest_name: Option<&str>,
+ index_field_id: i32,
+ shards: &[VindexIndexShard],
+) -> Result<()> {
+ let Some(index_manifest_name) = index_manifest_name else {
+ return Ok(());
+ };
+ let path = format!(
+ "{}/manifest/{}",
+ table.location().trim_end_matches('/'),
+ index_manifest_name
+ );
+ let entries = IndexManifest::read(table.file_io(), &path).await?;
+ for entry in entries {
+ if entry.kind != FileKind::Add {
+ continue;
+ }
+ let Some(meta) = entry.index_file.global_index_meta else {
+ continue;
+ };
+ if meta.index_field_id != index_field_id {
+ continue;
+ }
+ if shards.iter().any(|shard| {
+ ranges_overlap(
+ meta.row_range_start,
+ meta.row_range_end,
+ shard.row_range_start,
+ shard.row_range_end,
+ )
+ }) {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Existing global index file '{}' overlaps requested row
range for field {}",
+ entry.index_file.file_name, index_field_id
+ ),
+ source: None,
+ });
+ }
+ }
+ Ok(())
+}
+
+async fn extract_vectors(
+ table: &Table,
+ shard: &VindexIndexShard,
+ index_column: &str,
+ dimension: i32,
+) -> Result<Vec<f32>> {
+ let split = DataSplitBuilder::new()
+ .with_snapshot(shard.snapshot_id)
+ .with_partition(shard.partition.clone())
+ .with_bucket(shard.source_bucket)
+ .with_bucket_path(shard.bucket_path.clone())
+ .with_total_buckets(shard.total_buckets)
+ .with_data_files(shard.files.clone())
+ .with_row_ranges(vec![RowRange::new(
+ shard.row_range_start,
+ shard.row_range_end,
+ )])
+ .build()?;
+
+ let mut read_builder = table.new_read_builder();
+ read_builder.with_projection(&[index_column, ROW_ID_FIELD_NAME])?;
+ let read = read_builder.new_read()?;
+ let batches = read.to_arrow(&[split])?.try_collect::<Vec<_>>().await?;
+ extract_vectors_from_batches(
+ &batches,
+ index_column,
+ dimension,
+ shard.row_range_start,
+ i64::from(checked_row_count(
+ shard.row_range_start,
+ shard.row_range_end,
+ )?),
+ )
+}
+
+fn extract_vectors_from_batches(
+ batches: &[RecordBatch],
+ index_column: &str,
+ dimension: i32,
+ row_range_start: i64,
+ expected_row_count: i64,
+) -> Result<Vec<f32>> {
+ let dimension = usize::try_from(dimension).map_err(|e| Error::DataInvalid {
+ message: format!("Invalid vindex dimension: {dimension}"),
+ source: Some(Box::new(e)),
+ })?;
+ let row_count = batches.iter().map(RecordBatch::num_rows).sum::<usize>();
+ let mut vectors = Vec::with_capacity(row_count * dimension);
+ let mut expected_row_id = row_range_start;
+ for batch in batches {
+ let vector_index =
+ batch
+ .schema()
+ .index_of(index_column)
+ .map_err(|e| Error::DataInvalid {
+ message: format!("Vector column '{index_column}' not found
in read batch: {e}"),
+ source: None,
+ })?;
+ let row_id_index =
+ batch
+ .schema()
+ .index_of(ROW_ID_FIELD_NAME)
+ .map_err(|e| Error::DataInvalid {
+ message: format!("_ROW_ID column not found in read batch:
{e}"),
+ source: None,
+ })?;
+ let column = batch.column(vector_index);
+ enum VectorLayout<'a> {
+ List(&'a ListArray),
+ Fixed(&'a FixedSizeListArray),
+ }
+ let layout = if let Some(a) =
column.as_any().downcast_ref::<ListArray>() {
+ VectorLayout::List(a)
+ } else if let Some(a) =
column.as_any().downcast_ref::<FixedSizeListArray>() {
+ VectorLayout::Fixed(a)
+ } else {
+ return Err(Error::DataInvalid {
+ message:
+ "vindex vector extraction requires Arrow List<Float32> or
FixedSizeList<Float32>"
+ .to_string(),
+ source: None,
+ });
+ };
+ let values = match layout {
+ VectorLayout::List(a) => a.values(),
+ VectorLayout::Fixed(a) => a.values(),
+ }
+ .as_any()
+ .downcast_ref::<Float32Array>()
+ .ok_or_else(|| Error::DataInvalid {
+ message: "vindex vector extraction requires Float32 vector
elements".to_string(),
+ source: None,
+ })?;
+ let row_ids = batch
+ .column(row_id_index)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .ok_or_else(|| Error::DataInvalid {
+ message: "vindex vector extraction requires non-null Int64
_ROW_ID".to_string(),
+ source: None,
+ })?;
+
+ for row in 0..batch.num_rows() {
+ if row_ids.is_null(row) {
+ return Err(Error::DataInvalid {
+ message: "vindex vector extraction found null
_ROW_ID".to_string(),
+ source: None,
+ });
+ }
+ let row_id = row_ids.value(row);
+ if row_id != expected_row_id {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "vindex vector extraction expected _ROW_ID {}, got {}",
+ expected_row_id, row_id
+ ),
+ source: None,
+ });
+ }
+ expected_row_id += 1;
+
+ let is_null = match layout {
+ VectorLayout::List(a) => a.is_null(row),
+ VectorLayout::Fixed(a) => a.is_null(row),
+ };
+ if is_null {
+ return Err(Error::DataInvalid {
+ message: "vindex vector extraction found null vector
row".to_string(),
+ source: None,
+ });
+ }
+ let (start, end) = match layout {
+ VectorLayout::List(a) => {
+ let offsets = a.value_offsets();
+ (offsets[row] as usize, offsets[row + 1] as usize)
+ }
+ VectorLayout::Fixed(a) => {
+ let len = a.value_length() as usize;
+ (row * len, (row + 1) * len)
+ }
+ };
+ if end - start != dimension {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "vindex vector dimension mismatch: expected {}, got
{}",
+ dimension,
+ end - start
+ ),
+ source: None,
+ });
+ }
+ for value_index in start..end {
+ if values.is_null(value_index) {
+ return Err(Error::DataInvalid {
+ message: "vindex vector extraction found null vector
element".to_string(),
+ source: None,
+ });
+ }
+ vectors.push(values.value(value_index));
+ }
+ }
+ }
+ let actual_row_count = expected_row_id - row_range_start;
+ if actual_row_count != expected_row_count {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "vindex vector extraction expected {} rows, got {}",
+ expected_row_count, actual_row_count
+ ),
+ source: None,
+ });
+ }
+ Ok(vectors)
+}
+
+fn checked_i32(value: u64, context: &str) -> Result<i32> {
+ i32::try_from(value).map_err(|_| Error::DataInvalid {
+ message: format!("{context}: {value}"),
+ source: None,
+ })
+}
+
+fn checked_row_count(row_range_start: i64, row_range_end: i64) -> Result<i32> {
+ if row_range_end < row_range_start {
+ return Err(Error::DataInvalid {
+ message: format!("Invalid vindex row range [{row_range_start},
{row_range_end}]"),
+ source: None,
+ });
+ }
+ i32::try_from(row_range_end - row_range_start + 1).map_err(|_|
Error::DataInvalid {
+ message: format!(
+ "vindex row count is too large for Rust IndexFileMeta:
[{row_range_start}, {row_range_end}]"
+ ),
+ source: None,
+ })
+}
+
+fn validate_vector_buffer(vectors: &[f32], row_count: i32, dimension: i32) ->
Result<()> {
+ if row_count <= 0 {
+ return Err(Error::DataInvalid {
+ message: format!("vindex shard row count must be positive, got:
{row_count}"),
+ source: None,
+ });
+ }
+ if dimension <= 0 {
+ return Err(Error::DataInvalid {
+ message: format!("vindex vector dimension must be positive, got:
{dimension}"),
+ source: None,
+ });
+ }
+ let row_count = row_count as usize;
+ let dimension = dimension as usize;
+ let expected_len = row_count
+ .checked_mul(dimension)
+ .ok_or_else(|| Error::DataInvalid {
+ message: format!(
+ "vindex vector buffer length overflows: row_count={row_count},
dimension={dimension}"
+ ),
+ source: None,
+ })?;
+ if vectors.len() != expected_len {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "vindex vector buffer length {} does not match row_count={}
and dimension={}",
+ vectors.len(),
+ row_count,
+ dimension
+ ),
+ source: None,
+ });
+ }
+ Ok(())
+}
+
+fn ranges_overlap(left_start: i64, left_end: i64, right_start: i64, right_end:
i64) -> bool {
+ left_start <= right_end && right_start <= left_end
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::catalog::Identifier;
+ use crate::io::FileIOBuilder;
+ use crate::spec::stats::BinaryTableStats;
+ use crate::spec::{ArrayType, FloatType, IntType, ManifestEntry, Schema,
TableSchema};
+ use arrow_array::builder::{Float32Builder, Int64Builder, ListBuilder};
+ use arrow_array::ArrayRef;
+ use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema
as ArrowSchema};
+ use chrono::{DateTime, Utc};
+ use std::sync::Arc;
+
+ fn data_file(name: &str, first_row_id: Option<i64>, row_count: i64) ->
DataFileMeta {
+ DataFileMeta {
+ file_name: name.to_string(),
+ file_size: 128,
+ row_count,
+ min_key: vec![],
+ max_key: vec![],
+ key_stats: BinaryTableStats::new(vec![], vec![], vec![]),
+ value_stats: BinaryTableStats::new(vec![], vec![], vec![]),
+ min_sequence_number: 0,
+ max_sequence_number: 0,
+ schema_id: 0,
+ level: 0,
+ extra_files: vec![],
+ creation_time: Some(
+ "2024-09-06T07:45:55.039+00:00"
+ .parse::<DateTime<Utc>>()
+ .unwrap(),
+ ),
+ delete_row_count: None,
+ embedded_index: None,
+ first_row_id,
+ write_cols: None,
+ external_path: None,
+ file_source: None,
+ value_stats_cols: None,
+ }
+ }
+
+ fn manifest_entry(file: DataFileMeta) -> ManifestEntry {
+ ManifestEntry::new(FileKind::Add, vec![], 0, 1, file, 2)
+ }
+
+ fn table_options(rows_per_shard: &str) -> HashMap<String, String> {
+ HashMap::from([
+ ("row-tracking.enabled".to_string(), "true".to_string()),
+ ("data-evolution.enabled".to_string(), "true".to_string()),
+ ("global-index.enabled".to_string(), "true".to_string()),
+ (
+ "global-index.row-count-per-shard".to_string(),
+ rows_per_shard.to_string(),
+ ),
+ ])
+ }
+
+ fn test_table(options: HashMap<String, String>) -> Table {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column(
+ "embedding",
+
DataType::Array(ArrayType::new(DataType::Float(FloatType::new()))),
+ )
+ .options(options)
+ .build()
+ .unwrap();
+ Table::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("default", "test_table"),
+ "memory:/test_vindex_builder".to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ )
+ }
+
+ fn plan(entries: Vec<ManifestEntry>, rows_per_shard: i64) ->
Result<Vec<VindexIndexShard>> {
+ let table = test_table(table_options(&rows_per_shard.to_string()));
+ let core = CoreOptions::new(table.schema().options());
+ plan_vindex_shards(
+ table.location(),
+ table.schema().partition_keys(),
+ table.schema().fields(),
+ &core,
+ 1,
+ entries,
+ rows_per_shard,
+ )
+ }
+
+ #[test]
+ fn test_planner_splits_single_file_across_shards() {
+ let shards = plan(vec![manifest_entry(data_file("a", Some(0), 25))],
10).unwrap();
+
+ assert_eq!(
+ shards
+ .iter()
+ .map(|s| (s.row_range_start, s.row_range_end))
+ .collect::<Vec<_>>(),
+ vec![(0, 9), (10, 19), (20, 24)]
+ );
+ }
+
+ #[test]
+ fn test_planner_rejects_missing_first_row_id() {
+ let err = plan(vec![manifest_entry(data_file("a", None, 5))], 10)
+ .expect_err("missing first_row_id should fail");
+ assert!(
+ matches!(err, Error::DataInvalid { message, .. } if
message.contains("missing first_row_id"))
+ );
+ }
+
+ #[test]
+ fn test_validate_vector_field_accepts_array_float() {
+ let field = DataField::new(
+ 0,
+ "embedding".to_string(),
+ DataType::Array(ArrayType::new(DataType::Float(FloatType::new()))),
+ );
+ assert!(validate_vector_field(&field).is_ok());
+ }
+
+ fn vector_batch(rows: Vec<Option<Vec<Option<f32>>>>, row_ids:
Vec<Option<i64>>) -> RecordBatch {
+ let mut vector_builder = ListBuilder::new(Float32Builder::new());
+ for row in rows {
+ match row {
+ Some(values) => {
+ for value in values {
+ match value {
+ Some(value) =>
vector_builder.values().append_value(value),
+ None => vector_builder.values().append_null(),
+ }
+ }
+ vector_builder.append(true);
+ }
+ None => vector_builder.append(false),
+ }
+ }
+ let mut row_id_builder = Int64Builder::new();
+ for row_id in row_ids {
+ match row_id {
+ Some(value) => row_id_builder.append_value(value),
+ None => row_id_builder.append_null(),
+ }
+ }
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new(
+ "embedding",
+ ArrowDataType::List(Arc::new(ArrowField::new(
+ "item",
+ ArrowDataType::Float32,
+ true,
+ ))),
+ true,
+ ),
+ ArrowField::new(ROW_ID_FIELD_NAME, ArrowDataType::Int64, true),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(vector_builder.finish()) as ArrayRef,
+ Arc::new(row_id_builder.finish()) as ArrayRef,
+ ],
+ )
+ .unwrap()
+ }
+
+ #[test]
+ fn test_extract_vectors_accepts_list_float32_and_row_ids() {
+ let batch = vector_batch(
+ vec![
+ Some(vec![Some(1.0), Some(2.0)]),
+ Some(vec![Some(3.0), Some(4.0)]),
+ ],
+ vec![Some(10), Some(11)],
+ );
+
+ let vectors = extract_vectors_from_batches(&[batch], "embedding", 2,
10, 2).unwrap();
+
+ assert_eq!(vectors, vec![1.0, 2.0, 3.0, 4.0]);
+ }
+
+ #[test]
+ fn test_extract_vectors_rejects_dimension_mismatch() {
+ let batch = vector_batch(vec![Some(vec![Some(1.0)])], vec![Some(0)]);
+
+ let err = extract_vectors_from_batches(&[batch], "embedding", 2, 0, 1)
+ .expect_err("dimension mismatch should fail");
+
+ assert!(
+ matches!(err, Error::DataInvalid { message, .. } if
message.contains("dimension mismatch"))
+ );
+ }
+}
diff --git a/crates/paimon/src/vindex/mod.rs b/crates/paimon/src/vindex/mod.rs
index aa4f7b8..c0d566a 100644
--- a/crates/paimon/src/vindex/mod.rs
+++ b/crates/paimon/src/vindex/mod.rs
@@ -17,11 +17,21 @@
pub mod reader;
+use crate::spec::{DataField, DataType};
+use paimon_vindex_core::index::VectorIndexConfig;
+use std::collections::HashMap;
+
pub const IVF_FLAT_IDENTIFIER: &str = "ivf-flat";
pub const IVF_PQ_IDENTIFIER: &str = "ivf-pq";
pub const IVF_HNSW_FLAT_IDENTIFIER: &str = "ivf-hnsw-flat";
pub const IVF_HNSW_SQ_IDENTIFIER: &str = "ivf-hnsw-sq";
+const DEFAULT_DIMENSION: &str = "128";
+const DEFAULT_METRIC: &str = "inner_product";
+const DEFAULT_NLIST: &str = "256";
+const DEFAULT_PQ_M: &str = "16";
+const DEFAULT_PQ_USE_OPQ: &str = "false";
+
pub fn is_vindex_index_type(index_type: &str) -> bool {
matches!(
index_type,
@@ -29,9 +39,294 @@ pub fn is_vindex_index_type(index_type: &str) -> bool {
)
}
+pub(crate) fn native_index_type(index_type: &str) -> Option<&'static str> {
+ match index_type {
+ IVF_FLAT_IDENTIFIER => Some("ivf_flat"),
+ IVF_PQ_IDENTIFIER => Some("ivf_pq"),
+ IVF_HNSW_FLAT_IDENTIFIER => Some("ivf_hnsw_flat"),
+ IVF_HNSW_SQ_IDENTIFIER => Some("ivf_hnsw_sq"),
+ _ => None,
+ }
+}
+
+#[derive(Debug)]
+pub(crate) struct VindexVectorIndexOptions {
+ pub config: VectorIndexConfig,
+ pub native_options: HashMap<String, String>,
+}
+
+impl VindexVectorIndexOptions {
+ pub fn new(
+ table_options: &HashMap<String, String>,
+ user_options: &HashMap<String, String>,
+ index_type: &str,
+ field: &DataField,
+ ) -> crate::Result<Self> {
+ let native_index_type =
+ native_index_type(index_type).ok_or_else(||
crate::Error::DataInvalid {
+ message: format!("Unsupported vindex index type:
{index_type}"),
+ source: None,
+ })?;
+
+ validate_user_option_keys(user_options, index_type, field.name())?;
+ validate_index_type_option(table_options, user_options,
native_index_type)?;
+
+ let mut native_options = HashMap::new();
+ native_options.insert("index.type".to_string(),
native_index_type.to_string());
+ native_options.insert(
+ "dimension".to_string(),
+ resolve_dimension(table_options, user_options, index_type, field)?,
+ );
+ native_options.insert(
+ "nlist".to_string(),
+ option_value(
+ table_options,
+ user_options,
+ field.name(),
+ index_type,
+ "nlist",
+ "nlist",
+ DEFAULT_NLIST,
+ ),
+ );
+ native_options.insert(
+ "metric".to_string(),
+ normalize_metric(&option_value(
+ table_options,
+ user_options,
+ field.name(),
+ index_type,
+ "metric",
+ "distance.metric",
+ DEFAULT_METRIC,
+ )),
+ );
+
+ if index_type == IVF_PQ_IDENTIFIER {
+ native_options.insert(
+ "pq.m".to_string(),
+ option_value(
+ table_options,
+ user_options,
+ field.name(),
+ index_type,
+ "pq.m",
+ "pq.m",
+ DEFAULT_PQ_M,
+ ),
+ );
+ native_options.insert(
+ "use-opq".to_string(),
+ option_value(
+ table_options,
+ user_options,
+ field.name(),
+ index_type,
+ "use-opq",
+ "pq.use-opq",
+ DEFAULT_PQ_USE_OPQ,
+ ),
+ );
+ }
+
+ for key in ["hnsw.m", "hnsw.ef-construction", "hnsw.max-level"] {
+ if let Some(value) = optional_value(
+ table_options,
+ user_options,
+ field.name(),
+ index_type,
+ key,
+ key,
+ ) {
+ native_options.insert(key.to_string(), value);
+ }
+ }
+
+ let config =
VectorIndexConfig::from_options(&native_options).map_err(|e| {
+ crate::Error::DataInvalid {
+ message: format!("Invalid vindex options: {e}"),
+ source: Some(Box::new(e)),
+ }
+ })?;
+ Ok(Self {
+ config,
+ native_options,
+ })
+ }
+
+ pub fn dimension(&self) -> usize {
+ self.config.dimension()
+ }
+}
+
+fn validate_index_type_option(
+ table_options: &HashMap<String, String>,
+ user_options: &HashMap<String, String>,
+ expected_native: &str,
+) -> crate::Result<()> {
+ for options in [table_options, user_options] {
+ if let Some(value) = options.get("index.type") {
+ let normalized = value.trim().to_ascii_lowercase().replace('-',
"_");
+ if normalized != expected_native {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!(
+ "Option 'index.type' is '{}', but procedure index_type
resolves to '{}'. \
+ Remove 'index.type' from options or set it to '{}'.",
+ value, expected_native, expected_native
+ ),
+ });
+ }
+ }
+ }
+ Ok(())
+}
+
+fn validate_user_option_keys(
+ user_options: &HashMap<String, String>,
+ index_type: &str,
+ field_name: &str,
+) -> crate::Result<()> {
+ let mut unknown = user_options
+ .keys()
+ .filter(|key| !is_supported_user_option_key(key, index_type,
field_name))
+ .cloned()
+ .collect::<Vec<_>>();
+ if unknown.is_empty() {
+ return Ok(());
+ }
+
+ unknown.sort();
+ Err(crate::Error::ConfigInvalid {
+ message: format!(
+ "Unknown vindex option(s) for index_type '{}': {}",
+ index_type,
+ unknown.join(", ")
+ ),
+ })
+}
+
+fn is_supported_user_option_key(key: &str, index_type: &str, field_name: &str)
-> bool {
+ if key == "index.type" {
+ return true;
+ }
+ if is_allowed_native_key(key, index_type) {
+ return true;
+ }
+
+ let index_prefix = format!("{index_type}.");
+ if let Some(suffix) = key.strip_prefix(&index_prefix) {
+ return is_allowed_paimon_suffix(suffix, index_type);
+ }
+
+ let field_prefix = format!("fields.{field_name}.");
+ if let Some(suffix) = key.strip_prefix(&field_prefix) {
+ return is_allowed_paimon_suffix(suffix, index_type);
+ }
+
+ false
+}
+
+fn is_allowed_native_key(key: &str, index_type: &str) -> bool {
+ match key {
+ "dimension" | "nlist" | "metric" => true,
+ "pq.m" | "use-opq" => index_type == IVF_PQ_IDENTIFIER,
+ "hnsw.m" | "hnsw.ef-construction" | "hnsw.max-level" => {
+ index_type == IVF_HNSW_FLAT_IDENTIFIER || index_type ==
IVF_HNSW_SQ_IDENTIFIER
+ }
+ _ => false,
+ }
+}
+
+fn is_allowed_paimon_suffix(suffix: &str, index_type: &str) -> bool {
+ match suffix {
+ "dimension" | "nlist" | "distance.metric" => true,
+ "pq.m" | "pq.use-opq" => index_type == IVF_PQ_IDENTIFIER,
+ "hnsw.m" | "hnsw.ef-construction" | "hnsw.max-level" => {
+ index_type == IVF_HNSW_FLAT_IDENTIFIER || index_type ==
IVF_HNSW_SQ_IDENTIFIER
+ }
+ _ => false,
+ }
+}
+
+fn resolve_dimension(
+ table_options: &HashMap<String, String>,
+ user_options: &HashMap<String, String>,
+ index_type: &str,
+ field: &DataField,
+) -> crate::Result<String> {
+ if let DataType::Vector(vector) = field.data_type() {
+ return Ok(vector.length().to_string());
+ }
+
+ Ok(option_value(
+ table_options,
+ user_options,
+ field.name(),
+ index_type,
+ "dimension",
+ "dimension",
+ DEFAULT_DIMENSION,
+ ))
+}
+
+fn option_value(
+ table_options: &HashMap<String, String>,
+ user_options: &HashMap<String, String>,
+ field_name: &str,
+ index_type: &str,
+ native_key: &str,
+ paimon_suffix: &str,
+ default_value: &str,
+) -> String {
+ optional_value(
+ table_options,
+ user_options,
+ field_name,
+ index_type,
+ native_key,
+ paimon_suffix,
+ )
+ .unwrap_or_else(|| default_value.to_string())
+}
+
+fn optional_value(
+ table_options: &HashMap<String, String>,
+ user_options: &HashMap<String, String>,
+ field_name: &str,
+ index_type: &str,
+ native_key: &str,
+ paimon_suffix: &str,
+) -> Option<String> {
+ for options in [user_options, table_options] {
+ for key in [
+ format!("fields.{field_name}.{paimon_suffix}"),
+ format!("{index_type}.{paimon_suffix}"),
+ native_key.to_string(),
+ ] {
+ if let Some(value) = options.get(&key) {
+ return Some(value.clone());
+ }
+ }
+ }
+ None
+}
+
+fn normalize_metric(metric: &str) -> String {
+ metric.trim().to_ascii_lowercase().replace('-', "_")
+}
+
#[cfg(test)]
mod tests {
use super::*;
+ use crate::spec::{ArrayType, FloatType, VectorType};
+
+ fn array_float_field() -> DataField {
+ DataField::new(
+ 7,
+ "embedding".to_string(),
+ DataType::Array(ArrayType::new(DataType::Float(FloatType::new()))),
+ )
+ }
#[test]
fn test_vindex_index_type_identifier_helper() {
@@ -44,4 +339,227 @@ mod tests {
assert!(!is_vindex_index_type("lumina"));
assert!(!is_vindex_index_type("IVF-FLAT"));
}
+
+ #[test]
+ fn test_vindex_options_map_java_prefixed_keys_to_native_config() {
+ let table_options = HashMap::new();
+ let user_options = HashMap::from([
+ ("ivf-pq.dimension".to_string(), "8".to_string()),
+ ("ivf-pq.nlist".to_string(), "4".to_string()),
+ ("ivf-pq.distance.metric".to_string(), "cosine".to_string()),
+ ("ivf-pq.pq.m".to_string(), "2".to_string()),
+ ("ivf-pq.pq.use-opq".to_string(), "true".to_string()),
+ ]);
+
+ let options = VindexVectorIndexOptions::new(
+ &table_options,
+ &user_options,
+ IVF_PQ_IDENTIFIER,
+ &array_float_field(),
+ )
+ .unwrap();
+
+ assert_eq!(options.dimension(), 8);
+ assert_eq!(
+ options.native_options.get("index.type").map(String::as_str),
+ Some("ivf_pq")
+ );
+ assert_eq!(
+ options.native_options.get("metric").map(String::as_str),
+ Some("cosine")
+ );
+ assert_eq!(
+ options.native_options.get("pq.m").map(String::as_str),
+ Some("2")
+ );
+ assert_eq!(
+ options.native_options.get("use-opq").map(String::as_str),
+ Some("true")
+ );
+ }
+
+ #[test]
+ fn test_vindex_options_field_options_override_shared_table_options() {
+ let table_options = HashMap::from([
+ ("ivf-flat.dimension".to_string(), "8".to_string()),
+ ("ivf-flat.nlist".to_string(), "4".to_string()),
+ ("fields.embedding.nlist".to_string(), "2".to_string()),
+ ]);
+ let user_options = HashMap::new();
+
+ let options = VindexVectorIndexOptions::new(
+ &table_options,
+ &user_options,
+ IVF_FLAT_IDENTIFIER,
+ &array_float_field(),
+ )
+ .unwrap();
+
+ assert_eq!(
+ options.native_options.get("nlist").map(String::as_str),
+ Some("2")
+ );
+ }
+
+ #[test]
+ fn test_vindex_options_vector_type_uses_type_dimension() {
+ let field = DataField::new(
+ 8,
+ "embedding".to_string(),
+ DataType::Vector(
+ VectorType::try_new(true, 16,
DataType::Float(FloatType::new())).unwrap(),
+ ),
+ );
+ let table_options = HashMap::from([
+ ("ivf-flat.dimension".to_string(), "128".to_string()),
+ ("ivf-flat.nlist".to_string(), "4".to_string()),
+ ]);
+ let user_options = HashMap::new();
+
+ let options = VindexVectorIndexOptions::new(
+ &table_options,
+ &user_options,
+ IVF_FLAT_IDENTIFIER,
+ &field,
+ )
+ .unwrap();
+
+ assert_eq!(options.dimension(), 16);
+ assert_eq!(
+ options.native_options.get("dimension").map(String::as_str),
+ Some("16")
+ );
+ }
+
+ #[test]
+ fn test_vindex_options_reject_mismatched_native_index_type() {
+ let table_options = HashMap::new();
+ let user_options = HashMap::from([
+ ("index.type".to_string(), "ivf_flat".to_string()),
+ ("dimension".to_string(), "8".to_string()),
+ ("nlist".to_string(), "4".to_string()),
+ ]);
+
+ let err = VindexVectorIndexOptions::new(
+ &table_options,
+ &user_options,
+ IVF_PQ_IDENTIFIER,
+ &array_float_field(),
+ )
+ .expect_err("mismatched index.type should be rejected");
+
+ assert!(matches!(err, crate::Error::ConfigInvalid { .. }));
+ }
+
+ #[test]
+ fn test_vindex_options_reject_invalid_pq_config() {
+ let table_options = HashMap::new();
+ let user_options = HashMap::from([
+ ("ivf-pq.dimension".to_string(), "7".to_string()),
+ ("ivf-pq.nlist".to_string(), "4".to_string()),
+ ("ivf-pq.pq.m".to_string(), "2".to_string()),
+ ]);
+
+ let err = VindexVectorIndexOptions::new(
+ &table_options,
+ &user_options,
+ IVF_PQ_IDENTIFIER,
+ &array_float_field(),
+ )
+ .expect_err("invalid native config should be rejected");
+
+ assert!(
+ matches!(err, crate::Error::DataInvalid { message, .. } if
message.contains("dimension 7 must be divisible by m 2"))
+ );
+ }
+
+ #[test]
+ fn test_vindex_options_reject_unknown_user_options() {
+ let table_options = HashMap::new();
+ let user_options = HashMap::from([
+ ("ivf-flat.dimension".to_string(), "8".to_string()),
+ ("ivf-flat.nlsit".to_string(), "4".to_string()),
+ ]);
+
+ let err = VindexVectorIndexOptions::new(
+ &table_options,
+ &user_options,
+ IVF_FLAT_IDENTIFIER,
+ &array_float_field(),
+ )
+ .expect_err("unknown user option should be rejected");
+
+ assert!(
+ matches!(err, crate::Error::ConfigInvalid { message } if
message.contains("ivf-flat.nlsit"))
+ );
+ }
+
+ #[test]
+ fn test_vindex_options_reject_non_applicable_user_options() {
+ let table_options = HashMap::new();
+ let user_options = HashMap::from([
+ ("ivf-flat.dimension".to_string(), "8".to_string()),
+ ("ivf-flat.nlist".to_string(), "4".to_string()),
+ ("ivf-flat.pq.m".to_string(), "2".to_string()),
+ ]);
+
+ let err = VindexVectorIndexOptions::new(
+ &table_options,
+ &user_options,
+ IVF_FLAT_IDENTIFIER,
+ &array_float_field(),
+ )
+ .expect_err("non-applicable user option should be rejected");
+
+ assert!(
+ matches!(err, crate::Error::ConfigInvalid { message } if
message.contains("ivf-flat.pq.m"))
+ );
+ }
+
+ #[test]
+ fn test_vindex_options_defaults_align_java_docs() {
+ let options = VindexVectorIndexOptions::new(
+ &HashMap::new(),
+ &HashMap::new(),
+ IVF_FLAT_IDENTIFIER,
+ &array_float_field(),
+ )
+ .unwrap();
+
+ assert_eq!(
+ options.native_options.get("dimension").map(String::as_str),
+ Some("128")
+ );
+ assert_eq!(
+ options.native_options.get("metric").map(String::as_str),
+ Some("inner_product")
+ );
+ assert_eq!(
+ options.native_options.get("nlist").map(String::as_str),
+ Some("256")
+ );
+ }
+
+ #[test]
+ fn test_native_index_type_helper() {
+ assert_eq!(native_index_type(IVF_FLAT_IDENTIFIER), Some("ivf_flat"));
+ assert_eq!(native_index_type(IVF_PQ_IDENTIFIER), Some("ivf_pq"));
+ assert_eq!(
+ native_index_type(IVF_HNSW_FLAT_IDENTIFIER),
+ Some("ivf_hnsw_flat")
+ );
+ assert_eq!(
+ native_index_type(IVF_HNSW_SQ_IDENTIFIER),
+ Some("ivf_hnsw_sq")
+ );
+ assert_eq!(native_index_type("btree"), None);
+ }
+
+ #[test]
+ fn test_array_field_helper_is_not_vector() {
+ assert!(matches!(
+ array_float_field().data_type(),
+ DataType::Array(_)
+ ));
+ }
}
diff --git a/docs/src/sql.md b/docs/src/sql.md
index 70b8099..b97a654 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -105,6 +105,12 @@ The following SQL data types are supported in CREATE TABLE
and mapped to their c
| `MAP(key, value)` | MapType | e.g. `MAP(STRING, INT)` |
| `STRUCT<field TYPE, ...>` | RowType | e.g. `STRUCT<city STRING, zip INT>` |
+For vector search tables created from SQL, use `ARRAY<FLOAT>` for embedding
+columns. Existing Paimon tables may also expose logical `VECTOR<FLOAT,N>`
+columns; DataFusion reads those as Arrow `FixedSizeList<Float32>`, and vindex
+index creation uses `N` as the vector dimension. `SHOW CREATE TABLE` currently
+does not round-trip `VECTOR` columns.
+
### Variant Usage
`VARIANT` stores semi-structured data using the same logical value + metadata
binary shape as Paimon Java. Use it for JSON-like fields whose schema may
differ row by row.
@@ -632,6 +638,106 @@ Rollback a table to a specific timestamp:
CALL sys.rollback_to_timestamp(table => 'paimon.my_db.my_table', timestamp =>
1234567890000);
```
+### create_global_index
+
+Build and commit a global index for a table column:
+
+```sql
+CALL sys.create_global_index(
+ table => 'paimon.my_db.my_table',
+ index_column => 'id',
+ index_type => 'btree'
+);
+```
+
+`index_type` defaults to `btree`. BTree indexes support scalar columns and do
+not accept the `options` argument yet.
+
+The current global-index builders require a row-tracking data-evolution table
+with global indexes enabled. They do not support primary-key tables or tables
+with deletion vectors enabled:
+
+```sql
+CREATE TABLE paimon.my_db.items (
+ id INT,
+ embedding ARRAY<FLOAT>
+) WITH (
+ 'bucket' = '1',
+ 'row-tracking.enabled' = 'true',
+ 'data-evolution.enabled' = 'true',
+ 'global-index.enabled' = 'true',
+ 'global-index.row-count-per-shard' = '100000'
+);
+```
+
+For vector indexes backed by vindex, set `index_type` to one of `ivf-flat`,
+`ivf-pq`, `ivf-hnsw-flat`, or `ivf-hnsw-sq`:
+
+```sql
+CALL sys.create_global_index(
+ table => 'paimon.my_db.items',
+ index_column => 'embedding',
+ index_type => 'ivf-flat',
+ options =>
'ivf-flat.dimension=4,ivf-flat.nlist=256,ivf-flat.distance.metric=inner_product'
+);
+```
+
+The `options` argument is a comma-separated `key=value` string. User options
+override table options. Use keys prefixed by the selected index type, or set
+field-level table options with `fields.<column>.<option>`:
+
+```sql
+CREATE TABLE paimon.my_db.image_items (
+ id INT,
+ embedding ARRAY<FLOAT>
+) WITH (
+ 'bucket' = '1',
+ 'row-tracking.enabled' = 'true',
+ 'data-evolution.enabled' = 'true',
+ 'global-index.enabled' = 'true',
+ 'fields.embedding.dimension' = '768',
+ 'fields.embedding.distance.metric' = 'cosine',
+ 'fields.embedding.nlist' = '1024'
+);
+```
+
+Supported vindex options:
+
+| Option | Default | Applies To | Description |
+|---|---:|---|---|
+| `<index-type>.dimension` | `128` | all vindex types | Vector dimension for
`ARRAY<FLOAT>` columns. Existing `VECTOR<FLOAT,N>` columns use `N` from the
type. |
+| `<index-type>.distance.metric` | `inner_product` | all vindex types |
Distance metric: `inner_product`, `cosine`, or `l2`. |
+| `<index-type>.nlist` | `256` | all vindex types | Number of IVF lists. |
+| `<index-type>.pq.m` | `16` | `ivf-pq` | Number of product-quantization
sub-vectors. The dimension must be divisible by this value. |
+| `<index-type>.pq.use-opq` | `false` | `ivf-pq` | Whether to enable OPQ
before PQ encoding. |
+| `<index-type>.hnsw.m` | native default | HNSW vindex types | HNSW graph
connectivity. |
+| `<index-type>.hnsw.ef-construction` | native default | HNSW vindex types |
HNSW construction beam width. |
+| `<index-type>.hnsw.max-level` | native default | HNSW vindex types | Maximum
HNSW graph level. |
+
+Native vindex aliases are also accepted in the `options` string: `dimension`,
+`metric`, `nlist`, `pq.m`, `use-opq`, and `hnsw.*`.
+
+Inspect committed index files with the `$table_indexes` system table:
+
+```sql
+SELECT index_type, index_field_name, row_count, row_range_start, row_range_end
+FROM paimon.my_db.items$table_indexes;
+```
+
+### drop_global_index
+
+Drop a committed BTree global index:
+
+```sql
+CALL sys.drop_global_index(
+ table => 'paimon.my_db.my_table',
+ index_column => 'id',
+ index_type => 'btree'
+);
+```
+
+Only BTree indexes can be dropped through this procedure currently.
+
### create_lumina_index
Build and commit a Lumina global vector index for a table column:
@@ -775,7 +881,11 @@ CROSS JOIN LATERAL vector_search(
## Vector Search
-Paimon supports approximate nearest neighbor (ANN) vector search via the
Lumina vector index. The `vector_search` table-valued function is registered as
a UDTF on the DataFusion session context.
+Paimon supports approximate nearest neighbor (ANN) vector search through global
+vector indexes. DataFusion can search vindex indexes created by
+`CALL sys.create_global_index` and Lumina indexes created by
+`CALL sys.create_lumina_index`. The `vector_search` table-valued function is
+registered as a UDTF on the DataFusion session context.
### Registration
@@ -808,7 +918,9 @@ Example:
SELECT * FROM vector_search('paimon.my_db.items', 'embedding', '[1.0, 0.0,
0.0, 0.0]', 10);
```
-The function performs ANN search across all Lumina vector index files for the
target column, merges results, and returns the top-k rows ordered by relevance
score. If no matching index is found, an empty result is returned.
+The function performs ANN search across all matching vector index files for the
+target column, merges results, and returns the top-k rows ordered by relevance
+score. If no matching index is found, an empty result is returned.
### Lateral Joins
@@ -838,17 +950,25 @@ The distance metric is configured at index creation time
via table options:
| `cosine` | Cosine similarity |
| `l2` | Euclidean (L2) distance |
-### Vector Index Options
+### Vindex Index Options
-Vector index behavior is configured via table options prefixed with `lumina.`:
+For vindex-backed search, build the index with
+`CALL sys.create_global_index` and an index type such as `ivf-flat` or
+`ivf-pq`. See [create_global_index](#create_global_index) for the supported
+index types, table requirements, and option keys.
+
+### Lumina Index Options
+
+Lumina index behavior is configured via table options prefixed with `lumina.`:
| Option | Description |
|---|---|
-| `lumina.dimension` | Vector dimension |
-| `lumina.metric` | Distance metric (`inner_product`, `cosine`, `l2`) |
-| `lumina.index-type` | Index type (default: `diskann`) |
+| `lumina.index.dimension` | Vector dimension |
+| `lumina.distance.metric` | Distance metric (`inner_product`, `cosine`, `l2`)
|
+| `lumina.index.type` | Index type (default: `diskann`) |
+| `lumina.encoding.type` | Encoding type (default: `pq`) |
-### Environment
+### Lumina Environment
The Lumina native library must be available at runtime. Set the
`LUMINA_LIB_PATH` environment variable to the path of the shared library, or
place it in the platform default location.
@@ -1249,6 +1369,31 @@ Columns:
| `total_buckets` | INT | Total bucket count for the partition (0 unless
catalog-tracked) |
| `done` | BOOLEAN | Whether the partition is marked done (false unless
catalog-tracked) |
+### $table_indexes
+
+View committed global index files, including BTree indexes, vector indexes, and
+deletion-vector metadata:
+
+```sql
+SELECT * FROM paimon.default.my_table$table_indexes;
+```
+
+Columns:
+
+| Column | Type | Description |
+|---|---|---|
+| `partition` | STRING | Partition spec for the indexed data, or `NULL` for
unpartitioned tables |
+| `bucket` | INT | Bucket id covered by the index file |
+| `index_type` | STRING | Index type, such as `btree`, `ivf-flat`, `lumina`,
or `DELETION_VECTORS` |
+| `file_name` | STRING | Index file name under the table index directory |
+| `file_size` | BIGINT | Index file size in bytes |
+| `row_count` | BIGINT | Number of rows covered by the index file |
+| `dv_ranges` | ARRAY | Deletion-vector ranges, only populated for
deletion-vector metadata |
+| `row_range_start` | BIGINT | First row id covered by the index file |
+| `row_range_end` | BIGINT | Last row id covered by the index file |
+| `index_field_id` | INT | Field id of the indexed column |
+| `index_field_name` | STRING | Name of the indexed column |
+
### $physical_files_size
Scan the table directory recursively and compute the total size of recognized
physical files on disk, categorized by file type. This table is a diagnostic
size summary; orphan cleanup needs file-level candidates and retention checks,
not just aggregate size differences.
@@ -1357,6 +1502,22 @@ rows (`DELETE` / `UPDATE_BEFORE`), deletion vectors,
cross-partition dynamic
bucket writes, or advanced aggregation options such as `ignore-retract`,
`distinct`, `nested-key`, `count-limit`, and sequence groups.
+### Global Index Options
+
+Set these options when building global indexes with
+`CALL sys.create_global_index`. The current DataFusion builders require
+row-tracking and data evolution, and reject primary-key tables and tables with
+deletion vectors enabled.
+
+| Option | Default | Description |
+|---|---:|---|
+| `row-tracking.enabled` | `false` | Enables stable row ids required by global
index files. |
+| `data-evolution.enabled` | `false` | Enables row-id-aware table evolution
and partial-column writes. |
+| `global-index.enabled` | `false` | Enables global index metadata and
global-index-aware reads. |
+| `global-index.row-count-per-shard` | `100000` | Maximum row count per vector
global-index shard. |
+| `sorted-index.records-per-range` | `100000` | Maximum row count per BTree
range. |
+| `global-index.search-mode` | `fast` | Global index coverage mode for reads:
`fast`, `full`, or `detail`. |
+
### Variant Shredding Options
Set these as table options when writing `VARIANT` columns to Parquet. The
@@ -1383,7 +1544,9 @@ the normal physical format without wrapping the writer.
| Option | Description |
|---|---|
| `'sequence.field' = 'col'` | Sequence field used to determine which record
wins during deduplication |
+| `'row-tracking.enabled' = 'true'` | Enable stable row ids |
| `'data-evolution.enabled' = 'true'` | Enable data evolution (partial-column
writes, row-level UPDATE/MERGE/DELETE) |
+| `'global-index.enabled' = 'true'` | Enable global index metadata and reads |
| `'deletion-vectors.enabled' = 'true'` | Enable deletion vectors |
| `'cross-partition-update.enabled' = 'true'` | Allow cross-partition updates |
| `'changelog-producer' = 'input'` | Changelog producer (PK tables with input
mode reject writes) |
