gaborkaszab commented on code in PR #18486:
URL: https://github.com/apache/iceberg/pull/18486#discussion_r4466467444
##########
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:
I was hesitant to do this, but the code seems somewhat simpler this way.
Thanks for the suggestion!
##########
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:
You're right. I added test coverage (the existing tests hard-coded what
bulkDecrypt() returns, had to make it more generic)
##########
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:
Indeed, I checked and nothing uses `DeletedRowsScanTask` as of now. Instead
of adding coverage here now, I'd prefer to simply drop this branch and let the
default branch throw. I get the concern that once Sparks adds support for that
type the decryptor will fail, but that is not something happening silently and
makes the implementation of DeletedRowsScanTask to also add coverage here.
--
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]
