zhengxiang378928908-code commented on code in PR #18602:
URL: https://github.com/apache/iceberg/pull/18602#discussion_r4348393893
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicRecordProcessor.java:
##########
@@ -160,14 +160,19 @@ public void collect(DynamicRecord inputData) {
isForward);
} else {
// Shuffled records with immediateUpdate=false go to the update side
output
+ // The schema, the spec and the row used for the key must describe the
same layout: the
+ // spec resolves partition source fields by ID in the schema, and the
schema reads them by
+ // position from the row. Use the table's layout only when both the
table schema and spec
+ // are known, otherwise stay with the record's own schema, spec and
row.
+ boolean keyByTable = foundSchema.resolvedTableSchema() != null &&
foundSpec != null;
int writerKey =
- hashKeyGenerator.generateKey(
- data,
- foundSchema.resolvedTableSchema() != null
- ? foundSchema.resolvedTableSchema()
- : data.schema(),
- foundSpec != null ? foundSpec : data.spec(),
- data.rowData());
+ keyByTable
Review Comment:
Good point, done in 26500608b. The table schema and spec are only both known
here when just the branch is missing, so the extra case wasn't worth a second
conversion of the row. The 322 tests in `org.apache.iceberg.flink.sink.dynamic`
still run with no failures (2 skipped).
One small difference, for the record: when a record goes to a new branch and
needs data conversion, the update path now keys it with the record's types,
while `emit()` keys later records with the table's types. With equality-field
distribution, an `int` key column written to a `long` table column hashes
differently for negative values (`Integer.hashCode` vs `Long.hashCode`), so in
upsert mode rows with the same key values may go to different writers,
depending on the write parallelism. The same can already happen when a schema
or spec update is needed. I haven't reproduced this end to end and didn't keep
a special case for it.
--
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]
