yhyyz opened a new issue, #18609:
URL: https://github.com/apache/iceberg/issues/18609
### Apache Iceberg version
1.12.0 (latest release)
### Query engine
Flink
### Please describe the bug 🐞
### Summary
The Flink `ConvertEqualityDeletes` maintenance task (added in 1.12.0) fails
on every conversion cycle when the equality field columns include a `DATE`
column. No cycle ever commits. Because the converter requires the partition
source columns to be part of the equality fields, this affects any table
partitioned by a date column, which is a very common layout for CDC and log
tables.
### Environment
- `iceberg-flink-runtime-1.20-1.12.0.jar`
- Flink 1.20.0
- Format version 3 table, Parquet data files
- Writer: Flink upsert sink writing to a `staging` branch
- Converter: a separate Flink job running `TableMaintenance` +
`ConvertEqualityDeletes` with `stagingBranch("staging")` and
`targetBranch("main")`
### Steps to reproduce
1. Create a v3 table:
```sql
CREATE TABLE t (
id BIGINT,
d DATE,
v STRING,
PRIMARY KEY (id, d) NOT ENFORCED
) PARTITIONED BY (d)
WITH ('format-version' = '3', 'write.upsert.enabled' = 'true');
```
2. With Flink, write to the `staging` branch: insert a row, then update the
same key, so that equality delete files are produced. The same failure also
happens with `write.upsert.enabled=false` (changelog input with `-U`/`+U`).
3. Start the converter:
```java
TableMaintenance.forTable(env, tableLoader)
.add(
ConvertEqualityDeletes.builder()
.stagingBranch("staging")
.targetBranch("main")
.equalityFieldColumns(List.of("id", "d"))
.scheduleOnInterval(Duration.ofSeconds(60)))
.append();
```
### Expected behavior
Equality deletes on `staging` are resolved to deletion vectors and committed
to `main`.
### Actual behavior
Every cycle fails in `EqualityConvertReader`. The committer logs `Skipping
commit ... a DV writer reported an error` and the `TaskResult` has
`success=false`:
```
java.lang.ClassCastException: class java.time.LocalDate cannot be cast to
class java.lang.Integer (java.time.LocalDate and java.lang.Integer are in
module java.base of loader 'bootstrap')
at
org.apache.iceberg.types.Conversions.toByteBuffer(Conversions.java:100)
at
org.apache.iceberg.types.Conversions.toByteBuffer(Conversions.java:87)
at
org.apache.iceberg.flink.maintenance.operator.StructLikeSerializer.writeField(StructLikeSerializer.java:135)
at
org.apache.iceberg.flink.maintenance.operator.StructLikeSerializer.serializeKey(StructLikeSerializer.java:63)
at
org.apache.iceberg.flink.maintenance.operator.EqualityConvertReader.processDataFile(EqualityConvertReader.java:168)
at
org.apache.iceberg.flink.maintenance.operator.EqualityConvertReader.processElement(EqualityConvertReader.java:115)
```
As a control, the same pipeline works on an unpartitioned table whose only
equality field is a `STRING` column. The converted `main` branch matched the
source data exactly (row count and checksums).
### Analysis
All references are to the `apache-iceberg-1.12.0` tag, `flink/v1.20`:
- `EqualityConvertReader` reads data files
([L157](https://github.com/apache/iceberg/blob/apache-iceberg-1.12.0/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java#L157))
and equality delete files
([L201](https://github.com/apache/iceberg/blob/apache-iceberg-1.12.0/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java#L201))
with `FormatModelRegistry.readBuilder(file.format(), Record.class, input)`.
This yields generic `Record`s, where a `DATE` value is a `java.time.LocalDate`.
- The record is then passed directly to
`StructLikeSerializer.serializeKey(record, keySchema.asStruct())`
([L168](https://github.com/apache/iceberg/blob/apache-iceberg-1.12.0/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java#L168),
[L204](https://github.com/apache/iceberg/blob/apache-iceberg-1.12.0/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java#L204)).
- `StructLikeSerializer.writeField` calls `struct.get(pos, Object.class)`
and hands the value to `Conversions.toByteBuffer(fieldType, value)`
([L128-L135](https://github.com/apache/iceberg/blob/apache-iceberg-1.12.0/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/StructLikeSerializer.java#L128-L135)).
- `Conversions.toByteBuffer` expects the internal representation: for `DATE`
it casts `(int) value` (days since epoch), so a `LocalDate` fails.
The code on `main` is the same, in all of `flink/v1.20`, `flink/v2.2` and
`flink/v2.3`.
The same mismatch between generic and internal representations exists for
`TIME`, `TIMESTAMP` and `TIMESTAMPTZ` (`LocalTime` / `LocalDateTime` /
`OffsetDateTime` versus `Long`). I only reproduced the `DATE` case, but tables
partitioned by a transform on a timestamp column (for example `days(ts)`) are
likely affected as well.
### Test coverage
The existing reader and end-to-end tests use `SimpleDataUtil.SCHEMA` (`id
INT`, `data STRING`). `TestStructLikeSerializer` covers `DATE` only for
partition encoding, and it passes the internal value (an `int`). Key
serialization is tested only for INT, STRING and DECIMAL, so this path was not
exercised.
### Possible fix
Before calling `serializeKey`, wrap the generic record with
`org.apache.iceberg.data.InternalRecordWrapper`, built from the projected read
schema. Do this in both the data file path (`readSchema`, which is `keySchema`
plus `_pos`) and the equality delete file path (`keySchema`):
```java
// processDataFile
InternalRecordWrapper wrapper = new
InternalRecordWrapper(readSchema.asStruct());
...
SerializedEqualityValues key =
fieldSerializer.serializeKey(wrapper.wrap(record), keySchema.asStruct());
// processDeleteFile
InternalRecordWrapper wrapper = new
InternalRecordWrapper(keySchema.asStruct());
...
SerializedEqualityValues key =
fieldSerializer.serializeKey(wrapper.wrap(record), keySchema.asStruct());
```
I have a fix with a regression test (a v3 table partitioned by a `DATE`
column, with `DATE`, `TIME`, `TIMESTAMP` and `TIMESTAMPTZ` equality fields).
The test fails with this `ClassCastException` on current `main` and passes with
the fix. I will open a PR.
### Willingness to contribute
- [x] I can contribute a fix for this bug independently
- [ ] I would be willing to contribute a fix for this bug with guidance from
the Iceberg community
- [ ] I cannot contribute a fix for this bug at this time
--
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]
