Data: Support reading default values from generic Avro readers #6004
There are no files selected for viewing
This file contains 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
This file contains 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
This file contains 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
This file contains 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
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -31,6 +31,7 @@ | |
| import java.util.LinkedList; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.Set; | ||
| import java.util.UUID; | ||
| import java.util.function.Supplier; | ||
| import org.apache.avro.Schema; | ||
|
|
@@ -41,8 +42,10 @@ | |
| import org.apache.avro.util.Utf8; | ||
| import org.apache.iceberg.MetadataColumns; | ||
| import org.apache.iceberg.common.DynConstructors; | ||
| import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; | ||
| import org.apache.iceberg.relocated.com.google.common.collect.Lists; | ||
| import org.apache.iceberg.relocated.com.google.common.collect.Maps; | ||
| import org.apache.iceberg.relocated.com.google.common.collect.Sets; | ||
| import org.apache.iceberg.types.Types; | ||
| import org.apache.iceberg.util.UUIDUtil; | ||
|
|
||
|
|
@@ -566,8 +569,10 @@ public Map<K, V> read(Decoder decoder, Object reuse) throws IOException { | |
|
|
||
| public abstract static class StructReader<S> implements ValueReader<S>, SupportsRowPosition { | ||
| private final ValueReader<?>[] readers; | ||
| private final int[] constantPositions; | ||
| private final Object[] constantValues; | ||
| private final int[] defaultPositions; | ||
| private final Object[] defaultValues; | ||
|
There was a problem hiding this comment. If we go with the approach in #9366, then we should no longer need changes here, which was what originally made this difficult. |
||
| private int posField = -1; | ||
|
|
||
| protected StructReader(List<ValueReader<?>> readers, Schema schema) { | ||
|
|
@@ -586,37 +591,68 @@ protected StructReader(List<ValueReader<?>> readers, Schema schema) { | |
| } | ||
|
|
||
| if (isDeletedColumnPos == null) { | ||
| this.constantPositions = new int[0]; | ||
| this.constantValues = new Object[0]; | ||
| } else { | ||
| this.constantPositions = new int[] {isDeletedColumnPos}; | ||
| this.constantValues = new Object[] {false}; | ||
| } | ||
| this.defaultPositions = new int[0]; | ||
| this.defaultValues = new Object[0]; | ||
| } | ||
|
|
||
| protected StructReader( | ||
| List<ValueReader<?>> readers, Types.StructType struct, Map<Integer, ?> idToConstant) { | ||
| this(readers, struct, idToConstant, ImmutableMap.of()); | ||
| } | ||
|
|
||
| protected StructReader( | ||
| List<ValueReader<?>> readers, | ||
| Types.StructType struct, | ||
| Map<Integer, ?> idToConstant, | ||
| Map<Integer, ?> idToDefault) { | ||
| this.readers = readers.toArray(new ValueReader[0]); | ||
|
|
||
| List<Types.NestedField> fields = struct.fields(); | ||
| List<Integer> constantPositionsList = Lists.newArrayListWithCapacity(fields.size()); | ||
| List<Object> constantValuesList = Lists.newArrayListWithCapacity(fields.size()); | ||
| List<Integer> defaultPositionList = Lists.newArrayListWithCapacity(fields.size()); | ||
| List<Object> defaultValuesList = Lists.newArrayListWithCapacity(fields.size()); | ||
| for (int pos = 0; pos < fields.size(); pos += 1) { | ||
| Types.NestedField field = fields.get(pos); | ||
| if (idToConstant.containsKey(field.fieldId())) { | ||
| constantPositionsList.add(pos); | ||
| constantValuesList.add(idToConstant.get(field.fieldId())); | ||
| } else if (field.fieldId() == MetadataColumns.ROW_POSITION.fieldId()) { | ||
| // track where the _pos field is located for setRowPositionSupplier | ||
| this.posField = pos; | ||
| } else if (field.fieldId() == MetadataColumns.IS_DELETED.fieldId()) { | ||
| constantPositionsList.add(pos); | ||
| constantValuesList.add(false); | ||
| } else if (field.initialDefault() != null) { | ||
| if (idToDefault.containsKey(field.fieldId())) { | ||
| // Add a constant value for fields that have a default value. | ||
| // In the {@link #read()} method, this will be leveraged only if there is no | ||
| // corresponding | ||
| // reader. | ||
| defaultPositionList.add(pos); | ||
| defaultValuesList.add(idToDefault.get(field.fieldId())); | ||
| } else { | ||
| // Throw an exception if the map does not contain a default value for that field. | ||
| throw new UnsupportedOperationException( | ||
| "Default value not found for field: " | ||
| + field.name() | ||
| + " (field ID: " | ||
| + field.fieldId() | ||
| + ")."); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| this.constantPositions = constantPositionsList.stream().mapToInt(Integer::intValue).toArray(); | ||
| this.constantValues = constantValuesList.toArray(); | ||
| this.defaultPositions = defaultPositionList.stream().mapToInt(Integer::intValue).toArray(); | ||
| this.defaultValues = defaultValuesList.toArray(); | ||
| } | ||
|
|
||
| @Override | ||
|
|
@@ -654,10 +690,20 @@ public S read(Decoder decoder, Object reuse) throws IOException { | |
| S struct = reuseOrCreate(reuse); | ||
|
|
||
| if (decoder instanceof ResolvingDecoder) { | ||
| Set<Integer> existingFieldPositionsSet = Sets.newHashSet(); | ||
| // This may not set all of the fields. nulls are set by default. | ||
| for (Schema.Field field : ((ResolvingDecoder) decoder).readFieldOrder()) { | ||
| Object reusedValue = get(struct, field.pos()); | ||
| set(struct, field.pos(), readers[field.pos()].read(decoder, reusedValue)); | ||
| existingFieldPositionsSet.add(field.pos()); | ||
| } | ||
|
|
||
| // Set default values | ||
|
There was a problem hiding this comment. Style: missing empty newline between control flow block and the following statement. |
||
| for (int i = 0; i < defaultPositions.length; i += 1) { | ||
| // Set default values only if the field does not exist in the data. | ||
| if (!existingFieldPositionsSet.contains(defaultPositions[i])) { | ||
| set(struct, defaultPositions[i], defaultValues[i]); | ||
| } | ||
| } | ||
|
|
||
|
There was a problem hiding this comment. Nit: unnecessary whitespace change. |
||
| } else { | ||
|
|
@@ -667,8 +713,8 @@ public S read(Decoder decoder, Object reuse) throws IOException { | |
| } | ||
| } | ||
|
|
||
| for (int i = 0; i < constantPositions.length; i += 1) { | ||
| set(struct, constantPositions[i], constantValues[i]); | ||
| } | ||
|
|
||
| return struct; | ||
|
|
||
This file contains 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
This file contains 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
Oops, something went wrong.
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.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I thought we talked about making these package-private until the implementation was complete so that it doesn't leak into the public API before the values are read correctly? Is there some reason why that wouldn't work?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The reason is that I need to instantiate a
NestedFieldwith default value in theTestReadDefaultValuesclass, but this class lives in the iceberg-data module, which is not the same package astypes.javain the API module, so package private doesn't work..I can annotate this method as
VisibleForTestingand/orBeta, do you think that works?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can you move the test class? We can't merge this unless these are not visible. Annotations don't change visibility.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It is hard to move the test class to another package since all the Avro dependencies will break in this case. I have updated the PR to make both
optionaland theNestedFieldconstructor protected and inheritedNestedFieldclass in the test class. However, that has the risk of having to downgradeNestedFiledconstructor to private in the future once this workaround is no longer needed.Another option is to keep both as private, and use reflection in the test to access them as a temporary workaround.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
kept the constructor as private and the newly introduced
NestedFieldAPIs as public.