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 02b02463 feat: enable DV merge engine and postpone fixed bucket writes
(#1101)
02b02463 is described below
commit 02b02463d2f4f997240512938a7af76de3bd5750
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Oct 11 14:40:01 2026 +0800
feat: enable DV merge engine and postpone fixed bucket writes (#1101)
---
crates/paimon/src/table/kv_file_reader.rs | 117 +++---
crates/paimon/src/table/kv_file_writer.rs | 52 ---
.../src/table/postpone_fixed_bucket_write.rs | 16 +-
.../table/postpone_fixed_bucket_write_builder.rs | 17 +-
crates/paimon/src/table/table_write.rs | 37 --
crates/paimon/tests/dv_merge_engine_write_test.rs | 405 +++++++++++++++++++++
6 files changed, 484 insertions(+), 160 deletions(-)
diff --git a/crates/paimon/src/table/kv_file_reader.rs
b/crates/paimon/src/table/kv_file_reader.rs
index a9d9cdf6..f50585d6 100644
--- a/crates/paimon/src/table/kv_file_reader.rs
+++ b/crates/paimon/src/table/kv_file_reader.rs
@@ -1732,61 +1732,76 @@ mod tests {
/// key with no replacement is resurrected.
#[tokio::test]
async fn dv_merge_on_read_applies_deletion_vectors_before_key_merge() {
- let file_io = test_file_io();
- let table_path = "memory:/dv_merge_on_read_with_dv";
- setup_dirs(&file_io, table_path).await;
- let table = pk_table(
- &file_io,
- table_path,
- &[
- ("deletion-vectors.enabled", "true"),
- ("deletion-vectors.merge-on-read", "true"),
- ],
- );
+ for (engine, expected) in [
+ ("deduplicate", 11),
+ ("partial-update", 11),
+ ("aggregation", 21),
+ ] {
+ let file_io = test_file_io();
+ let table_path = "memory:/dv_merge_on_read_with_dv";
+ setup_dirs(&file_io, table_path).await;
+ let table = pk_table(
+ &file_io,
+ table_path,
+ &[
+ ("merge-engine", engine),
+ ("deletion-vectors.enabled", "true"),
+ ("deletion-vectors.merge-on-read", "true"),
+ ],
+ );
+ let table = if engine == "aggregation" {
+ table.copy_with_options(HashMap::from([(
+ "fields.value.aggregate-function".into(),
+ "sum".into(),
+ )]))
+ } else {
+ table
+ };
- write_commit(&table, &int_batch(vec![1, 2], vec![Some(10),
Some(20)])).await;
- write_commit(&table, &int_batch(vec![1, 3], vec![Some(11),
Some(30)])).await;
+ write_commit(&table, &int_batch(vec![1, 2], vec![Some(10),
Some(20)])).await;
+ write_commit(&table, &int_batch(vec![1, 3], vec![Some(11),
Some(30)])).await;
- let all_files = table
- .new_read_builder()
- .new_scan()
- .with_scan_all_files()
- .plan()
- .await
- .unwrap();
- let mut files = all_files
- .splits()
- .iter()
- .flat_map(|split| split.data_files().iter().cloned())
- .collect::<Vec<_>>();
- files.sort_by_key(|file| file.min_sequence_number);
- assert_eq!(files.len(), 2);
- files[0].level = 1;
+ let all_files = table
+ .new_read_builder()
+ .new_scan()
+ .with_scan_all_files()
+ .plan()
+ .await
+ .unwrap();
+ let mut files = all_files
+ .splits()
+ .iter()
+ .flat_map(|split| split.data_files().iter().cloned())
+ .collect::<Vec<_>>();
+ files.sort_by_key(|file| file.min_sequence_number);
+ assert_eq!(files.len(), 2);
+ files[0].level = 1;
- let deletion_file = write_deletion_file(&file_io, table_path,
&[1]).await;
- let split = DataSplitBuilder::new()
- .with_snapshot(2)
- .with_partition(BinaryRow::new(0))
- .with_bucket(0)
- .with_bucket_path(format!("{table_path}/bucket-0"))
- .with_total_buckets(1)
- .with_data_files(files)
- .with_data_deletion_files(vec![Some(deletion_file), None])
- .with_raw_convertible(false)
- .build()
- .unwrap();
+ let deletion_file = write_deletion_file(&file_io, table_path,
&[1]).await;
+ let split = DataSplitBuilder::new()
+ .with_snapshot(2)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path(format!("{table_path}/bucket-0"))
+ .with_total_buckets(1)
+ .with_data_files(files)
+ .with_data_deletion_files(vec![Some(deletion_file), None])
+ .with_raw_convertible(false)
+ .build()
+ .unwrap();
- let batches = table
- .new_read_builder()
- .new_read()
- .unwrap()
- .to_arrow(&[split])
- .unwrap()
- .try_collect::<Vec<_>>()
- .await
- .unwrap();
- assert_eq!(int_column(&batches, "id"), vec![1, 3]);
- assert_eq!(int_column(&batches, "value"), vec![11, 30]);
+ let batches = table
+ .new_read_builder()
+ .new_read()
+ .unwrap()
+ .to_arrow(&[split])
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(int_column(&batches, "id"), vec![1, 3]);
+ assert_eq!(int_column(&batches, "value"), vec![expected, 30]);
+ }
}
/// Disjoint files in one split are consumed as sequential merge groups. A
diff --git a/crates/paimon/src/table/kv_file_writer.rs
b/crates/paimon/src/table/kv_file_writer.rs
index d12c2b26..33845d4e 100644
--- a/crates/paimon/src/table/kv_file_writer.rs
+++ b/crates/paimon/src/table/kv_file_writer.rs
@@ -110,7 +110,6 @@ pub(crate) struct KeyValueWriteConfig {
pub sequence_field_indices: Vec<usize>,
/// Merge engine for deduplication.
pub merge_engine: MergeEngine,
- pub deletion_vectors_enabled: bool,
/// File indexes follow the sorted, merged data rows and never changelog
rows.
pub file_index_options: Option<Arc<FileIndexOptions>>,
}
@@ -145,29 +144,11 @@ impl KeyValueFileWriter {
partial_update.validate_write_mode(true, &config.table_name)?;
partial_update
.validated_aggregate_functions(&config.value_fields,
&config.primary_keys)?;
-
- if config.deletion_vectors_enabled {
- return Err(crate::Error::Unsupported {
- message: format!(
- "Table '{}' uses merge-engine=partial-update with
deletion-vectors.enabled=true, which is not supported yet",
- config.table_name
- ),
- });
- }
}
if config.merge_engine == MergeEngine::Aggregation {
AggregationConfig::new(&config.table_options)
.validate_runtime_mode(true, &config.table_name)?;
-
- if config.deletion_vectors_enabled {
- return Err(crate::Error::Unsupported {
- message: format!(
- "Table '{}' uses merge-engine=aggregation with
deletion-vectors.enabled=true, which is not supported yet",
- config.table_name
- ),
- });
- }
}
let paths = super::data_file_path_factory::DataFilePathFactory::new(
@@ -1346,7 +1327,6 @@ mod tests {
],
sequence_field_indices: vec![1],
merge_engine,
- deletion_vectors_enabled: false,
file_index_options: None,
}
}
@@ -2518,22 +2498,6 @@ mod tests {
);
}
- #[test]
- fn test_new_rejects_partial_update_with_deletion_vectors() {
- let mut config = test_write_config(MergeEngine::PartialUpdate);
- config.deletion_vectors_enabled = true;
-
- let err =
KeyValueFileWriter::new(FileIOBuilder::new("memory").build().unwrap(), config,
0)
- .err()
- .unwrap();
-
- assert!(matches!(
- err,
- crate::Error::Unsupported { message }
- if message.contains("deletion-vectors.enabled=true")
- ));
- }
-
#[test]
fn test_new_rejects_unsupported_partial_update_options() {
let mut config = test_write_config(MergeEngine::PartialUpdate);
@@ -2814,22 +2778,6 @@ mod tests {
arrow_select::concat::concat_batches(&schema, &batches).unwrap()
}
- #[test]
- fn test_new_rejects_aggregation_with_deletion_vectors() {
- let mut config = test_write_config(MergeEngine::Aggregation);
- config.deletion_vectors_enabled = true;
-
- let err =
KeyValueFileWriter::new(FileIOBuilder::new("memory").build().unwrap(), config,
0)
- .err()
- .unwrap();
-
- assert!(matches!(
- err,
- crate::Error::Unsupported { message }
- if message.contains("deletion-vectors.enabled=true")
- ));
- }
-
#[test]
fn test_new_rejects_unsupported_aggregation_options() {
let mut config = test_write_config(MergeEngine::Aggregation);
diff --git a/crates/paimon/src/table/postpone_fixed_bucket_write.rs
b/crates/paimon/src/table/postpone_fixed_bucket_write.rs
index 6f2f59a9..44565fc6 100644
--- a/crates/paimon/src/table/postpone_fixed_bucket_write.rs
+++ b/crates/paimon/src/table/postpone_fixed_bucket_write.rs
@@ -20,7 +20,6 @@ use super::postpone_fixed_bucket_router::{
validate_postpone_fixed_bucket_table, PostponeFixedBucketRouter,
};
use crate::resource::ResourceContext;
-use crate::spec::CoreOptions;
use crate::table::{CommitMessage, PostponeBucketPlan, Table, TableCommit,
TableWrite};
use crate::Result;
use arrow_array::RecordBatch;
@@ -55,7 +54,7 @@ impl PostponeFixedBucketTableWrite {
overwrite: bool,
resources: Option<ResourceContext>,
) -> Result<Self> {
- validate_postpone_fixed_bucket_write(table)?;
+ validate_postpone_fixed_bucket_table(table)?;
let mut inner = TableWrite::new(table, commit_user)?;
if let Some(resources) = resources {
inner = inner.with_resources(resources);
@@ -231,19 +230,6 @@ impl PostponeFixedBucketTableCommit {
}
}
-fn validate_postpone_fixed_bucket_write(table: &Table) -> Result<()> {
- validate_postpone_fixed_bucket_table(table)?;
- if CoreOptions::new(table.schema().options()).deletion_vectors_enabled() {
- return Err(crate::Error::Unsupported {
- message: format!(
- "Table '{}' cannot use postpone fixed-bucket writes with
deletion-vectors.enabled=true because deletion-vector scans skip the level-0
files produced by batch writers; use the normal postpone writer or disable
deletion vectors",
- table.identifier().full_name()
- ),
- });
- }
- Ok(())
-}
-
#[cfg(test)]
mod tests {
use crate::catalog::Identifier;
diff --git a/crates/paimon/src/table/postpone_fixed_bucket_write_builder.rs
b/crates/paimon/src/table/postpone_fixed_bucket_write_builder.rs
index 6f7e3d3c..530fa496 100644
--- a/crates/paimon/src/table/postpone_fixed_bucket_write_builder.rs
+++ b/crates/paimon/src/table/postpone_fixed_bucket_write_builder.rs
@@ -160,16 +160,23 @@ mod tests {
"deletion-vectors.enabled".to_string(),
"true".to_string(),
)]));
- let error = dv_table
+ let mut write = dv_table
.new_postpone_fixed_bucket_write_builder()
.unwrap()
.with_bucket_plan(bucket_plan(&dv_table, 1))
.new_write()
- .err()
.unwrap();
- assert!(matches!(error, crate::Error::Unsupported { ref message }
- if message.contains("postpone fixed-bucket writes")
- && message.contains("deletion-vectors.enabled=true")));
+ write
+ .write_arrow_batch(&make_batch(vec![5], vec![50]))
+ .await
+ .unwrap();
+ let messages = write.prepare_commit().await.unwrap();
+ assert_eq!(messages.len(), 1);
+ assert_eq!(messages[0].bucket, 0);
+ assert_eq!(messages[0].total_buckets, Some(1));
+ assert_eq!(messages[0].new_files[0].level, 0);
+ assert_eq!(messages[0].new_files[0].row_count, 1);
+ write.close().await;
}
#[tokio::test]
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index 5c153f25..23675fe1 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -1634,8 +1634,6 @@ impl TableWrite {
value_fields: self.table.schema().fields().to_vec(),
sequence_field_indices:
self.sequence_field_indices.clone(),
merge_engine: self.merge_engine,
- deletion_vectors_enabled:
CoreOptions::new(self.table.schema().options())
- .deletion_vectors_enabled(),
file_index_options: self.file_index_options.clone(),
},
next_seq,
@@ -3629,41 +3627,6 @@ pub(in crate::table) mod tests {
.unwrap();
}
- #[tokio::test]
- async fn
test_rejects_partial_update_with_deletion_vectors_when_creating_writer() {
- let file_io = test_file_io();
- let table_path = "memory:/test_partial_update_dv_table";
- setup_dirs(&file_io, table_path).await;
-
- let table = Table::new(
- file_io,
- Identifier::new("default", "test_partial_update_dv_table"),
- table_path.to_string(),
- TableSchema::new(
- 0,
- &Schema::builder()
- .column("id", DataType::Int(IntType::new()))
- .column("value", DataType::Int(IntType::new()))
- .primary_key(["id"])
- .option("bucket", "1")
- .option("merge-engine", "partial-update")
- .option("deletion-vectors.enabled", "true")
- .build()
- .unwrap(),
- ),
- None,
- );
-
- let mut table_write = TableWrite::new(&table,
"test-user".to_string()).unwrap();
- let err = table_write
- .write_arrow_batch(&make_batch(vec![1], vec![10]))
- .await
- .unwrap_err();
- assert!(
- matches!(err, crate::Error::Unsupported { message } if
message.contains("deletion-vectors.enabled=true"))
- );
- }
-
#[tokio::test]
async fn test_write_partitioned() {
let file_io = test_file_io();
diff --git a/crates/paimon/tests/dv_merge_engine_write_test.rs
b/crates/paimon/tests/dv_merge_engine_write_test.rs
new file mode 100644
index 00000000..3af0e91c
--- /dev/null
+++ b/crates/paimon/tests/dv_merge_engine_write_test.rs
@@ -0,0 +1,405 @@
+// 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 std::sync::Arc;
+
+use arrow_array::{Array, ArrayRef, Int32Array, RecordBatch, StringArray};
+use futures::TryStreamExt;
+use paimon::spec::{DataType, Datum, IntType, PredicateBuilder, Schema,
TableSchema, VarCharType};
+use paimon::table::{CommitMessage, Table, TableCommit};
+
+use common::incremental_helpers::{memory_table, persist_table_schema,
setup_dirs};
+
+type Row = (i32, Option<i32>, Option<String>);
+
+async fn table(engine: &str, bucket: &str, input_changelog: bool) -> Table {
+ let mut builder = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .column("text", DataType::VarChar(VarCharType::string_type()))
+ .primary_key(["id"])
+ .option("bucket", bucket)
+ .option("merge-engine", engine)
+ .option("deletion-vectors.enabled", "true")
+ .option("source.split.target-size", "1b")
+ .option("source.split.open-file-cost", "1b");
+ if engine == "aggregation" {
+ builder = builder
+ .option("fields.value.aggregate-function", "sum")
+ .option("fields.text.aggregate-function", "last_non_null_value");
+ }
+ if input_changelog {
+ builder = builder.option("changelog-producer", "input");
+ }
+ let (io, table) = memory_table(
+ "memory:/dv_merge_engine",
+ TableSchema::new(0, &builder.build().unwrap()),
+ );
+ setup_dirs(&io, table.location()).await;
+ persist_table_schema(&io, table.location(), table.schema()).await;
+ table
+}
+
+fn batch(ids: Vec<i32>, values: Vec<Option<i32>>, texts: Vec<Option<&str>>) ->
RecordBatch {
+ RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(ids)) as ArrayRef),
+ ("value", Arc::new(Int32Array::from(values)) as ArrayRef),
+ ("text", Arc::new(StringArray::from(texts)) as ArrayRef),
+ ])
+ .unwrap()
+}
+
+async fn write(table: &Table, batches: Vec<RecordBatch>, identifier: i64) ->
Vec<CommitMessage> {
+ let builder = table
+ .new_write_builder()
+ .with_commit_user("dv-test")
+ .unwrap();
+ let mut writer = builder.new_write().unwrap();
+ for batch in batches {
+ writer.write_arrow_batch(&batch).await.unwrap();
+ }
+ let messages = writer.prepare_commit().await.unwrap();
+ TableCommit::new(table.clone(), "dv-test".into())
+ .commit_with_identifier(messages.clone(), identifier)
+ .await
+ .unwrap();
+ writer.close().await;
+ messages
+}
+
+async fn rows(table: &Table) -> Vec<Row> {
+ let builder = table.new_read_builder();
+ 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 rows = Vec::new();
+ for batch in batches {
+ let ids = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let values = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let texts = batch
+ .column(2)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ for row in 0..batch.num_rows() {
+ rows.push((
+ ids.value(row),
+ values.is_valid(row).then(|| values.value(row)),
+ texts.is_valid(row).then(|| texts.value(row).to_string()),
+ ));
+ }
+ }
+ rows.sort();
+ rows
+}
+
+#[tokio::test]
+async fn dv_merge_engines_fold_inputs_and_restore_across_commits() {
+ for engine in ["partial-update", "aggregation"] {
+ for bucket in ["1", "-1"] {
+ for input_changelog in [false, true] {
+ let table = table(engine, bucket, input_changelog).await;
+ let first = write(
+ &table,
+ vec![
+ batch(
+ vec![1, 2],
+ vec![Some(10), Some(20)],
+ vec![Some("first"), Some("second")],
+ ),
+ batch(vec![1], vec![None], vec![Some("patched")]),
+ ],
+ 1,
+ )
+ .await;
+ assert_eq!(
+ first
+ .iter()
+ .flat_map(|m| &m.new_files)
+ .map(|f| f.row_count)
+ .sum::<i64>(),
+ 2
+ );
+ assert!(first
+ .iter()
+ .flat_map(|m| &m.new_files)
+ .all(|f| f.level == 0));
+ assert_eq!(
+ first
+ .iter()
+ .flat_map(|m| &m.new_changelog_files)
+ .map(|f| f.row_count)
+ .sum::<i64>(),
+ if input_changelog { 3 } else { 0 }
+ );
+ // Java does not expose uncompacted level-0 files by default
in DV mode.
+ assert!(rows(&table).await.is_empty());
+ let visible =
table.copy_with_options(std::collections::HashMap::from([(
+ "deletion-vectors.merge-on-read".into(),
+ "true".into(),
+ )]));
+ assert_eq!(
+ rows(&visible).await,
+ vec![
+ (1, Some(10), Some("patched".into())),
+ (2, Some(20), Some("second".into())),
+ ]
+ );
+ write(
+ &table,
+ vec![batch(
+ vec![1, 3],
+ vec![Some(3), Some(30)],
+ vec![None, Some("third")],
+ )],
+ 2,
+ )
+ .await;
+ assert_eq!(
+ rows(&visible).await,
+ vec![
+ (
+ 1,
+ Some(if engine == "aggregation" { 13 } else { 3 }),
+ Some("patched".into())
+ ),
+ (2, Some(20), Some("second".into())),
+ (3, Some(30), Some("third".into())),
+ ]
+ );
+ assert!(rows(&table).await.is_empty());
+ assert_eq!(
+ table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap()
+ .id(),
+ 2
+ );
+ }
+ }
+ }
+}
+
+#[tokio::test]
+async fn dv_merge_engine_filters_are_applied_after_merging() {
+ for engine in ["partial-update", "aggregation"] {
+ let table = table(engine, "1", false).await;
+ write(
+ &table,
+ vec![batch(vec![1], vec![Some(5)], vec![Some("old")])],
+ 1,
+ )
+ .await;
+ write(&table, vec![batch(vec![1], vec![Some(6)], vec![None])],
2).await;
+ let visible =
table.copy_with_options(std::collections::HashMap::from([(
+ "deletion-vectors.merge-on-read".into(),
+ "true".into(),
+ )]));
+ let predicates = PredicateBuilder::new(visible.schema().fields());
+ for (predicate, expected) in [
+ (
+ predicates.greater_than("value", Datum::Int(10)).unwrap(),
+ if engine == "aggregation" { 1 } else { 0 },
+ ),
+ (
+ predicates
+ .equal("text", Datum::String("old".into()))
+ .unwrap(),
+ 1,
+ ),
+ ] {
+ let mut builder = visible.new_read_builder();
+ builder
+ .with_projection(&["id"])
+ .unwrap()
+ .with_filter(predicate)
+ .with_limit(1);
+ let plan = builder.new_scan().plan().await.unwrap();
+ let batches = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(
+ batches.iter().map(RecordBatch::num_rows).sum::<usize>(),
+ expected
+ );
+ for batch in batches {
+ assert_eq!(batch.schema().fields().len(), 1);
+ let ids = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ assert!(ids.values().iter().all(|id| *id == 1));
+ }
+ }
+ }
+}
+
+#[tokio::test]
+async fn dv_merge_engines_keep_inputs_across_buffer_flushes() {
+ for engine in ["partial-update", "aggregation"] {
+ let table =
+ table(engine, "1", true)
+ .await
+ .copy_with_options(std::collections::HashMap::from([
+ ("write.parquet-buffer-size".into(), "1b".into()),
+ ("deletion-vectors.merge-on-read".into(), "true".into()),
+ ]));
+ let messages = write(
+ &table,
+ vec![
+ batch(vec![1], vec![Some(1)], vec![Some("first")]),
+ batch(vec![1], vec![Some(2)], vec![None]),
+ batch(vec![1], vec![Some(3)], vec![Some("last")]),
+ ],
+ 1,
+ )
+ .await;
+ assert_eq!(messages.iter().flat_map(|m| &m.new_files).count(), 3);
+ assert_eq!(
+ messages
+ .iter()
+ .flat_map(|m| &m.new_changelog_files)
+ .map(|f| f.row_count)
+ .sum::<i64>(),
+ 3
+ );
+ assert_eq!(
+ rows(&table).await,
+ vec![(
+ 1,
+ Some(if engine == "aggregation" { 6 } else { 3 }),
+ Some("last".into())
+ )]
+ );
+ }
+}
+
+#[tokio::test]
+async fn dv_postpone_fixed_bucket_writers_merge_and_overwrite() {
+ for engine in ["deduplicate", "partial-update", "aggregation"] {
+ let table =
+ table(engine, "-2", true)
+ .await
+ .copy_with_options(std::collections::HashMap::from([(
+ "postpone.default-bucket-num".into(),
+ "3".into(),
+ )]));
+ for (identifier, inputs, expected) in [
+ (
+ 1,
+ vec![
+ batch(vec![1], vec![Some(10)], vec![Some("first")]),
+ batch(vec![1], vec![Some(20)], vec![None]),
+ ],
+ if engine == "aggregation" { 30 } else { 20 },
+ ),
+ (
+ 2,
+ vec![batch(vec![1], vec![Some(30)], vec![Some("after")])],
+ if engine == "aggregation" { 60 } else { 30 },
+ ),
+ ] {
+ let builder = table
+ .new_postpone_fixed_bucket_write_builder()
+ .unwrap()
+ .with_commit_user("fixed-dv")
+ .unwrap();
+ let mut writer = builder.new_write().unwrap();
+ for data in inputs {
+ writer.write_arrow_batch(&data).await.unwrap();
+ }
+ let messages = writer.prepare_commit().await.unwrap();
+ assert!(messages
+ .iter()
+ .all(|m| m.bucket >= 0 && m.bucket < 3 && m.total_buckets ==
Some(3)));
+ assert!(messages
+ .iter()
+ .flat_map(|m| &m.new_files)
+ .all(|f| f.level == 0 && f.row_count == 1));
+ builder
+ .new_commit()
+ .commit_with_identifier(messages, identifier)
+ .await
+ .unwrap();
+ writer.close().await;
+ assert!(rows(&table).await.is_empty());
+ let visible =
table.copy_with_options(std::collections::HashMap::from([(
+ "deletion-vectors.merge-on-read".into(),
+ "true".into(),
+ )]));
+ let text = if identifier == 2 {
+ Some("after".into())
+ } else if engine == "deduplicate" {
+ None
+ } else {
+ Some("first".into())
+ };
+ assert_eq!(rows(&visible).await, vec![(1, Some(expected), text)]);
+ }
+ let replacement =
table.copy_with_options(std::collections::HashMap::from([(
+ "postpone.default-bucket-num".into(),
+ "5".into(),
+ )]));
+ let builder = replacement
+ .new_postpone_fixed_bucket_write_builder()
+ .unwrap()
+ .with_commit_user("fixed-dv-overwrite")
+ .unwrap()
+ .with_overwrite();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(&batch(vec![1], vec![Some(100)], vec![None]))
+ .await
+ .unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ assert!(messages
+ .iter()
+ .all(|m| m.total_buckets == Some(5) &&
m.new_changelog_files.is_empty()));
+ builder.new_commit().commit(messages).await.unwrap();
+ writer.close().await;
+ let visible =
table.copy_with_options(std::collections::HashMap::from([(
+ "deletion-vectors.merge-on-read".into(),
+ "true".into(),
+ )]));
+ assert_eq!(rows(&visible).await, vec![(1, Some(100), None)]);
+ }
+}
