mfellner opened a new issue, #3401:
URL: https://github.com/apache/iceberg-rust/issues/3401
Related to #2411 (v3 tracking).
Related to #2752 (COW rewrite primitive), #2185 and #3300 (overwrite commit
side), #3046 (RewriteFilesAction), #1607 (RewriteFiles). The #3300 review
thread covers row lineage of surviving manifest entries on overwrite, which is
a different case from rows rewritten into replacement files.
### Apache Iceberg Rust version
main at a76678363
### Describe the bug
Thanks @u70b3 for the COW rewrite primitive in #2752. On a v3 table,
`CowRewriteBuilder::rewrite()` does not carry row lineage into the replacement
files.
- The planner scans with `select_all()`
(`crates/iceberg/src/cow_rewrite/plan.rs:61`). The rewriter's input batches
contain only the table's data columns (`["id"]` in the reproduction below), so
`_row_id` and `_last_updated_sequence_number` are not written to the
replacement file.
- The replacement `DataFile` has `first_row_id() == None`.
The v3 spec's "Row lineage assignment" section says that when an existing
row is moved to a different data file, its non-null `_row_id` must be copied
into the new file. For a row that was not modified, its non-null
`_last_updated_sequence_number` must be copied too.
On read, the replacement rows get new lineage values. The reproduction below
shows this with the source file still live, because `rewrite()` does not commit
and no overwrite action exists on main yet. Code reading, not run: a commit
that removes the source file would assign the replacement a `first_row_id` from
the snapshot's `next-row-id` (`crates/iceberg/src/transaction/snapshot.rs:440`
and `:471`), and readers assign data-file `first_row_id` by manifest
inheritance (`crates/iceberg/src/spec/manifest/reader.rs:75`). The same new
values would be expected.
Neither the planner nor the writer checks the table format version (code
reading: there is no format-version check in `crates/iceberg/src/cow_rewrite/`
outside tests).
### To Reproduce
Run on a snapshot of main at a76678363 with the production code unchanged.
Append the test module below to `crates/iceberg/src/cow_rewrite/mod.rs`, then
run:
```
cargo test -p iceberg --lib repro_cow_rewrite_v3_row_lineage -- --nocapture
```
The test:
1. Creates a v3 table with one `id: int` column.
2. Writes ids 1, 2, 3 and fast-appends the file. Scanning `id, _row_id,
_last_updated_sequence_number` returns `(1, 0, 1), (2, 1, 1), (3, 2, 1)`.
3. Runs `CowRewriteBuilder` with a rewriter that removes id 2 and reports
the batch as changed.
4. Fast-appends `added_data_files` as a stand-in for a commit. The source
file stays live, so rows appear twice.
Output:
```
BEFORE (id, _row_id, _last_updated_sequence_number): [(1, 0, 1), (2, 1, 1),
(3, 2, 1)]
rewriter input columns: ["id"]
removed=1 added=1 added first_row_id=[None]
AFTER simulated commit: [(1, 0, 1), (1, 3, 2), (2, 1, 1), (3, 2, 1), (3, 4,
2)]
```
In the AFTER line, `(1, 3, 2)` and `(3, 4, 2)` are the replacement file's
rows for ids 1 and 3, which were not modified. Their `_row_id` values are new
(3 and 4, not 0 and 2), and their `_last_updated_sequence_number` is 2, not 1.
The original rows still hold the source values.
<details>
<summary>Test module (append to
crates/iceberg/src/cow_rewrite/mod.rs)</summary>
```rust
#[cfg(test)]
mod repro_lineage {
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use arrow_array::{ArrayRef, Int32Array, Int64Array, RecordBatch};
use arrow_schema::{DataType, Field, Schema as ArrowSchema};
use futures::TryStreamExt;
use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
use tempfile::TempDir;
use crate::cow_rewrite::{CowBatchRewrite, CowBatchRewriter,
CowRewriteBuilder};
use crate::io::LocalFsStorageFactory;
use crate::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder};
use crate::spec::{FormatVersion, NestedField, PrimitiveType, Schema,
Type};
use crate::table::Table;
use crate::transaction::{ApplyTransactionAction, Transaction};
use crate::{Catalog, CatalogBuilder, NamespaceIdent, Result,
TableCreation};
struct DropId2 {
seen_columns: Mutex<Vec<String>>,
}
impl CowBatchRewriter for DropId2 {
fn rewrite_batch(&self, batch: RecordBatch) ->
Result<CowBatchRewrite> {
*self.seen_columns.lock().unwrap() = batch
.schema()
.fields()
.iter()
.map(|f| f.name().clone())
.collect();
let ids = batch
.column(0)
.as_any()
.downcast_ref::<Int32Array>()
.unwrap();
let keep: Vec<i32> = ids.values().iter().copied().filter(|v| *v
!= 2).collect();
let changed = keep.len() != ids.len();
let output = RecordBatch::try_new(batch.schema(), vec![
Arc::new(Int32Array::from(keep)) as ArrayRef,
])
.unwrap();
Ok(CowBatchRewrite {
output: Some(output),
changed,
})
}
}
async fn read_lineage(table: &Table) -> Result<Vec<(i32, i64, i64)>> {
let stream = table
.scan()
.select(["id", "_row_id", "_last_updated_sequence_number"])
.build()?
.to_arrow()
.await?;
let batches: Vec<RecordBatch> = stream.try_collect().await?;
let mut rows = vec![];
for b in batches {
let id =
b.column_by_name("id").unwrap().as_any().downcast_ref::<Int32Array>().unwrap();
let rid =
b.column_by_name("_row_id").unwrap().as_any().downcast_ref::<Int64Array>().unwrap();
let lu = arrow_cast::cast(
b.column_by_name("_last_updated_sequence_number").unwrap(),
&DataType::Int64,
)?;
let lu = lu.as_any().downcast_ref::<Int64Array>().unwrap();
for i in 0..b.num_rows() {
rows.push((id.value(i), rid.value(i), lu.value(i)));
}
}
rows.sort();
Ok(rows)
}
#[tokio::test]
async fn repro_cow_rewrite_v3_row_lineage() -> Result<()> {
let temp_dir = TempDir::new().unwrap();
let warehouse = format!("file://{}",
temp_dir.path().join("warehouse").display());
let catalog = MemoryCatalogBuilder::default()
.with_storage_factory(Arc::new(LocalFsStorageFactory))
.load("memory",
HashMap::from([(MEMORY_CATALOG_WAREHOUSE.to_string(), warehouse)]))
.await?;
let namespace = NamespaceIdent::new("ns".to_string());
catalog.create_namespace(&namespace, HashMap::new()).await?;
let schema = Schema::builder()
.with_fields(vec![
NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
])
.build()?;
let table = catalog
.create_table(&namespace, TableCreation::builder()
.name("v3_cow".to_string())
.schema(schema)
.format_version(FormatVersion::V3)
.build())
.await?;
let arrow_schema = Arc::new(ArrowSchema::new(vec![
Field::new("id", DataType::Int32,
false).with_metadata(HashMap::from([(
PARQUET_FIELD_ID_META_KEY.to_string(),
"1".to_string(),
)])),
]));
let batch = RecordBatch::try_new(arrow_schema, vec![
Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef,
])?;
let files = super::writer::write_replacement_batches(
&table,
table.metadata().current_schema().clone(),
None,
futures::stream::iter(vec![Ok(batch)]),
)
.await?;
let tx = Transaction::new(&table);
let tx = tx.fast_append().add_data_files(files).apply(tx)?;
let table = tx.commit(&catalog).await?;
let before = read_lineage(&table).await?;
println!("BEFORE (id, _row_id, _last_updated_sequence_number):
{before:?}");
assert_eq!(before, vec![(1, 0, 1), (2, 1, 1), (3, 2, 1)]);
let rewriter = Arc::new(DropId2 {
seen_columns: Mutex::new(vec![]),
});
let result = CowRewriteBuilder::new(&table)
.with_rewriter(rewriter.clone())
.rewrite()
.await?;
println!("rewriter input columns: {:?}",
rewriter.seen_columns.lock().unwrap());
println!(
"removed={} added={} added first_row_id={:?}",
result.removed_data_files.len(),
result.added_data_files.len(),
result.added_data_files.iter().map(|f|
f.first_row_id()).collect::<Vec<_>>()
);
assert_eq!(result.added_data_files.len(), 1);
assert_eq!(result.added_data_files[0].first_row_id(), None);
// Simulated commit: the replacement is appended while the source
file is still live,
// because no overwrite commit action exists yet.
let tx = Transaction::new(&table);
let tx = tx
.fast_append()
.add_data_files(result.added_data_files.clone())
.apply(tx)?;
let table = tx.commit(&catalog).await?;
let after = read_lineage(&table).await?;
println!("AFTER simulated commit: {after:?}");
Ok(())
}
}
```
</details>
### Expected behavior
Rows that a rewrite moves keep their `_row_id`. Rows that the rewrite does
not modify keep their `_last_updated_sequence_number`. For v3 tables, the
primitive should carry both values into the replacement files, or give the
commit step what it needs to write them, so that row lineage after commit
matches the spec.
### Willingness to contribute
I would be willing to contribute a fix with guidance from the Iceberg
community (any contribution would be agent-assisted and I would disclose that).
Found during an agent-assisted audit of v3 support (see #2411). The
reproduction above was run against main at a76678363.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
