anuragmantri commented on code in PR #18486:
URL: https://github.com/apache/iceberg/pull/18486#discussion_r4416412320
##########
core/src/main/java/org/apache/iceberg/encryption/InputFilesDecryptor.java:
##########
@@ -83,6 +100,39 @@ private static Collection<ContentFile<?>>
referencedFiles(Iterable<FileScanTask>
return files.values();
}
+ private static Collection<ContentFile<?>> changelogReferencedFiles(
+ Iterable<ChangelogScanTask> tasks) {
+ Map<String, ContentFile<?>> files = Maps.newHashMap();
+ for (ChangelogScanTask task : tasks) {
+ if (task instanceof AddedRowsScanTask addedRowsTask) {
+ Stream.concat(Stream.of(addedRowsTask.file()),
addedRowsTask.deletes().stream())
+ .forEach(file -> files.put(file.location(), file));
+ } else if (task instanceof DeletedDataFileScanTask deletedDataFileTask) {
+ Stream.concat(
+ Stream.of(deletedDataFileTask.file()),
+ deletedDataFileTask.existingDeletes().stream())
+ .forEach(file -> files.put(file.location(), file));
+ } else if (task instanceof DeletedRowsScanTask) {
+ throw new UnsupportedOperationException("Deleted rows scan task is not
supported yet");
Review Comment:
Nothing creates a `DeletedRowsScanTask` yet, so this can't fire today. It
does copy a Spark reader gap into core, though, and once changelog scans
support deletes every reader would fail here. Could we return `file()`,
`addedDeletes()` and `existingDeletes()` instead? Spark still throws from
`openChangelogScanTask`.
##########
core/src/main/java/org/apache/iceberg/encryption/InputFilesDecryptor.java:
##########
@@ -83,6 +100,39 @@ private static Collection<ContentFile<?>>
referencedFiles(Iterable<FileScanTask>
return files.values();
}
+ private static Collection<ContentFile<?>> changelogReferencedFiles(
Review Comment:
Now that this lives in core, could `TestInputFilesDecryptor` cover the
changelog and position delete tasks and the unsupported-type error? It only
covers `FileScanTask` today.
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/BaseReader.java:
##########
@@ -148,12 +148,8 @@ public boolean next() throws IOException {
}
}
} catch (IOException | RuntimeException e) {
- if (currentTask != null && !currentTask.isDataTask()) {
- String filePaths =
- referencedFiles(currentTask)
- .map(ContentFile::location)
- .collect(Collectors.joining(", "));
- LOG.error("Error reading file(s): {}", filePaths, e);
+ if (currentTask instanceof ContentScanTask<?> contentTask &&
!currentTask.isDataTask()) {
+ LOG.error("Error reading file: {}", contentTask.file().location(), e);
Review Comment:
Fair enough on the logging, let's drop it.
This is more of a question:
I'd rather not add two public factories that only Spark calls, though. Could
we widen `fromTasks``it to `Iterable<? extends ScanTask>` and branch on the
task type inside?
The Flink callers still compile. `BaseReader` can then call
`InputFilesDecryptor.fromTasks(taskGroup.tasks(), this.fileIO)`, and the
abstract method, the two accessors and the overrides go away. I tried this on
your branch. Core and Spark 4.2 compile, and `TestInputFilesDecryptor` and
`TestBaseReader` pass. A short Javadoc on `fromTasks` would be nice too.
--
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]
