dramaticlly commented on code in PR #10935:
URL: https://github.com/apache/iceberg/pull/10935#discussion_r1733658370
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/ChangelogRowReader.java:
##########
@@ -112,13 +149,62 @@ private CloseableIterable<InternalRow>
openChangelogScanTask(ChangelogScanTask t
CloseableIterable<InternalRow> openAddedRowsScanTask(AddedRowsScanTask task)
{
String filePath = task.file().path().toString();
SparkDeleteFilter deletes = new SparkDeleteFilter(filePath,
task.deletes(), counter());
- return deletes.filter(rows(task, deletes.requiredSchema()));
+ int[] indexes = indexesInRow(deletes.requiredSchema());
+
+ return CloseableIterable.transform(
+ deletes.filter(rows(task, deletes.requiredSchema())),
+ row -> {
+ InternalRow expectedRow = new GenericInternalRow(columns.length);
+
+ for (int i = 0; i < columns.length; i++) {
+ expectedRow.update(i, row.get(indexes[i], sparkColumnTypes[i]));
+ }
+
+ return expectedRow;
+ });
Review Comment:
Speak of which, i saw this update to `InternalRow` has been repeated here
and below for deletedDataFileScanTask, do you know if we can put it into a
method instead?
--
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]
