Core, Flink: Refactor InputFilesDecryptor and use EncryptingFileIO.bulkDecrypt() - #18198
Open
gaborkaszab wants to merge 1 commit into
Open
gaborkaszab wants to merge 1 commit into
gaborkaszab wants to merge 1 commit into
Conversation
pvary
reviewed
Sep 21, 2026
|
|
||
| private Map<String, InputFile> inputFiles() { | ||
| if (lazyInputFiles == null) { | ||
| this.lazyInputFiles = encryptingIO.bulkDecrypt(referencedFiles); |
Contributor
There was a problem hiding this comment.
Do we need to synchronize this?
pvary
reviewed
Sep 21, 2026
| public InputFile getInputFile(String location) { | ||
| return decryptedInputFiles.get(location); | ||
| InputFile inputFile = inputFiles().get(location); | ||
| Preconditions.checkArgument( |
Contributor
There was a problem hiding this comment.
It is a behavioural change. Seems acceptable, but I would like to make sure that we are ok with this.
pvary
changed the title
pvary
reviewed
Sep 21, 2026
| void before() { | ||
| this.dataInputFile = Mockito.mock(InputFile.class); | ||
| this.deleteInputFile = Mockito.mock(InputFile.class); | ||
| this.encryptingIO = Mockito.mock(EncryptingFileIO.class); |
Contributor
There was a problem hiding this comment.
Maybe use InMemoryFileIO?
this.encryptingIO = Mockito.spy(EncryptingFileIO.combine(new InMemoryFileIO(), PlaintextEncryptionManager.instance()));
pvary
reviewed
Sep 21, 2026
| EncryptedFiles.encryptedInput( | ||
| io.newInputFile(entry.getKey()), entry.getValue())); | ||
| this( | ||
| () -> referencedFiles(combinedTask.files()).iterator(), |
Contributor
There was a problem hiding this comment.
do we need this lambda? Shall we just call referencedFiles when lazyInputFiles is initialized?
…kDecrypt()
Changes this contains:
- More generic interface with parameters:
- Iterable<FileScanTask>
- EncryptingFileIO
- Contructs InputFiles lazily when first requested
- EncryptionManager is wrapped into EncryptingFileIO input parameter
- Use EncryptingFileIO.bulkDecrypt() instead of EncryptionManager.decrypt()
- Throw if file location is not found
- Removed usage of old contructor and getInputFile(FileScanTask)
With this design, this class is generic enough to be wired into other
readers as a common implementation for creating InputFiles with bulk
decryption taken into account. Later, GenericReader and Spark's readers can
swap to use this.
One hidden improvement: Previously this class created InputFiles using
location + key metadata without the size. While this works, when it comes
to reading, it requires an extra RPC to get the size of the file.
Now, with bulkDecrypt(), the location + size + key metadata are used
together saving an extra RPC per file.
gaborkaszab
force-pushed
the
main_InputFileDecryptor_refactor
branch
from
September 21, 2026 18:49
55ba635 to
4e54fc1
Compare
gaborkaszab
commented
Sep 21, 2026
| .getValue() | ||
| .getMessage()) | ||
| .contains("File does not exist: "); | ||
| .contains("Failed to open input stream for file: "); |
Contributor
Author
There was a problem hiding this comment.
This is an error scenario test that verifies that reading from missing files is handled gracefully. The error text changes because we create InputFile differently: via the newInputFile variant that accepts a length.
Originally the test failed when doing an extra RPC to get the length before reading the file:
NotFoundException: File does not exist: /tmp/.../generic-appender-partition-test-...
at HadoopInputFile.lazyStat(HadoopInputFile.java:166)
at HadoopInputFile.getLength(HadoopInputFile.java:177)
at Parquet$ReadBuilder.<init>(Parquet.java:1334) <-- reads length up front
at Parquet.read(Parquet.java:1254)
at ParquetFormatModel.readBuilder(ParquetFormatModel.java:140)
at RowDataFileScanTaskReader.newIterable(RowDataFileScanTaskReader.java:108)
While now the extra RPC is gone, this fails when trying to actually open the file:
NotFoundException: Failed to open input stream for file: /tmp/.../generic-appender-partition-test-...
at HadoopInputFile.newStream(HadoopInputFile.java:187)
at EagerInputFile.newStream(EagerInputFile.java:76)
at ParquetIO$ParquetInputFile.newStream(ParquetIO.java:254)
at ParquetFileReader.<init>(ParquetFileReader.java:961)
at ReadConf.newReader(ReadConf.java:194)
I think this is acceptable. WDYT @pvary ?
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Changes this contains:
With this design, this class is generic enough to be wired into other readers as a common implementation for creating InputFiles with bulk decryption taken into account. Later, GenericReader and Spark's readers can swap to use this.
One hidden improvement: Previously this class created InputFiles using location + key metadata without the size. While this works, when it comes to reading, it requires an extra RPC to get the size of the file. Now, with bulkDecrypt(), the location + size + key metadata are used together saving an extra RPC per file.