JeonDaehong commented on code in PR #1164:
URL: https://github.com/apache/paimon-rust/pull/1164#discussion_r4442815312
##########
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:
Fixed in 7270d1e1, both comparisons use `not_distinct` now.
`single_run_read_merges_repeated_null_keys` covers batch size 64 and 1, and
both returned the two NULL-key versions before. While merging main I also kept
lookup merges on sort-merge and made the fallback use the schema with the level
column.
--
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]
