flyrain commented on code in PR #11045:
URL: https://github.com/apache/iceberg/pull/11045#discussion_r1739888471
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java:
##########
@@ -183,21 +185,26 @@ private boolean shouldComputeUpdateImages(ProcedureInput
input) {
}
private Dataset<Row> removeCarryoverRows(Dataset<Row> df, boolean
netChanges) {
- Predicate<String> columnsToKeep;
- if (netChanges) {
- Set<String> metadataColumn =
- Sets.newHashSet(
- MetadataColumns.CHANGE_TYPE.name(),
- MetadataColumns.CHANGE_ORDINAL.name(),
- MetadataColumns.COMMIT_SNAPSHOT_ID.name());
-
- columnsToKeep = column -> !metadataColumn.contains(column);
- } else {
- columnsToKeep = column ->
!column.equals(MetadataColumns.CHANGE_TYPE.name());
- }
+ Set<String> metadataColumn =
+ netChanges
+ ? Sets.newHashSet(
+ MetadataColumns.CHANGE_TYPE.name(),
+ MetadataColumns.CHANGE_ORDINAL.name(),
+ MetadataColumns.COMMIT_SNAPSHOT_ID.name())
+ : Sets.newHashSet(MetadataColumns.CHANGE_TYPE.name());
+
+ Predicate<StructField> columnsToDiscard =
+ field ->
+ metadataColumn.contains(field.name())
+ // avoid sort on incomparable columns
+ || field.dataType() instanceof MapType
+ || field.dataType() instanceof BinaryType;
Column[] repartitionSpec =
-
Arrays.stream(df.columns()).filter(columnsToKeep).map(df::col).toArray(Column[]::new);
+ Arrays.stream(df.schema().fields())
+ .filter(Predicate.not(columnsToDiscard))
Review Comment:
This breaks the
[assumption](https://github.com/apache/iceberg/blob/79620e198009fa243c278c66fd442d107b46206a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/RemoveCarryoverIterator.java#L28-L28)
of `RemoveCarryoverIterator`. For example, this is an order can happen now,
but it will break `RemoveCarryoverIterator`, as we didn't remove the carryover
rows, row1 and row3.
```
row1: (id=1, map=('a', 'a'), op='DELETE')
row2: (id=1, map=('a', 'b')', op='INSERT')
row3: (id=1, map=('a', 'a'), op='INSERT')
```
--
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]
