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 d04eb4bd feat(python): expose prepared sorted global index builds
(#1088)
d04eb4bd is described below
commit d04eb4bd6a792bcfe3085bbf1092ae2048f19621
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Oct 10 13:31:17 2026 +0800
feat(python): expose prepared sorted global index builds (#1088)
---
.../python/python/pypaimon_rust/datafusion.pyi | 13 +
bindings/python/src/context.rs | 1 +
bindings/python/src/index_build.rs | 118 +++++
bindings/python/src/lib.rs | 1 +
bindings/python/src/predicate.rs | 76 ++-
bindings/python/src/table.rs | 6 +
bindings/python/src/write.rs | 9 +
bindings/python/tests/test_read.py | 33 +-
crates/paimon/src/spec/core_options.rs | 37 +-
.../paimon/src/spec/data_evolution_index_source.rs | 104 ++++
crates/paimon/src/spec/mod.rs | 2 +
crates/paimon/src/spec/predicate.rs | 142 ++++--
crates/paimon/src/table/partition_filter.rs | 21 +-
.../src/table/sorted_global_index_build_builder.rs | 96 ++--
.../preparation_tests.rs | 568 +++++++++++++++++++++
.../sorted_global_index_build_builder/writer.rs | 35 +-
crates/paimon/src/table/table_commit.rs | 5 +
.../table_commit/global_index_source_check.rs | 186 +++++++
.../src/table/table_commit/row_tracking_tests.rs | 13 +-
crates/paimon/src/table/vector_search_common.rs | 19 +-
20 files changed, 1339 insertions(+), 146 deletions(-)
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index 4ff2065d..0974cea2 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -223,6 +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_batch_write_builder(self) -> "BatchWriteBuilder": ...
def new_postpone_fixed_bucket_write_builder(self) ->
"PostponeFixedBucketWriteBuilder": ...
def new_stream_write_builder(self) -> "StreamWriteBuilder": ...
@@ -259,6 +260,18 @@ 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": ...
+ def build(self) -> List["CommitMessage"]:
+ """Build unpublished index files; the caller decides whether to commit
or abort."""
+ ...
+ def execute(self) -> int:
+ """Build and publish indexes, rejecting a change to the planned
snapshot."""
+ ...
+
class CommitMessage:
@staticmethod
def deserialize(data: bytes, *, version: int = 14) -> "CommitMessage":
diff --git a/bindings/python/src/context.rs b/bindings/python/src/context.rs
index 1c50299c..37a1f03e 100644
--- a/bindings/python/src/context.rs
+++ b/bindings/python/src/context.rs
@@ -726,6 +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::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
new file mode 100644
index 00000000..2b216153
--- /dev/null
+++ b/bindings/python/src/index_build.rs
@@ -0,0 +1,118 @@
+// 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.
+
+//! Parameter and commit-message wrappers for the core sorted-index builder.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use paimon::spec::Predicate;
+use paimon::table::{SortedGlobalIndexBuildBuilder, Table};
+use paimon_datafusion::runtime::runtime;
+use pyo3::prelude::*;
+use pyo3::types::PyDict;
+
+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 {
+ table: Arc<Table>,
+ column: Option<String>,
+ index_type: String,
+ options: HashMap<String, String>,
+ partition_filters: Vec<Predicate>,
+}
+
+impl PySortedGlobalIndexBuildBuilder {
+ pub(crate) fn new(table: Arc<Table>) -> Self {
+ Self {
+ table,
+ column: 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);
+ }
+ builder
+ .with_index_type(&self.index_type)
+ .with_options(self.options.clone());
+ for filter in &self.partition_filters {
+ builder.with_partition_filter(filter.clone())?;
+ }
+ Ok(builder)
+ }
+}
+
+#[pymethods]
+impl PySortedGlobalIndexBuildBuilder {
+ fn with_index_column(mut slf: PyRefMut<'_, Self>, column: String) ->
PyRefMut<'_, Self> {
+ slf.column = Some(column);
+ slf
+ }
+
+ fn with_index_type(mut slf: PyRefMut<'_, Self>, index_type: String) ->
PyRefMut<'_, Self> {
+ slf.index_type = index_type;
+ slf
+ }
+
+ fn with_options(
+ mut slf: PyRefMut<'_, Self>,
+ options: HashMap<String, String>,
+ ) -> PyRefMut<'_, Self> {
+ slf.options = options;
+ slf
+ }
+
+ fn with_partition_filter<'py>(
+ mut slf: PyRefMut<'py, Self>,
+ predicate: &Bound<'_, PyDict>,
+ ) -> PyResult<PyRefMut<'py, Self>> {
+ let filter = dict_to_table_predicate(predicate, slf.table.schema(),
true)?;
+ // Validation and conjunction belong to the core builder.
+ let mut builder = slf.builder().map_err(to_py_err)?;
+ builder
+ .with_partition_filter(filter.clone())
+ .map_err(to_py_err)?;
+ slf.partition_filters.push(filter);
+ Ok(slf)
+ }
+
+ fn build(&self, py: Python<'_>) -> PyResult<Vec<PyCommitMessage>> {
+ let builder = self.builder().map_err(to_py_err)?;
+ let messages = py
+ .detach(|| runtime().block_on(builder.build()))
+ .map_err(to_py_err)?;
+ Ok(messages.into_iter().map(PyCommitMessage::new).collect())
+ }
+
+ fn execute(&self, py: Python<'_>) -> PyResult<usize> {
+ let builder = self.builder().map_err(to_py_err)?;
+ py.detach(|| runtime().block_on(builder.execute()))
+ .map_err(to_py_err)
+ }
+}
diff --git a/bindings/python/src/lib.rs b/bindings/python/src/lib.rs
index 900e19ef..2478c02d 100644
--- a/bindings/python/src/lib.rs
+++ b/bindings/python/src/lib.rs
@@ -21,6 +21,7 @@ mod blob;
mod blob_uri_reader;
mod context;
mod error;
+mod index_build;
mod merge_into;
mod oss_cpp;
mod predicate;
diff --git a/bindings/python/src/predicate.rs b/bindings/python/src/predicate.rs
index f3f3ff7a..1986bad5 100644
--- a/bindings/python/src/predicate.rs
+++ b/bindings/python/src/predicate.rs
@@ -374,7 +374,7 @@ const METHOD_NOT_SUPPORTED: &[&str] = &["not"];
///
/// Errors:
/// - `ValueError` for unknown fields, missing keys, wrong literal counts,
`None`
-/// literals, empty/missing `children`, non-dict children, or non-list
+/// comparison literals outside IN/NOT IN, empty/missing `children`,
non-dict children, or non-list
/// `literals`/`children`.
/// - `NotImplementedError` for unsupported operators or unsupported literal
types.
pub(crate) fn dict_to_predicate(
@@ -482,7 +482,7 @@ fn leaf_to_predicate(
// Convert literals (DataType-driven), wrapping NotImplemented type
messages
// with field context.
- let to_datums = |obj: Option<Bound<'_, PyAny>>| -> PyResult<Vec<Datum>> {
+ let to_nullable_datums = |obj: Option<Bound<'_, PyAny>>| ->
PyResult<Vec<Option<Datum>>> {
let mut out = Vec::new();
if let Some(obj) = obj {
let list = obj
@@ -490,18 +490,29 @@ fn leaf_to_predicate(
.map_err(|_| PyValueError::new_err("'literals' must be a
list"))?;
for item in list.iter() {
if item.is_none() {
- return Err(PyValueError::new_err(
- "None is not a valid comparison literal; use
isNull/isNotNull",
- ));
+ out.push(None);
+ continue;
}
- out.push(
+ out.push(Some(
py_to_datum(&item, &data_type)
.map_err(|e| with_field_context(e, &field,
&data_type))?,
- );
+ ));
}
}
Ok(out)
};
+ let to_datums = |obj| -> PyResult<Vec<Datum>> {
+ to_nullable_datums(obj)?
+ .into_iter()
+ .map(|value| {
+ value.ok_or_else(|| {
+ PyValueError::new_err(
+ "None is not a valid comparison literal; use
isNull/isNotNull",
+ )
+ })
+ })
+ .collect()
+ };
let result = match method {
"equal" => pb.equal(&field, one(to_datums(literals_obj)?)?),
@@ -518,20 +529,8 @@ fn leaf_to_predicate(
ensure_no_literals(method, literals_obj)?;
pb.is_not_null(&field)
}
- "in" => {
- let ds = to_datums(literals_obj)?;
- if ds.is_empty() {
- return Err(PyValueError::new_err("'in' requires at least 1
literal"));
- }
- pb.is_in(&field, ds)
- }
- "notIn" => {
- let ds = to_datums(literals_obj)?;
- if ds.is_empty() {
- return Err(PyValueError::new_err("'notIn' requires at least 1
literal"));
- }
- pb.is_not_in(&field, ds)
- }
+ "in" => pb.is_in_with_nulls(&field, to_nullable_datums(literals_obj)?),
+ "notIn" => pb.is_not_in_with_nulls(&field,
to_nullable_datums(literals_obj)?),
"startsWith" => pb.starts_with(&field, one(to_datums(literals_obj)?)?),
"endsWith" => pb.ends_with(&field, one(to_datums(literals_obj)?)?),
"contains" => pb.contains(&field, one(to_datums(literals_obj)?)?),
@@ -1069,6 +1068,41 @@ mod tests {
});
}
+ #[test]
+ fn nullable_and_empty_set_literals_use_core_semantics() {
+ Python::attach(|py| {
+ let fields = test_fields();
+ let pb = PredicateBuilder::new(&fields);
+ for method in ["in", "notIn"] {
+ let dict = leaf_dict(py, method, "id", &[]);
+ let expected = if method == "in" {
+ pb.is_in("id", vec![]).unwrap()
+ } else {
+ pb.is_not_in("id", vec![]).unwrap()
+ };
+ assert_eq!(dict_to_predicate(&dict, &fields, true).unwrap(),
expected);
+ dict.set_item(
+ "literals",
+ PyList::new(
+ py,
+ [
+ py.None(),
+
1i32.into_pyobject(py).unwrap().into_any().unbind(),
+ ],
+ )
+ .unwrap(),
+ )
+ .unwrap();
+ let expected = if method == "in" {
+ pb.is_in("id", vec![Datum::Int(1)]).unwrap()
+ } else {
+ Predicate::AlwaysFalse
+ };
+ assert_eq!(dict_to_predicate(&dict, &fields, true).unwrap(),
expected);
+ }
+ });
+ }
+
#[test]
fn wrong_literal_count_raises_value_error() {
Python::attach(|py| {
diff --git a/bindings/python/src/table.rs b/bindings/python/src/table.rs
index fd39b0c9..a2771e41 100644
--- a/bindings/python/src/table.rs
+++ b/bindings/python/src/table.rs
@@ -206,6 +206,12 @@ 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_batch_write_builder(&self) -> PyBatchWriteBuilder {
PyBatchWriteBuilder::new(Arc::clone(&self.inner))
}
diff --git a/bindings/python/src/write.rs b/bindings/python/src/write.rs
index da7d6b9b..6f6901d4 100644
--- a/bindings/python/src/write.rs
+++ b/bindings/python/src/write.rs
@@ -1459,6 +1459,15 @@ pub struct PyCommitMessage {
origin: Option<MessageOrigin>,
}
+impl PyCommitMessage {
+ pub(crate) fn new(inner: CommitMessage) -> Self {
+ Self {
+ inner,
+ origin: None,
+ }
+ }
+}
+
#[pymethods]
impl PyCommitMessage {
/// Export the Java `CommitMessageSerializer` v14 body (no version header).
diff --git a/bindings/python/tests/test_read.py
b/bindings/python/tests/test_read.py
index abd0cf16..b98e5a59 100644
--- a/bindings/python/tests/test_read.py
+++ b/bindings/python/tests/test_read.py
@@ -1046,14 +1046,39 @@ def test_filter_out_of_range_raises():
{"method": "equal", "field": "small", "literals": [9999]})
-def test_filter_wrong_literal_count_raises():
[email protected]("literals", [[], [1, 2]])
+def test_filter_wrong_literal_count_raises(literals):
with tempfile.TemporaryDirectory() as warehouse:
table = _make_table_with_data(warehouse)
b = table.new_read_builder()
with pytest.raises(ValueError):
- b.with_filter({"method": "equal", "field": "id", "literals": [1,
2]})
- with pytest.raises(ValueError):
- b.with_filter({"method": "in", "field": "id", "literals": []})
+ b.with_filter({"method": "equal", "field": "id", "literals":
literals})
+
+
[email protected]("method, literals, expected_ids", [
+ ("in", [], []),
+ ("notIn", [], [2, 3]),
+ ("in", [None], []),
+ ("notIn", [None], []),
+ ("in", [None, 10], [2]),
+ ("notIn", [None, 10], []),
+ ("in", [10], [2]),
+ ("notIn", [10], [3]),
+])
+def test_filter_set_predicates_follow_java_null_semantics(method, literals,
expected_ids):
+ with tempfile.TemporaryDirectory() as warehouse:
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": warehouse})
+ ctx.sql("CREATE SCHEMA paimon.setdb")
+ ctx.sql("CREATE TABLE paimon.setdb.t (id INT, value INT)")
+ ctx.sql("INSERT INTO paimon.setdb.t VALUES (1, NULL), (2, 10), (3,
20)")
+ table = PaimonCatalog({"warehouse": warehouse}).get_table("setdb.t")
+ builder = table.new_read_builder().with_filter({
+ "method": method, "field": "value", "literals": literals,
+ }).with_projection(["id"])
+ batches = builder.new_read().read(builder.new_scan().plan().splits())
+ actual_ids = sorted(value for batch in batches for value in
batch.column(0).to_pylist())
+ assert actual_ids == expected_ids
def test_filter_compound_with_unsupported_child_fails():
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index b48208f2..79354eea 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -36,6 +36,7 @@ const GLOBAL_INDEX_THREAD_NUM_OPTION: &str =
"global-index.thread-num";
const GLOBAL_INDEX_VINDEX_READ_THREAD_NUM_OPTION: &str =
"global-index.vindex.read-thread-num";
const GLOBAL_INDEX_COLUMN_UPDATE_ACTION_OPTION: &str =
"global-index.column-update-action";
pub(crate) const INDEX_FILE_IN_DATA_FILE_DIR_OPTION: &str =
"index-file-in-data-file-dir";
+const SORTED_INDEX_RECORDS_PER_FILE_OPTION: &str =
"sorted-index.records-per-file";
const SORTED_INDEX_RECORDS_PER_RANGE_OPTION: &str =
"sorted-index.records-per-range";
const BTREE_INDEX_RECORDS_PER_RANGE_OPTION: &str =
"btree-index.records-per-range";
const BTREE_INDEX_CACHE_SIZE_OPTION: &str = "btree-index.cache-size";
@@ -163,6 +164,7 @@ const DEFAULT_DYNAMIC_BUCKET_TARGET_ROW_NUM: i64 =
2_000_000;
pub(crate) const MAX_DYNAMIC_BUCKETS: i32 = 32768;
const DYNAMIC_BUCKET_MAX_BUCKETS_OPTION: &str = "dynamic-bucket.max-buckets";
const DEFAULT_GLOBAL_INDEX_ROW_COUNT_PER_SHARD: i64 = 100_000;
+const DEFAULT_SORTED_INDEX_RECORDS_PER_FILE: i64 = 25_000_000;
const DEFAULT_GLOBAL_INDEX_THREAD_NUM: i64 = 32;
pub(crate) const DEFAULT_GLOBAL_INDEX_VINDEX_READ_THREAD_NUM: usize = 64;
const MAX_GLOBAL_INDEX_VINDEX_READ_THREAD_NUM: i64 =
tokio::sync::Semaphore::MAX_PERMITS as i64;
@@ -1120,17 +1122,18 @@ impl<'a> CoreOptions<'a> {
}
pub fn sorted_index_records_per_range(&self) -> crate::Result<i64> {
- let option = if self
- .options
- .contains_key(SORTED_INDEX_RECORDS_PER_RANGE_OPTION)
- {
- SORTED_INDEX_RECORDS_PER_RANGE_OPTION
- } else {
- BTREE_INDEX_RECORDS_PER_RANGE_OPTION
- };
+ // Java SortedIndexOptions uses records-per-file as the primary key.
+ let option = [
+ SORTED_INDEX_RECORDS_PER_FILE_OPTION,
+ SORTED_INDEX_RECORDS_PER_RANGE_OPTION,
+ BTREE_INDEX_RECORDS_PER_RANGE_OPTION,
+ ]
+ .into_iter()
+ .find(|option| self.options.contains_key(*option))
+ .unwrap_or(SORTED_INDEX_RECORDS_PER_FILE_OPTION);
let value = self
.parse_i64_option(option)?
- .unwrap_or(DEFAULT_GLOBAL_INDEX_ROW_COUNT_PER_SHARD);
+ .unwrap_or(DEFAULT_SORTED_INDEX_RECORDS_PER_FILE);
if value <= 0 {
return Err(crate::Error::DataInvalid {
message: format!("Option '{}' must be greater than 0, got:
{}", option, value),
@@ -2344,7 +2347,7 @@ mod tests {
);
assert_eq!(
core_options.sorted_index_records_per_range().unwrap(),
- 100_000
+ 25_000_000
);
assert_eq!(
core_options.btree_index_fallback_scan_max_size().unwrap(),
@@ -2631,6 +2634,7 @@ mod tests {
#[test]
fn test_sorted_index_records_per_range_rejects_invalid_values() {
for option in [
+ SORTED_INDEX_RECORDS_PER_FILE_OPTION,
SORTED_INDEX_RECORDS_PER_RANGE_OPTION,
BTREE_INDEX_RECORDS_PER_RANGE_OPTION,
] {
@@ -2676,6 +2680,19 @@ mod tests {
.unwrap(),
10
);
+ let mut primary = options;
+ primary.insert(SORTED_INDEX_RECORDS_PER_FILE_OPTION.into(),
"30".into());
+ assert_eq!(
+ CoreOptions::new(&primary)
+ .sorted_index_records_per_range()
+ .unwrap(),
+ 30
+ );
+ // A bad primary value must fail rather than silently using an alias.
+ primary.insert(SORTED_INDEX_RECORDS_PER_FILE_OPTION.into(),
"0".into());
+ assert!(CoreOptions::new(&primary)
+ .sorted_index_records_per_range()
+ .is_err());
}
#[test]
diff --git a/crates/paimon/src/spec/data_evolution_index_source.rs
b/crates/paimon/src/spec/data_evolution_index_source.rs
new file mode 100644
index 00000000..510888dd
--- /dev/null
+++ b/crates/paimon/src/spec/data_evolution_index_source.rs
@@ -0,0 +1,104 @@
+// 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.
+
+//! Java DataEvolutionIndexSourceMeta, carried by GlobalIndexMeta.source_meta.
+
+use crate::{Error, Result};
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct DataEvolutionIndexSourceMeta {
+ scan_snapshot_id: i64,
+}
+
+impl DataEvolutionIndexSourceMeta {
+ const MAGIC: [u8; 4] = *b"DEIX";
+ const VERSION: i32 = 1;
+
+ pub fn new(scan_snapshot_id: i64) -> Result<Self> {
+ if scan_snapshot_id <= 0 {
+ return Err(invalid("Scan snapshot id must be positive."));
+ }
+ Ok(Self { scan_snapshot_id })
+ }
+
+ pub fn scan_snapshot_id(&self) -> i64 {
+ self.scan_snapshot_id
+ }
+
+ pub fn serialize(&self) -> Vec<u8> {
+ let mut bytes = Vec::with_capacity(16);
+ bytes.extend_from_slice(&Self::MAGIC);
+ bytes.extend_from_slice(&Self::VERSION.to_be_bytes());
+ bytes.extend_from_slice(&self.scan_snapshot_id.to_be_bytes());
+ bytes
+ }
+
+ pub fn is_data_evolution_meta(bytes: Option<&[u8]>) -> bool {
+ bytes.is_some_and(|bytes| bytes.starts_with(&Self::MAGIC))
+ }
+
+ pub fn deserialize(bytes: &[u8]) -> Result<Self> {
+ if bytes.len() != 16 || !Self::is_data_evolution_meta(Some(bytes)) {
+ return Err(invalid("Invalid data-evolution index source
metadata."));
+ }
+ let version = i32::from_be_bytes(bytes[4..8].try_into().unwrap());
+ if version != Self::VERSION {
+ return Err(invalid(format!(
+ "Unsupported data-evolution index source version: {version}."
+ )));
+ }
+ Self::new(i64::from_be_bytes(bytes[8..16].try_into().unwrap()))
+ }
+}
+
+fn invalid(message: impl Into<String>) -> Error {
+ Error::DataInvalid {
+ message: message.into(),
+ source: None,
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn java_wire_format_and_invalid_frames() {
+ let java = [0x44, 0x45, 0x49, 0x58, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0,
42];
+ let source = DataEvolutionIndexSourceMeta::new(42).unwrap();
+ assert_eq!(source.serialize(), java);
+ assert_eq!(
+ DataEvolutionIndexSourceMeta::deserialize(&java).unwrap(),
+ source
+ );
+ for length in 0..16 {
+
assert!(DataEvolutionIndexSourceMeta::deserialize(&java[..length]).is_err());
+ }
+ for offset in [0, 7, 15] {
+ let mut invalid = java;
+ invalid[offset] = 0;
+
assert!(DataEvolutionIndexSourceMeta::deserialize(&invalid).is_err());
+ }
+ let mut trailing = java.to_vec();
+ trailing.push(0);
+ assert!(DataEvolutionIndexSourceMeta::deserialize(&trailing).is_err());
+ assert!(!DataEvolutionIndexSourceMeta::is_data_evolution_meta(None));
+ assert!(!DataEvolutionIndexSourceMeta::is_data_evolution_meta(Some(
+ &[0, 0, 0, 1]
+ )));
+ }
+}
diff --git a/crates/paimon/src/spec/mod.rs b/crates/paimon/src/spec/mod.rs
index 2a0df88b..8ef2af95 100644
--- a/crates/paimon/src/spec/mod.rs
+++ b/crates/paimon/src/spec/mod.rs
@@ -83,6 +83,8 @@ pub use index_file_meta::*;
mod pk_index_source;
pub use pk_index_source::*;
+mod data_evolution_index_source;
+pub use data_evolution_index_source::*;
mod index_manifest;
pub use index_manifest::{IndexManifest, IndexManifestEntry};
diff --git a/crates/paimon/src/spec/predicate.rs
b/crates/paimon/src/spec/predicate.rs
index 3cfa72c2..e3107e8e 100644
--- a/crates/paimon/src/spec/predicate.rs
+++ b/crates/paimon/src/spec/predicate.rs
@@ -432,6 +432,28 @@ impl Predicate {
Predicate::Not(inner) => *inner,
Predicate::AlwaysTrue => Predicate::AlwaysFalse,
Predicate::AlwaysFalse => Predicate::AlwaysTrue,
+ Predicate::Leaf {
+ column,
+ index,
+ data_type,
+ op,
+ literals,
+ } if literals.is_empty()
+ && matches!(op, PredicateOperator::In |
PredicateOperator::NotIn) =>
+ {
+ // Java negates the leaf function, retaining its NULL
exclusion.
+ Predicate::Leaf {
+ column,
+ index,
+ data_type,
+ op: if op == PredicateOperator::In {
+ PredicateOperator::NotIn
+ } else {
+ PredicateOperator::In
+ },
+ literals,
+ }
+ }
other => Predicate::Not(Box::new(other)),
}
}
@@ -923,26 +945,8 @@ fn parse_rest_leaf(
"GREATER_OR_EQUAL" => binary(one_literal()?, &|l|
builder.greater_or_equal(field, l)),
"LESS_THAN" => binary(one_literal()?, &|l| builder.less_than(field,
l)),
"LESS_OR_EQUAL" => binary(one_literal()?, &|l|
builder.less_or_equal(field, l)),
- "IN" => {
- // Java In: null literals never match; an empty/all-null list
matches nothing.
- let datums: Vec<Datum> =
literals.iter().flatten().cloned().collect();
- if datums.is_empty() {
- Ok(Predicate::AlwaysFalse)
- } else {
- builder.is_in(field, datums)
- }
- }
- "NOT_IN" => {
- // Java NotIn: any null literal fails every row; an empty list
keeps
- // exactly the non-null rows.
- if literals.iter().any(Option::is_none) {
- Ok(Predicate::AlwaysFalse)
- } else if literals.is_empty() {
- builder.is_not_null(field)
- } else {
- builder.is_not_in(field,
literals.iter().flatten().cloned().collect())
- }
- }
+ "IN" => builder.is_in_with_nulls(field, literals),
+ "NOT_IN" => builder.is_not_in_with_nulls(field, literals),
"STARTS_WITH" => binary(one_literal()?, &|l|
builder.starts_with(field, l)),
"ENDS_WITH" => binary(one_literal()?, &|l| builder.ends_with(field,
l)),
"CONTAINS" => binary(one_literal()?, &|l| builder.contains(field, l)),
@@ -1405,19 +1409,31 @@ impl PredicateBuilder {
// -- set operators --
pub fn is_in(&self, field: &str, literals: Vec<Datum>) ->
Result<Predicate> {
- if literals.is_empty() {
- return Ok(Predicate::AlwaysFalse);
- }
self.leaf(field, PredicateOperator::In, literals)
}
pub fn is_not_in(&self, field: &str, literals: Vec<Datum>) ->
Result<Predicate> {
- if literals.is_empty() {
- return Ok(Predicate::AlwaysTrue);
- }
self.leaf(field, PredicateOperator::NotIn, literals)
}
+ /// Java IN ignores NULL literals; an empty/all-NULL list matches nothing.
+ pub fn is_in_with_nulls(&self, field: &str, literals: Vec<Option<Datum>>)
-> Result<Predicate> {
+ self.is_in(field, literals.into_iter().flatten().collect())
+ }
+
+ /// Java NOT IN fails every row if any literal is NULL.
+ pub fn is_not_in_with_nulls(
+ &self,
+ field: &str,
+ literals: Vec<Option<Datum>>,
+ ) -> Result<Predicate> {
+ if literals.iter().any(Option::is_none) {
+ self.resolve_field(field)?;
+ return Ok(Predicate::AlwaysFalse);
+ }
+ self.is_not_in(field, literals.into_iter().flatten().collect())
+ }
+
// -- string operators --
/// `field LIKE 'pat%'` shape. Empty pattern → `IsNotNull(field)` (every
@@ -1643,14 +1659,8 @@ impl PredicateBuilder {
});
}
PredicateOperator::In | PredicateOperator::NotIn => {
- if !literals.is_empty() {
- return Ok(());
- }
- // Empty IN is handled at is_in()/is_not_in() level; this
guards
- // against direct leaf() misuse.
- return Err(Error::ConfigInvalid {
- message: format!("{op} expects at least 1 literal, got 0"),
- });
+ // Java retains empty set leaves so negation preserves NULL
semantics.
+ return Ok(());
}
PredicateOperator::ArraysOverlap |
PredicateOperator::ArrayContainsAll => return Ok(()),
PredicateOperator::Between | PredicateOperator::NotBetween => (2,
literals.len()),
@@ -2566,15 +2576,50 @@ mod tests {
// ======================== Empty IN / NOT IN handling
========================
#[test]
- fn test_in_empty_returns_always_false() {
+ fn test_empty_set_negation_preserves_java_null_semantics() {
let pb = PredicateBuilder::new(&test_fields());
- assert_eq!(pb.is_in("id", vec![]).unwrap(), Predicate::AlwaysFalse);
+ assert_eq!(
+ Predicate::negate(pb.is_in("id", vec![]).unwrap()),
+ pb.is_not_in("id", vec![]).unwrap()
+ );
+ assert_eq!(
+ Predicate::negate(pb.is_not_in("id", vec![]).unwrap()),
+ pb.is_in("id", vec![]).unwrap()
+ );
}
#[test]
- fn test_not_in_empty_returns_always_true() {
- let pb = PredicateBuilder::new(&test_fields());
- assert_eq!(pb.is_not_in("id", vec![]).unwrap(), Predicate::AlwaysTrue);
+ fn nullable_set_literals_follow_java() {
+ let fields = vec![DataField::new(
+ 0,
+ "id".into(),
+ DataType::Int(crate::spec::IntType::new()),
+ )];
+ let pb = PredicateBuilder::new(&fields);
+ let predicates = [
+ pb.is_in_with_nulls("id", vec![]).unwrap(),
+ pb.is_not_in_with_nulls("id", vec![]).unwrap(),
+ pb.is_in_with_nulls("id", vec![None, Some(Datum::Int(1))])
+ .unwrap(),
+ pb.is_not_in_with_nulls("id", vec![None, Some(Datum::Int(1))])
+ .unwrap(),
+ ];
+ for (value, expected) in [
+ (None, [false, false, false, false]),
+ (Some(1), [false, true, true, false]),
+ (Some(2), [false, true, false, false]),
+ ] {
+ let mut row = TestBinaryRowBuilder::new(1);
+ if let Some(value) = value {
+ row.write_int(0, value);
+ } else {
+ row.set_null_at(0);
+ }
+ let row = row.build();
+ for (predicate, expected) in predicates.iter().zip(expected) {
+ assert_eq!(eval_row(predicate, &row).unwrap(), expected);
+ }
+ }
}
// ======================== Constant absorption ========================
@@ -3825,9 +3870,6 @@ mod tests {
("NOT_BETWEEN", "[1,null]"),
// NOT_IN with any null literal fails every row.
("NOT_IN", "[1,null]"),
- // IN with only null literals matches nothing.
- ("IN", "[null]"),
- ("IN", "[]"),
] {
let json = rest_leaf_json(function, "id", literals);
assert!(
@@ -3839,6 +3881,20 @@ mod tests {
);
}
+ for literals in ["[]", "[null]"] {
+ let parsed =
+ Predicate::from_rest_json(&rest_leaf_json("IN", "id",
literals), &fields).unwrap();
+ for value in [None, Some(1)] {
+ let mut row = TestBinaryRowBuilder::new(fields.len() as i32);
+ if let Some(value) = value {
+ row.write_int(0, value);
+ } else {
+ row.set_null_at(0);
+ }
+ assert!(!eval_row(&parsed, &row.build()).unwrap());
+ }
+ }
+
// IN drops null literals and keeps the rest.
let parsed =
Predicate::from_rest_json(&rest_leaf_json("IN", "id",
"[1,null,2]"), &fields).unwrap();
@@ -3854,7 +3910,7 @@ mod tests {
assert!(matches!(
&parsed,
Predicate::Leaf {
- op: PredicateOperator::IsNotNull,
+ op: PredicateOperator::NotIn,
..
}
));
diff --git a/crates/paimon/src/table/partition_filter.rs
b/crates/paimon/src/table/partition_filter.rs
index 17ad4635..ce5dda04 100644
--- a/crates/paimon/src/table/partition_filter.rs
+++ b/crates/paimon/src/table/partition_filter.rs
@@ -21,12 +21,29 @@
use crate::predicate_stats::data_leaf_may_match;
use crate::spec::{
- eval_row, extract_datum, BinaryRow, BinaryRowBuilder, DataField, Datum,
ManifestFileMeta,
- Predicate, PredicateBuilder, PredicateOperator,
+ eval_row, extract_datum, field_idx_to_partition_idx, BinaryRow,
BinaryRowBuilder, DataField,
+ Datum, ManifestFileMeta, Predicate, PredicateBuilder, PredicateOperator,
};
use crate::table::stats_filter::FileStatsRows;
use std::collections::HashSet;
+/// Explicit partition filters use table-field indices and must not contain
data predicates.
+pub(super) fn validate_partition_filter(
+ table: &super::Table,
+ filter: &Predicate,
+) -> crate::Result<()> {
+ let mapping =
+ field_idx_to_partition_idx(table.schema().fields(),
table.schema().partition_keys());
+ if table.schema().partition_keys().is_empty() ||
!filter.references_only_mapped_fields(&mapping)
+ {
+ return Err(crate::Error::ConfigInvalid {
+ message: "Partition filter must reference only partition keys of a
partitioned table"
+ .to_string(),
+ });
+ }
+ Ok(())
+}
+
/// Per-field min/max bounds for stats-based manifest pruning.
#[derive(Debug, Clone)]
pub(crate) struct FieldBounds {
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 043dd3bf..7b8e08e0 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder.rs
@@ -37,8 +37,8 @@ use
super::sorted_global_index_options::SortedIndexWriteOptions;
use crate::btree::key_serde::KeyComparator;
use crate::btree::{make_key_comparator, serialize_datum};
use crate::fm_index::{FMOptions, FMWriteOptions};
-use crate::spec::{CoreOptions, DataType, Datum};
-use crate::table::{CommitMessage, RowRange, SnapshotManager, Table,
TableCommit};
+use crate::spec::{CoreOptions, DataType, Datum, Predicate};
+use crate::table::{CommitMessage, RowRange, Table, TableCommit};
use crate::{Error, Result};
use std::collections::HashMap;
@@ -68,6 +68,7 @@ pub struct SortedGlobalIndexBuildBuilder<'a> {
index_column: Option<IndexColumns>,
index_type: String,
options: HashMap<String, String>,
+ partition_filter: Option<Predicate>,
}
/// Backward-compatible name retained for callers that used the original
@@ -81,6 +82,7 @@ impl<'a> SortedGlobalIndexBuildBuilder<'a> {
index_column: None,
index_type: BTREE_GLOBAL_INDEX_TYPE.to_string(),
options: HashMap::new(),
+ partition_filter: None,
}
}
@@ -107,7 +109,55 @@ impl<'a> SortedGlobalIndexBuildBuilder<'a> {
self
}
+ /// Restrict the build to partitions using table-field indices.
+ /// Multiple calls combine the partition predicates 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)
+ }
+
+ /// 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(),
+ format!(
+ "global-index-{}-create-{}",
+ self.index_type,
+ uuid::Uuid::new_v4()
+ ),
+ )
+ }
+
+ 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()?;
@@ -159,21 +209,15 @@ impl<'a> SortedGlobalIndexBuildBuilder<'a> {
.collect::<Vec<_>>();
let extra_field_ids =
(!extra_field_ids.is_empty()).then_some(extra_field_ids);
- 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 sorted global index without a
snapshot".to_string(),
- source: None,
- })?;
+ let Some(snapshot) =
self.table.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)
@@ -198,7 +242,7 @@ impl<'a> SortedGlobalIndexBuildBuilder<'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(
@@ -214,14 +258,7 @@ impl<'a> SortedGlobalIndexBuildBuilder<'a> {
)
.await?;
- let commit = TableCommit::new(
- self.table.clone(),
- format!(
- "global-index-{}-create-{}",
- index_type,
- uuid::Uuid::new_v4()
- ),
- );
+ let commit = self.new_commit();
let shard_count = shards.len();
let mut messages = Vec::with_capacity(shard_count);
for shard in shards {
@@ -243,13 +280,12 @@ impl<'a> SortedGlobalIndexBuildBuilder<'a> {
messages.push(message);
}
- commit
- .commit_if_latest_snapshot(messages, snapshot.id())
- .await?;
-
- Ok(shard_count)
+ Ok((Some(snapshot.id()), messages))
}
}
#[cfg(test)]
mod tests;
+
+#[cfg(test)]
+mod preparation_tests;
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
new file mode 100644
index 00000000..b23a7a49
--- /dev/null
+++
b/crates/paimon/src/table/sorted_global_index_build_builder/preparation_tests.rs
@@ -0,0 +1,568 @@
+// 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 super::*;
+use crate::catalog::Identifier;
+use crate::io::FileIOBuilder;
+use crate::spec::{IntType, PredicateBuilder, Schema, TableSchema, VarCharType};
+use crate::table::TableWrite;
+use arrow_array::{Int32Array, Int64Array, RecordBatch, StringArray};
+use arrow_schema::{DataType as ArrowType, Field, Schema as ArrowSchema};
+use futures::TryStreamExt;
+use std::sync::Arc;
+
+fn table(partitioned: bool, external: bool) -> Table {
+ let location = format!("memory:/index-preparation-{}",
uuid::Uuid::new_v4());
+ let mut options = HashMap::from([
+ ("row-tracking.enabled".into(), "true".into()),
+ ("data-evolution.enabled".into(), "true".into()),
+ ("global-index.enabled".into(), "true".into()),
+ ("global-index.search-mode".into(), "full".into()),
+ ("sorted-index.records-per-range".into(), "2".into()),
+ ]);
+ if external {
+ options.insert(
+ "global-index.external-path".into(),
+ format!("{location}-external"),
+ );
+ }
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("name", DataType::VarChar(VarCharType::string_type()))
+ .column("pt", DataType::Int(IntType::new()))
+ .options(options);
+ if partitioned {
+ schema = schema.partition_keys(vec!["pt"]);
+ }
+ Table::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("default", "indexes"),
+ location,
+ TableSchema::new(0, &schema.build().unwrap()),
+ None,
+ )
+}
+
+async fn append(table: &Table, ids: Vec<i32>, names: Vec<Option<&str>>, pts:
Vec<Option<i32>>) {
+ let batch = RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![
+ Field::new("id", ArrowType::Int32, true),
+ Field::new("name", ArrowType::Utf8, true),
+ Field::new("pt", ArrowType::Int32, true),
+ ])),
+ vec![
+ Arc::new(Int32Array::from(ids)),
+ Arc::new(StringArray::from(names)),
+ Arc::new(Int32Array::from(pts)),
+ ],
+ )
+ .unwrap();
+ let mut writer = TableWrite::new(table, "data".into()).unwrap();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ TableCommit::new(table.clone(), "data".into())
+ .commit(messages)
+ .await
+ .unwrap();
+}
+
+async fn read_ids(table: &Table, predicate: Predicate) -> Vec<i32> {
+ let mut builder = table.new_read_builder();
+ builder
+ .with_filter(predicate)
+ .with_projection(&["id"])
+ .unwrap();
+ let plan = builder.new_scan().plan().await.unwrap();
+ let batches = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ let mut ids = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .to_vec()
+ })
+ .collect::<Vec<_>>();
+ ids.sort_unstable();
+ ids
+}
+
+async fn update_name(table: &Table, value: &str) -> Vec<CommitMessage> {
+ let batch = RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![
+ Field::new("_ROW_ID", ArrowType::Int64, false),
+ Field::new("name", ArrowType::Utf8, true),
+ ])),
+ vec![
+ Arc::new(Int64Array::from(vec![0])),
+ Arc::new(StringArray::from(vec![value])),
+ ],
+ )
+ .unwrap();
+ table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .update_by_arrow_with_row_id(vec![batch])
+ .await
+ .unwrap()
+}
+
+#[tokio::test]
+async fn staged_indexes_reject_target_column_changes_and_preserve_files() {
+ for kind in ["btree", "bitmap"] {
+ for same_commit in [false, true] {
+ let table = table(false, false);
+ append(
+ &table,
+ vec![1, 2],
+ vec![Some("a"), Some("b")],
+ vec![None; 2],
+ )
+ .await;
+ let mut builder = table.new_sorted_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();
+ let updates = update_name(&table, "changed").await;
+ let commit = TableCommit::new(table.clone(), "publish".into());
+ if same_commit {
+ messages.extend(updates);
+ } else {
+ commit.commit(updates).await.unwrap();
+ }
+ let before = table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap()
+ .id();
+ assert!(commit
+ .commit(messages)
+ .await
+ .unwrap_err()
+ .to_string()
+ .contains("Global index source conflict"));
+ assert_eq!(
+ table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap()
+ .id(),
+ before
+ );
+ assert!(table
+ .file_io()
+ .exists(&format!("{}/index/{}", table.location(),
file.file_name))
+ .await
+ .unwrap());
+ }
+ }
+}
+
+#[tokio::test]
+async fn
staged_indexes_reject_deleted_partial_columns_and_removed_source_ranges() {
+ for kind in ["btree", "bitmap"] {
+ for remove_all in [false, true] {
+ let table = table(false, false);
+ append(
+ &table,
+ vec![1, 2],
+ vec![Some("a"), Some("b")],
+ vec![None; 2],
+ )
+ .await;
+ 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();
+ builder.with_index_column("name").with_index_type(kind);
+ let messages = builder.build().await.unwrap();
+ if remove_all {
+ commit.overwrite(vec![], None).await.unwrap();
+ } else {
+ let deletions = updates
+ .into_iter()
+ .map(|update| {
+ let mut message =
+ CommitMessage::new(update.partition,
update.bucket, vec![]);
+ message.deleted_files = update.new_files;
+ message
+ })
+ .collect();
+ commit.commit(deletions).await.unwrap();
+ }
+ let error = commit.commit(messages).await.unwrap_err().to_string();
+ assert!(
+ error.contains(if remove_all {
+ "Global index row ID existence conflict"
+ } else {
+ "Global index source conflict"
+ }),
+ "{error}"
+ );
+ }
+ }
+}
+
+#[tokio::test]
+async fn staged_index_survives_append_and_unrelated_column_update() {
+ let table = table(false, false);
+ append(
+ &table,
+ vec![1, 2],
+ vec![Some("a"), Some("b")],
+ vec![None; 2],
+ )
+ .await;
+ let mut builder = table.new_sorted_global_index_build_builder();
+ builder.with_index_column("name");
+ let messages = builder.build().await.unwrap();
+ assert_eq!(
+ crate::spec::DataEvolutionIndexSourceMeta::deserialize(
+ messages[0].new_index_files[0]
+ .global_index_meta
+ .as_ref()
+ .unwrap()
+ .source_meta
+ .as_ref()
+ .unwrap()
+ )
+ .unwrap()
+ .scan_snapshot_id(),
+ 1
+ );
+ let batch = RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![
+ Field::new("_ROW_ID", ArrowType::Int64, false),
+ Field::new("id", ArrowType::Int32, true),
+ ])),
+ vec![
+ Arc::new(Int64Array::from(vec![0])),
+ Arc::new(Int32Array::from(vec![10])),
+ ],
+ )
+ .unwrap();
+ let updates = table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .update_by_arrow_with_row_id(vec![batch])
+ .await
+ .unwrap();
+ let commit = TableCommit::new(table.clone(), "data".into());
+ commit.commit(updates).await.unwrap();
+ append(&table, vec![3], vec![Some("a")], vec![None]).await;
+ commit.commit(messages).await.unwrap();
+ let predicate = PredicateBuilder::new(table.schema().fields())
+ .equal("name", Datum::String("a".into()))
+ .unwrap();
+ assert_eq!(read_ids(&table, predicate).await, vec![3, 10]);
+}
+
+#[tokio::test]
+async fn equivalent_empty_partition_encodings_share_source_checks() {
+ for partition in [
+ vec![],
+ vec![0, 0, 0, 0],
+ crate::spec::EMPTY_SERIALIZED_ROW.to_vec(),
+ ] {
+ 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();
+ builder.with_index_column("name");
+ let mut messages = builder.build().await.unwrap();
+ for message in &mut messages {
+ message.partition = partition.clone();
+ }
+ let commit = TableCommit::new(table.clone(), "publish".into());
+ if changed {
+ commit
+ .commit(update_name(&table, "changed").await)
+ .await
+ .unwrap();
+ assert!(commit
+ .commit(messages)
+ .await
+ .unwrap_err()
+ .to_string()
+ .contains("Global index source conflict"));
+ } else {
+ commit.commit(messages).await.unwrap();
+ let predicate = PredicateBuilder::new(table.schema().fields())
+ .equal("name", Datum::String("a".into()))
+ .unwrap();
+ assert_eq!(read_ids(&table, predicate).await, vec![1]);
+ }
+ }
+ }
+}
+
+#[tokio::test]
+async fn prepared_files_are_invisible_until_explicit_commit() {
+ for kind in ["btree", "bitmap"] {
+ let table = table(false, false);
+ append(
+ &table,
+ vec![3, 1, 2, 4],
+ vec![Some("c"), Some("a"), None, Some("a")],
+ vec![None; 4],
+ )
+ .await;
+ let before = table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ let mut builder = table.new_sorted_global_index_build_builder();
+ builder.with_index_column("name").with_index_type(kind);
+ let messages = builder.build().await.unwrap();
+ assert_eq!(messages.len(), 2);
+ assert!(messages
+ .iter()
+ .all(|m| m.new_files.is_empty() && m.new_index_files.len() == 1));
+ let after_build = table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(before.id(), after_build.id());
+ assert!(after_build.index_manifest().is_none());
+ for message in &messages {
+ let file = &message.new_index_files[0];
+ assert_eq!(file.index_type, kind);
+ assert_eq!(file.row_count, 2);
+ assert!(table
+ .file_io()
+ .exists(&format!("{}/index/{}", table.location(),
file.file_name))
+ .await
+ .unwrap());
+ assert_eq!(
+ CommitMessage::deserialize(14, &message.serialize().unwrap())
+ .unwrap()
+ .new_index_files,
+ message.new_index_files
+ );
+ }
+ TableCommit::new(table.clone(), "publish".into())
+ .commit(messages)
+ .await
+ .unwrap();
+ let after = table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(after.id(), before.id() + 1);
+ assert!(after.index_manifest().is_some());
+ let predicate = PredicateBuilder::new(table.schema().fields())
+ .equal("name", Datum::String("a".into()))
+ .unwrap();
+ assert_eq!(read_ids(&table, predicate).await, vec![1, 4]);
+ assert!(builder.build().await.unwrap().is_empty());
+ assert_eq!(builder.execute().await.unwrap(), 0);
+ assert_eq!(
+ table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap()
+ .id(),
+ after.id()
+ );
+ }
+}
+
+#[tokio::test]
+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();
+ builder.with_index_column("name").with_index_type(kind);
+ let messages = builder.build().await.unwrap();
+ let path = messages[0].new_index_files[0]
+ .external_path
+ .as_ref()
+ .unwrap();
+ assert!(path.starts_with(
+ table
+ .schema()
+ .options()
+ .get("global-index.external-path")
+ .unwrap()
+ ));
+ assert!(table.file_io().exists(path).await.unwrap());
+ assert!(!table
+ .file_io()
+ .exists(&format!(
+ "{}/index/{}",
+ table.location(),
+ messages[0].new_index_files[0].file_name
+ ))
+ .await
+ .unwrap());
+ TableCommit::new(table.clone(), "discard".into())
+ .abort(&messages)
+ .await
+ .unwrap();
+ assert!(!table.file_io().exists(path).await.unwrap());
+ assert_eq!(
+ table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap()
+ .id(),
+ 1
+ );
+ let replacements = builder.build().await.unwrap();
+ TableCommit::new(table.clone(), "publish".into())
+ .commit(replacements)
+ .await
+ .unwrap();
+ let predicate = PredicateBuilder::new(table.schema().fields())
+ .is_null("name")
+ .unwrap();
+ assert_eq!(read_ids(&table, predicate).await, vec![2]);
+ }
+}
+
+#[tokio::test]
+async fn partition_builds_only_cover_selected_partitions_and_new_rows() {
+ for kind in ["btree", "bitmap"] {
+ let table = table(true, false);
+ append(
+ &table,
+ vec![1, 2, 3],
+ vec![Some("a"), Some("b"), Some("c")],
+ vec![Some(0), Some(1), None],
+ )
+ .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();
+ builder
+ .with_index_column("id")
+ .with_index_type(kind)
+ .with_partition_filter(selected.clone())
+ .unwrap();
+ let first = builder.build().await.unwrap();
+ assert_eq!(
+ first
+ .iter()
+ .map(|m| m.new_index_files.iter().map(|f|
f.row_count).sum::<i64>())
+ .sum::<i64>(),
+ 1
+ );
+ TableCommit::new(table.clone(), "partition-one".into())
+ .commit(first)
+ .await
+ .unwrap();
+ assert!(builder.build().await.unwrap().is_empty());
+ append(
+ &table,
+ vec![4, 5],
+ vec![Some("d"), Some("e")],
+ vec![Some(1), Some(0)],
+ )
+ .await;
+ assert_eq!(builder.execute().await.unwrap(), 1);
+ assert!(builder.build().await.unwrap().is_empty());
+
+ let mut all = table.new_sorted_global_index_build_builder();
+ all.with_index_column("id").with_index_type(kind);
+ let remaining = all.build().await.unwrap();
+ assert_eq!(
+ remaining
+ .iter()
+ .flat_map(|m| &m.new_index_files)
+ .map(|f| f.row_count)
+ .sum::<i64>(),
+ 3
+ );
+ TableCommit::new(table.clone(), "all-partitions".into())
+ .commit(remaining)
+ .await
+ .unwrap();
+ assert!(all.build().await.unwrap().is_empty());
+ assert_eq!(read_ids(&table, selected).await, vec![2, 4]);
+ }
+}
+
+#[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();
+ builder.with_index_column("id");
+ assert!(builder.build().await.unwrap().is_empty());
+ assert_eq!(builder.execute().await.unwrap(), 0);
+ assert!(table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .is_none());
+ append(&table, vec![1], vec![Some("a")], vec![Some(1)]).await;
+ let predicates = PredicateBuilder::new(table.schema().fields());
+ for value in [1, 2] {
+ builder
+ .with_partition_filter(predicates.equal("pt",
Datum::Int(value)).unwrap())
+ .unwrap();
+ }
+ assert!(builder.build().await.unwrap().is_empty());
+ assert_eq!(builder.execute().await.unwrap(), 0);
+ assert_eq!(
+ table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap()
+ .id(),
+ 1
+ );
+ let data_filter = predicates.equal("id", Datum::Int(1)).unwrap();
+ assert!(builder.with_partition_filter(data_filter).is_err());
+ let unpartitioned = self::table(false, false);
+ let filter = PredicateBuilder::new(unpartitioned.schema().fields())
+ .is_null("pt")
+ .unwrap();
+ assert!(unpartitioned
+ .new_sorted_global_index_build_builder()
+ .with_partition_filter(filter)
+ .is_err());
+}
diff --git
a/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs
b/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs
index fdacd5b7..683ccc31 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder/writer.rs
@@ -85,19 +85,24 @@ impl SortedGlobalIndexBuildBuilder<'_> {
sort_index_rows(&mut rows, &cmp)?;
}
- self.table
- .file_io()
- .mkdirs(&format!(
- "{}/{INDEX_DIR}/",
- self.table.location().trim_end_matches('/')
- ))
- .await?;
let file_name = format!("{index_type}-global-index-{}.index",
uuid::Uuid::new_v4());
- let index_path = format!(
- "{}/{INDEX_DIR}/{}",
- self.table.location().trim_end_matches('/'),
- file_name
- );
+ // Global index placement is a table property, as in Java
GlobalIndexFileFactory.
+ let external_path =
super::super::external_path::new_index_external_path(
+ self.table.schema().options(),
+ false,
+ "",
+ &file_name,
+ )?;
+ let index_path = external_path.clone().unwrap_or_else(|| {
+ format!(
+ "{}/{INDEX_DIR}/{}",
+ self.table.location().trim_end_matches('/'),
+ file_name,
+ )
+ });
+ if let Some((directory, _)) = index_path.rsplit_once('/') {
+ self.table.file_io().mkdirs(directory).await?;
+ }
let write_result: Result<(u64, Vec<u8>, i64)> = async {
let output = self.table.file_io().new_output(&index_path)?;
let writer = output.writer().await?;
@@ -251,7 +256,7 @@ impl SortedGlobalIndexBuildBuilder<'_> {
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,
@@ -263,7 +268,9 @@ impl SortedGlobalIndexBuildBuilder<'_> {
.map(|field| field.id())
.collect()
}),
- 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/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index c508694d..698bea47 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -48,6 +48,7 @@ const BATCH_COMMIT_IDENTIFIER: i64 = i64::MAX;
/// Java RollingFileWriter.CHECK_ROLLING_RECORD_CNT.
const CHECK_ROLLING_RECORD_COUNT: usize = 1000;
+mod global_index_source_check;
mod row_tracking;
const DELETION_VECTORS_INDEX_TYPE: &str = "DELETION_VECTORS";
@@ -1743,6 +1744,8 @@ impl TableCommit {
let previous =
Self::read_prev_index_entries(file_io, &manifest_dir,
latest_snapshot).await?;
+ self.check_global_index_sources(latest_snapshot, entries,
new_index_entries)
+ .await?;
let mut index_entries = new_index_entries.clone();
index_entries.extend(self.global_index_update_entries(
&previous,
@@ -1804,6 +1807,8 @@ impl TableCommit {
let previous =
Self::read_prev_index_entries(file_io, &manifest_dir,
latest_snapshot).await?;
+ self.check_global_index_sources(latest_snapshot, &entries,
&new_index_entries)
+ .await?;
let mut all = previous.clone();
match partition_filter.as_ref() {
None => all.clear(),
diff --git a/crates/paimon/src/table/table_commit/global_index_source_check.rs
b/crates/paimon/src/table/table_commit/global_index_source_check.rs
new file mode 100644
index 00000000..e0d83a28
--- /dev/null
+++ b/crates/paimon/src/table/table_commit/global_index_source_check.rs
@@ -0,0 +1,186 @@
+// 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.
+
+//! Validate unpublished global indexes against their source data on every
commit attempt.
+
+use super::*;
+use crate::spec::DataEvolutionIndexSourceMeta;
+use crate::table::{merge_row_ranges, RowRange};
+
+impl TableCommit {
+ pub(super) async fn check_global_index_sources(
+ &self,
+ latest_snapshot: &Option<Snapshot>,
+ data_entries: &[ManifestEntry],
+ index_entries: &[IndexManifestEntry],
+ ) -> Result<()> {
+ if !self.data_evolution_enabled {
+ return Ok(());
+ }
+ let additions = index_entries
+ .iter()
+ .filter(|entry| {
+ entry.kind == FileKind::Add &&
entry.index_file.global_index_meta.is_some()
+ })
+ .cloned()
+ .collect::<Vec<_>>();
+ if additions.is_empty() {
+ return Ok(());
+ }
+ let partition_fields = self.table.schema().partition_fields();
+ let filter = if partition_fields.is_empty() {
+ None
+ } else {
+ Some(PartitionFilter::from_partition_set(
+ additions
+ .iter()
+ .map(|entry| entry.partition.clone())
+ .collect(),
+ &partition_fields,
+ )?)
+ };
+ let mut current = self
+ .scan_snapshot_entries(latest_snapshot, filter.as_ref())
+ .await?;
+ current.extend_from_slice(data_entries);
+ let current = merge_active_entries(current);
+ let mut ranges: HashMap<PartitionBucketKey, Vec<RowRange>> =
HashMap::new();
+ for entry in current {
+ if let Some((start, end)) = entry.file().row_id_range() {
+ ranges
+ .entry((partition_key(entry.partition()), entry.bucket()))
+ .or_default()
+ .push(RowRange::new(start, end));
+ }
+ }
+ let ranges = ranges
+ .into_iter()
+ .map(|(key, ranges)| (key, merge_row_ranges(ranges)))
+ .collect::<HashMap<_, _>>();
+ let mut sources = Vec::new();
+ for entry in &additions {
+ let meta = entry.index_file.global_index_meta.as_ref().unwrap();
+ if !ranges
+ .get(&(partition_key(&entry.partition), entry.bucket))
+ .is_some_and(|ranges| {
+ ranges.iter().any(|range| {
+ range.from() <= meta.row_range_start &&
meta.row_range_end <= range.to()
+ })
+ })
+ {
+ return Err(index_conflict(format!(
+ "Global index row ID existence conflict: index file '{}'
is not fully covered by current data files.",
+ entry.index_file.file_name
+ )));
+ }
+ if
DataEvolutionIndexSourceMeta::is_data_evolution_meta(meta.source_meta.as_deref())
{
+ let source = DataEvolutionIndexSourceMeta::deserialize(
+ meta.source_meta.as_deref().unwrap(),
+ )?;
+ let source_id = source.scan_snapshot_id();
+ let latest = latest_snapshot.as_ref().ok_or_else(|| {
+ index_conflict("Global index source conflict: no snapshot
exists.")
+ })?;
+ if source_id > latest.id() {
+ return Err(index_conflict("Global index source conflict:
source snapshot is newer than the latest snapshot."));
+ }
+ // Missing/expired snapshots cannot prove that the indexed
values are unchanged.
+ if source_id != latest.id() {
+ self.snapshot_manager.get_snapshot(source_id).await?;
+ }
+ sources.push((source_id, entry.clone()));
+ }
+ }
+ if sources.is_empty() {
+ return Ok(());
+ }
+ self.check_index_column_changes(&sources, data_entries, None)
+ .await?;
+ let earliest = sources.iter().map(|(id, _)| *id).min().unwrap();
+ let latest = latest_snapshot.as_ref().unwrap();
+ for id in earliest + 1..=latest.id() {
+ let snapshot = if id == latest.id() {
+ latest.clone()
+ } else {
+ self.snapshot_manager.get_snapshot(id).await?
+ };
+ if snapshot.commit_kind() == &CommitKind::COMPACT {
+ continue;
+ }
+ let changes = self.read_delta_entries(filter.as_ref(),
&snapshot).await?;
+ self.check_index_column_changes(&sources, &changes, Some(id))
+ .await?;
+ }
+ Ok(())
+ }
+
+ async fn check_index_column_changes(
+ &self,
+ sources: &[(i64, IndexManifestEntry)],
+ changes: &[ManifestEntry],
+ snapshot_id: Option<i64>,
+ ) -> Result<()> {
+ for change in changes {
+ let Some((start, end)) = change.file().row_id_range() else {
+ continue;
+ };
+ let overlapping = sources
+ .iter()
+ .filter(|(source_id, index)| {
+ let meta =
index.index_file.global_index_meta.as_ref().unwrap();
+ snapshot_id.is_none_or(|id| id > *source_id)
+ && same_index_partition(&index.partition,
change.partition())
+ && index.bucket == change.bucket()
+ && ranges_overlap(start, end, meta.row_range_start,
meta.row_range_end)
+ })
+ .map(|(_, index)| index)
+ .collect::<Vec<_>>();
+ if overlapping.is_empty() {
+ continue;
+ }
+ let written_fields = self.write_field_ids(change.file()).await?;
+ for index in overlapping {
+ let meta =
index.index_file.global_index_meta.as_ref().unwrap();
+ if std::iter::once(&meta.index_field_id)
+ .chain(meta.extra_field_ids.iter().flatten())
+ .any(|id| written_fields.contains(id))
+ {
+ return Err(index_conflict(format!(
+ "Global index source conflict: indexed values changed
after building index file '{}'.",
+ index.index_file.file_name
+ )));
+ }
+ }
+ }
+ Ok(())
+ }
+}
+
+fn index_conflict(message: impl Into<String>) -> crate::Error {
+ crate::Error::DataInvalid {
+ message: message.into(),
+ source: None,
+ }
+}
+
+fn partition_key(partition: &[u8]) -> Vec<u8> {
+ if is_empty_partition(partition) {
+ Vec::new()
+ } else {
+ partition.to_vec()
+ }
+}
diff --git a/crates/paimon/src/table/table_commit/row_tracking_tests.rs
b/crates/paimon/src/table/table_commit/row_tracking_tests.rs
index 598ee627..16f89ccb 100644
--- a/crates/paimon/src/table/table_commit/row_tracking_tests.rs
+++ b/crates/paimon/src/table/table_commit/row_tracking_tests.rs
@@ -423,19 +423,22 @@ async fn
row_tracking_ignore_index_update_preserves_index_metadata() {
let commit = TableCommit::new(table, "test-user".into());
let mut file = test_data_file("initial.parquet", 10);
file.file_source = Some(0);
- let mut seed = CommitMessage::new(EMPTY_SERIALIZED_ROW.clone(), 0,
vec![file]);
- let index = test_global_index_file("global-name.index", 1, 0, 9);
- seed.new_index_files.push(index.clone());
+ let seed = CommitMessage::new(EMPTY_SERIALIZED_ROW.clone(), 0, vec![file]);
commit.commit(vec![seed]).await.unwrap();
+ // Global indexes reference row IDs assigned by a completed data commit.
+ let mut indexes = CommitMessage::new(EMPTY_SERIALIZED_ROW.clone(), 0,
vec![]);
+ let index = test_global_index_file("global-name.index", 1, 0, 9);
+ indexes.new_index_files.push(index.clone());
+ commit.commit(vec![indexes]).await.unwrap();
let mut partial = test_data_file("updated.parquet", 10);
partial.file_source = Some(0);
partial.first_row_id = Some(0);
partial.write_cols = Some(vec!["name".into()]);
let mut update = CommitMessage::new(EMPTY_SERIALIZED_ROW.clone(), 0,
vec![partial]);
- update.check_from_snapshot = Some(1);
+ update.check_from_snapshot = Some(2);
commit.commit(vec![update]).await.unwrap();
let snapshot = latest_snapshot(&io, path).await.unwrap();
- assert_eq!(snapshot.id(), 2);
+ assert_eq!(snapshot.id(), 3);
assert_eq!(snapshot.next_row_id(), Some(10));
let index_entries = IndexManifest::read(
&io,
diff --git a/crates/paimon/src/table/vector_search_common.rs
b/crates/paimon/src/table/vector_search_common.rs
index 321bd914..10dcab6f 100644
--- a/crates/paimon/src/table/vector_search_common.rs
+++ b/crates/paimon/src/table/vector_search_common.rs
@@ -18,9 +18,7 @@
//! Shared vector-search option validation, index I/O helpers, and result
ordering.
use crate::lumina::is_lumina_index_type;
-use crate::spec::{
- field_idx_to_partition_idx, CoreOptions, DataField, Predicate,
ROW_ID_FIELD_NAME,
-};
+use crate::spec::{CoreOptions, DataField, Predicate, ROW_ID_FIELD_NAME};
use crate::table::pk_vector_position_read::{PKEY_VECTOR_POSITION_COLUMN,
SEARCH_SCORE_COLUMN};
use crate::table::read_builder::resolve_projected_fields;
use crate::table::Table;
@@ -30,20 +28,7 @@ use arrow_array::{Int64Array, RecordBatch};
use arrow_select::interleave::interleave_record_batch;
use std::collections::HashMap;
-/// Explicit partition filters use table-field indices, like scalar filters.
-/// Reject mixed predicates rather than silently discarding their data
conjuncts.
-pub(super) fn validate_partition_filter(table: &Table, filter: &Predicate) ->
crate::Result<()> {
- let mapping =
- field_idx_to_partition_idx(table.schema().fields(),
table.schema().partition_keys());
- if table.schema().partition_keys().is_empty() ||
!filter.references_only_mapped_fields(&mapping)
- {
- return Err(crate::Error::ConfigInvalid {
- message: "Partition filter must reference only partition keys of a
partitioned table"
- .to_string(),
- });
- }
- Ok(())
-}
+pub(super) use super::partition_filter::validate_partition_filter;
pub(super) fn add_filter(current: &mut Option<Predicate>, filter: Predicate) {
*current = Some(match current.take() {
