JingsongLi commented on code in PR #1164:
URL: https://github.com/apache/paimon-rust/pull/1164#discussion_r4435252863
##########
crates/paimon/src/table/kv_file_reader.rs:
##########
@@ -184,6 +184,85 @@ fn ensure_merge_input_limit(input_stream_count: usize,
limit: Option<usize>) ->
Ok(())
}
+/// Handles retract rows of an unmerged single run the way the merge path
+/// handles a lone version of a key: first-row without `ignore-delete` rejects
+/// them, otherwise they are omitted.
+fn drop_lone_retracts(
+ batch: RecordBatch,
+ value_kind_index: usize,
+ reject: bool,
+) -> crate::Result<RecordBatch> {
+ // The merge cursor reads a missing or NULL `_VALUE_KIND` as INSERT.
+ let Some(kinds) = batch
+ .column(value_kind_index)
+ .as_any()
+ .downcast_ref::<Int8Array>()
+ else {
+ return Ok(batch);
+ };
+ if kinds.null_count() == 0 && kinds.values().iter().all(|&kind| kind == 0
|| kind == 2) {
+ return Ok(batch);
+ }
+ let mut keep = Vec::with_capacity(kinds.len());
+ for kind in kinds.iter() {
+ let is_add = RowKind::from_value(kind.unwrap_or(0))?.is_add();
+ if !is_add && reject {
+ return Err(Error::Unsupported {
+ message: "merge-engine=first-row does not support DELETE or
UPDATE_BEFORE rows; set ignore-delete=true to ignore them".to_string(),
+ });
+ }
+ keep.push(is_add);
+ }
+ arrow_select::filter::filter_record_batch(&batch,
&BooleanArray::from(keep)).map_err(|e| {
+ Error::DataInvalid {
+ message: format!("Failed to drop retract rows: {e}"),
+ source: Some(Box::new(e)),
+ }
+ })
+}
+
+/// Whether `batch` repeats a primary key, either in two adjacent rows or in
its
+/// first row and the last row of `previous`. A key type the comparison kernel
+/// does not support counts as repeated, so the caller merges.
+fn repeats_a_key(
+ previous: Option<&RecordBatch>,
+ batch: &RecordBatch,
+ key_indices: &[usize],
+) -> bool {
+ let rows = batch.num_rows();
+ if rows == 0 {
+ return false;
+ }
+ let previous = previous.filter(|previous| previous.num_rows() > 0);
+ let mut at_boundary = previous.is_some();
+ let mut adjacent: Option<BooleanArray> = None;
+ for &index in key_indices {
+ let column = batch.column(index);
+ if let Some(previous) = previous.filter(|_| at_boundary) {
+ let last = previous.column(index).slice(previous.num_rows() - 1,
1);
+ match arrow_ord::cmp::eq(&last, &column.slice(0, 1)) {
+ Ok(equal) => at_boundary = equal.true_count() == 1,
+ Err(_) => return true,
+ }
+ }
+ if rows > 1 {
+ let Ok(equal) =
+ arrow_ord::cmp::eq(&column.slice(0, rows - 1),
&column.slice(1, rows - 1))
Review Comment:
[P2] Detect repeated nullable keys with null-safe equality
`cmp::eq` returns NULL for `NULL == NULL`, so `true_count()` misses
duplicate keys with a NULL component. This differs from SortMerge's encoded-key
equality and lets the shortcut return multiple versions of one nullable primary
key. The boundary comparison at line 243 has the same issue.
An independent public prepare/commit/plan/read probe reconstructs historical
aggregation Parquet containing two NULL-key versions, commits accurate file
metadata/statistics plus a disjoint key-2 file, and asserts one normal two-file
non-raw split. With `primary-key.nullable=true` and `read.batch-size=1` or
`64`, this head returns three rows, including both NULL-key versions; the
unchanged probe passes on exact merge-base fb5e8f7e with two rows. Before #839,
the native writer accepted nullable Arrow input and aggregation flush retained
repeated versions; today's writer guarantees cannot exclude such historical
data.
Use null-safe equality such as `arrow_ord::cmp::not_distinct` in both
comparisons, or conservatively fall back to merging nullable keys. Please cover
within-batch and cross-batch NULL-key repetitions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
