Repository navigation
Data: Add a Record stitcher for column files - #18388
anuragmantri wants to merge 12 commits into
Conversation
This change introduces the interface for column files and also integrates it to the schema for TrackedFile.
…lkDecrypt()
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.
Lazy initialization of the input files is guarded against concurrent
access, and the concurrency test uses a barrier with a timeout.
anuragmantri
changed the title
anuragmantri
changed the title
|
@gaborkaszab, this 192d90b commit adds a stitcher for generic reads, this is on top of #18324 so it's still in draft. |
gaborkaszab
left a comment
gaborkaszab
left a comment
There was a problem hiding this comment.
Thanks for the PR, @anuragmantri !
I went through the non-test part as a first pass.
| .build(); | ||
| } | ||
|
|
||
| private Map<String, InputFile> inputFiles(DataFile file) { |
There was a problem hiding this comment.
Just for the record, as you wrote in the description, InputFilesDecryptor is meant.to be used in this class to provide the InputFiles. As a result this function will be gone.
|
|
||
| @Override | ||
| public Record copy() { | ||
| return toGenericRecord().copy(); |
There was a problem hiding this comment.
Isn't this a double copy? toGenericRecord seems itself a full copy, and then we call copy() on it. Same for the other variation below.
| } | ||
| } | ||
|
|
||
| private static class StitchedRecord implements Record { |
There was a problem hiding this comment.
Many of the method implementation seems identical to StitchedInternalRow. Maybe have a common base class sharing the common functionality?
| public Stitcher<Record> build(Schema projection, List<Schema> verticalSplits) { | ||
| StitchLayout layout = StitchLayout.of(projection, verticalSplits); | ||
| Types.StructType struct = projection.asStruct(); | ||
| Map<String, Integer> positions = Maps.newHashMap(); |
There was a problem hiding this comment.
positions seems something that is specific for StitchedRecord. Is it needed because GenericRecord has some name-based operations?
I think it should be internal to StitchedRecord. Is it feasible not to pass as a param but construct internally?
This branch has not been deployed
2 participants
Depends on #18324
This PR is only the top commit 192d90b
This PR adds
org.apache.iceberg.data.GenericStitchers, modeled onSparkStitchersin #18324.Recordreads each field from the part thatStitchLayoutassigns it to.setandsetFieldwrite through to that part, as the Spark row does, andcopy()returns a detachedGenericRecord.GenericReadernow resolves column files toInputFiles, unencrypted only.inputFiles(DataFile)is the method the plannedInputFilesDecryptorwiring will replace.ColumnFileTestHelpersmoves from Spark to core test sources so the data tests can use it.Test plan:
GenericReaderreturns column-file values instead of stale base values, keeps a struct from the base intact, and still works for a projection of only base fields._posthrough the stitched record.TestGenericStitchersvalidates the record's contract: fields resolve in projection order by position and name, a set value reads back,copy()stays intact when the parts change (readers reuse part records), and the parts list is not kept.AI Disclosure
StitcherRegistrynames and wire it intoGenericReader.