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 2457d30c [core] Support row-tracking sequence projections and
predicates (#1051)
2457d30c is described below
commit 2457d30c8f902e34a1f6d55134ce970e52c3cfb6
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Oct 7 08:53:14 2026 +0800
[core] Support row-tracking sequence projections and predicates (#1051)
---
bindings/python/src/predicate.rs | 26 +-
bindings/python/src/read.rs | 3 +-
bindings/python/src/write.rs | 2 +-
bindings/python/tests/test_read.py | 50 +-
crates/paimon/src/arrow/filtering.rs | 6 +-
crates/paimon/src/arrow/format/orc.rs | 6 +-
crates/paimon/src/arrow/format/parquet.rs | 6 +-
crates/paimon/src/arrow/residual.rs | 27 +-
crates/paimon/src/predicate_stats.rs | 10 +-
crates/paimon/src/spec/core_options.rs | 7 -
crates/paimon/src/spec/predicate.rs | 10 +-
crates/paimon/src/spec/schema.rs | 47 +-
crates/paimon/src/table/data_evolution_reader.rs | 9 +-
crates/paimon/src/table/data_file_reader.rs | 43 +-
crates/paimon/src/table/kv_file_reader.rs | 19 +-
crates/paimon/src/table/read_builder.rs | 84 ++-
crates/paimon/src/table/row_sidecar.rs | 10 +-
crates/paimon/src/table/sort_merge.rs | 59 +-
crates/paimon/src/table/stats_filter.rs | 9 +-
crates/paimon/src/table/table_scan.rs | 29 +
.../tests/data_evolution_row_sidecar_test.rs | 32 +-
.../tests/data_evolution_system_columns_test.rs | 599 +++++++++++++++++++++
22 files changed, 950 insertions(+), 143 deletions(-)
diff --git a/bindings/python/src/predicate.rs b/bindings/python/src/predicate.rs
index f164feda..f3f3ff7a 100644
--- a/bindings/python/src/predicate.rs
+++ b/bindings/python/src/predicate.rs
@@ -15,7 +15,9 @@
// specific language governing permissions and limitations
// under the License.
-use paimon::spec::{DataField, DataType, Datum, DecimalType, Predicate,
PredicateBuilder};
+use paimon::spec::{
+ DataField, DataType, Datum, DecimalType, Predicate, PredicateBuilder,
TableSchema,
+};
use pyo3::exceptions::{PyNotImplementedError, PyValueError};
use pyo3::prelude::*;
use pyo3::types::PyTzInfoAccess;
@@ -419,29 +421,13 @@ pub(crate) fn dict_to_predicate(
}
}
-/// Convert against table fields plus the synthesized `_ROW_ID` field.
+/// Use the core read schema, including only metadata the table provides.
pub(crate) fn dict_to_table_predicate(
node: &Bound<'_, PyDict>,
- fields: &[DataField],
+ schema: &TableSchema,
case_sensitive: bool,
) -> PyResult<Predicate> {
- use paimon::spec::{BigIntType, ROW_ID_FIELD_ID, ROW_ID_FIELD_NAME};
- let mut fields = fields.to_vec();
- let has_row_id = fields.iter().any(|field| {
- if case_sensitive {
- field.name() == ROW_ID_FIELD_NAME
- } else {
- field.name().eq_ignore_ascii_case(ROW_ID_FIELD_NAME)
- }
- });
- if !has_row_id {
- fields.push(DataField::new(
- ROW_ID_FIELD_ID,
- ROW_ID_FIELD_NAME.into(),
- DataType::BigInt(BigIntType::with_nullable(true)),
- ));
- }
- dict_to_predicate(node, &fields, case_sensitive)
+ dict_to_predicate(node, &schema.fields_with_read_metadata(),
case_sensitive)
}
/// Resolve a leaf's field name to its schema [`DataType`] under the given case
diff --git a/bindings/python/src/read.rs b/bindings/python/src/read.rs
index 8cbaf515..33c951d1 100644
--- a/bindings/python/src/read.rs
+++ b/bindings/python/src/read.rs
@@ -281,8 +281,7 @@ impl PyReadBuilder {
mut slf: PyRefMut<'py, Self>,
predicate: &Bound<'_, PyDict>,
) -> PyResult<PyRefMut<'py, Self>> {
- let filter =
- dict_to_table_predicate(predicate, slf.table.schema().fields(),
slf.case_sensitive)?;
+ let filter = dict_to_table_predicate(predicate, slf.table.schema(),
slf.case_sensitive)?;
slf.filter = Some(filter);
Ok(slf)
}
diff --git a/bindings/python/src/write.rs b/bindings/python/src/write.rs
index 6c619e65..23d0cd62 100644
--- a/bindings/python/src/write.rs
+++ b/bindings/python/src/write.rs
@@ -464,7 +464,7 @@ impl UpdateContext {
read_columns: Option<Vec<String>>,
) -> PyResult<Vec<PyCommitMessage>> {
let predicate = predicate
- .map(|predicate| dict_to_table_predicate(predicate,
self.table.schema().fields(), true))
+ .map(|predicate| dict_to_table_predicate(predicate,
self.table.schema(), true))
.transpose()?;
let callback_error = Arc::new(Mutex::new(None));
let assignments = crate::update_assignment::from_python(
diff --git a/bindings/python/tests/test_read.py
b/bindings/python/tests/test_read.py
index 27642786..7a6903c6 100644
--- a/bindings/python/tests/test_read.py
+++ b/bindings/python/tests/test_read.py
@@ -405,22 +405,60 @@ def
test_row_id_filter_with_projection_and_data_predicate(data_evolution, case_s
assert rows.to_pylist() == [{"name": "b", "_ROW_ID": 1}]
[email protected]("predicate_field", ["_row_id", "_ROW_ID"])
-def
test_case_insensitive_filter_uses_real_lowercase_row_id_column(predicate_field):
[email protected]("user_field", ["_row_id", "_sequence_number"])
[email protected]("uppercase", [False, True])
+def test_case_insensitive_filter_uses_untracked_user_column(user_field,
uppercase):
with tempfile.TemporaryDirectory() as warehouse:
ctx = SQLContext()
ctx.register_catalog("paimon", {"warehouse": warehouse})
ctx.sql("CREATE SCHEMA paimon.realrowid")
- ctx.sql("CREATE TABLE paimon.realrowid.t (_row_id BIGINT, id INT)")
+ ctx.sql(f"CREATE TABLE paimon.realrowid.t ({user_field} BIGINT, id
INT)")
ctx.sql("INSERT INTO paimon.realrowid.t VALUES (11, 1), (22, 2)")
table = PaimonCatalog({"warehouse":
warehouse}).get_table("realrowid.t")
builder = table.new_read_builder().with_case_sensitive(False)
- builder.with_projection(["_row_id", "id"])
+ # Projection and predicate resolution must use the same available
+ # fields. Untracked append tables have no virtual tracking columns.
+ field = user_field.upper() if uppercase else user_field
+ builder.with_projection([field, "id"])
builder.with_filter({
- "method": "equal", "field": predicate_field, "literals": [22],
+ "method": "equal", "field": field, "literals": [22],
})
batches = builder.new_read().read(builder.new_scan().plan().splits())
- assert pa.Table.from_batches(batches).to_pylist() == [{"_row_id": 22,
"id": 2}]
+ assert pa.Table.from_batches(batches).to_pylist() == [{user_field: 22,
"id": 2}]
+
+
[email protected]("user_field", ["_row_id", "_sequence_number"])
[email protected]("uppercase", [False, True])
+def test_case_insensitive_tracking_metadata_collision_is_ambiguous(user_field,
uppercase):
+ with tempfile.TemporaryDirectory() as warehouse:
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": warehouse})
+ ctx.sql("CREATE SCHEMA paimon.trackingcollision")
+ ctx.sql(f"""CREATE TABLE paimon.trackingcollision.t ({user_field}
BIGINT, id INT)
+ WITH ('row-tracking.enabled' = 'true')""")
+ table = PaimonCatalog({"warehouse":
warehouse}).get_table("trackingcollision.t")
+ builder = table.new_read_builder().with_case_sensitive(False)
+ field = user_field.upper() if uppercase else user_field
+ with pytest.raises(ValueError, match="Ambiguous"):
+ builder.with_filter({"method": "equal", "field": field,
"literals": [22]})
+ with pytest.raises(ValueError, match="Ambiguous"):
+ builder.with_projection([field]).new_read().read([])
+
+
+def test_primary_key_sequence_metadata_does_not_require_row_tracking():
+ with tempfile.TemporaryDirectory() as warehouse:
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": warehouse})
+ ctx.sql("CREATE SCHEMA paimon.pksequence")
+ ctx.sql("""CREATE TABLE paimon.pksequence.t (id INT, v INT, PRIMARY
KEY (id))
+ WITH ('bucket' = '1', 'row-tracking.enabled' = 'false')""")
+ ctx.sql("INSERT INTO paimon.pksequence.t VALUES (1, 11), (2, 22)")
+ table = PaimonCatalog({"warehouse":
warehouse}).get_table("pksequence.t")
+ builder = table.new_read_builder().with_case_sensitive(False)
+ builder.with_projection(["id", "_sequence_number"])
+ builder.with_filter({"method": "equal", "field": "_sequence_number",
"literals": [1]})
+ result =
pa.Table.from_batches(builder.new_read().read(builder.new_scan().plan().splits()))
+ assert result.to_pylist() == [{"id": 2, "_SEQUENCE_NUMBER": 1}]
def test_row_tracking_append_row_ranges_keep_global_row_ids():
diff --git a/crates/paimon/src/arrow/filtering.rs
b/crates/paimon/src/arrow/filtering.rs
index b502e7f7..3ae911a2 100644
--- a/crates/paimon/src/arrow/filtering.rs
+++ b/crates/paimon/src/arrow/filtering.rs
@@ -17,7 +17,7 @@
use crate::arrow::schema_evolution::create_index_mapping;
pub(crate) use crate::predicate_stats::{predicates_may_match_with_schema,
StatsAccessor};
-use crate::spec::{is_row_id_column, DataField, Predicate, PredicateOperator};
+use crate::spec::{is_row_tracking_column, DataField, Predicate,
PredicateOperator};
/// Remap predicates from table-level indices to file-level indices.
/// Predicates referencing fields not present in the file are resolved based on
@@ -44,10 +44,10 @@ fn remap_predicate(predicate: &Predicate, mapping:
&[Option<usize>]) -> Predicat
op,
literals,
} => {
- // `_ROW_ID` is not a file column and has no per-file position, so
+ // Row-tracking metadata has no logical-table position, so
// mapping its placeholder index would collapse the leaf to a
// constant. Keep it; the residual resolves it by name.
- if is_row_id_column(column) {
+ if is_row_tracking_column(column) {
return predicate.clone();
}
match mapping.get(*index).copied().flatten() {
diff --git a/crates/paimon/src/arrow/format/orc.rs
b/crates/paimon/src/arrow/format/orc.rs
index 85c77780..79bd02f9 100644
--- a/crates/paimon/src/arrow/format/orc.rs
+++ b/crates/paimon/src/arrow/format/orc.rs
@@ -17,7 +17,9 @@
use super::{FilePredicates, FormatFileReader, FormatFileWriter,
FormatWriteResult};
use crate::io::{FileRead, FileWrite, OutputFile};
-use crate::spec::{is_row_id_column, DataField, DataType, Datum, Predicate,
PredicateOperator};
+use crate::spec::{
+ is_row_tracking_column, DataField, DataType, Datum, Predicate,
PredicateOperator,
+};
use crate::table::{ArrowRecordBatchStream, RowRange};
use crate::Error;
use arrow_array::RecordBatch;
@@ -547,7 +549,7 @@ fn build_orc_leaf_predicate(
return None;
};
// Not in the file, and its index would push the wrong column down.
- if is_row_id_column(column) {
+ if is_row_tracking_column(column) {
return None;
}
let file_field = file_fields.get(*index)?;
diff --git a/crates/paimon/src/arrow/format/parquet.rs
b/crates/paimon/src/arrow/format/parquet.rs
index 81b9f15b..44efbc8c 100644
--- a/crates/paimon/src/arrow/format/parquet.rs
+++ b/crates/paimon/src/arrow/format/parquet.rs
@@ -29,8 +29,8 @@ use crate::arrow::{ReadBudget, RowFilter, RowFilterContext};
use crate::io::{FileRead, OutputFile};
use crate::spec::stats::BinaryTableStats;
use crate::spec::{
- is_row_id_column, BinaryRowBuilder, CoreOptions, DataField, DataType,
Datum, MetadataStatsMode,
- Predicate, PredicateOperator,
+ is_row_tracking_column, BinaryRowBuilder, CoreOptions, DataField,
DataType, Datum,
+ MetadataStatsMode, Predicate, PredicateOperator,
};
use crate::table::{ArrowRecordBatchStream, RowRange};
use crate::Error;
@@ -1469,7 +1469,7 @@ fn parquet_leaf_row_filter_accepted(
}
// Not in the file, so the decoder cannot evaluate it. Rejecting the leaf
// rejects any enclosing predicate too, leaving it to the post-scan
residual.
- if is_row_id_column(column) {
+ if is_row_tracking_column(column) {
return Ok(false);
}
let Some(file_field) = file_fields.get(index) else {
diff --git a/crates/paimon/src/arrow/residual.rs
b/crates/paimon/src/arrow/residual.rs
index f1e5fe2a..eaf463a8 100644
--- a/crates/paimon/src/arrow/residual.rs
+++ b/crates/paimon/src/arrow/residual.rs
@@ -45,7 +45,9 @@
//! must not reference any vortex-specific types.
use crate::arrow::format::FilePredicates;
-use crate::spec::{is_row_id_column, DataField, DataType, Datum, Predicate,
PredicateOperator};
+use crate::spec::{
+ is_row_tracking_column, DataField, DataType, Datum, Predicate,
PredicateOperator,
+};
use crate::Error;
use arrow_array::{
Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Datum as
ArrowDatum, Decimal128Array,
@@ -168,10 +170,10 @@ fn evaluate_predicate_mask(
} => {
// Resolve the batch column by NAME, but pick that name carefully:
a
// real column may be renamed in the file, so its batch name comes
- // from `file_fields[index]`; `_ROW_ID` is absent from
`file_fields`
+ // from `file_fields[index]`; row-tracking metadata is absent from
the logical `file_fields`
// and only its own name is meaningful. Java's `PredicateRemapper`
// rebinds by name too.
- let file_field = if is_row_id_column(column) {
+ let file_field = if is_row_tracking_column(column) {
None
} else {
match file_fields.get(*index) {
@@ -196,9 +198,14 @@ fn evaluate_predicate_mask(
// Backstop for residuals applied to a predicate-free
reader
// (PK merge output, vector search); readers that own their
// predicates reject earlier.
- return
Err(crate::table::row_id_predicate::unsupported_row_id_filter(
- "this read",
- ));
+ if crate::spec::is_row_id_column(field_name) {
+ return
Err(crate::table::row_id_predicate::unsupported_row_id_filter(
+ "this read",
+ ));
+ }
+ return Err(Error::Unsupported {
+ message: format!("filtering on '{field_name}' requires
this read to supply the row-tracking column"),
+ });
}
// A real column missing here is a reader bug: it did not widen
// its scan to the predicate columns.
@@ -265,8 +272,12 @@ pub(crate) fn widen_scan_fields(
collect_predicate_leaf_refs(predicate, &mut refs);
}
for (name, index) in refs {
+ if name == crate::spec::SEQUENCE_NUMBER_FIELD_NAME {
+ push_unique_scan_field(&mut fields,
&crate::spec::sequence_number_data_field());
+ continue;
+ }
// Not read from the file, so there is nothing to widen with.
- if is_row_id_column(name) {
+ if is_row_tracking_column(name) {
continue;
}
if let Some(field) = fp.file_fields.get(index) {
@@ -281,7 +292,7 @@ pub(crate) fn widen_scan_fields(
/// Collect every leaf as `(column name, leaf index)`.
///
/// Callers that resolve a leaf positionally must check the name first — see
-/// [`crate::spec::is_row_id_column`].
+/// [`crate::spec::is_row_tracking_column`].
pub(crate) fn collect_predicate_leaf_refs<'a>(
predicate: &'a Predicate,
refs: &mut Vec<(&'a str, usize)>,
diff --git a/crates/paimon/src/predicate_stats.rs
b/crates/paimon/src/predicate_stats.rs
index c30c5465..a111a658 100644
--- a/crates/paimon/src/predicate_stats.rs
+++ b/crates/paimon/src/predicate_stats.rs
@@ -15,7 +15,9 @@
// specific language governing permissions and limitations
// under the License.
-use crate::spec::{is_row_id_column, DataField, DataType, Datum, Predicate,
PredicateOperator};
+use crate::spec::{
+ is_row_tracking_column, DataField, DataType, Datum, Predicate,
PredicateOperator,
+};
use std::cmp::Ordering;
pub(crate) trait StatsAccessor {
@@ -494,8 +496,8 @@ fn predicate_may_match_with_schema<T: StatsAccessor>(
Predicate::Not(inner) => {
!predicate_must_match_with_schema(inner, stats, field_mapping,
file_fields)
}
- // `_ROW_ID` has no column stats, so never prune on it.
- Predicate::Leaf { column, .. } if is_row_id_column(column) => true,
+ // Row-tracking metadata has no logical column stats, so never prune
on it.
+ Predicate::Leaf { column, .. } if is_row_tracking_column(column) =>
true,
Predicate::Leaf {
index,
data_type,
@@ -540,7 +542,7 @@ fn predicate_must_match_with_schema<T: StatsAccessor>(
!predicate_may_match_with_schema(inner, stats, field_mapping,
file_fields)
}
// Stats cannot decide `_ROW_ID`, so it never provably matches.
- Predicate::Leaf { column, .. } if is_row_id_column(column) => false,
+ Predicate::Leaf { column, .. } if is_row_tracking_column(column) =>
false,
Predicate::Leaf {
index,
data_type,
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index f67a5e37..beee183a 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -773,13 +773,6 @@ impl<'a> CoreOptions<'a> {
.unwrap_or(false)
}
- /// Whether a missing optional read target may fall back to the primary
file.
- pub fn scan_ignore_lost_file(&self) -> bool {
- self.options
- .get("scan.ignore-lost-files")
- .is_some_and(|value| value.eq_ignore_ascii_case("true"))
- }
-
/// Whether normal Data Evolution files also write an aligned ROW sidecar.
pub fn data_evolution_row_sidecar_enabled(&self) -> crate::Result<bool> {
match self.options.get("data-evolution.row-sidecar.enabled") {
diff --git a/crates/paimon/src/spec/predicate.rs
b/crates/paimon/src/spec/predicate.rs
index 2abe34eb..73a078c2 100644
--- a/crates/paimon/src/spec/predicate.rs
+++ b/crates/paimon/src/spec/predicate.rs
@@ -26,7 +26,7 @@
use crate::error::*;
use crate::spec::binary_row::BinaryRow;
use crate::spec::types::DataType;
-use crate::spec::{is_row_id_column, DataField};
+use crate::spec::{is_row_tracking_column, DataField};
use std::cmp::Ordering;
use std::fmt;
@@ -468,7 +468,7 @@ impl Predicate {
op,
literals,
} => {
- if is_row_id_column(column) {
+ if is_row_tracking_column(column) {
return None;
}
let new_index = (*mapping.get(*index)?)?;
@@ -510,7 +510,7 @@ impl Predicate {
pub(crate) fn references_only_mapped_fields(&self, mapping:
&[Option<usize>]) -> bool {
match self {
Predicate::Leaf { column, index, .. } => {
- !is_row_id_column(column) &&
mapping.get(*index).is_some_and(Option::is_some)
+ !is_row_tracking_column(column) &&
mapping.get(*index).is_some_and(Option::is_some)
}
Predicate::And(children) | Predicate::Or(children) => children
.iter()
@@ -522,7 +522,7 @@ impl Predicate {
/// Project leaf field indices from table schema space into a smaller
field space.
///
- /// A `_ROW_ID` leaf never maps — see [`is_row_id_column`].
+ /// A row-tracking metadata leaf never maps — see
[`is_row_tracking_column`].
///
/// Unlike [`Self::remap_field_index`], mixed `AND` subtrees keep the
children
/// that can be projected and drop the rest. `OR` and `NOT` still require
all
@@ -542,7 +542,7 @@ impl Predicate {
op,
literals,
} => {
- if is_row_id_column(column) {
+ if is_row_tracking_column(column) {
return None;
}
let new_index = (*mapping.get(*index)?)?;
diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs
index b6f73e09..2377b4be 100644
--- a/crates/paimon/src/spec/schema.rs
+++ b/crates/paimon/src/spec/schema.rs
@@ -99,6 +99,26 @@ impl TableSchema {
&self.fields
}
+ /// Logical fields followed by the metadata available to table reads.
+ /// Row tracking provides ROW_ID and SEQUENCE_NUMBER; primary-key files
+ /// provide SEQUENCE_NUMBER independently of row tracking. Keeping this
+ /// schema shared makes projection and predicate name resolution agree.
+ pub fn fields_with_read_metadata(&self) -> Vec<DataField> {
+ let mut fields = self.fields.clone();
+ let row_tracking =
CoreOptions::new(&self.options).row_tracking_enabled();
+ if row_tracking && !fields.iter().any(|field| field.name() ==
ROW_ID_FIELD_NAME) {
+ fields.push(row_id_data_field());
+ }
+ if (row_tracking || !self.primary_keys.is_empty())
+ && !fields
+ .iter()
+ .any(|field| field.name() == SEQUENCE_NUMBER_FIELD_NAME)
+ {
+ fields.push(sequence_number_data_field());
+ }
+ fields
+ }
+
pub fn highest_field_id(&self) -> i32 {
self.highest_field_id
}
@@ -1147,19 +1167,28 @@ pub(crate) fn row_id_data_field() -> DataField {
)
}
-/// `_ROW_ID` is synthesized by the reader and is not a table column, so
-/// `PredicateBuilder` cannot resolve it and callers hand-build the leaf with a
-/// placeholder index. Every index-based resolution must recognize it by name
-/// instead, or it binds the predicate to whatever field sits at that index.
-///
-/// The other reserved names are excluded on purpose. `_SEQUENCE_NUMBER` and
-/// `_VALUE_KIND` are physical columns of a KV file and do have a position;
none
-/// of them is ever referenced by a predicate. Widening this to every reserved
-/// name would instead break a schema that predates their rejection.
+/// `_ROW_ID` conditions alone can be converted to positional row ranges.
pub(crate) fn is_row_id_column(name: &str) -> bool {
name == ROW_ID_FIELD_NAME
}
+/// Java's row-tracking version column. A missing or NULL physical value is
+/// supplied from the data file's max sequence number.
+pub(crate) fn sequence_number_data_field() -> DataField {
+ DataField::new(
+ SEQUENCE_NUMBER_FIELD_ID,
+ SEQUENCE_NUMBER_FIELD_NAME.to_string(),
+ DataType::BigInt(crate::spec::BigIntType::with_nullable(false)),
+ )
+}
+
+/// Row-tracking metadata has no index in the logical table schema. Resolve
+/// predicates by reserved name instead of interpreting a placeholder index as
+/// a user column. `_VALUE_KIND` remains a physical KV column.
+pub(crate) fn is_row_tracking_column(name: &str) -> bool {
+ is_row_id_column(name) || name == SEQUENCE_NUMBER_FIELD_NAME
+}
+
/// Must match Java Paimon's `SpecialFields.ROW_KIND` (Integer.MAX_VALUE - 4).
pub const ROW_KIND_FIELD_ID: i32 = i32::MAX - 4;
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index 123a138f..3e6ecb83 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -1343,7 +1343,7 @@ fn predicate_references_any_field(
Predicate::Leaf { column, index, .. } => {
// Never a BLOB column; resolving its placeholder index would force
// every BLOB to be resolved before filtering.
- if crate::spec::is_row_id_column(column) {
+ if crate::spec::is_row_tracking_column(column) {
return false;
}
field_names.contains(column)
@@ -2462,6 +2462,13 @@ fn build_source_plan_with_row_id_pushdown(
data_fields: info.data_fields.clone(),
read_fields: Vec::new(),
});
+ // Java rowTypeWithRowTracking adds this metadata provider even
+ // when write_cols omits the physical sequence column. The newest
+ // normal file supplies the group version, independently of the
+ // requested data columns.
+ normal_providers
+ .entry(crate::spec::SEQUENCE_NUMBER_FIELD_ID)
+ .or_insert(source_idx);
for &field_id in &info.field_ids {
// first normal file that carries the id wins (preserve
existing semantics)
normal_providers.entry(field_id).or_insert(source_idx);
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index 808f795b..d12e49df 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -537,27 +537,46 @@ impl DataFileReader {
let projects_row_id = read_type
.iter()
.any(|field| field.name() == ROW_ID_FIELD_NAME);
- let row_id_residual = projects_row_id
+ let row_tracking_enabled =
+
crate::spec::CoreOptions::new(&self.table_options).row_tracking_enabled();
+ let mut refs = Vec::new();
+ for predicate in &predicates {
+ crate::arrow::residual::collect_predicate_leaf_refs(predicate,
&mut refs);
+ }
+ let metadata_residual = (projects_row_id
&& predicates
.iter()
- .any(|predicate| !matches!(predicate, Predicate::AlwaysTrue));
- if row_id_residual {
+ .any(|predicate| !matches!(predicate, Predicate::AlwaysTrue)))
+ || (row_tracking_enabled
+ && refs
+ .iter()
+ .any(|(name, _)| *name ==
crate::spec::SEQUENCE_NUMBER_FIELD_NAME));
+ if metadata_residual {
// Decoder-side filtering can drop rows before positional `_ROW_ID`
// attachment. Read predicate-only columns without pushing the
filter
// down, attach the original ids, then evaluate the exact
predicate.
- let mut refs = Vec::new();
- for predicate in &predicates {
- crate::arrow::residual::collect_predicate_leaf_refs(predicate,
&mut refs);
- }
for (name, index) in refs {
- if name != ROW_ID_FIELD_NAME {
+ if name == crate::spec::SEQUENCE_NUMBER_FIELD_NAME {
+ crate::arrow::residual::push_unique_scan_field(
+ &mut read_type,
+ &crate::spec::sequence_number_data_field(),
+ );
+ } else if name == ROW_ID_FIELD_NAME {
+ crate::arrow::residual::push_unique_scan_field(
+ &mut read_type,
+ &crate::spec::row_id_data_field(),
+ );
+ } else {
if let Some(field) = table_fields.get(index) {
crate::arrow::residual::push_unique_scan_field(&mut
read_type, field);
}
}
}
}
- let residual_predicates = row_id_residual.then(||
crate::arrow::format::FilePredicates {
+ let projects_row_id = read_type
+ .iter()
+ .any(|field| field.name() == ROW_ID_FIELD_NAME);
+ let residual_predicates = metadata_residual.then(||
crate::arrow::format::FilePredicates {
predicates: predicates.clone(),
row_filter_factory: None,
file_fields: table_fields.clone(),
@@ -567,7 +586,7 @@ impl DataFileReader {
// post-filter until expression adaptation is proven for that path.
// Positional `_ROW_ID` materialization must see the unfiltered row
// stream, so the engine hook stays disabled for this projection.
- let row_filter_factory = (data_fields.is_none() && !projects_row_id)
+ let row_filter_factory = (data_fields.is_none() && !projects_row_id &&
!metadata_residual)
.then(|| self.row_filter_factory.clone())
.flatten();
let file_io = self.file_io.clone();
@@ -581,8 +600,6 @@ impl DataFileReader {
let read_timing = self.read_timing.clone();
let target_schema = build_target_arrow_schema(&read_type)?;
- let row_tracking_enabled =
-
crate::spec::CoreOptions::new(&table_options).row_tracking_enabled();
let file_fields = data_fields.clone().unwrap_or_else(||
table_fields.clone());
let data_schema_fields = data_schema_fields_for_file(
&file_fields,
@@ -616,7 +633,7 @@ impl DataFileReader {
mosaic_prefetch,
)?;
// Remap predicates from table-level to file-level indices.
- let file_predicates = if row_id_residual {
+ let file_predicates = if metadata_residual {
None
} else {
let remapped = crate::arrow::filtering::remap_predicates_to_file(
diff --git a/crates/paimon/src/table/kv_file_reader.rs
b/crates/paimon/src/table/kv_file_reader.rs
index 0e67eae3..3b603133 100644
--- a/crates/paimon/src/table/kv_file_reader.rs
+++ b/crates/paimon/src/table/kv_file_reader.rs
@@ -506,7 +506,10 @@ impl KeyValueFileReader {
let user_fields = crate::arrow::residual::widen_scan_fields(
&user_fields,
residual_file_predicates.as_ref(),
- );
+ )
+ .into_iter()
+ .filter(|field| !matches!(field.id(), SEQUENCE_NUMBER_FIELD_ID |
VALUE_KIND_FIELD_ID))
+ .collect();
let user_fields = widen_partial_update_sequence_group_fields(
self.config.merge_engine,
&self.config.table_options,
@@ -529,6 +532,13 @@ impl KeyValueFileReader {
// Indices within internal_schema (offset 2 for _SEQ and _VK).
let seq_index = 0;
let value_kind_index = 1;
+ let mut predicate_refs = Vec::new();
+ for predicate in &self.config.predicates {
+ crate::arrow::residual::collect_predicate_leaf_refs(predicate,
&mut predicate_refs);
+ }
+ let needs_sequence = predicate_refs
+ .iter()
+ .any(|(name, _)| *name == SEQUENCE_NUMBER_FIELD_NAME);
let key_indices: Vec<usize> = self
.config
.primary_keys
@@ -547,6 +557,7 @@ impl KeyValueFileReader {
.filter(|(index, field)| {
!key_names.contains(field.name())
&& (*index >= 2
+ || (needs_sequence && field.id() ==
SEQUENCE_NUMBER_FIELD_ID)
|| self
.config
.read_type
@@ -1221,9 +1232,9 @@ mod tests {
vec![Datum::Long(102)],
);
let mut read_builder = table.new_read_builder();
- read_builder
- .with_projection(&["id", "value", crate::spec::ROW_ID_FIELD_NAME])
- .unwrap();
+ // Untracked PK tables do not advertise ROW_ID. A hand-built predicate
+ // must still be rejected rather than binding its placeholder index.
+ read_builder.with_projection(&["id", "value"]).unwrap();
read_builder.with_filter(row_id);
let plan = read_builder.new_scan().plan().await.unwrap();
let err = read_builder
diff --git a/crates/paimon/src/table/read_builder.rs
b/crates/paimon/src/table/read_builder.rs
index b3fd6730..dfffa31d 100644
--- a/crates/paimon/src/table/read_builder.rs
+++ b/crates/paimon/src/table/read_builder.rs
@@ -389,7 +389,7 @@ impl<'a> PaimonReadBuilder<'a> {
let projection_names: Vec<String> = columns.iter().map(|c|
(*c).to_string()).collect();
validate_projection_possible(
self.table.identifier().full_name(),
- self.table.schema.fields(),
+ &self.table.schema.fields_with_read_metadata(),
&projection_names,
)?;
self.projection_names = Some(projection_names);
@@ -606,7 +606,7 @@ impl<'a> PaimonReadBuilder<'a> {
) -> Result<Vec<DataField>> {
resolve_projected_fields(
self.table.identifier().full_name(),
- self.table.schema.fields(),
+ &self.table.schema.fields_with_read_metadata(),
projection_names,
self.case_sensitive,
)
@@ -632,9 +632,6 @@ pub(super) fn validate_projection_possible(
.map(|f| f.name().to_ascii_lowercase())
.collect();
for name in projection_names {
- if name == crate::spec::ROW_ID_FIELD_NAME {
- continue;
- }
if !folded_names.contains(&name.to_ascii_lowercase()) {
return Err(Error::ColumnNotExist {
full_name: full_name.clone(),
@@ -693,11 +690,6 @@ pub(super) fn resolve_projected_fields(
});
}
- if name == crate::spec::ROW_ID_FIELD_NAME {
- resolved.push(crate::spec::row_id_data_field());
- continue;
- }
-
let field = if case_sensitive {
sensitive_index.get(name.as_str()).copied()
} else {
@@ -726,7 +718,10 @@ pub(super) fn resolve_projected_fields(
pub(super) fn projected_read_field_ids_from_fields(fields: &[DataField]) ->
HashSet<i32> {
fields
.iter()
- .filter(|field| !is_system_projection_field(field.id()))
+ .filter(|field| {
+ field.id() == crate::spec::SEQUENCE_NUMBER_FIELD_ID
+ || !is_system_projection_field(field.id())
+ })
.map(|field| field.id())
.collect::<HashSet<_>>()
}
@@ -748,9 +743,14 @@ fn projected_read_field_ids_with_predicates(
crate::arrow::residual::collect_predicate_leaf_refs(predicate, &mut
refs);
}
for (name, index) in refs {
+ if name == crate::spec::SEQUENCE_NUMBER_FIELD_NAME {
+ // The version provider need not write any projected user column.
+ field_ids.insert(crate::spec::SEQUENCE_NUMBER_FIELD_ID);
+ continue;
+ }
// Not in `table_fields`, and excluded from this set anyway — files
never
// list it in `write_cols`.
- if crate::spec::is_row_id_column(name) {
+ if crate::spec::is_row_tracking_column(name) {
continue;
}
let Some(field) = table_fields.get(index) else {
@@ -1652,6 +1652,66 @@ mod tests {
assert!(matches!(err, crate::Error::ConfigInvalid { .. }));
}
+ #[test]
+ fn test_metadata_projection_case_insensitive_collision_is_ambiguous() {
+ for (user, metadata) in [
+ ("_sequence_number", "_SEQUENCE_NUMBER"),
+ ("_row_id", "_ROW_ID"),
+ ] {
+ let untracked = TableSchema::new(
+ 0,
+ &Schema::builder()
+ .column(user, DataType::Int(IntType::new()))
+ .build()
+ .unwrap(),
+ );
+ let tracked = untracked.copy_with_options(HashMap::from([(
+ "row-tracking.enabled".into(),
+ "true".into(),
+ )]));
+ let fields = tracked.fields_with_read_metadata();
+ let metadata_fields = fields
+ .iter()
+ .filter(|field| field.id() != 0)
+ .cloned()
+ .collect::<Vec<_>>();
+ for name in [user, metadata] {
+ let err =
+ super::resolve_projected_fields("db.t".into(), &fields,
&[name.into()], false)
+ .unwrap_err();
+ assert!(
+ matches!(err, crate::Error::ConfigInvalid { message } if
message.contains("Ambiguous"))
+ );
+ let resolved = super::resolve_projected_fields(
+ "db.t".into(),
+ &metadata_fields,
+ &[name.into()],
+ false,
+ )
+ .unwrap();
+ assert_eq!(resolved[0].name(), metadata);
+ let resolved = super::resolve_projected_fields(
+ "db.t".into(),
+ &untracked.fields_with_read_metadata(),
+ &[name.into()],
+ false,
+ )
+ .unwrap();
+ assert_eq!(resolved[0].name(), user);
+ assert_eq!(resolved[0].id(), 0);
+ }
+ let fields = super::resolve_projected_fields(
+ "db.t".into(),
+ &fields,
+ &[user.into(), metadata.into()],
+ true,
+ )
+ .unwrap();
+ assert_eq!(fields[0].id(), 0);
+ assert_ne!(fields[1].id(), 0);
+ }
+ }
+
#[test]
fn test_resolve_projection_case_insensitive_dedups_by_folded_name() {
// With case-insensitive matching, `["Name","name"]` both resolve to
the
diff --git a/crates/paimon/src/table/row_sidecar.rs
b/crates/paimon/src/table/row_sidecar.rs
index 5cbdc953..9e7dd87d 100644
--- a/crates/paimon/src/table/row_sidecar.rs
+++ b/crates/paimon/src/table/row_sidecar.rs
@@ -241,14 +241,8 @@ pub(super) async fn read_target(
) -> Result<(String, u64)> {
if let Some(name) = selected_sidecar(file, local_ranges, options)? {
let path = file.aligned_file_path(bucket_path, name);
- match io.get_status(&path).await {
- Ok(status) => return Ok((path, status.size)),
- Err(error) => {
- if !options.scan_ignore_lost_file() {
- return Err(error);
- }
- }
- }
+ let status = io.get_status(&path).await?;
+ return Ok((path, status.size));
}
Ok((file.data_file_path(bucket_path), file.file_size as u64))
}
diff --git a/crates/paimon/src/table/sort_merge.rs
b/crates/paimon/src/table/sort_merge.rs
index 2641d68c..665ff7d9 100644
--- a/crates/paimon/src/table/sort_merge.rs
+++ b/crates/paimon/src/table/sort_merge.rs
@@ -251,6 +251,7 @@ pub(crate) struct PartialUpdateMergeFunction {
ignore_delete: bool,
remove_record_on_delete: bool,
sequence_group_partial_delete: HashSet<usize>,
+ sequence_group_enabled: bool,
sequence_groups: Vec<RuntimeSequenceGroup>,
grouped_fields: HashSet<usize>,
aggregators: Option<Mutex<FieldAggregatorSlots>>,
@@ -258,6 +259,32 @@ pub(crate) struct PartialUpdateMergeFunction {
type FieldAggregatorSlots = Vec<Option<Box<dyn FieldAggregator>>>;
+/// KV metadata belongs to the merged record, outside user-field aggregation.
+/// Java merge functions carry the latest effective sequence separately and
+/// normalize surviving records to INSERT, even when the last input retracts.
+fn merged_metadata_column(
+ field: &arrow_schema::Field,
+ row: &MergeRow,
+ deleted: bool,
+) -> Option<ArrayRef> {
+ match field.name().as_str() {
+ crate::spec::SEQUENCE_NUMBER_FIELD_NAME => {
+ Some(std::sync::Arc::new(Int64Array::from(vec![
+ row.sequence_number,
+ ])))
+ }
+ crate::spec::VALUE_KIND_FIELD_NAME => {
+ let kind = if deleted {
+ RowKind::Delete
+ } else {
+ RowKind::Insert
+ };
+ Some(std::sync::Arc::new(Int8Array::from(vec![kind as i8])))
+ }
+ _ => None,
+ }
+}
+
#[derive(Debug, Clone)]
struct RuntimeSequenceGroup {
sequence_indices: Vec<usize>,
@@ -333,6 +360,7 @@ impl PartialUpdateMergeFunction {
remove_record_on_delete: PartialUpdateConfig::new(table_options)
.remove_record_on_delete(),
sequence_group_partial_delete: HashSet::new(),
+ sequence_group_enabled: false,
sequence_groups: Vec::new(),
grouped_fields: HashSet::new(),
aggregators: None,
@@ -349,6 +377,7 @@ impl PartialUpdateMergeFunction {
let config = PartialUpdateConfig::new(table_options);
config.validate_read_mode(true, table_name)?;
let groups = config.validated_sequence_groups(table_fields,
primary_keys)?;
+ let sequence_group_enabled = !groups.is_empty();
let declared_group_sequences: HashSet<&str> = groups
.iter()
.flat_map(|group| group.sequence_fields.iter().map(String::as_str))
@@ -441,6 +470,7 @@ impl PartialUpdateMergeFunction {
ignore_delete: CoreOptions::new(table_options).ignore_delete(),
remove_record_on_delete: config.remove_record_on_delete(),
sequence_group_partial_delete,
+ sequence_group_enabled,
sequence_groups,
grouped_fields,
aggregators: aggregators
@@ -506,6 +536,7 @@ impl PartialUpdateMergeFunction {
let mut saw_add = false;
let mut current_delete_row = false;
let mut last_retract_source = None;
+ let mut latest_row = None;
for row_idx in ordered_row_indices {
let row = &rows[row_idx];
@@ -514,6 +545,7 @@ impl PartialUpdateMergeFunction {
if self.ignore_delete {
continue;
}
+ latest_row = Some(row);
if !saw_add && last_retract_source.is_none() {
// Java initializes the row from the first retract before
// applying sequence-group retractions. That row also
@@ -530,7 +562,7 @@ impl PartialUpdateMergeFunction {
}
last_retract_source = Some((row.batch_idx, row.row_idx));
current_delete_row = false;
- if !self.sequence_groups.is_empty() {
+ if self.sequence_group_enabled {
let mut full_delete = false;
for (group_idx, group) in
self.sequence_groups.iter().enumerate() {
let sequence_is_empty =
group.sequence_indices.iter().all(|&index| {
@@ -618,6 +650,7 @@ impl PartialUpdateMergeFunction {
}
saw_add = true;
current_delete_row = false;
+ latest_row = Some(row);
for (output_col_idx, selected) in
selected_by_col.iter_mut().enumerate() {
if self.grouped_fields.contains(&output_col_idx) {
@@ -712,6 +745,9 @@ impl PartialUpdateMergeFunction {
.iter()
.enumerate()
.map(|(output_col_idx, field)| {
+ if let Some(column) = merged_metadata_column(field,
latest_row.unwrap(), !saw_add) {
+ return Ok(column);
+ }
let column = match aggregators
.as_ref()
.and_then(|aggregators| aggregators.get(output_col_idx))
@@ -822,8 +858,8 @@ fn compare_sequence_group_rows(
/// Reference: Java
`org.apache.paimon.mergetree.compact.aggregate.AggregateMergeFunction`.
#[derive(Debug)]
pub(crate) struct AggregateMergeFunction {
- /// One slot per output column. `None` marks primary-key columns that are
- /// copied through; `Some` holds the aggregator that owns the column.
+ /// One slot per output column. `None` marks keys or KV metadata;
+ /// `Some` holds the aggregator that owns a user value column.
aggregators: Mutex<Vec<Option<Box<dyn FieldAggregator>>>>,
remove_record_on_delete: bool,
}
@@ -855,6 +891,12 @@ impl AggregateMergeFunction {
.iter()
.map(|field| -> crate::Result<Option<Box<dyn FieldAggregator>>> {
let name = field.name();
+ if matches!(
+ field.id(),
+ crate::spec::SEQUENCE_NUMBER_FIELD_ID |
crate::spec::VALUE_KIND_FIELD_ID
+ ) {
+ return Ok(None);
+ }
let agg_name: &str = if seq_set.contains(name) {
"last_value"
} else if pk_set.contains(name) {
@@ -939,15 +981,20 @@ impl AggregateMergeFunction {
}
}
- // Use the last sorted row to source primary-key column values: every
- // row in the group shares the same PK by construction, so any row
- // works; picking the last one keeps the slice cheap to compute.
+ // Java getResult uses the last KV for keys and record metadata.
let pk_source = rows.last().unwrap();
let output_columns: Vec<ArrayRef> = aggregators
.iter()
.enumerate()
.map(|(col_idx, slot)| -> crate::Result<ArrayRef> {
+ if let Some(column) = merged_metadata_column(
+ output_schema.field(col_idx),
+ pk_source,
+ current_delete_row,
+ ) {
+ return Ok(column);
+ }
if current_delete_row {
return Ok(batch_buffer[pk_source.batch_idx]
.column_for_output(col_idx, source_output_col_indices)
diff --git a/crates/paimon/src/table/stats_filter.rs
b/crates/paimon/src/table/stats_filter.rs
index 8c778b2b..accc320f 100644
--- a/crates/paimon/src/table/stats_filter.rs
+++ b/crates/paimon/src/table/stats_filter.rs
@@ -24,7 +24,8 @@ use crate::predicate_stats::{
predicates_may_match_with_schema, StatsAccessor,
};
use crate::spec::{
- extract_datum, is_row_id_column, BinaryRow, DataField, DataFileMeta,
DataType, Datum, Predicate,
+ extract_datum, is_row_tracking_column, BinaryRow, DataField, DataFileMeta,
DataType, Datum,
+ Predicate,
};
use std::collections::HashMap;
use std::sync::Arc;
@@ -601,8 +602,8 @@ fn data_evolution_predicate_may_match(
file_stats,
row_count,
),
- // `_ROW_ID` has no column stats, so never prune a group on it.
- Predicate::Leaf { column, .. } if is_row_id_column(column) => true,
+ // Row-tracking metadata has no logical column stats, so never prune a
group on it.
+ Predicate::Leaf { column, .. } if is_row_tracking_column(column) =>
true,
Predicate::Leaf {
index,
data_type,
@@ -667,7 +668,7 @@ fn data_evolution_predicate_must_match(
row_count,
),
// Stats cannot decide `_ROW_ID`, so it never provably matches.
- Predicate::Leaf { column, .. } if is_row_id_column(column) => false,
+ Predicate::Leaf { column, .. } if is_row_tracking_column(column) =>
false,
Predicate::Leaf {
index,
data_type,
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index f31d64fd..a6479071 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -1012,6 +1012,11 @@ async fn prune_data_evolution_group_by_read_fields(
} else {
None
};
+ let sequence_idx = if read_field_ids.contains(&SEQUENCE_NUMBER_FIELD_ID) {
+ Some(data_evolution_sequence_file_index(&group)?)
+ } else {
+ None
+ };
let mut keep = Vec::with_capacity(group.len());
for (idx, file) in group.iter().enumerate() {
@@ -1035,6 +1040,11 @@ async fn prune_data_evolution_group_by_read_fields(
keep.push(anchor_idx);
}
}
+ if let Some(sequence_idx) = sequence_idx {
+ if !keep.contains(&sequence_idx) {
+ keep.push(sequence_idx);
+ }
+ }
if keep.is_empty() {
keep.push(data_evolution_representative_file(&group)?);
@@ -1047,6 +1057,10 @@ async fn prune_data_evolution_group_by_read_fields(
}
}
+ if sequence_idx.is_some() {
+ // Preserve original order for equal-sequence metadata providers.
+ keep.sort_unstable();
+ }
let mut files = group.into_iter().map(Some).collect::<Vec<_>>();
Ok(keep
.into_iter()
@@ -1054,6 +1068,21 @@ async fn prune_data_evolution_group_by_read_fields(
.collect())
}
+/// The newest normal file supplies row-tracking sequence metadata. It must
+/// survive user-column pruning even if it only wrote an unprojected column.
+fn data_evolution_sequence_file_index(group: &[DataFileMeta]) ->
crate::Result<usize> {
+ group
+ .iter()
+ .enumerate()
+ .filter(|(_, file)| is_normal_data_file(file))
+ .min_by_key(|(idx, file)|
(std::cmp::Reverse(file.max_sequence_number), *idx))
+ .map(|(idx, _)| idx)
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: "Data-evolution sequence metadata requires a normal data
file".into(),
+ source: None,
+ })
+}
+
#[derive(Debug, Clone)]
pub struct TableScan<'a>(TableScanKind<'a>);
diff --git a/crates/paimon/tests/data_evolution_row_sidecar_test.rs
b/crates/paimon/tests/data_evolution_row_sidecar_test.rs
index c8814e3b..b2d924ee 100644
--- a/crates/paimon/tests/data_evolution_row_sidecar_test.rs
+++ b/crates/paimon/tests/data_evolution_row_sidecar_test.rs
@@ -249,7 +249,7 @@ async fn
selection_thresholds_are_inclusive_but_full_reads_keep_primary() {
}
#[tokio::test]
-async fn missing_sidecar_only_falls_back_when_ignore_lost_file_is_enabled() {
+async fn missing_sidecar_is_an_error_for_selected_reads() {
let table = table(&[]).await;
let messages = append(&table, 0, 100).await;
let file = &messages[0].new_files[0];
@@ -258,35 +258,17 @@ async fn
missing_sidecar_only_falls_back_when_ignore_lost_file_is_enabled() {
&file.extra_files[0],
);
table.file_io().delete_file(&sidecar).await.unwrap();
- let strict = table.copy_with_options(HashMap::from([(
- "scan.ignore-lost-files".into(),
- "false".into(),
- )]));
- for reader_table in [&table, &strict] {
- let error = read(reader_table, Some(vec![RowRange::new(5, 5)]))
- .await
- .unwrap_err();
- assert!(error.to_string().contains(".row"), "{error}");
- }
- let tolerant = table.copy_with_options(HashMap::from([(
- "scan.ignore-lost-files".into(),
- "true".into(),
- )]));
- assert_eq!(
- rows(
- &read(&tolerant, Some(vec![RowRange::new(5, 5)]))
- .await
- .unwrap()
- ),
- vec![(5, "v5".into(), 5)]
- );
+ let error = read(&table, Some(vec![RowRange::new(5, 5)]))
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains(".row"), "{error}");
// Whole-file reads never require the auxiliary file.
assert_eq!(rows(&read(&table, None).await.unwrap()).len(), 100);
}
#[tokio::test]
-async fn corrupt_sidecar_is_not_hidden_by_ignore_lost_file() {
- let table = table(&[("scan.ignore-lost-files", "true")]).await;
+async fn corrupt_sidecar_is_an_error_for_selected_reads() {
+ let table = table(&[]).await;
let messages = append(&table, 0, 100).await;
let file = &messages[0].new_files[0];
let path = file.aligned_file_path(
diff --git a/crates/paimon/tests/data_evolution_system_columns_test.rs
b/crates/paimon/tests/data_evolution_system_columns_test.rs
new file mode 100644
index 00000000..d87847fc
--- /dev/null
+++ b/crates/paimon/tests/data_evolution_system_columns_test.rs
@@ -0,0 +1,599 @@
+// 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.
+
+mod common;
+
+use arrow_array::{Array, ArrayRef, Int32Array, Int64Array, Int8Array,
RecordBatch};
+use common::incremental_helpers::{memory_table, persist_table_schema,
setup_dirs};
+use futures::TryStreamExt;
+use paimon::spec::{
+ BigIntType, DataField, DataType, Datum, IntType, Predicate,
PredicateOperator, Schema,
+ TableSchema,
+};
+use paimon::table::Table;
+use std::sync::Arc;
+
+async fn table() -> Table {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("v", DataType::Int(IntType::new()))
+ .column("w", DataType::Int(IntType::new()))
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .build()
+ .unwrap();
+ create_table(schema).await
+}
+
+async fn create_table(schema: Schema) -> Table {
+ let (io, table) = memory_table("memory:/system_columns",
TableSchema::new(0, &schema));
+ setup_dirs(&io, table.location()).await;
+ persist_table_schema(&io, table.location(), table.schema()).await;
+ table
+}
+
+async fn append(table: &Table, start: i32) {
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ let batch = RecordBatch::try_from_iter([
+ (
+ "id",
+ Arc::new(Int32Array::from(vec![start, start + 1])) as ArrayRef,
+ ),
+ ("v", Arc::new(Int32Array::from(vec![10, 20])) as ArrayRef),
+ ("w", Arc::new(Int32Array::from(vec![100, 200])) as ArrayRef),
+ ])
+ .unwrap();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ writer.close().await;
+ builder.new_commit().commit(messages).await.unwrap();
+}
+
+async fn update(table: &Table, field: &str, row_id: i64, value: i32) {
+ let builder = table.new_write_builder();
+ let messages = builder
+ .new_update()
+ .unwrap()
+ .update_by_arrow_with_row_id(vec![RecordBatch::try_from_iter([
+ (
+ "_ROW_ID",
+ Arc::new(Int64Array::from(vec![row_id])) as ArrayRef,
+ ),
+ (field, Arc::new(Int32Array::from(vec![value])) as ArrayRef),
+ ])
+ .unwrap()])
+ .await
+ .unwrap();
+ builder.new_commit().commit(messages).await.unwrap();
+}
+
+fn sequence_field() -> DataField {
+ DataField::new(
+ paimon::spec::SEQUENCE_NUMBER_FIELD_ID,
+ "_SEQUENCE_NUMBER".into(),
+ DataType::BigInt(BigIntType::with_nullable(false)),
+ )
+}
+
+async fn sequences(table: &Table) -> paimon::Result<Vec<i64>> {
+ let mut builder = table.new_read_builder();
+ builder.with_read_type(vec![sequence_field()]);
+ let plan = builder.new_scan().plan().await?;
+ let batches: Vec<_> = builder
+ .new_read()?
+ .to_arrow(plan.splits())?
+ .try_collect()
+ .await?;
+ let mut values = Vec::new();
+ for batch in batches {
+ let column = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ assert_eq!(column.null_count(), 0);
+ values.extend(column.values().iter().copied());
+ }
+ values.sort_unstable();
+ Ok(values)
+}
+
+#[tokio::test]
+async fn partial_files_supply_the_latest_sequence_without_a_physical_column() {
+ let table = table().await;
+ append(&table, 0).await;
+ assert_eq!(sequences(&table).await.unwrap(), vec![1, 1]);
+ update(&table, "v", 0, 77).await;
+ assert_eq!(sequences(&table).await.unwrap(), vec![2, 2]);
+ update(&table, "w", 1, 88).await;
+ assert_eq!(sequences(&table).await.unwrap(), vec![3, 3]);
+}
+
+fn sequence_predicate(op: PredicateOperator, values: &[i64]) -> Predicate {
+ // A system column has no position in the logical schema. This deliberately
+ // aliases the id column to catch accidental index-based stats/filtering.
+ Predicate::Leaf {
+ column: "_SEQUENCE_NUMBER".into(),
+ index: 0,
+ data_type: DataType::BigInt(BigIntType::with_nullable(false)),
+ op,
+ literals: values.iter().map(|v| Datum::Long(*v)).collect(),
+ }
+}
+
+async fn matching_ids(table: &Table, predicate: Predicate, project_sequence:
bool) -> Vec<i32> {
+ let mut builder = table.new_read_builder();
+ let projection = if project_sequence {
+ vec!["id", "_SEQUENCE_NUMBER"]
+ } else {
+ vec!["id"]
+ };
+ builder
+ .with_projection(&projection)
+ .unwrap()
+ .with_filter(predicate);
+ let plan = builder.new_scan().plan().await.unwrap();
+ let batches: Vec<_> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut ids = Vec::new();
+ for batch in batches {
+ assert_eq!(batch.num_columns(), projection.len());
+ let values = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ ids.extend(values.values().iter().copied());
+ }
+ ids.sort_unstable();
+ ids
+}
+
+#[tokio::test]
+async fn sequence_predicates_use_metadata_before_and_after_partial_updates() {
+ let table = table().await;
+ append(&table, 100).await;
+ append(&table, 200).await;
+ for project in [false, true] {
+ assert_eq!(
+ matching_ids(
+ &table,
+ sequence_predicate(PredicateOperator::Eq, &[1]),
+ project
+ )
+ .await,
+ vec![100, 101]
+ );
+ assert_eq!(
+ matching_ids(
+ &table,
+ sequence_predicate(PredicateOperator::Eq, &[2]),
+ project
+ )
+ .await,
+ vec![200, 201]
+ );
+ }
+ update(&table, "v", 0, 77).await;
+ for project in [false, true] {
+ assert_eq!(
+ matching_ids(
+ &table,
+ sequence_predicate(PredicateOperator::Eq, &[3]),
+ project
+ )
+ .await,
+ vec![100, 101]
+ );
+ assert_eq!(
+ matching_ids(
+ &table,
+ sequence_predicate(PredicateOperator::Lt, &[3]),
+ project
+ )
+ .await,
+ vec![200, 201]
+ );
+ assert_eq!(
+ matching_ids(
+ &table,
+ sequence_predicate(PredicateOperator::IsNull, &[]),
+ project
+ )
+ .await,
+ Vec::<i32>::new()
+ );
+ assert_eq!(
+ matching_ids(
+ &table,
+ sequence_predicate(PredicateOperator::IsNotNull, &[]),
+ project
+ )
+ .await,
+ vec![100, 101, 200, 201]
+ );
+ }
+ let id = paimon::spec::PredicateBuilder::new(table.schema().fields())
+ .equal("id", Datum::Int(200))
+ .unwrap();
+ assert_eq!(
+ matching_ids(
+ &table,
+ Predicate::or(vec![sequence_predicate(PredicateOperator::Eq,
&[3]), id]),
+ false
+ )
+ .await,
+ vec![100, 101, 200]
+ );
+ assert_eq!(
+ matching_ids(
+ &table,
+ Predicate::Not(Box::new(sequence_predicate(PredicateOperator::Eq,
&[3]))),
+ false
+ )
+ .await,
+ vec![200, 201]
+ );
+}
+
+#[tokio::test]
+async fn physical_versions_are_kept_and_null_versions_use_manifest_metadata() {
+ use parquet::arrow::ArrowWriter;
+ let table = table().await;
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ let batch = RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![100, 101])) as ArrayRef),
+ ("v", Arc::new(Int32Array::from(vec![10, 20])) as ArrayRef),
+ ("w", Arc::new(Int32Array::from(vec![100, 200])) as ArrayRef),
+ ])
+ .unwrap();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ let mut messages = writer.prepare_commit().await.unwrap();
+ writer.close().await;
+ let schema = batch.schema();
+ let mut columns: Vec<_> = schema
+ .fields()
+ .iter()
+ .zip(batch.columns())
+ .map(|(field, array)| (field.name().as_str(), array.clone()))
+ .collect();
+ columns.push((
+ "_SEQUENCE_NUMBER",
+ Arc::new(Int64Array::from(vec![Some(7), None])) as ArrayRef,
+ ));
+ let physical = RecordBatch::try_from_iter(columns).unwrap();
+ let mut parquet = ArrowWriter::try_new(Vec::new(), physical.schema(),
None).unwrap();
+ parquet.write(&physical).unwrap();
+ let data = bytes::Bytes::from(parquet.into_inner().unwrap());
+ let file = &mut messages[0].new_files[0];
+ file.file_size = data.len() as i64;
+ table
+ .file_io()
+ .new_output(&format!("{}/bucket-0/{}", table.location(),
file.file_name))
+ .unwrap()
+ .write(data)
+ .await
+ .unwrap();
+ builder.new_commit().commit(messages).await.unwrap();
+ assert_eq!(sequences(&table).await.unwrap(), vec![1, 7]);
+ assert_eq!(
+ matching_ids(
+ &table,
+ sequence_predicate(PredicateOperator::Eq, &[7]),
+ false
+ )
+ .await,
+ vec![100]
+ );
+ assert_eq!(
+ matching_ids(
+ &table,
+ sequence_predicate(PredicateOperator::Eq, &[1]),
+ false
+ )
+ .await,
+ vec![101]
+ );
+}
+
+#[tokio::test]
+async fn blob_versions_cannot_replace_the_normal_metadata_provider() {
+ use arrow_array::LargeBinaryArray;
+ use paimon::spec::{BinaryRow, BlobType};
+ use paimon::table::DataSplitBuilder;
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("payload", DataType::Blob(BlobType::new()))
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .build()
+ .unwrap();
+ let table = create_table(schema).await;
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(
+ &RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![100, 101])) as ArrayRef),
+ (
+ "payload",
+ Arc::new(LargeBinaryArray::from(vec![
+ Some(b"a".as_slice()),
+ Some(b"b".as_slice()),
+ ])) as ArrayRef,
+ ),
+ ])
+ .unwrap(),
+ )
+ .await
+ .unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ writer.close().await;
+ builder.new_commit().commit(messages).await.unwrap();
+ let mut files = table
+ .new_read_builder()
+ .new_scan()
+ .plan()
+ .await
+ .unwrap()
+ .splits()[0]
+ .data_files()
+ .to_vec();
+ let blob = files
+ .iter_mut()
+ .find(|file| file.file_name.ends_with(".blob"))
+ .unwrap();
+ blob.max_sequence_number = 99;
+ // An unrequested payload file must not be read just to supply metadata.
+ table
+ .file_io()
+ .delete_file(&format!("{}/bucket-0/{}", table.location(),
blob.file_name))
+ .await
+ .unwrap();
+ let split = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_total_buckets(1)
+ .with_bucket_path(format!("{}/bucket-0", table.location()))
+ .with_data_files(files)
+ .build()
+ .unwrap();
+ let mut read = table.new_read_builder();
+ read.with_projection(&["id", "_SEQUENCE_NUMBER"]).unwrap();
+ let batches: Vec<_> = read
+ .new_read()
+ .unwrap()
+ .to_arrow(&[split])
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 2);
+ for batch in batches {
+ let seq = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ assert_eq!(seq.values().as_ref(), &[1, 1]);
+ }
+}
+
+#[tokio::test]
+async fn primary_key_metadata_never_uses_the_user_aggregate_function() {
+ for engine in ["deduplicate", "first-row", "partial-update",
"aggregation"] {
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("v", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .option("merge-engine", engine);
+ if engine == "aggregation" {
+ schema = schema.option("fields.default-aggregate-function", "sum");
+ }
+ let schema = schema.build().unwrap();
+ let table = create_table(schema).await;
+ let table = if engine == "first-row" {
+ // A Java pk-clustering table permits first-row DVs. Reproduce the
+ // resolved catalog options and read level-0 files without
compaction.
+ table.copy_with_options(std::collections::HashMap::from([
+ ("deletion-vectors.enabled".into(), "true".into()),
+ ("deletion-vectors.merge-on-read".into(), "true".into()),
+ ("pk-clustering-override".into(), "true".into()),
+ ]))
+ } else {
+ table
+ };
+ for _ in 0..3 {
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(
+ &RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![100])) as
ArrayRef),
+ ("v", Arc::new(Int32Array::from(vec![10])) as
ArrayRef),
+ ])
+ .unwrap(),
+ )
+ .await
+ .unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ writer.close().await;
+ builder.new_commit().commit(messages).await.unwrap();
+ }
+ let sequence = if engine == "first-row" { 0 } else { 2 };
+ assert_eq!(
+ matching_ids(
+ &table,
+ sequence_predicate(PredicateOperator::Eq, &[sequence]),
+ false
+ )
+ .await,
+ vec![100]
+ );
+ assert_eq!(
+ matching_ids(
+ &table,
+ sequence_predicate(PredicateOperator::NotEq, &[sequence]),
+ true
+ )
+ .await,
+ Vec::<i32>::new()
+ );
+ let mut reader = table.new_read_builder();
+ reader.with_projection(&["v", "_SEQUENCE_NUMBER"]).unwrap();
+ let plan = reader.new_scan().plan().await.unwrap();
+ let batches: Vec<_> = reader
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(batches.len(), 1);
+ let value = batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ assert_eq!(
+ value.value(0),
+ if engine == "aggregation" { 30 } else { 10 }
+ );
+ let seq = batches[0]
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ assert_eq!(seq.value(0), sequence);
+ }
+}
+
+async fn retract_metadata(engine: &str) {
+ for kind in [1_i8, 3] {
+ for one_file in [false, true] {
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("g", DataType::Int(IntType::new()))
+ .column("v", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .option("merge-engine", engine);
+ schema = if engine == "aggregation" {
+ schema
+ .option("fields.default-aggregate-function", "sum")
+ .option("aggregation.remove-record-on-delete", "false")
+ } else {
+ schema.option("fields.g.sequence-group", "v")
+ };
+ let table = create_table(schema.build().unwrap()).await;
+ let operations = [(1, 10, 0_i8), (2, 5, kind)];
+ for rows in operations.chunks(if one_file { 2 } else { 1 }) {
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(
+ &RecordBatch::try_from_iter([
+ (
+ "id",
+ Arc::new(Int32Array::from(vec![100;
rows.len()])) as ArrayRef,
+ ),
+ (
+ "g",
+ Arc::new(Int32Array::from(
+ rows.iter().map(|r|
r.0).collect::<Vec<_>>(),
+ )) as ArrayRef,
+ ),
+ (
+ "v",
+ Arc::new(Int32Array::from(
+ rows.iter().map(|r|
r.1).collect::<Vec<_>>(),
+ )) as ArrayRef,
+ ),
+ (
+ "_VALUE_KIND",
+ Arc::new(Int8Array::from(
+ rows.iter().map(|r|
r.2).collect::<Vec<_>>(),
+ )) as ArrayRef,
+ ),
+ ])
+ .unwrap(),
+ )
+ .await
+ .unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ writer.close().await;
+ builder.new_commit().commit(messages).await.unwrap();
+ }
+ assert_eq!(
+ sequences(&table).await.unwrap(),
+ vec![1],
+ "{engine}, kind={kind}, one_file={one_file}"
+ );
+ assert_eq!(
+ matching_ids(
+ &table,
+ sequence_predicate(PredicateOperator::Eq, &[1]),
+ false
+ )
+ .await,
+ vec![100]
+ );
+ let mut reader = table.new_read_builder();
+ reader.with_projection(&["v", "_SEQUENCE_NUMBER"]).unwrap();
+ let plan = reader.new_scan().plan().await.unwrap();
+ let batches: Vec<_> = reader
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let values = batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ if engine == "aggregation" {
+ assert_eq!(values.value(0), 5);
+ } else {
+ assert!(values.is_null(0));
+ }
+ }
+ }
+}
+
+#[tokio::test]
+async fn aggregation_retractions_keep_the_latest_sequence() {
+ retract_metadata("aggregation").await;
+}
+
+#[tokio::test]
+async fn partial_update_retractions_keep_the_latest_sequence() {
+ retract_metadata("partial-update").await;
+}
