Skip to content

ORC: fill initial defaults for missing id-bound fields #261

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
cbb330 wants to merge 1 commit into from
Closed

ORC: fill initial defaults for missing id-bound fields #261

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion orc/src/main/java/org/apache/iceberg/data/orc/GenericOrcReaders.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
import java.util.Map;
import java.util.UUID;
import org.apache.iceberg.data.GenericRecord;
import org.apache.iceberg.data.IdentityPartitionConverters;
import org.apache.iceberg.data.Record;
import org.apache.iceberg.orc.OrcValueReader;
import org.apache.iceberg.orc.OrcValueReaders;
Expand Down Expand Up @@ -238,7 +239,8 @@ protected StructReader(
List<OrcValueReader<?>> readers,
Types.StructType structType,
Map<Integer, ?> idToConstant) {
super(orcType, readers, structType, idToConstant);
super(
orcType, readers, structType, idToConstant, IdentityPartitionConverters::convertConstant);
this.template = GenericRecord.create(structType);
}

Expand Down
33 changes: 29 additions & 4 deletions orc/src/main/java/org/apache/iceberg/orc/ORCSchemaUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -261,18 +261,37 @@ public static Schema convert(TypeDescription orcSchema) {
*/
public static TypeDescription buildOrcProjection(
Schema schema, TypeDescription originalOrcSchema) {
return buildOrcProjection(schema, originalOrcSchema, false);
}

static TypeDescription buildOrcProjection(
Schema schema, TypeDescription originalOrcSchema, boolean applyDefaults) {
final Map<Integer, OrcField> icebergToOrc = icebergToOrcMapping("root", originalOrcSchema);
return buildOrcProjection(Integer.MIN_VALUE, schema.asStruct(), true, icebergToOrc);
return buildOrcProjection(
Integer.MIN_VALUE, schema.asStruct(), true, applyDefaults, icebergToOrc);
}

private static boolean isOmittableDefault(
Types.NestedField field, boolean applyDefaults, Map<Integer, OrcField> mapping) {
return applyDefaults && field.initialDefault() != null && !mapping.containsKey(field.fieldId());
}

private static TypeDescription buildOrcProjection(
Integer fieldId, Type type, boolean isRequired, Map<Integer, OrcField> mapping) {
Integer fieldId,
Type type,
boolean isRequired,
boolean applyDefaults,
Map<Integer, OrcField> mapping) {
final TypeDescription orcType;

switch (type.typeId()) {
case STRUCT:
orcType = TypeDescription.createStruct();
for (Types.NestedField nestedField : type.asStructType().fields()) {
if (isOmittableDefault(nestedField, applyDefaults, mapping)) {
continue;
}

// Using suffix _r to avoid potential underlying issues in ORC reader
// with reused column names between ORC and Iceberg;
// e.g. renaming column c -> d and adding new column d
Expand All @@ -285,6 +304,7 @@ private static TypeDescription buildOrcProjection(
nestedField.fieldId(),
nestedField.type(),
isRequired && nestedField.isRequired(),
applyDefaults,
mapping);
orcType.addField(name, childType);
}
Expand All @@ -296,16 +316,21 @@ private static TypeDescription buildOrcProjection(
list.elementId(),
list.elementType(),
isRequired && list.isElementRequired(),
applyDefaults,
mapping);
orcType = TypeDescription.createList(elementType);
break;
case MAP:
Types.MapType map = (Types.MapType) type;
TypeDescription keyType =
buildOrcProjection(map.keyId(), map.keyType(), isRequired, mapping);
buildOrcProjection(map.keyId(), map.keyType(), isRequired, applyDefaults, mapping);
TypeDescription valueType =
buildOrcProjection(
map.valueId(), map.valueType(), isRequired && map.isValueRequired(), mapping);
map.valueId(),
map.valueType(),
isRequired && map.isValueRequired(),
applyDefaults,
mapping);
orcType = TypeDescription.createMap(keyType, valueType);
break;
default:
Expand Down
2 changes: 1 addition & 1 deletion orc/src/main/java/org/apache/iceberg/orc/OrcIterable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,7 @@ public CloseableIterator<T> iterator() {
TypeDescription fileSchema = orcFileReader.getSchema();
final TypeDescription readOrcSchema;
if (ORCSchemaUtil.hasIds(fileSchema)) {
readOrcSchema = ORCSchemaUtil.buildOrcProjection(schema, fileSchema);
readOrcSchema = ORCSchemaUtil.buildOrcProjection(schema, fileSchema, true);
} else {
if (nameMapping == null) {
nameMapping = MappingUtil.create(schema);
Expand Down
15 changes: 15 additions & 0 deletions orc/src/main/java/org/apache/iceberg/orc/OrcValueReaders.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,9 +21,11 @@
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.function.BiFunction;
import org.apache.iceberg.MetadataColumns;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
import org.apache.orc.TypeDescription;
import org.apache.orc.storage.ql.exec.vector.BytesColumnVector;
Expand Down Expand Up @@ -183,6 +185,15 @@ protected StructReader(
List<OrcValueReader<?>> readers,
Types.StructType struct,
Map<Integer, ?> idToConstant) {
this(orcType, readers, struct, idToConstant, null);
}

protected StructReader(
TypeDescription orcType,
List<OrcValueReader<?>> readers,
Types.StructType struct,
Map<Integer, ?> idToConstant,
BiFunction<Type, Object, Object> convertConstant) {
List<Types.NestedField> fields = struct.fields();
this.readers = new OrcValueReader[fields.size()];
this.isConstantOrMetadataField = new boolean[fields.size()];
Expand All @@ -208,6 +219,10 @@ protected StructReader(
this.isConstantOrMetadataField[pos] = false;
this.orcFieldIndex[pos] = fieldIdToOrcIndex.getOrDefault(field.fieldId(), -1);
this.readers[pos] = fileReader;
} else if (field.initialDefault() != null && convertConstant != null) {
this.isConstantOrMetadataField[pos] = true;
this.readers[pos] =
constants(convertConstant.apply(field.type(), field.initialDefault()));
} else if (MetadataColumns.isMetadataColumn(field.name())) {
this.isConstantOrMetadataField[pos] = true;
this.readers[pos] = constants(null);
Expand Down
135 changes: 135 additions & 0 deletions orc/src/test/java/org/apache/iceberg/orc/TestBuildOrcProjection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,11 @@
import static org.apache.iceberg.types.Types.NestedField.optional;
import static org.apache.iceberg.types.Types.NestedField.required;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;

import org.apache.iceberg.Schema;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.types.Types;
import org.apache.orc.TypeDescription;
import org.assertj.core.api.Assertions;
Expand Down Expand Up @@ -159,4 +162,136 @@ public void testRequiredNestedFieldMissingInFile() {
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Field 4 of type long is required and was not found.");
}

@Test
public void testTopLevelScalarDefaultOmittedWhenFileIdentityIsTrustworthy() {
Schema baseSchema = new Schema(required(1, "id", Types.LongType.get()));
TypeDescription baseOrcSchema = ORCSchemaUtil.convert(baseSchema);

Schema evolvedSchema =
new Schema(
required(1, "id", Types.LongType.get()),
Types.NestedField.optional("country")
.withId(2)
.ofType(Types.StringType.get())
.withInitialDefault(Expressions.lit("US"))
.build());

// The file carries embedded field IDs, so the absent field can be identified safely.
TypeDescription projection =
ORCSchemaUtil.buildOrcProjection(evolvedSchema, baseOrcSchema, true);
assertEquals(1, projection.getChildren().size());
assertNotNull(projection.findSubtype("id"));
assertFalse(
"defaulted column must be omitted from the read projection",
projection.getFieldNames().contains("country_r2"));
}

@Test
public void testTopLevelScalarDefaultSynthesizedWithoutTrustedFileIdentity() {
Schema baseSchema = new Schema(required(1, "id", Types.LongType.get()));
TypeDescription baseOrcSchema = ORCSchemaUtil.convert(baseSchema);

Schema evolvedSchema =
new Schema(
required(1, "id", Types.LongType.get()),
Types.NestedField.optional("country")
.withId(2)
.ofType(Types.StringType.get())
.withInitialDefault(Expressions.lit("US"))
.build());

// Without trustworthy file identity, synthesize NULL rather than guessing that the field is
// absent and applying its default.
TypeDescription projection = ORCSchemaUtil.buildOrcProjection(evolvedSchema, baseOrcSchema);
assertEquals(2, projection.getChildren().size());
assertEquals(2, projection.findSubtype("country_r2").getId());
assertEquals(
TypeDescription.Category.STRING, projection.findSubtype("country_r2").getCategory());
}

@Test
public void testTopLevelRequiredScalarDefaultOmitted() {
Schema baseSchema = new Schema(required(1, "id", Types.LongType.get()));
TypeDescription baseOrcSchema = ORCSchemaUtil.convert(baseSchema);

// A required top-level field that is absent from the file but declares a default must be
// omitted (then filled), not rejected by the required-missing check.
Schema evolvedSchema =
new Schema(
required(1, "id", Types.LongType.get()),
Types.NestedField.required("code")
.withId(2)
.ofType(Types.IntegerType.get())
.withInitialDefault(Expressions.lit(7))
.build());

TypeDescription projection =
ORCSchemaUtil.buildOrcProjection(evolvedSchema, baseOrcSchema, true);
assertEquals(1, projection.getChildren().size());
assertFalse(
"required defaulted column must be omitted, not throw",
projection.getFieldNames().contains("code_r2"));
}

@Test
public void testNestedScalarDefaultOmitted() {
Schema baseSchema =
new Schema(
required(1, "id", Types.LongType.get()),
required(2, "s", Types.StructType.of(required(3, "a", Types.LongType.get()))));
TypeDescription baseOrcSchema = ORCSchemaUtil.convert(baseSchema);

// A scalar default on a field nested inside a struct is omitted so the reader can fill it.
// The present sibling "a" keeps the struct non-empty.
Schema evolvedSchema =
new Schema(
required(1, "id", Types.LongType.get()),
required(
2,
"s",
Types.StructType.of(
required(3, "a", Types.LongType.get()),
Types.NestedField.optional("b")
.withId(4)
.ofType(Types.StringType.get())
.withInitialDefault(Expressions.lit("x"))
.build())));

TypeDescription projection =
ORCSchemaUtil.buildOrcProjection(evolvedSchema, baseOrcSchema, true);
TypeDescription nested = projection.findSubtype("s");
assertEquals(1, nested.getChildren().size());
assertFalse("nested defaulted column must be omitted", nested.getFieldNames().contains("b_r4"));
}

@Test
public void testNestedStructEmptiedByOmit() {
// Base file: s { a }. Project only a new defaulted subfield s { b default 'x' } (drop a). Every
// projected subfield of s is absent + defaulted, so the nested read struct is omitted down to
// empty; the reader fills b (see TestOrcDefaultValues end-to-end coverage).
Schema baseSchema =
new Schema(
required(1, "id", Types.LongType.get()),
required(2, "s", Types.StructType.of(required(3, "a", Types.LongType.get()))));
TypeDescription baseOrcSchema = ORCSchemaUtil.convert(baseSchema);

Schema evolvedSchema =
new Schema(
required(1, "id", Types.LongType.get()),
optional(
2,
"s",
Types.StructType.of(
Types.NestedField.optional("b")
.withId(4)
.ofType(Types.StringType.get())
.withInitialDefault(Expressions.lit("x"))
.build())));

TypeDescription projection =
ORCSchemaUtil.buildOrcProjection(evolvedSchema, baseOrcSchema, true);
TypeDescription nested = projection.findSubtype("s");
assertEquals(0, nested.getChildren().size());
}
}
Loading
Loading