vbhanuchander-lang commented on issue #17455:
URL: https://github.com/apache/iceberg/issues/17455#issuecomment-5247428352
I went looking for the equality-delete sequencing described here and could
not find it, because I don't think this sink writes equality deletes at all. If
that's right, the duplicates are real but the mechanism is simpler — and worse
— than same-snapshot sequence numbers.
**The write path is append-only.**
`RecordUtils.createTableWriter` builds a writer factory *with*
equality-delete capability:
```java
// RecordUtils.java:141
writerFactory =
new GenericFileWriterFactory.Builder(table)
.dataSchema(table.schema())
.equalityFieldIds(Ints.toArray(identifierFieldIds))
.equalityDeleteRowSchema(TypeUtil.select(table.schema(), ...))
...
```
but then hands it to a `TaskWriter` that can only append:
```java
// RecordUtils.java:161
writer = table.spec().isUnpartitioned()
? new UnpartitionedWriter<>(...) // single RollingFileWriter,
data only
: new PartitionedAppendWriter(...); // extends
PartitionedFanoutWriter, data only
```
Neither writes deletes, so `equalityFieldIds` and `equalityDeleteRowSchema`
are inert here. And `IcebergWriter.write` has no operation handling at all —
every non-tombstone record is appended:
```java
// IcebergWriter.java:60
if (record.value() != null) {
Record row = convertToRow(record);
writer.write(row);
}
```
A grep across `kafka-connect/` for `DeltaWriter`, `EqualityDeleteWriter`,
`PositionDelete` and `deleteWriter` returns nothing outside tests.
**The commit side is ready for deletes, which is what makes this look like
upsert support.** `Coordinator` branches on whether any delete files arrived:
```java
// Coordinator.java:307
if (deleteFiles.isEmpty()) {
... table.newAppend() ...
} else {
RowDelta deltaOp = table.newRowDelta()...;
dataFiles.forEach(deltaOp::addRows);
deleteFiles.forEach(deltaOp::addDeletes);
```
The `RowDelta` branch exists but is unreachable from this sink, because the
writers never produce delete files.
**What that means for this report**
If the above is right, then a crash is not required to see duplicates, and
the fix is not about snapshot separation:
- Two updates to the same primary key produce two data rows, in one snapshot
or across two. Nothing ever removes the earlier one.
- `iceberg.tables.default-id-columns` / `iceberg.table.<name>.id-columns`
are documented as "columns that identify a row in the table (primary key)",
which reads as identity semantics. They select `identifierFieldIds` for the
(inert) delete schema and otherwise do not deduplicate anything.
So the user-visible defect is arguably broader than the title: configuring
`id-columns` on this sink does not give you upserts, and nothing says so.
@LiorV10 — does that match what you saw? Specifically, do you get duplicate
PK rows *without* a crash, just from two updates to the same key? That would
confirm it, and it's much easier to reproduce than the crash path.
For maintainers: I'm happy to work on this, but the shape depends on intent,
and I don't want to guess at a design decision:
1. If the sink is **meant** to be append-only, the honest fix is
documentation plus removing the inert
`equalityFieldIds`/`equalityDeleteRowSchema` configuration so the next reader
isn't misled, and being explicit that `id-columns` is not an upsert key.
2. If upserts are **intended**, this needs a delta writer on the write path
(the `RowDelta` branch in `Coordinator` is already waiting for it), which is a
feature rather than a bug fix and should probably be discussed before anyone
writes it.
Is one of those the direction you'd want? I have the repo built and can take
either.
--
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]
