diff --git a/api/src/main/java/org/apache/iceberg/Accessors.java b/api/src/main/java/org/apache/iceberg/Accessors.java index 0b36730fbb4b..6095cb05f35f 100644 --- a/api/src/main/java/org/apache/iceberg/Accessors.java +++ b/api/src/main/java/org/apache/iceberg/Accessors.java @@ -213,8 +213,18 @@ public Map> schema( @Override public Map> struct( Types.StructType struct, List>> fieldResults) { + return buildAccessors(struct.fields(), fieldResults); + } + + @Override + public Map> file( + Types.FileType file, List>> fieldResults) { + return buildAccessors(file.fields(), fieldResults); + } + + private Map> buildAccessors( + List fields, List>> fieldResults) { Map> accessors = Maps.newHashMap(); - List fields = struct.fields(); for (int i = 0; i < fieldResults.size(); i += 1) { Types.NestedField field = fields.get(i); Map> result = fieldResults.get(i); diff --git a/api/src/main/java/org/apache/iceberg/Schema.java b/api/src/main/java/org/apache/iceberg/Schema.java index 3e59998be476..2a5fdd5f83b7 100644 --- a/api/src/main/java/org/apache/iceberg/Schema.java +++ b/api/src/main/java/org/apache/iceberg/Schema.java @@ -67,7 +67,8 @@ public class Schema implements Serializable { Type.TypeID.VARIANT, 3, Type.TypeID.UNKNOWN, 3, Type.TypeID.GEOMETRY, 3, - Type.TypeID.GEOGRAPHY, 3); + Type.TypeID.GEOGRAPHY, 3, + Type.TypeID.FILE, 4); private final StructType struct; private final int schemaId; @@ -578,20 +579,33 @@ private List reassignIds(List columns, TypeUtil.GetID if (getID == null) { return columns; } - Type res = - TypeUtil.assignIds( - StructType.of(columns), - oldId -> { - int newId = getID.get(oldId); - if (newId != oldId) { - idsToReassigned.put(oldId, newId); - idsToOriginal.put(newId, oldId); - } - return newId; - }); + + TypeUtil.GetID tracked = + new TypeUtil.GetID() { + @Override + public int get(int oldId) { + return track(oldId, getID.get(oldId)); + } + + @Override + public int get(int oldId, int numReserved) { + return track(oldId, getID.get(oldId, numReserved)); + } + }; + + Type res = TypeUtil.assignIds(StructType.of(columns), tracked); return res.asStructType().fields(); } + private int track(int oldId, int newId) { + if (newId != oldId) { + idsToReassigned.put(oldId, newId); + idsToOriginal.put(newId, oldId); + } + + return newId; + } + /** * Check the compatibility of the schema with a format version. * diff --git a/api/src/main/java/org/apache/iceberg/types/AssignFreshIds.java b/api/src/main/java/org/apache/iceberg/types/AssignFreshIds.java index f3759f1d72f3..26fd72bf639e 100644 --- a/api/src/main/java/org/apache/iceberg/types/AssignFreshIds.java +++ b/api/src/main/java/org/apache/iceberg/types/AssignFreshIds.java @@ -48,7 +48,16 @@ class AssignFreshIds extends TypeUtil.CustomOrderSchemaVisitor { this.nextId = nextId; } - private int idFor(String fullName) { + private int idFor(String fullName, Type type) { + Integer existingId = baseId(fullName); + if (existingId != null) { + return existingId; + } + + return type.isFileType() ? nextId.get(Types.FileType.NUM_NESTED_FIELDS) : nextId.get(); + } + + private Integer baseId(String fullName) { if (baseSchema != null && fullName != null) { Types.NestedField field = baseSchema.findField(fullName); if (field != null) { @@ -56,7 +65,7 @@ private int idFor(String fullName) { } } - return nextId.get(); + return null; } private String name(int id) { @@ -80,15 +89,17 @@ public Type struct(Types.StructType struct, Iterable futures) { // assign IDs for this struct's fields first List newIds = Lists.newArrayListWithExpectedSize(length); for (int i = 0; i < length; i += 1) { - newIds.add(idFor(name(fields.get(i).fieldId()))); + Types.NestedField field = fields.get(i); + newIds.add(idFor(name(field.fieldId()), field.type())); } List newFields = Lists.newArrayListWithExpectedSize(length); Iterator types = futures.iterator(); for (int i = 0; i < length; i += 1) { Types.NestedField field = fields.get(i); - Type type = types.next(); - newFields.add(Types.NestedField.from(field).withId(newIds.get(i)).ofType(type).build()); + int newId = newIds.get(i); + Type type = TypeUtil.assignedType(field.type(), newId, types.next()); + newFields.add(Types.NestedField.from(field).withId(newId).ofType(type).build()); } return Types.StructType.of(newFields); @@ -101,22 +112,25 @@ public Type field(Types.NestedField field, Supplier future) { @Override public Type list(Types.ListType list, Supplier future) { - int newId = idFor(name(list.elementId())); + int newId = idFor(name(list.elementId()), list.elementType()); + Type elementType = TypeUtil.assignedType(list.elementType(), newId, future.get()); if (list.isElementOptional()) { - return Types.ListType.ofOptional(newId, future.get()); + return Types.ListType.ofOptional(newId, elementType); } else { - return Types.ListType.ofRequired(newId, future.get()); + return Types.ListType.ofRequired(newId, elementType); } } @Override public Type map(Types.MapType map, Supplier keyFuture, Supplier valueFuture) { - int newKeyId = idFor(name(map.keyId())); - int newValueId = idFor(name(map.valueId())); + int newKeyId = idFor(name(map.keyId()), map.keyType()); + int newValueId = idFor(name(map.valueId()), map.valueType()); + Type keyType = TypeUtil.assignedType(map.keyType(), newKeyId, keyFuture.get()); + Type valueType = TypeUtil.assignedType(map.valueType(), newValueId, valueFuture.get()); if (map.isValueOptional()) { - return Types.MapType.ofOptional(newKeyId, newValueId, keyFuture.get(), valueFuture.get()); + return Types.MapType.ofOptional(newKeyId, newValueId, keyType, valueType); } else { - return Types.MapType.ofRequired(newKeyId, newValueId, keyFuture.get(), valueFuture.get()); + return Types.MapType.ofRequired(newKeyId, newValueId, keyType, valueType); } } @@ -125,6 +139,12 @@ public Type variant(Types.VariantType variant) { return variant; } + @Override + public Type file(Types.FileType file, Iterable futures) { + // nested fields are rebuilt from the new id assigned to the field that holds this type + return file; + } + @Override public Type primitive(Type.PrimitiveType primitive) { return primitive; diff --git a/api/src/main/java/org/apache/iceberg/types/AssignIds.java b/api/src/main/java/org/apache/iceberg/types/AssignIds.java index fd5ac7ff67b9..e22bddba5180 100644 --- a/api/src/main/java/org/apache/iceberg/types/AssignIds.java +++ b/api/src/main/java/org/apache/iceberg/types/AssignIds.java @@ -31,8 +31,8 @@ class AssignIds extends TypeUtil.CustomOrderSchemaVisitor { this.getID = getID; } - private int idFor(int id) { - return getID.get(id); + private int idFor(int id, Type type) { + return type.isFileType() ? getID.get(id, Types.FileType.NUM_NESTED_FIELDS) : getID.get(id); } @Override @@ -48,15 +48,16 @@ public Type struct(Types.StructType struct, Iterable futures) { // assign IDs for this struct's fields first List newIds = Lists.newArrayListWithExpectedSize(length); for (Types.NestedField field : fields) { - newIds.add(idFor(field.fieldId())); + newIds.add(idFor(field.fieldId(), field.type())); } List newFields = Lists.newArrayListWithExpectedSize(length); Iterator types = futures.iterator(); for (int i = 0; i < length; i += 1) { Types.NestedField field = fields.get(i); - Type type = types.next(); - newFields.add(Types.NestedField.from(field).withId(newIds.get(i)).ofType(type).build()); + int newId = newIds.get(i); + Type type = TypeUtil.assignedType(field.type(), newId, types.next()); + newFields.add(Types.NestedField.from(field).withId(newId).ofType(type).build()); } return Types.StructType.of(newFields); @@ -69,22 +70,25 @@ public Type field(Types.NestedField field, Supplier future) { @Override public Type list(Types.ListType list, Supplier future) { - int newId = idFor(list.elementId()); + int newId = idFor(list.elementId(), list.elementType()); + Type elementType = TypeUtil.assignedType(list.elementType(), newId, future.get()); if (list.isElementOptional()) { - return Types.ListType.ofOptional(newId, future.get()); + return Types.ListType.ofOptional(newId, elementType); } else { - return Types.ListType.ofRequired(newId, future.get()); + return Types.ListType.ofRequired(newId, elementType); } } @Override public Type map(Types.MapType map, Supplier keyFuture, Supplier valueFuture) { - int newKeyId = idFor(map.keyId()); - int newValueId = idFor(map.valueId()); + int newKeyId = idFor(map.keyId(), map.keyType()); + int newValueId = idFor(map.valueId(), map.valueType()); + Type keyType = TypeUtil.assignedType(map.keyType(), newKeyId, keyFuture.get()); + Type valueType = TypeUtil.assignedType(map.valueType(), newValueId, valueFuture.get()); if (map.isValueOptional()) { - return Types.MapType.ofOptional(newKeyId, newValueId, keyFuture.get(), valueFuture.get()); + return Types.MapType.ofOptional(newKeyId, newValueId, keyType, valueType); } else { - return Types.MapType.ofRequired(newKeyId, newValueId, keyFuture.get(), valueFuture.get()); + return Types.MapType.ofRequired(newKeyId, newValueId, keyType, valueType); } } @@ -93,6 +97,12 @@ public Type variant(Types.VariantType variant) { return variant; } + @Override + public Type file(Types.FileType file, Iterable futures) { + // nested fields are rebuilt from the new id assigned to the field that holds this type + return file; + } + @Override public Type primitive(Type.PrimitiveType primitive) { return primitive; diff --git a/api/src/main/java/org/apache/iceberg/types/CheckCompatibility.java b/api/src/main/java/org/apache/iceberg/types/CheckCompatibility.java index 3b3a38ff5aeb..bbca4137bca5 100644 --- a/api/src/main/java/org/apache/iceberg/types/CheckCompatibility.java +++ b/api/src/main/java/org/apache/iceberg/types/CheckCompatibility.java @@ -260,6 +260,16 @@ public List variant(Types.VariantType readVariant) { return ImmutableList.of(String.format(": %s cannot be read as a %s", currentType, readVariant)); } + @Override + public List file(Types.FileType readFile, Iterable> fieldErrorLists) { + if (currentType.isFileType()) { + // the nested fields are derived from the enclosing id, so matching ids means matching fields + return NO_ERRORS; + } + + return ImmutableList.of(String.format(": %s cannot be read as a %s", currentType, readFile)); + } + @Override public List primitive(Type.PrimitiveType readPrimitive) { if (currentType.equals(readPrimitive)) { diff --git a/api/src/main/java/org/apache/iceberg/types/Comparators.java b/api/src/main/java/org/apache/iceberg/types/Comparators.java index ab59c895686d..67d7d3543527 100644 --- a/api/src/main/java/org/apache/iceberg/types/Comparators.java +++ b/api/src/main/java/org/apache/iceberg/types/Comparators.java @@ -83,6 +83,8 @@ private static Comparator internal(Type type) { return forType(type.asPrimitiveType()); } else if (type.isStructType()) { return (Comparator) forType(type.asStructType()); + } else if (type.isFileType()) { + return (Comparator) forType(type.asFileType().asStruct()); } else if (type.isListType()) { return (Comparator) forType(type.asListType()); } else if (type.isMapType()) { diff --git a/api/src/main/java/org/apache/iceberg/types/FindTypeVisitor.java b/api/src/main/java/org/apache/iceberg/types/FindTypeVisitor.java index 64faebb48243..0f43358e1029 100644 --- a/api/src/main/java/org/apache/iceberg/types/FindTypeVisitor.java +++ b/api/src/main/java/org/apache/iceberg/types/FindTypeVisitor.java @@ -85,6 +85,21 @@ public Type variant(Types.VariantType variant) { return null; } + @Override + public Type file(Types.FileType file, List fieldResults) { + if (predicate.test(file)) { + return file; + } + + for (Type fieldType : fieldResults) { + if (fieldType != null) { + return fieldType; + } + } + + return null; + } + @Override public Type primitive(Type.PrimitiveType primitive) { if (predicate.test(primitive)) { diff --git a/api/src/main/java/org/apache/iceberg/types/GetProjectedIds.java b/api/src/main/java/org/apache/iceberg/types/GetProjectedIds.java index 1ec70b8578bc..de5ee564cf31 100644 --- a/api/src/main/java/org/apache/iceberg/types/GetProjectedIds.java +++ b/api/src/main/java/org/apache/iceberg/types/GetProjectedIds.java @@ -45,9 +45,14 @@ public Set struct(Types.StructType struct, List> fieldResu return fieldIds; } + @Override + public Set file(Types.FileType file, List> fieldResults) { + return fieldIds; + } + @Override public Set field(Types.NestedField field, Set fieldResult) { - if ((includeStructIds && field.type().isStructType()) + if ((includeStructIds && (field.type().isStructType() || field.type().isFileType())) || field.type().isPrimitiveType() || field.type().isVariantType()) { fieldIds.add(field.fieldId()); diff --git a/api/src/main/java/org/apache/iceberg/types/IndexById.java b/api/src/main/java/org/apache/iceberg/types/IndexById.java index a7b96eb381f7..3f0381262f79 100644 --- a/api/src/main/java/org/apache/iceberg/types/IndexById.java +++ b/api/src/main/java/org/apache/iceberg/types/IndexById.java @@ -38,6 +38,12 @@ public Map struct( return index; } + @Override + public Map file( + Types.FileType file, List> fieldResults) { + return index; + } + @Override public Map field( Types.NestedField field, Map fieldResult) { diff --git a/api/src/main/java/org/apache/iceberg/types/IndexByName.java b/api/src/main/java/org/apache/iceberg/types/IndexByName.java index 9ca2a1d3396c..1eb9a2f1f1f1 100644 --- a/api/src/main/java/org/apache/iceberg/types/IndexByName.java +++ b/api/src/main/java/org/apache/iceberg/types/IndexByName.java @@ -113,7 +113,7 @@ public void beforeListElement(Types.NestedField elementField) { // only add "element" to the short name if the element is not a struct, so that names are more // natural // for example, locations.latitude instead of locations.element.latitude - if (!elementField.type().isStructType()) { + if (!hasNestedFields(elementField)) { shortFieldNames.push(elementField.name()); } } @@ -123,7 +123,7 @@ public void afterListElement(Types.NestedField elementField) { fieldNames.pop(); // only remove "element" if it was added - if (!elementField.type().isStructType()) { + if (!hasNestedFields(elementField)) { shortFieldNames.pop(); } } @@ -143,7 +143,7 @@ public void beforeMapValue(Types.NestedField valueField) { fieldNames.push(valueField.name()); // only add "value" to the name if the value is not a struct, so that names are more natural - if (!valueField.type().isStructType()) { + if (!hasNestedFields(valueField)) { shortFieldNames.push(valueField.name()); } } @@ -153,11 +153,15 @@ public void afterMapValue(Types.NestedField valueField) { fieldNames.pop(); // only remove "value" if it was added - if (!valueField.type().isStructType()) { + if (!hasNestedFields(valueField)) { shortFieldNames.pop(); } } + private static boolean hasNestedFields(Types.NestedField field) { + return field.type().isStructType() || field.type().isFileType(); + } + @Override public Map schema(Schema schema, Map structResult) { return nameToId; @@ -194,6 +198,11 @@ public Map variant(Types.VariantType variant) { return nameToId; } + @Override + public Map file(Types.FileType file, List> fieldResults) { + return nameToId; + } + @Override public Map primitive(Type.PrimitiveType primitive) { return nameToId; diff --git a/api/src/main/java/org/apache/iceberg/types/IndexParents.java b/api/src/main/java/org/apache/iceberg/types/IndexParents.java index 6e611d47e912..5202f40d5914 100644 --- a/api/src/main/java/org/apache/iceberg/types/IndexParents.java +++ b/api/src/main/java/org/apache/iceberg/types/IndexParents.java @@ -47,7 +47,16 @@ public Map schema(Schema schema, Map structR @Override public Map struct( Types.StructType struct, List> fieldResults) { - for (Types.NestedField field : struct.fields()) { + return indexFields(struct.fields()); + } + + @Override + public Map file(Types.FileType file, List> fieldResults) { + return indexFields(file.fields()); + } + + private Map indexFields(List fields) { + for (Types.NestedField field : fields) { Integer parentId = idStack.peek(); if (parentId != null) { // fields in the root struct are not added diff --git a/api/src/main/java/org/apache/iceberg/types/JavaHash.java b/api/src/main/java/org/apache/iceberg/types/JavaHash.java index 1988a90322e4..ceef07a4d4ad 100644 --- a/api/src/main/java/org/apache/iceberg/types/JavaHash.java +++ b/api/src/main/java/org/apache/iceberg/types/JavaHash.java @@ -31,6 +31,8 @@ static JavaHash forType(Type type) { return (JavaHash) JavaHashes.strings(); case STRUCT: return (JavaHash) JavaHashes.struct(type.asStructType()); + case FILE: + return (JavaHash) JavaHashes.struct(type.asFileType().asStruct()); case LIST: return (JavaHash) JavaHashes.list(type.asListType()); default: diff --git a/api/src/main/java/org/apache/iceberg/types/PruneColumns.java b/api/src/main/java/org/apache/iceberg/types/PruneColumns.java index 56f01cf34bb5..7ce5ca87ddd8 100644 --- a/api/src/main/java/org/apache/iceberg/types/PruneColumns.java +++ b/api/src/main/java/org/apache/iceberg/types/PruneColumns.java @@ -52,7 +52,16 @@ public Type schema(Schema schema, Type structResult) { @Override public Type struct(Types.StructType struct, List fieldResults) { - List fields = struct.fields(); + return project(struct.fields(), fieldResults, struct); + } + + @Override + public Type file(Types.FileType file, List fieldResults) { + return project(file.fields(), fieldResults, file); + } + + private Type project( + List fields, List fieldResults, Type unchangedResult) { List selectedFields = Lists.newArrayListWithExpectedSize(fields.size()); boolean sameTypes = true; @@ -79,7 +88,7 @@ public Type struct(Types.StructType struct, List fieldResults) { if (!selectedFields.isEmpty()) { if (selectedFields.size() == fields.size() && sameTypes) { - return struct; + return unchangedResult; } else { return Types.StructType.of(selectedFields); } @@ -95,6 +104,8 @@ public Type field(Types.NestedField field, Type fieldResult) { return field.type(); } else if (field.type().isStructType()) { return projectSelectedStruct(fieldResult); + } else if (field.type().isFileType()) { + return projectSelectedFile(fieldResult); } else { Preconditions.checkArgument( !field.type().isNestedType(), @@ -120,6 +131,8 @@ public Type list(Types.ListType list, Type elementResult) { } else if (list.elementType().isStructType()) { StructType projectedStruct = projectSelectedStruct(elementResult); return projectList(list, projectedStruct); + } else if (list.elementType().isFileType()) { + return projectList(list, projectSelectedFile(elementResult)); } else { Preconditions.checkArgument( list.elementType().isPrimitiveType(), @@ -142,6 +155,8 @@ public Type map(Types.MapType map, Type ignored, Type valueResult) { } else if (map.valueType().isStructType()) { Type projectedStruct = projectSelectedStruct(valueResult); return projectMap(map, projectedStruct); + } else if (map.valueType().isFileType()) { + return projectMap(map, projectSelectedFile(valueResult)); } else { Preconditions.checkArgument( map.valueType().isPrimitiveType(), @@ -169,6 +184,18 @@ public Type primitive(Type.PrimitiveType primitive) { return null; } + /** + * Returns the projection of a selected file, which is a file when every nested field is projected + * and a struct when only some are. + */ + private Type projectSelectedFile(Type projectedField) { + if (projectedField == null) { + // no nested fields were selected but the file was, return an empty struct + return Types.StructType.of(); + } + return projectedField; + } + private ListType projectList(ListType list, Type elementResult) { Preconditions.checkArgument( elementResult != null, "Cannot project a list when the element result is null"); diff --git a/api/src/main/java/org/apache/iceberg/types/ReassignDoc.java b/api/src/main/java/org/apache/iceberg/types/ReassignDoc.java index 86527fb3897f..de63e94ffc6b 100644 --- a/api/src/main/java/org/apache/iceberg/types/ReassignDoc.java +++ b/api/src/main/java/org/apache/iceberg/types/ReassignDoc.java @@ -96,6 +96,12 @@ public Type variant(Types.VariantType variant) { return variant; } + @Override + public Type file(Types.FileType file, Iterable fieldTypes) { + // the nested fields of a file cannot carry docs + return file; + } + @Override public Type primitive(Type.PrimitiveType primitive) { return primitive; diff --git a/api/src/main/java/org/apache/iceberg/types/ReassignIds.java b/api/src/main/java/org/apache/iceberg/types/ReassignIds.java index 3d114f093f6b..ef7308ff2fb7 100644 --- a/api/src/main/java/org/apache/iceberg/types/ReassignIds.java +++ b/api/src/main/java/org/apache/iceberg/types/ReassignIds.java @@ -50,7 +50,7 @@ public Type schema(Schema schema, Supplier future) { } } - private int id(Types.StructType sourceStruct, String name) { + private int id(Types.StructType sourceStruct, String name, Type type) { Types.NestedField sourceField = caseSensitive ? sourceStruct.field(name) : sourceStruct.caseInsensitiveField(name); @@ -59,7 +59,7 @@ private int id(Types.StructType sourceStruct, String name) { } if (assignId != null) { - return assignId.get(); + return type.isFileType() ? assignId.get(Types.FileType.NUM_NESTED_FIELDS) : assignId.get(); } throw new IllegalArgumentException("Field " + name + " not found in source schema"); @@ -78,8 +78,9 @@ public Type struct(Types.StructType struct, Iterable fieldTypes) { List newFields = Lists.newArrayListWithExpectedSize(length); for (int i = 0; i < length; i += 1) { Types.NestedField field = fields.get(i); - int fieldId = id(sourceStruct, field.name()); - newFields.add(Types.NestedField.from(field).withId(fieldId).ofType(types.get(i)).build()); + int fieldId = id(sourceStruct, field.name(), field.type()); + Type type = TypeUtil.assignedType(field.type(), fieldId, types.get(i)); + newFields.add(Types.NestedField.from(field).withId(fieldId).ofType(type).build()); } return Types.StructType.of(newFields); @@ -120,10 +121,12 @@ public Type list(Types.ListType list, Supplier elementTypeFuture) { this.sourceType = sourceList.elementType(); try { + Type elementType = + TypeUtil.assignedType(list.elementType(), sourceElementId, elementTypeFuture.get()); if (list.isElementOptional()) { - return Types.ListType.ofOptional(sourceElementId, elementTypeFuture.get()); + return Types.ListType.ofOptional(sourceElementId, elementType); } else { - return Types.ListType.ofRequired(sourceElementId, elementTypeFuture.get()); + return Types.ListType.ofRequired(sourceElementId, elementType); } } finally { @@ -141,10 +144,10 @@ public Type map(Types.MapType map, Supplier keyTypeFuture, Supplier try { this.sourceType = sourceMap.keyType(); - Type keyType = keyTypeFuture.get(); + Type keyType = TypeUtil.assignedType(map.keyType(), sourceKeyId, keyTypeFuture.get()); this.sourceType = sourceMap.valueType(); - Type valueType = valueTypeFuture.get(); + Type valueType = TypeUtil.assignedType(map.valueType(), sourceValueId, valueTypeFuture.get()); if (map.isValueOptional()) { return Types.MapType.ofOptional(sourceKeyId, sourceValueId, keyType, valueType); @@ -162,6 +165,12 @@ public Type variant(Types.VariantType variant) { return variant; } + @Override + public Type file(Types.FileType file, Iterable fieldTypes) { + // nested fields are rebuilt from the id assigned to the field that holds this type + return file; + } + @Override public Type primitive(Type.PrimitiveType primitive) { return primitive; // nothing to reassign diff --git a/api/src/main/java/org/apache/iceberg/types/ReplaceTypeById.java b/api/src/main/java/org/apache/iceberg/types/ReplaceTypeById.java index 1c94bd57c114..9e767e8babce 100644 --- a/api/src/main/java/org/apache/iceberg/types/ReplaceTypeById.java +++ b/api/src/main/java/org/apache/iceberg/types/ReplaceTypeById.java @@ -37,7 +37,16 @@ public Type schema(Schema schema, Type structResult) { @Override public Type struct(Types.StructType struct, List fieldResults) { - List fields = struct.fields(); + return replaceFieldTypes(struct.fields(), fieldResults, struct); + } + + @Override + public Type file(Types.FileType file, List fieldResults) { + return replaceFieldTypes(file.fields(), fieldResults, file); + } + + private Type replaceFieldTypes( + List fields, List fieldResults, Type unchangedResult) { List newFields = Lists.newArrayListWithExpectedSize(fields.size()); boolean hasChanged = false; @@ -56,7 +65,7 @@ public Type struct(Types.StructType struct, List fieldResults) { return Types.StructType.of(newFields); } - return struct; + return unchangedResult; } @Override diff --git a/api/src/main/java/org/apache/iceberg/types/Type.java b/api/src/main/java/org/apache/iceberg/types/Type.java index bed478d938e7..d27305f7ead3 100644 --- a/api/src/main/java/org/apache/iceberg/types/Type.java +++ b/api/src/main/java/org/apache/iceberg/types/Type.java @@ -50,6 +50,7 @@ enum TypeID { LIST(List.class), MAP(Map.class), VARIANT(Variant.class), + FILE(StructLike.class), UNKNOWN(Object.class); private final Class javaClass; @@ -89,6 +90,10 @@ default Types.VariantType asVariantType() { throw new IllegalArgumentException("Not a variant type: " + this); } + default Types.FileType asFileType() { + throw new IllegalArgumentException("Not a file type: " + this); + } + default boolean isNestedType() { return false; } @@ -109,6 +114,10 @@ default boolean isVariantType() { return false; } + default boolean isFileType() { + return false; + } + default NestedType asNestedType() { throw new IllegalArgumentException("Not a nested type: " + this); } diff --git a/api/src/main/java/org/apache/iceberg/types/TypeUtil.java b/api/src/main/java/org/apache/iceberg/types/TypeUtil.java index 8e39ae7a43bc..241089c4e8ac 100644 --- a/api/src/main/java/org/apache/iceberg/types/TypeUtil.java +++ b/api/src/main/java/org/apache/iceberg/types/TypeUtil.java @@ -72,6 +72,24 @@ public static Schema project(Schema schema, Set fieldIds) { return new Schema(Collections.emptyList(), schema.getAliases()); } + /** + * Returns a type's nested fields as a struct. + * + *

Unlike {@link Type#asStructType()}, this also accepts a file type and returns the struct of + * its derived nested fields. Use this where a file is stored and read as a group of its nested + * fields, such as in the Avro and Parquet layers. + * + * @param type a struct or file type + * @return the type's nested fields as a struct + */ + public static Types.StructType asStructType(Type type) { + if (type.isFileType()) { + return type.asFileType().asStruct(); + } + + return type.asStructType(); + } + public static Types.StructType project(Types.StructType struct, Set fieldIds) { Preconditions.checkNotNull(struct, "Struct cannot be null"); Preconditions.checkNotNull(fieldIds, "Field ids cannot be null"); @@ -460,6 +478,14 @@ public static Type assignIds(Type type, GetID getId) { return TypeUtil.visit(type, new AssignIds(getId)); } + static Type assignedType(Type original, int newId, Type visited) { + if (original.isFileType()) { + return Types.FileType.of(newId); + } + + return visited; + } + public static Type find(Schema schema, Predicate predicate) { return visit(schema, new FindTypeVisitor(predicate)); } @@ -639,11 +665,48 @@ private static int estimateSize(Type type) { /** Interface for passing a function that assigns column IDs. */ public interface NextID { int get(); + + default int get(int numReserved) { + int id = get(); + if (numReserved > 0) { + for (int offset = 1; offset <= numReserved; offset += 1) { + int reserved = get(); + Preconditions.checkState( + reserved == id + offset, + "Cannot reserve %s IDs after %s: assigned %s", + numReserved, + id, + reserved); + } + } + + return id; + } } /** Interface for passing a function that assigns column IDs from the previous Id. */ public interface GetID { int get(int oldId); + + /** + * Assigns a new ID and reserves the IDs that immediately follow it. + * + *

Implementations must override this method to assign IDs for types with derived field IDs. + * + * @param oldId an existing field ID + * @param numReserved number of IDs after the new ID that must not be assigned + * @return a new field ID + */ + default int get(int oldId, int numReserved) { + if (numReserved > 0) { + throw new UnsupportedOperationException( + String.format( + "Cannot reserve %s IDs after %s: reserving IDs is not supported", + numReserved, oldId)); + } + + return get(oldId); + } } /** @@ -674,22 +737,40 @@ private ReassignConflictingIds(Set conflictingIds, Set allUsed @Override public int get(int oldId) { - if (conflictingIds.contains(oldId)) { - return nextAvailableId(); + return get(oldId, 0); + } + + @Override + public int get(int oldId, int numReserved) { + // only the reserved IDs are checked because a field that is not conflicting keeps its ID + if (conflictingIds.contains(oldId) || !isRangeAvailable(oldId + 1, oldId + numReserved)) { + return nextAvailableId(numReserved); } else { return oldId; } } - private int nextAvailableId() { + private int nextAvailableId(int numReserved) { int candidateId = nextId.incrementAndGet(); - while (allUsedIds.contains(candidateId)) { + while (!isRangeAvailable(candidateId, candidateId + numReserved)) { candidateId = nextId.incrementAndGet(); } + nextId.addAndGet(numReserved); + return candidateId; } + + private boolean isRangeAvailable(int firstId, int lastId) { + for (int id = firstId; id <= lastId; id += 1) { + if (allUsedIds.contains(id)) { + return false; + } + } + + return true; + } } public static class SchemaVisitor { @@ -753,6 +834,10 @@ public T variant(Types.VariantType variant) { throw new UnsupportedOperationException("Unsupported type: variant"); } + public T file(Types.FileType file, List fieldResults) { + throw new UnsupportedOperationException("Unsupported type: file"); + } + public T primitive(Type.PrimitiveType primitive) { return null; } @@ -766,18 +851,11 @@ public static T visit(Type type, SchemaVisitor visitor) { switch (type.typeId()) { case STRUCT: Types.StructType struct = type.asNestedType().asStructType(); - List results = Lists.newArrayListWithExpectedSize(struct.fields().size()); - for (Types.NestedField field : struct.fields()) { - visitor.beforeField(field); - T result; - try { - result = visit(field.type(), visitor); - } finally { - visitor.afterField(field); - } - results.add(visitor.field(field, result)); - } - return visitor.struct(struct, results); + return visitor.struct(struct, visitFields(struct.fields(), visitor)); + + case FILE: + Types.FileType file = type.asFileType(); + return visitor.file(file, visitFields(file.fields(), visitor)); case LIST: Types.ListType list = type.asNestedType().asListType(); @@ -824,6 +902,21 @@ public static T visit(Type type, SchemaVisitor visitor) { } } + private static List visitFields(List fields, SchemaVisitor visitor) { + List results = Lists.newArrayListWithExpectedSize(fields.size()); + for (Types.NestedField field : fields) { + visitor.beforeField(field); + T result; + try { + result = visit(field.type(), visitor); + } finally { + visitor.afterField(field); + } + results.add(visitor.field(field, result)); + } + return results; + } + public static class CustomOrderSchemaVisitor { public T schema(Schema schema, Supplier structResult) { return null; @@ -849,6 +942,10 @@ public T variant(Types.VariantType variant) { throw new UnsupportedOperationException("Unsupported type: variant"); } + public T file(Types.FileType file, Iterable fieldResults) { + throw new UnsupportedOperationException("Unsupported type: file"); + } + public T primitive(Type.PrimitiveType primitive) { return null; } @@ -906,13 +1003,11 @@ public static T visit(Type type, CustomOrderSchemaVisitor visitor) { switch (type.typeId()) { case STRUCT: Types.StructType struct = type.asNestedType().asStructType(); - List> results = - Lists.newArrayListWithExpectedSize(struct.fields().size()); - for (Types.NestedField field : struct.fields()) { - results.add(new VisitFieldFuture<>(field, visitor)); - } + return visitor.struct(struct, fieldFutures(struct.fields(), visitor)); - return visitor.struct(struct, Iterables.transform(results, VisitFieldFuture::get)); + case FILE: + Types.FileType file = type.asFileType(); + return visitor.file(file, fieldFutures(file.fields(), visitor)); case LIST: Types.ListType list = type.asNestedType().asListType(); @@ -933,6 +1028,15 @@ public static T visit(Type type, CustomOrderSchemaVisitor visitor) { } } + private static Iterable fieldFutures( + List fields, CustomOrderSchemaVisitor visitor) { + List> results = Lists.newArrayListWithExpectedSize(fields.size()); + for (Types.NestedField field : fields) { + results.add(new VisitFieldFuture<>(field, visitor)); + } + return Iterables.transform(results, VisitFieldFuture::get); + } + static int decimalMaxPrecision(int numBytes) { Preconditions.checkArgument( numBytes >= 0 && numBytes < 24, "Unsupported decimal length: %s", numBytes); diff --git a/api/src/main/java/org/apache/iceberg/types/Types.java b/api/src/main/java/org/apache/iceberg/types/Types.java index f082915920ea..b6a71b3fd631 100644 --- a/api/src/main/java/org/apache/iceberg/types/Types.java +++ b/api/src/main/java/org/apache/iceberg/types/Types.java @@ -1155,6 +1155,150 @@ private Map lazyFieldsById() { } } + public static final class FileType extends NestedType { + public static final String NAME = "file"; + public static final int NUM_NESTED_FIELDS = 6; + + private static final String URI = "uri"; + private static final String OFFSET = "offset"; + private static final String SIZE = "size"; + private static final String CONTENT_TYPE = "content_type"; + private static final String CHECKSUM = "checksum"; + private static final String INLINE = "inline"; + + public static FileType of(int enclosingId) { + return new FileType(enclosingId); + } + + private final int enclosingId; + + // lazy values + private transient List fieldList = null; + private transient Map fieldsByName = null; + private transient Map fieldsByLowerCaseName = null; + private transient Map fieldsById = null; + + private FileType(int enclosingId) { + this.enclosingId = enclosingId; + } + + /** Returns the ID of the field that holds this type. */ + public int enclosingId() { + return enclosingId; + } + + @Override + public TypeID typeId() { + return TypeID.FILE; + } + + @Override + public boolean isFileType() { + return true; + } + + @Override + public FileType asFileType() { + return this; + } + + @Override + public List fields() { + if (fieldList == null) { + this.fieldList = + ImmutableList.of( + NestedField.optional(enclosingId + 1, URI, StringType.get()), + NestedField.optional(enclosingId + 2, OFFSET, LongType.get()), + NestedField.optional(enclosingId + 3, SIZE, LongType.get()), + NestedField.optional(enclosingId + 4, CONTENT_TYPE, StringType.get()), + NestedField.optional(enclosingId + 5, CHECKSUM, StringType.get()), + NestedField.optional(enclosingId + 6, INLINE, BinaryType.get())); + } + return fieldList; + } + + public NestedField field(String name) { + return lazyFieldsByName().get(name); + } + + @Override + public NestedField field(int id) { + return lazyFieldsById().get(id); + } + + public NestedField caseInsensitiveField(String name) { + return lazyFieldsByLowerCaseName().get(name.toLowerCase(Locale.ROOT)); + } + + @Override + public Type fieldType(String name) { + NestedField field = field(name); + if (field != null) { + return field.type(); + } + return null; + } + + /** Returns the nested fields of this type as a struct. */ + public StructType asStruct() { + return StructType.of(fields()); + } + + @Override + public String toString() { + return NAME; + } + + @Override + public boolean equals(Object other) { + if (this == other) { + return true; + } else if (!(other instanceof FileType)) { + return false; + } + + return enclosingId == ((FileType) other).enclosingId; + } + + @Override + public int hashCode() { + return Objects.hash(FileType.class, enclosingId); + } + + private Map lazyFieldsByName() { + if (fieldsByName == null) { + ImmutableMap.Builder builder = ImmutableMap.builder(); + for (NestedField field : fields()) { + builder.put(field.name(), field); + } + this.fieldsByName = builder.build(); + } + return fieldsByName; + } + + private Map lazyFieldsByLowerCaseName() { + if (fieldsByLowerCaseName == null) { + ImmutableMap.Builder builder = ImmutableMap.builder(); + for (NestedField field : fields()) { + builder.put(field.name().toLowerCase(Locale.ROOT), field); + } + this.fieldsByLowerCaseName = builder.build(); + } + return fieldsByLowerCaseName; + } + + private Map lazyFieldsById() { + if (fieldsById == null) { + ImmutableMap.Builder builder = ImmutableMap.builder(); + for (NestedField field : fields()) { + builder.put(field.fieldId(), field); + } + this.fieldsById = builder.build(); + } + return fieldsById; + } + } + public static class ListType extends NestedType { public static ListType ofOptional(int elementId, Type elementType) { Preconditions.checkNotNull(elementType, "Element type cannot be null"); diff --git a/api/src/main/java/org/apache/iceberg/util/StructProjection.java b/api/src/main/java/org/apache/iceberg/util/StructProjection.java index 9db90a061cab..d18251b32f55 100644 --- a/api/src/main/java/org/apache/iceberg/util/StructProjection.java +++ b/api/src/main/java/org/apache/iceberg/util/StructProjection.java @@ -121,12 +121,17 @@ private StructProjection(StructType structType, StructType projection, boolean a positionMap[pos] = i; switch (projectedField.type().typeId()) { case STRUCT: + // the data field may be a file when only some of its nested fields are projected nestedProjections[pos] = new StructProjection( - dataField.type().asStructType(), + TypeUtil.asStructType(dataField.type()), projectedField.type().asStructType(), allowMissing); break; + case FILE: + // a projected file is always complete, so its fields need no reordering + nestedProjections[pos] = null; + break; case MAP: MapType projectedMap = projectedField.type().asMapType(); MapType originalMap = dataField.type().asMapType(); diff --git a/api/src/test/java/org/apache/iceberg/TestAccessors.java b/api/src/test/java/org/apache/iceberg/TestAccessors.java index 7b4feb845f12..3eb662030eb4 100644 --- a/api/src/test/java/org/apache/iceberg/TestAccessors.java +++ b/api/src/test/java/org/apache/iceberg/TestAccessors.java @@ -247,4 +247,16 @@ public void testEmptySchema() { Schema emptySchema = new Schema(); assertThat(emptySchema.accessorForField(17)).isNull(); } + + @Test + void fileNestedFields() { + Schema schema = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "photo", Types.FileType.of(2))); + StructLike row = Row.of(1L, Row.of("s3://bucket/key", 4L, 1024L, "image/png", null, null)); + + assertThat(schema.accessorForField(3).get(row)).isEqualTo("s3://bucket/key"); + assertThat(schema.accessorForField(5).get(row)).isEqualTo(1024L); + assertThat(schema.accessorForField(8).get(row)).isNull(); + } } diff --git a/api/src/test/java/org/apache/iceberg/TestSchema.java b/api/src/test/java/org/apache/iceberg/TestSchema.java index 7abc3505d52e..e410ee682991 100644 --- a/api/src/test/java/org/apache/iceberg/TestSchema.java +++ b/api/src/test/java/org/apache/iceberg/TestSchema.java @@ -180,6 +180,32 @@ public void testUnknownSupport() { .doesNotThrowAnyException(); } + @Test + void fileSupport() { + // this needs a different schema because a file reserves the six ids that follow it + Schema schemaWithFile = + new Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.optional(2, "top", Types.FileType.of(2)), + Types.NestedField.optional( + 9, "arr", Types.ListType.ofOptional(10, Types.FileType.of(10)))); + int minVersion = MIN_FORMAT_VERSIONS.get(Type.TypeID.FILE); + + for (int version = 1; version < minVersion; version += 1) { + int unsupportedVersion = version; + assertThatThrownBy(() -> Schema.checkCompatibility(schemaWithFile, unsupportedVersion)) + .isInstanceOf(IllegalStateException.class) + .hasMessage( + "Invalid schema for v%s:\n" + + "- Invalid type for top: file is not supported until v%s\n" + + "- Invalid type for arr.element: file is not supported until v%s", + unsupportedVersion, minVersion, minVersion); + } + + assertThatCode(() -> Schema.checkCompatibility(schemaWithFile, minVersion)) + .doesNotThrowAnyException(); + } + @ParameterizedTest @MethodSource("supportedTypes") public void testTypeSupported(Type type, int supportedVersion) { diff --git a/api/src/test/java/org/apache/iceberg/expressions/TestExpressionBinding.java b/api/src/test/java/org/apache/iceberg/expressions/TestExpressionBinding.java index 24e58ad1e808..ef3d2bc98e39 100644 --- a/api/src/test/java/org/apache/iceberg/expressions/TestExpressionBinding.java +++ b/api/src/test/java/org/apache/iceberg/expressions/TestExpressionBinding.java @@ -100,6 +100,19 @@ public void testCaseSensitiveReference() { .hasMessageContaining("Cannot find field 'X' in struct"); } + @Test + void fileNestedFieldReference() { + StructType struct = + StructType.of( + required(0, "id", Types.LongType.get()), optional(1, "photo", Types.FileType.of(1))); + + Expression bound = Binder.bind(struct, equal("photo.uri", "s3://bucket/key")); + + BoundPredicate predicate = TestHelpers.assertAndUnwrap(bound); + assertThat(predicate.ref().fieldId()).isEqualTo(2); + assertThat(predicate.ref().type()).isEqualTo(Types.StringType.get()); + } + @Test public void testMultipleReferences() { Expression expr = or(and(equal("x", 7), lessThan("y", 100)), greaterThan("z", -100)); diff --git a/api/src/test/java/org/apache/iceberg/types/TestComparators.java b/api/src/test/java/org/apache/iceberg/types/TestComparators.java index 691e3f04a074..8ce7ac681e74 100644 --- a/api/src/test/java/org/apache/iceberg/types/TestComparators.java +++ b/api/src/test/java/org/apache/iceberg/types/TestComparators.java @@ -219,4 +219,18 @@ public void testNested() { TestHelpers.Row.of( "a", TestHelpers.Row.of("b", 1), ImmutableList.of(1, 2), ImmutableMap.of("c", 4))); } + + @Test + public void testFile() { + Comparator comparator = + Comparators.forType( + Types.StructType.of(Types.NestedField.optional(2, "photo", Types.FileType.of(2)))); + + assertComparesCorrectly( + comparator, TestHelpers.Row.of(photo("s3://a")), TestHelpers.Row.of(photo("s3://b"))); + } + + private static StructLike photo(String uri) { + return TestHelpers.Row.of(uri, 0L, 1L, "image/png", "abc", null); + } } diff --git a/api/src/test/java/org/apache/iceberg/types/TestFileType.java b/api/src/test/java/org/apache/iceberg/types/TestFileType.java new file mode 100644 index 000000000000..6d4491646268 --- /dev/null +++ b/api/src/test/java/org/apache/iceberg/types/TestFileType.java @@ -0,0 +1,333 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg.types; + +import static org.apache.iceberg.types.Types.NestedField.optional; +import static org.apache.iceberg.types.Types.NestedField.required; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.iceberg.Schema; +import org.apache.iceberg.StructLike; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet; +import org.junit.jupiter.api.Test; + +class TestFileType { + private static final Types.FileType FILE = Types.FileType.of(5); + + @Test + void nestedFieldsAreDerivedFromTheEnclosingId() { + assertThat(FILE.fields()) + .containsExactly( + optional(6, "uri", Types.StringType.get()), + optional(7, "offset", Types.LongType.get()), + optional(8, "size", Types.LongType.get()), + optional(9, "content_type", Types.StringType.get()), + optional(10, "checksum", Types.StringType.get()), + optional(11, "inline", Types.BinaryType.get())); + assertThat(FILE.enclosingId()).isEqualTo(5); + assertThat(Types.FileType.NUM_NESTED_FIELDS).isEqualTo(FILE.fields().size()); + } + + @Test + void isItsOwnNestedType() { + assertThat(FILE.typeId()).isEqualTo(Type.TypeID.FILE); + assertThat(FILE.isNestedType()).isTrue(); + assertThat(FILE.asNestedType()).isSameAs(FILE); + + assertThat(FILE.isStructType()).isFalse(); + assertThatThrownBy(FILE::asStructType) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Not a struct type: file"); + assertThat(FILE.asStruct()).isEqualTo(Types.StructType.of(FILE.fields())); + } + + @Test + void isDistinguishableFromAStruct() { + assertThat(FILE.isFileType()).isTrue(); + assertThat(FILE.asFileType()).isSameAs(FILE); + + Types.StructType struct = Types.StructType.of(FILE.fields()); + assertThat(struct.isFileType()).isFalse(); + assertThatThrownBy(struct::asFileType) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageStartingWith("Not a file type:"); + } + + @Test + void persistsAsASingleTypeName() { + assertThat(FILE.toString()).isEqualTo(Types.FileType.NAME).isEqualTo("file"); + assertThat(optional(5, "photo", FILE)).hasToString("5: photo: optional file"); + } + + @Test + void isNotEqualToAStructWithTheSameFields() { + Types.StructType struct = Types.StructType.of(FILE.fields()); + + assertThat(FILE).isNotEqualTo(struct); + assertThat(struct).isNotEqualTo(FILE); + assertThat(FILE.hashCode()).isNotEqualTo(struct.hashCode()); + } + + @Test + void isNotEqualToAFileHeldByADifferentField() { + assertThat(FILE).isEqualTo(Types.FileType.of(5)).isNotEqualTo(Types.FileType.of(12)); + assertThat(FILE.hashCode()).isNotEqualTo(Types.FileType.of(12).hashCode()); + } + + @Test + void freshIdsReserveTheNestedIdBlock() { + Schema schema = + new Schema( + required(1, "id", Types.LongType.get()), + optional(2, "photo", Types.FileType.of(2)), + optional(9, "data", Types.StringType.get())); + + Schema assigned = TypeUtil.assignFreshIds(schema, new AtomicInteger(0)::incrementAndGet); + + assertThat(assigned.findField("id").fieldId()).isEqualTo(1); + assertThat(assigned.findField("photo").fieldId()).isEqualTo(2); + assertThat(assigned.findField("photo").type()).isEqualTo(Types.FileType.of(2)); + assertThat(assigned.findField("photo.uri").fieldId()).isEqualTo(3); + assertThat(assigned.findField("photo.inline").fieldId()).isEqualTo(8); + assertThat(assigned.findField("data").fieldId()).isEqualTo(9); + assertThat(assigned.highestFieldId()).isEqualTo(9); + } + + @Test + void freshIdsHandleAdjacentFileColumns() { + Schema schema = + new Schema( + optional(1, "photo", Types.FileType.of(1)), + optional(8, "thumbnail", Types.FileType.of(8))); + + Schema assigned = TypeUtil.assignFreshIds(schema, new AtomicInteger(0)::incrementAndGet); + + assertThat(assigned.findField("photo").type()).isEqualTo(Types.FileType.of(1)); + assertThat(assigned.findField("thumbnail").type()).isEqualTo(Types.FileType.of(8)); + assertThat(assigned.highestFieldId()).isEqualTo(14); + assertThat(TypeUtil.indexById(assigned.asStruct())).hasSize(14); + } + + @Test + void freshIdsReuseBaseSchemaIdsWithoutReserving() { + Schema base = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "photo", Types.FileType.of(2))); + Schema updated = + new Schema( + required(11, "id", Types.LongType.get()), + optional(12, "photo", Types.FileType.of(12)), + optional(19, "data", Types.StringType.get())); + + Schema assigned = TypeUtil.assignFreshIds(updated, base, new AtomicInteger(8)::incrementAndGet); + + assertThat(assigned.findField("id").fieldId()).isEqualTo(1); + assertThat(assigned.findField("photo").fieldId()).isEqualTo(2); + assertThat(assigned.findField("photo.uri").fieldId()).isEqualTo(3); + assertThat(assigned.findField("data").fieldId()).isEqualTo(9); + } + + @Test + void freshIdsReserveForFilesInListsAndMaps() { + Schema schema = + new Schema( + optional(1, "photos", Types.ListType.ofOptional(2, Types.FileType.of(2))), + optional( + 9, + "byName", + Types.MapType.ofOptional(10, 11, Types.StringType.get(), Types.FileType.of(11)))); + + Schema assigned = TypeUtil.assignFreshIds(schema, new AtomicInteger(0)::incrementAndGet); + + assertThat(assigned.findField("photos.element").type()).isEqualTo(Types.FileType.of(3)); + assertThat(assigned.findField("photos.element.uri").fieldId()).isEqualTo(4); + assertThat(assigned.findField("byName.value").type()).isEqualTo(Types.FileType.of(11)); + assertThat(assigned.findField("byName.value.uri").fieldId()).isEqualTo(12); + assertThat(assigned.highestFieldId()).isEqualTo(17); + assertThat(TypeUtil.indexById(assigned.asStruct())).hasSize(17); + } + + @Test + void freshIdsRejectAnAssignerThatSkipsTheReservedIds() { + Schema schema = new Schema(optional(1, "photo", Types.FileType.of(1))); + AtomicInteger counter = new AtomicInteger(0); + + assertThatThrownBy(() -> TypeUtil.assignFreshIds(schema, () -> counter.addAndGet(10))) + .isInstanceOf(IllegalStateException.class) + .hasMessage("Cannot reserve 6 IDs after 10: assigned 20"); + } + + @Test + void assignedIdsRejectAnAssignerThatCannotReserve() { + Schema schema = new Schema(optional(1, "photo", Types.FileType.of(1))); + + assertThatThrownBy(() -> TypeUtil.assignIds(schema.asStruct(), oldId -> oldId + 10)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("Cannot reserve 6 IDs after 1: reserving IDs is not supported"); + } + + @Test + void reassignedConflictingIdsReserveTheNestedIdBlock() { + List columns = + ImmutableList.of( + required(1, "id", Types.LongType.get()), optional(2, "photo", Types.FileType.of(2))); + + Schema schema = + new Schema( + columns, + TypeUtil.reassignConflictingIds( + ImmutableSet.of(2), ImmutableSet.of(1, 2, 3, 4, 5, 6, 7, 8))); + + Types.NestedField photo = schema.findField("photo"); + assertThat(photo.fieldId()).isEqualTo(9); + assertThat(photo.type()).isEqualTo(Types.FileType.of(9)); + assertThat(schema.findField("photo.uri").fieldId()).isEqualTo(10); + assertThat(schema.findField("photo.inline").fieldId()).isEqualTo(15); + assertThat(schema.idsToReassigned()).containsEntry(2, 9).doesNotContainKey(3); + assertThat(schema.idsToOriginal()).containsEntry(9, 2).doesNotContainKey(10); + } + + @Test + void reassignedConflictingIdsMoveAFileWhenTheNestedIdsAreInUse() { + List columns = + ImmutableList.of( + required(1, "id", Types.LongType.get()), optional(2, "photo", Types.FileType.of(2))); + + // 5 falls inside the derived block 3-8 even though the file's own id is not conflicting + Schema schema = + new Schema(columns, TypeUtil.reassignConflictingIds(ImmutableSet.of(), ImmutableSet.of(5))); + + assertThat(schema.findField("id").fieldId()).isEqualTo(1); + assertThat(schema.findField("photo").type()).isEqualTo(Types.FileType.of(6)); + assertThat(schema.findField("photo.uri").fieldId()).isEqualTo(7); + assertThat(schema.findField("photo.inline").fieldId()).isEqualTo(12); + } + + @Test + void reassignedConflictingIdsKeepAFileWhenOnlyItsOwnIdIsInUse() { + List columns = ImmutableList.of(optional(2, "photo", Types.FileType.of(2))); + + Schema schema = + new Schema(columns, TypeUtil.reassignConflictingIds(ImmutableSet.of(), ImmutableSet.of(2))); + + assertThat(schema.findField("photo").type()).isEqualTo(Types.FileType.of(2)); + assertThat(schema.findField("photo.uri").fieldId()).isEqualTo(3); + assertThat(schema.findField("photo.inline").fieldId()).isEqualTo(8); + } + + @Test + void reassignedConflictingIdsSkipBlocksThatOverlapUsedIds() { + List columns = ImmutableList.of(optional(2, "photo", Types.FileType.of(2))); + + Schema schema = + new Schema( + columns, TypeUtil.reassignConflictingIds(ImmutableSet.of(), ImmutableSet.of(3, 9))); + + assertThat(schema.findField("photo").type()).isEqualTo(Types.FileType.of(10)); + assertThat(schema.findField("photo.uri").fieldId()).isEqualTo(11); + assertThat(schema.findField("photo.inline").fieldId()).isEqualTo(16); + assertThat(TypeUtil.indexById(schema.asStruct()).keySet()).doesNotContain(3, 9); + } + + @Test + void reassignedIdsComeFromTheSourceSchema() { + Schema source = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "photo", Types.FileType.of(2))); + Schema unassigned = + new Schema( + required(11, "id", Types.LongType.get()), optional(12, "photo", Types.FileType.of(12))); + + Schema reassigned = TypeUtil.reassignIds(unassigned, source); + + assertThat(reassigned.asStruct()).isEqualTo(source.asStruct()); + assertThat(reassigned.findField("photo").type()).isEqualTo(Types.FileType.of(2)); + } + + @Test + void refreshedIdsReserveTheNestedIdBlockForNewFileColumns() { + Schema source = new Schema(required(1, "id", Types.LongType.get())); + Schema unassigned = + new Schema( + required(11, "id", Types.LongType.get()), + optional(12, "photo", Types.FileType.of(12)), + optional(19, "data", Types.StringType.get())); + + Schema reassigned = TypeUtil.reassignOrRefreshIds(unassigned, source); + + assertThat(reassigned.findField("id").fieldId()).isEqualTo(1); + Types.NestedField photo = reassigned.findField("photo"); + assertThat(photo.type()).isEqualTo(Types.FileType.of(photo.fieldId())); + assertThat(reassigned.findField("photo.uri").fieldId()).isEqualTo(photo.fieldId() + 1); + assertThat(reassigned.findField("data").fieldId()) + .isEqualTo(photo.fieldId() + Types.FileType.NUM_NESTED_FIELDS + 1); + assertThat(TypeUtil.indexById(reassigned.asStruct())).hasSize(9); + } + + @Test + void nestedFieldsAreNamedWithoutTheListElementSegment() { + Schema schema = + new Schema(optional(9, "photos", Types.ListType.ofOptional(10, Types.FileType.of(10)))); + + assertThat(schema.findField("photos.uri").fieldId()).isEqualTo(11); + assertThat(schema.findField("photos.element.uri").fieldId()).isEqualTo(11); + } + + @Test + void isHashedByItsNestedFieldsRatherThanItsIdentity() { + JavaHash hash = JavaHash.forType(FILE); + + // an identity-hashed row stands in for the row types that do not implement hashCode + assertThat(hash.hash(identityHashedFile("s3://bucket/a"))) + .isEqualTo(hash.hash(identityHashedFile("s3://bucket/a"))); + assertThat(hash.hash(identityHashedFile("s3://bucket/a"))) + .isNotEqualTo(hash.hash(identityHashedFile("s3://bucket/b"))); + } + + private static StructLike identityHashedFile(String uri) { + return new IdentityHashedRow(uri, 0L, 1L, "image/png", "abc", null); + } + + private static class IdentityHashedRow implements StructLike { + private final Object[] values; + + private IdentityHashedRow(Object... values) { + this.values = values; + } + + @Override + public int size() { + return values.length; + } + + @Override + public T get(int pos, Class javaClass) { + return javaClass.cast(values[pos]); + } + + @Override + public void set(int pos, T value) { + values[pos] = value; + } + } +} diff --git a/api/src/test/java/org/apache/iceberg/types/TestReadabilityChecks.java b/api/src/test/java/org/apache/iceberg/types/TestReadabilityChecks.java index 20299cdafce2..1aad90ac240a 100644 --- a/api/src/test/java/org/apache/iceberg/types/TestReadabilityChecks.java +++ b/api/src/test/java/org/apache/iceberg/types/TestReadabilityChecks.java @@ -286,6 +286,27 @@ public void testIncompatibleStructAndPrimitive() { .contains("struct cannot be read as a string"); } + @Test + void incompatibleFileAndStruct() { + Schema fileSchema = new Schema(optional(1, "photo", Types.FileType.of(1))); + Schema structSchema = + new Schema(optional(1, "photo", Types.StructType.of(Types.FileType.of(1).fields()))); + + List asFile = CheckCompatibility.readCompatibilityErrors(fileSchema, structSchema); + assertThat(asFile).hasSize(1); + assertThat(asFile.get(0)) + .as("Should complain that a struct cannot be read as a file") + .contains("cannot be read as a file"); + + List asStruct = CheckCompatibility.readCompatibilityErrors(structSchema, fileSchema); + assertThat(asStruct).hasSize(1); + assertThat(asStruct.get(0)) + .as("Should complain that a file cannot be read as a struct") + .contains("file cannot be read as a struct"); + + assertThat(CheckCompatibility.readCompatibilityErrors(fileSchema, fileSchema)).isEmpty(); + } + @Test public void testMultipleErrors() { // required field is optional and cannot be promoted to the read type diff --git a/api/src/test/java/org/apache/iceberg/types/TestSerializableTypes.java b/api/src/test/java/org/apache/iceberg/types/TestSerializableTypes.java index 2363bd8dc66b..bb0aff3a8982 100644 --- a/api/src/test/java/org/apache/iceberg/types/TestSerializableTypes.java +++ b/api/src/test/java/org/apache/iceberg/types/TestSerializableTypes.java @@ -100,6 +100,17 @@ public void testStructs() throws Exception { .isEqualTo(Types.DecimalType.of(38, 2)); } + @Test + public void testFiles() throws Exception { + Types.FileType file = Types.FileType.of(5); + + Type copy = TestHelpers.roundTripSerialize(file); + + assertThat(copy).as("File serialization should be equal to starting type").isEqualTo(file); + assertThat(copy.isFileType()).as("File serialization should preserve the file type").isTrue(); + assertThat(copy.asFileType().enclosingId()).isEqualTo(5); + } + @Test public void testMaps() throws Exception { Type[] maps = diff --git a/api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java b/api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java index d540d239614e..e98b9ccd1717 100644 --- a/api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java +++ b/api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java @@ -1167,4 +1167,62 @@ public void testReplaceFieldTypesNoMatchReturnsSameSchema() { Schema result = TypeUtil.replaceFieldTypes(schema, ImmutableMap.of(99, Types.LongType.get())); assertThat(result).isSameAs(schema); } + + private static Schema fileSchema() { + return new Schema( + required(1, "id", Types.LongType.get()), optional(2, "photo", Types.FileType.of(2))); + } + + @Test + void reassignDocKeepsTheFileType() { + Schema schema = new Schema(optional(2, "photo", Types.FileType.of(2))); + Schema docSourceSchema = new Schema(optional(2, "photo", Types.FileType.of(2), "image")); + + Schema reassignedSchema = TypeUtil.reassignDoc(schema, docSourceSchema); + + assertThat(reassignedSchema.findField("photo").type()).isEqualTo(Types.FileType.of(2)); + assertThat(reassignedSchema.findField("photo").doc()).isEqualTo("image"); + } + + @Test + void projectKeepsTheFileTypeWhenAllNestedFieldsRemain() { + Schema projected = TypeUtil.project(fileSchema(), Sets.newHashSet(3, 4, 5, 6, 7, 8)); + + assertThat(projected.findField("photo").type()).isEqualTo(Types.FileType.of(2)); + } + + @Test + void projectDropsTheFileTypeWhenNestedFieldsArePruned() { + Schema projected = TypeUtil.project(fileSchema(), Sets.newHashSet(3)); + + assertThat(projected.findField("photo").type().isFileType()).isFalse(); + assertThat(projected.findField("photo").type().asStructType().fields()) + .containsExactly(optional(3, "uri", Types.StringType.get())); + } + + @Test + void selectKeepsTheFileTypeForAWholeFileColumn() { + Schema selected = fileSchema().select("photo"); + + assertThat(selected.findField("photo").type()).isEqualTo(Types.FileType.of(2)); + assertThat(selected.findField("id")).isNull(); + } + + @Test + void selectDropsTheFileTypeForASingleNestedField() { + Schema selected = fileSchema().select("photo.uri"); + + assertThat(selected.findField("photo.uri").fieldId()).isEqualTo(3); + assertThat(selected.findField("photo").type().asStructType().fields()) + .containsExactly(optional(3, "uri", Types.StringType.get())); + } + + @Test + void replaceFieldTypesDropsTheFileTypeWhenANestedFieldChanges() { + Schema replaced = + TypeUtil.replaceFieldTypes(fileSchema(), ImmutableMap.of(3, Types.BinaryType.get())); + + assertThat(replaced.findField("photo").type().isFileType()).isFalse(); + assertThat(replaced.findField("photo.uri").type()).isEqualTo(Types.BinaryType.get()); + } } diff --git a/api/src/test/java/org/apache/iceberg/types/TestTypes.java b/api/src/test/java/org/apache/iceberg/types/TestTypes.java index 2fb224aefb15..4646f02bc2d7 100644 --- a/api/src/test/java/org/apache/iceberg/types/TestTypes.java +++ b/api/src/test/java/org/apache/iceberg/types/TestTypes.java @@ -62,6 +62,11 @@ public void fromTypeName() { assertThat(Types.fromTypeName("geography ( srid:4269 , karney )")) .isEqualTo(Types.GeographyType.of("srid:4269", EdgeAlgorithm.KARNEY)); + // a file is not resolvable by name because its nested ids come from the enclosing field + assertThatExceptionOfType(IllegalArgumentException.class) + .isThrownBy(() -> Types.fromTypeName("file")) + .withMessage("Cannot parse type string to primitive: file"); + assertThatExceptionOfType(IllegalArgumentException.class) .isThrownBy(() -> Types.fromTypeName("abcdefghij")) .withMessage("Cannot parse type string to primitive: abcdefghij"); @@ -92,6 +97,10 @@ public void fromPrimitiveString() { .isThrownBy(() -> Types.fromPrimitiveString("Variant")) .withMessage("Cannot parse type string: variant is not a primitive type"); + assertThatExceptionOfType(IllegalArgumentException.class) + .isThrownBy(() -> Types.fromPrimitiveString("file")) + .withMessage("Cannot parse type string to primitive: file"); + assertThatExceptionOfType(IllegalArgumentException.class) .isThrownBy(() -> Types.fromPrimitiveString("abcdefghij")) .withMessage("Cannot parse type string to primitive: abcdefghij"); diff --git a/api/src/test/java/org/apache/iceberg/util/TestStructProjection.java b/api/src/test/java/org/apache/iceberg/util/TestStructProjection.java index 579918c75505..cc0823035d9b 100644 --- a/api/src/test/java/org/apache/iceberg/util/TestStructProjection.java +++ b/api/src/test/java/org/apache/iceberg/util/TestStructProjection.java @@ -48,6 +48,11 @@ class TestStructProjection { private static final StructType DATA_STRUCT_MISSING_NESTED_FIELD = TypeUtil.selectNot(PROJECTED_STRUCT, Set.of(4)); + private static final StructType FILE_STRUCT = + StructType.of( + NestedField.required(1, "id", Types.LongType.get()), + NestedField.optional(2, "photo", Types.FileType.of(2))); + @Test void createAllowMissingAllowsMissingOptionalFieldInNestedStruct() { Row row = Row.of(1L, Row.of("John", "Doe")); @@ -69,4 +74,21 @@ void createStillThrowsForMissingOptionalFieldInNestedStruct() { .isInstanceOf(IllegalArgumentException.class) .hasMessageStartingWith("Cannot find field"); } + + @Test + void projectsAWholeFileColumn() { + StructType projected = TypeUtil.select(FILE_STRUCT, Set.of(2)); + + assertThat(projected.field("photo").type()).isEqualTo(Types.FileType.of(2)); + assertThat(StructProjection.create(FILE_STRUCT, projected).projectedFields()).isEqualTo(1); + } + + @Test + void projectsASingleNestedFieldOfAFileColumn() { + StructType projected = TypeUtil.select(FILE_STRUCT, Set.of(3)); + + assertThat(projected.field("photo").type()) + .isEqualTo(StructType.of(NestedField.optional(3, "uri", Types.StringType.get()))); + assertThat(StructProjection.create(FILE_STRUCT, projected).projectedFields()).isEqualTo(1); + } } diff --git a/core/src/main/java/org/apache/iceberg/MetricsConfig.java b/core/src/main/java/org/apache/iceberg/MetricsConfig.java index 87dae4c95d2e..cb9a5a6b24f0 100644 --- a/core/src/main/java/org/apache/iceberg/MetricsConfig.java +++ b/core/src/main/java/org/apache/iceberg/MetricsConfig.java @@ -26,6 +26,7 @@ import java.io.Serializable; import java.util.Iterator; +import java.util.List; import java.util.Map; import java.util.Set; import java.util.function.Supplier; @@ -183,7 +184,17 @@ public Set schema(Schema schema, Supplier> structResult) { @Override public Set struct(Types.StructType struct, Iterable> fieldResults) { - Iterator fields = struct.fields().iterator(); + return collectIds(struct.fields(), fieldResults); + } + + @Override + public Set file(Types.FileType file, Iterable> fieldResults) { + return collectIds(file.fields(), fieldResults); + } + + private Set collectIds( + List structFields, Iterable> fieldResults) { + Iterator fields = structFields.iterator(); while (shouldContinue() && fields.hasNext()) { Types.NestedField field = fields.next(); if (metricsEligible(field.type())) { diff --git a/core/src/main/java/org/apache/iceberg/PartitionData.java b/core/src/main/java/org/apache/iceberg/PartitionData.java index b1c6752a4d54..353a1fec4d28 100644 --- a/core/src/main/java/org/apache/iceberg/PartitionData.java +++ b/core/src/main/java/org/apache/iceberg/PartitionData.java @@ -208,6 +208,7 @@ public static Object[] copyData(Types.StructType type, Object[] data) { case STRUCT: case LIST: case MAP: + case FILE: throw new IllegalArgumentException("Unsupported type in partition data: " + type); case BINARY: case FIXED: diff --git a/core/src/main/java/org/apache/iceberg/SchemaParser.java b/core/src/main/java/org/apache/iceberg/SchemaParser.java index 7481af0284f6..647f43e349b2 100644 --- a/core/src/main/java/org/apache/iceberg/SchemaParser.java +++ b/core/src/main/java/org/apache/iceberg/SchemaParser.java @@ -88,6 +88,7 @@ private static void toJson( generator.writeStringField(NAME, field.name()); generator.writeBooleanField(REQUIRED, field.isRequired()); generator.writeFieldName(TYPE); + checkDerivedIds(field.type(), field.fieldId()); toJson(field.type(), generator); if (field.doc() != null) { generator.writeStringField(DOC, field.doc()); @@ -117,6 +118,7 @@ static void toJson(Types.ListType list, JsonGenerator generator) throws IOExcept generator.writeNumberField(ELEMENT_ID, list.elementId()); generator.writeFieldName(ELEMENT); + checkDerivedIds(list.elementType(), list.elementId()); toJson(list.elementType(), generator); generator.writeBooleanField(ELEMENT_REQUIRED, !list.isElementOptional()); @@ -130,18 +132,30 @@ static void toJson(Types.MapType map, JsonGenerator generator) throws IOExceptio generator.writeNumberField(KEY_ID, map.keyId()); generator.writeFieldName(KEY); + checkDerivedIds(map.keyType(), map.keyId()); toJson(map.keyType(), generator); generator.writeNumberField(VALUE_ID, map.valueId()); generator.writeFieldName(VALUE); + checkDerivedIds(map.valueType(), map.valueId()); toJson(map.valueType(), generator); generator.writeBooleanField(VALUE_REQUIRED, !map.isValueOptional()); generator.writeEndObject(); } + private static void checkDerivedIds(Type type, int enclosingId) { + if (type.isFileType()) { + Preconditions.checkArgument( + type.asFileType().enclosingId() == enclosingId, + "Invalid file type: nested field IDs are derived from %s, not %s", + enclosingId, + type.asFileType().enclosingId()); + } + } + static void toJson(Type type, JsonGenerator generator) throws IOException { - if (type.isPrimitiveType() || type.isVariantType()) { + if (type.isPrimitiveType() || type.isVariantType() || type.isFileType()) { generator.writeString(type.toString()); } else { Type.NestedType nested = type.asNestedType(); @@ -176,8 +190,19 @@ public static String toJson(Schema schema, boolean pretty) { } private static Type typeFromJson(JsonNode json) { + return typeFromJson(json, null); + } + + private static Type typeFromJson(JsonNode json, Integer enclosingId) { if (json.isTextual()) { - return Types.fromTypeName(json.asText()); + String typeName = json.asText(); + if (Types.FileType.NAME.equalsIgnoreCase(typeName)) { + Preconditions.checkArgument( + enclosingId != null, "Cannot parse file type without an enclosing field ID"); + return Types.FileType.of(enclosingId); + } + + return Types.fromTypeName(typeName); } else if (json.isObject()) { JsonNode typeObj = json.get(TYPE); if (typeObj != null) { @@ -232,7 +257,7 @@ private static Types.StructType structFromJson(JsonNode json) { int id = JsonUtil.getInt(ID, field); String name = JsonUtil.getString(NAME, field); - Type type = typeFromJson(JsonUtil.get(TYPE, field)); + Type type = typeFromJson(JsonUtil.get(TYPE, field), id); Literal initialDefault = defaultFromJson(INITIAL_DEFAULT, type, field); Literal writeDefault = defaultFromJson(WRITE_DEFAULT, type, field); @@ -254,7 +279,7 @@ private static Types.StructType structFromJson(JsonNode json) { private static Types.ListType listFromJson(JsonNode json) { int elementId = JsonUtil.getInt(ELEMENT_ID, json); - Type elementType = typeFromJson(JsonUtil.get(ELEMENT, json)); + Type elementType = typeFromJson(JsonUtil.get(ELEMENT, json), elementId); boolean isRequired = JsonUtil.getBool(ELEMENT_REQUIRED, json); if (isRequired) { @@ -266,10 +291,10 @@ private static Types.ListType listFromJson(JsonNode json) { private static Types.MapType mapFromJson(JsonNode json) { int keyId = JsonUtil.getInt(KEY_ID, json); - Type keyType = typeFromJson(JsonUtil.get(KEY, json)); + Type keyType = typeFromJson(JsonUtil.get(KEY, json), keyId); int valueId = JsonUtil.getInt(VALUE_ID, json); - Type valueType = typeFromJson(JsonUtil.get(VALUE, json)); + Type valueType = typeFromJson(JsonUtil.get(VALUE, json), valueId); boolean isRequired = JsonUtil.getBool(VALUE_REQUIRED, json); diff --git a/core/src/main/java/org/apache/iceberg/SchemaUpdate.java b/core/src/main/java/org/apache/iceberg/SchemaUpdate.java index 1fa6ebbe8fef..b676686aaff3 100644 --- a/core/src/main/java/org/apache/iceberg/SchemaUpdate.java +++ b/core/src/main/java/org/apache/iceberg/SchemaUpdate.java @@ -63,6 +63,7 @@ class SchemaUpdate implements UpdateSchema { private final Map addedNameToId = Maps.newHashMap(); private final Multimap moves = Multimaps.newListMultimap(Maps.newHashMap(), Lists::newArrayList); + private final TypeUtil.NextID nextColumnId = this::assignNewColumnId; private int lastColumnId; private boolean allowIncompatibleChanges = false; private Set identifierFieldNames; @@ -163,7 +164,7 @@ private void internalAddColumn( fullName); // assign new IDs in order - int newId = assignNewColumnId(); + int newId = assignNewColumnId(type); // update tracking for moves addedNameToId.put(caseSensitivityAwareName(fullName), newId); @@ -176,7 +177,7 @@ private void internalAddColumn( .withName(name) .isOptional(isOptional) .withId(newId) - .ofType(TypeUtil.assignFreshIds(type, this::assignNewColumnId)) + .ofType(assignedType(type, newId)) .withDoc(doc) .withInitialDefault(defaultValue) .withWriteDefault(defaultValue) @@ -186,10 +187,25 @@ private void internalAddColumn( parentToAddedIds.put(parentId, newId); } + private int assignNewColumnId(Type type) { + return type.isFileType() + ? nextColumnId.get(Types.FileType.NUM_NESTED_FIELDS) + : nextColumnId.get(); + } + + private Type assignedType(Type type, int fieldId) { + if (type.isFileType()) { + return Types.FileType.of(fieldId); + } + + return TypeUtil.assignFreshIds(type, nextColumnId); + } + @Override public UpdateSchema deleteColumn(String name) { Types.NestedField field = findField(name); Preconditions.checkArgument(field != null, "Cannot delete missing column: %s", name); + checkNotNestedInFile(name, field.fieldId()); Preconditions.checkArgument( !parentToAddedIds.containsKey(field.fieldId()), "Cannot delete a column that has additions: %s", @@ -205,6 +221,7 @@ public UpdateSchema deleteColumn(String name) { public UpdateSchema renameColumn(String name, String newName) { Types.NestedField field = findField(name); Preconditions.checkArgument(field != null, "Cannot rename missing column: %s", name); + checkNotNestedInFile(name, field.fieldId()); Preconditions.checkArgument(newName != null, "Cannot rename a column to null"); Preconditions.checkArgument( !deletes.contains(field.fieldId()), @@ -241,6 +258,7 @@ public UpdateSchema makeColumnOptional(String name) { private void internalUpdateColumnRequirement(String name, boolean isOptional) { Types.NestedField field = findForUpdate(name); Preconditions.checkArgument(field != null, "Cannot update missing column: %s", name); + checkNotNestedInFile(name, field.fieldId()); if ((!isOptional && field.isRequired()) || (isOptional && field.isOptional())) { // if the change is a noop, allow it even if allowIncompatibleChanges is false @@ -273,6 +291,7 @@ private void internalUpdateColumnRequirement(String name, boolean isOptional) { public UpdateSchema updateColumn(String name, Type.PrimitiveType newType) { Types.NestedField field = findForUpdate(name); Preconditions.checkArgument(field != null, "Cannot update missing column: %s", name); + checkNotNestedInFile(name, field.fieldId()); Preconditions.checkArgument( !deletes.contains(field.fieldId()), "Cannot update a column that will be deleted: %s", @@ -301,6 +320,7 @@ public UpdateSchema updateColumn(String name, Type.PrimitiveType newType) { public UpdateSchema updateColumnDoc(String name, String doc) { Types.NestedField field = findForUpdate(name); Preconditions.checkArgument(field != null, "Cannot update missing column: %s", name); + checkNotNestedInFile(name, field.fieldId()); Preconditions.checkArgument( !deletes.contains(field.fieldId()), "Cannot update a column that will be deleted: %s", @@ -322,6 +342,7 @@ public UpdateSchema updateColumnDoc(String name, String doc) { public UpdateSchema updateColumnDefault(String name, Literal newDefault) { Types.NestedField field = findForUpdate(name); Preconditions.checkArgument(field != null, "Cannot update missing column: %s", name); + checkNotNestedInFile(name, field.fieldId()); Preconditions.checkArgument( !deletes.contains(field.fieldId()), "Cannot update a column that will be deleted: %s", @@ -396,6 +417,15 @@ private boolean isAdded(String name) { return addedNameToId.containsKey(caseSensitivityAwareName(name)); } + private void checkNotNestedInFile(String name, int fieldId) { + Integer parentId = idToParent.get(fieldId); + Types.NestedField parent = parentId != null ? schema.findField(parentId) : null; + Preconditions.checkArgument( + parent == null || !parent.type().isFileType(), + "Cannot change a nested field of a file column: %s", + name); + } + private Types.NestedField findForUpdate(String name) { Types.NestedField existing = findField(name); if (existing != null) { @@ -653,6 +683,19 @@ public Type struct(Types.StructType struct, List fieldResults) { return struct; } + @Override + public Type file(Types.FileType file, List fieldResults) { + for (int i = 0; i < fieldResults.size(); i += 1) { + Types.NestedField field = file.fields().get(i); + Preconditions.checkArgument( + fieldResults.get(i) == field.type() && updates.get(field.fieldId()) == null, + "Cannot change the nested fields of a file column: %s", + file); + } + + return file; + } + @Override public Type field(Types.NestedField field, Type fieldResult) { // the API validates deletes, updates, and additions don't conflict diff --git a/core/src/main/java/org/apache/iceberg/SingleValueParser.java b/core/src/main/java/org/apache/iceberg/SingleValueParser.java index c7f07ea1a2d4..bd64e7763b6b 100644 --- a/core/src/main/java/org/apache/iceberg/SingleValueParser.java +++ b/core/src/main/java/org/apache/iceberg/SingleValueParser.java @@ -180,6 +180,8 @@ public static Object fromJson(Type type, JsonNode defaultValue) { return mapFromJson(type, defaultValue); case STRUCT: return structFromJson(type, defaultValue); + case FILE: + return fromJson(type.asFileType().asStruct(), defaultValue); default: throw new UnsupportedOperationException(String.format("Type: %s is not supported", type)); } @@ -410,6 +412,9 @@ public static void toJson(Type type, Object defaultValue, JsonGenerator generato } generator.writeEndObject(); break; + case FILE: + toJson(type.asFileType().asStruct(), defaultValue, generator); + break; default: throw new UnsupportedOperationException(String.format("Type: %s is not supported", type)); } diff --git a/core/src/main/java/org/apache/iceberg/avro/AvroSchemaWithTypeVisitor.java b/core/src/main/java/org/apache/iceberg/avro/AvroSchemaWithTypeVisitor.java index 45892d3de151..339bad4bc40f 100644 --- a/core/src/main/java/org/apache/iceberg/avro/AvroSchemaWithTypeVisitor.java +++ b/core/src/main/java/org/apache/iceberg/avro/AvroSchemaWithTypeVisitor.java @@ -24,6 +24,7 @@ import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.TypeUtil; import org.apache.iceberg.types.Types; public abstract class AvroSchemaWithTypeVisitor { @@ -35,7 +36,8 @@ public static T visit( public static T visit(Type iType, Schema schema, AvroSchemaWithTypeVisitor visitor) { switch (schema.getType()) { case RECORD: - return visitRecord(iType != null ? iType.asStructType() : null, schema, visitor); + // a file is stored as a record of its nested fields + return visitRecord(iType != null ? TypeUtil.asStructType(iType) : null, schema, visitor); case UNION: return visitUnion(iType, schema, visitor); diff --git a/core/src/main/java/org/apache/iceberg/avro/AvroWithPartnerVisitor.java b/core/src/main/java/org/apache/iceberg/avro/AvroWithPartnerVisitor.java index 83ddc9be5e29..f208348ace85 100644 --- a/core/src/main/java/org/apache/iceberg/avro/AvroWithPartnerVisitor.java +++ b/core/src/main/java/org/apache/iceberg/avro/AvroWithPartnerVisitor.java @@ -24,6 +24,7 @@ import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.TypeUtil; import org.apache.iceberg.types.Types; public class AvroWithPartnerVisitor { @@ -46,7 +47,7 @@ public static FieldIDAccessors get() { @Override public Type fieldPartner(Type partner, Integer fieldId, String name) { - Types.NestedField field = partner.asStructType().field(fieldId); + Types.NestedField field = TypeUtil.asStructType(partner).field(fieldId); return field != null ? field.type() : null; } diff --git a/core/src/main/java/org/apache/iceberg/avro/AvroWithTypeByStructureVisitor.java b/core/src/main/java/org/apache/iceberg/avro/AvroWithTypeByStructureVisitor.java index 27b7ca6842a7..92702a873562 100644 --- a/core/src/main/java/org/apache/iceberg/avro/AvroWithTypeByStructureVisitor.java +++ b/core/src/main/java/org/apache/iceberg/avro/AvroWithTypeByStructureVisitor.java @@ -19,6 +19,7 @@ package org.apache.iceberg.avro; import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.TypeUtil; import org.apache.iceberg.types.Types; import org.apache.iceberg.util.Pair; @@ -55,7 +56,7 @@ protected Type mapValueType(Type mapType) { @Override protected Pair fieldNameAndType(Type structType, int pos) { - Types.NestedField field = structType.asStructType().fields().get(pos); + Types.NestedField field = TypeUtil.asStructType(structType).fields().get(pos); return Pair.of(field.name(), field.type()); } diff --git a/core/src/main/java/org/apache/iceberg/avro/BuildAvroProjection.java b/core/src/main/java/org/apache/iceberg/avro/BuildAvroProjection.java index f4dd2f41302d..f8754dbc255a 100644 --- a/core/src/main/java/org/apache/iceberg/avro/BuildAvroProjection.java +++ b/core/src/main/java/org/apache/iceberg/avro/BuildAvroProjection.java @@ -29,6 +29,7 @@ import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.TypeUtil; import org.apache.iceberg.types.Types; /** @@ -61,7 +62,7 @@ public Schema record(Schema record, List names, Iterable s "Cannot project non-struct: %s", current); - Types.StructType struct = current.asNestedType().asStructType(); + Types.StructType struct = TypeUtil.asStructType(current); boolean hasChange = false; List fields = record.getFields(); @@ -132,7 +133,7 @@ public Schema record(Schema record, List names, Iterable s @Override public Schema.Field field(Schema.Field field, Supplier fieldResult) { - Types.StructType struct = current.asNestedType().asStructType(); + Types.StructType struct = TypeUtil.asStructType(current); int fieldId = AvroSchemaUtil.getFieldId(field); Types.NestedField expectedField = struct.field(fieldId); diff --git a/core/src/main/java/org/apache/iceberg/avro/GenericAvroReader.java b/core/src/main/java/org/apache/iceberg/avro/GenericAvroReader.java index fc2d44f47060..58d7cd8a44d1 100644 --- a/core/src/main/java/org/apache/iceberg/avro/GenericAvroReader.java +++ b/core/src/main/java/org/apache/iceberg/avro/GenericAvroReader.java @@ -31,6 +31,7 @@ import org.apache.iceberg.common.DynClasses; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.TypeUtil; import org.apache.iceberg.types.Types; import org.apache.iceberg.util.Pair; @@ -112,7 +113,7 @@ public ValueReader record(Type partner, Schema record, List> f return ValueReaders.skipStruct(fieldResults); } - Types.StructType expected = partner.asStructType(); + Types.StructType expected = TypeUtil.asStructType(partner); List>> readPlan = ValueReaders.buildReadPlan(expected, record, fieldResults, idToConstant); diff --git a/core/src/main/java/org/apache/iceberg/avro/InternalReader.java b/core/src/main/java/org/apache/iceberg/avro/InternalReader.java index af3c4f1a822b..d68554732980 100644 --- a/core/src/main/java/org/apache/iceberg/avro/InternalReader.java +++ b/core/src/main/java/org/apache/iceberg/avro/InternalReader.java @@ -31,6 +31,7 @@ import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.TypeUtil; import org.apache.iceberg.types.Types; import org.apache.iceberg.util.Pair; @@ -116,7 +117,7 @@ public ValueReader record( return ValueReaders.skipStruct(fieldResults); } - Types.StructType expected = partner.second().asStructType(); + Types.StructType expected = TypeUtil.asStructType(partner.second()); List>> readPlan = ValueReaders.buildReadPlan(expected, record, fieldResults, idToConstant); @@ -243,7 +244,7 @@ public static AccessByID instance() { @Override public Pair fieldPartner( Pair partner, Integer fieldId, String name) { - Types.NestedField field = partner.second().asStructType().field(fieldId); + Types.NestedField field = TypeUtil.asStructType(partner.second()).field(fieldId); return field != null ? Pair.of(field.fieldId(), field.type()) : null; } diff --git a/core/src/main/java/org/apache/iceberg/avro/NameMappingWithAvroSchema.java b/core/src/main/java/org/apache/iceberg/avro/NameMappingWithAvroSchema.java index 96892ee9c008..485f21ee22b9 100644 --- a/core/src/main/java/org/apache/iceberg/avro/NameMappingWithAvroSchema.java +++ b/core/src/main/java/org/apache/iceberg/avro/NameMappingWithAvroSchema.java @@ -25,6 +25,7 @@ import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.TypeUtil; import org.apache.iceberg.types.Types; public class NameMappingWithAvroSchema extends AvroWithTypeByStructureVisitor { @@ -34,7 +35,7 @@ public MappedFields record( List fields = Lists.newArrayListWithExpectedSize(fieldResults.size()); for (int i = 0; i < fieldResults.size(); i += 1) { - Types.NestedField field = struct.asStructType().fields().get(i); + Types.NestedField field = TypeUtil.asStructType(struct).fields().get(i); MappedFields result = fieldResults.get(i); fields.add(MappedField.of(field.fieldId(), field.name(), result)); } diff --git a/core/src/main/java/org/apache/iceberg/avro/TypeToSchema.java b/core/src/main/java/org/apache/iceberg/avro/TypeToSchema.java index d46821df38ff..33fe83d9a7e5 100644 --- a/core/src/main/java/org/apache/iceberg/avro/TypeToSchema.java +++ b/core/src/main/java/org/apache/iceberg/avro/TypeToSchema.java @@ -98,18 +98,27 @@ void cacheSchema(Type struct, Schema schema) { @Override public Schema struct(Types.StructType struct, List fieldSchemas) { + return recordFor(struct, struct, fieldSchemas); + } + + @Override + public Schema file(Types.FileType file, List fieldSchemas) { + return recordFor(file, file.asStruct(), fieldSchemas); + } + + private Schema recordFor(Type type, Types.StructType structView, List fieldSchemas) { + List structFields = structView.fields(); Integer fieldId = fieldIds.peek(); - String recordName = namesFunction.apply(fieldId, struct); + String recordName = namesFunction.apply(fieldId, structView); if (recordName == null) { recordName = "r" + fieldId; } - Schema recordSchema = lookupSchema(struct, recordName); + Schema recordSchema = lookupSchema(type, recordName); if (recordSchema != null) { return recordSchema; } - List structFields = struct.fields(); List fields = Lists.newArrayListWithExpectedSize(fieldSchemas.size()); for (int i = 0; i < structFields.size(); i += 1) { Types.NestedField structField = structFields.get(i); @@ -131,7 +140,7 @@ public Schema struct(Types.StructType struct, List fieldSchemas) { recordSchema = Schema.createRecord(recordName, null, null, false, fields); - cacheSchema(struct, recordName, recordSchema); + cacheSchema(type, recordName, recordSchema); return recordSchema; } diff --git a/core/src/main/java/org/apache/iceberg/data/avro/PlannedDataReader.java b/core/src/main/java/org/apache/iceberg/data/avro/PlannedDataReader.java index 747907a2fb97..6e888c94a210 100644 --- a/core/src/main/java/org/apache/iceberg/data/avro/PlannedDataReader.java +++ b/core/src/main/java/org/apache/iceberg/data/avro/PlannedDataReader.java @@ -35,6 +35,7 @@ import org.apache.iceberg.data.GenericDataUtil; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.TypeUtil; import org.apache.iceberg.types.Types; import org.apache.iceberg.util.Pair; @@ -96,7 +97,7 @@ public ValueReader record(Type partner, Schema record, List> f return ValueReaders.skipStruct(fieldReaders); } - Types.StructType expected = partner.asStructType(); + Types.StructType expected = TypeUtil.asStructType(partner); List>> readPlan = ValueReaders.buildReadPlan( expected, record, fieldReaders, idToConstant, GenericDataUtil::internalToGeneric); diff --git a/core/src/main/java/org/apache/iceberg/mapping/MappingUtil.java b/core/src/main/java/org/apache/iceberg/mapping/MappingUtil.java index 72b2a6a783bf..e2685f917528 100644 --- a/core/src/main/java/org/apache/iceberg/mapping/MappingUtil.java +++ b/core/src/main/java/org/apache/iceberg/mapping/MappingUtil.java @@ -276,10 +276,20 @@ public MappedFields schema(Schema schema, MappedFields structResult) { @Override public MappedFields struct(Types.StructType struct, List fieldResults) { + return mapFields(struct.fields(), fieldResults); + } + + @Override + public MappedFields file(Types.FileType file, List fieldResults) { + return mapFields(file.fields(), fieldResults); + } + + private MappedFields mapFields( + List structFields, List fieldResults) { List fields = Lists.newArrayListWithExpectedSize(fieldResults.size()); for (int i = 0; i < fieldResults.size(); i += 1) { - Types.NestedField field = struct.fields().get(i); + Types.NestedField field = structFields.get(i); MappedFields result = fieldResults.get(i); fields.add(MappedField.of(field.fieldId(), field.name(), result)); } diff --git a/core/src/main/java/org/apache/iceberg/schema/SchemaWithPartnerVisitor.java b/core/src/main/java/org/apache/iceberg/schema/SchemaWithPartnerVisitor.java index 694bfb2f6242..a7316481ac2c 100644 --- a/core/src/main/java/org/apache/iceberg/schema/SchemaWithPartnerVisitor.java +++ b/core/src/main/java/org/apache/iceberg/schema/SchemaWithPartnerVisitor.java @@ -50,22 +50,12 @@ public static T visit( switch (type.typeId()) { case STRUCT: Types.StructType struct = type.asNestedType().asStructType(); - List results = Lists.newArrayListWithExpectedSize(struct.fields().size()); - for (Types.NestedField field : struct.fields()) { - P fieldPartner = - partner != null - ? accessors.fieldPartner(partner, field.fieldId(), field.name()) - : null; - visitor.beforeField(field, fieldPartner); - T result; - try { - result = visit(field.type(), fieldPartner, visitor, accessors); - } finally { - visitor.afterField(field, fieldPartner); - } - results.add(visitor.field(field, fieldPartner, result)); - } - return visitor.struct(struct, partner, results); + return visitor.struct( + struct, partner, visitFields(struct.fields(), partner, visitor, accessors)); + + case FILE: + Types.FileType file = type.asFileType(); + return visitor.file(file, partner, visitFields(file.fields(), partner, visitor, accessors)); case LIST: Types.ListType list = type.asNestedType().asListType(); @@ -115,6 +105,27 @@ public static T visit( } } + private static List visitFields( + List fields, + P partner, + SchemaWithPartnerVisitor visitor, + PartnerAccessors

accessors) { + List results = Lists.newArrayListWithExpectedSize(fields.size()); + for (Types.NestedField field : fields) { + P fieldPartner = + partner != null ? accessors.fieldPartner(partner, field.fieldId(), field.name()) : null; + visitor.beforeField(field, fieldPartner); + T result; + try { + result = visit(field.type(), fieldPartner, visitor, accessors); + } finally { + visitor.afterField(field, fieldPartner); + } + results.add(visitor.field(field, fieldPartner, result)); + } + return results; + } + public void beforeField(Types.NestedField field, P partnerField) {} public void afterField(Types.NestedField field, P partnerField) {} @@ -167,6 +178,10 @@ public R variant(Types.VariantType variant, P partner) { throw new UnsupportedOperationException("Unsupported type: variant"); } + public R file(Types.FileType file, P partner, List fieldResults) { + throw new UnsupportedOperationException("Unsupported type: file"); + } + public R primitive(Type.PrimitiveType primitive, P partner) { return null; } diff --git a/core/src/main/java/org/apache/iceberg/schema/UnionByNameVisitor.java b/core/src/main/java/org/apache/iceberg/schema/UnionByNameVisitor.java index c3b9a50b2081..2c81e62ec55b 100644 --- a/core/src/main/java/org/apache/iceberg/schema/UnionByNameVisitor.java +++ b/core/src/main/java/org/apache/iceberg/schema/UnionByNameVisitor.java @@ -83,7 +83,7 @@ public Boolean struct( } List fields = struct.fields(); - Types.StructType partnerStruct = findFieldType(partnerId).asStructType(); + Types.StructType partnerStruct = findFieldsByName(partnerId); IntStream.range(0, missingPositions.size()) .forEach( pos -> { @@ -103,6 +103,12 @@ public Boolean struct( return false; } + @Override + public Boolean file(Types.FileType file, Integer partnerId, List missingPositions) { + // the nested fields of a file are derived, so there is nothing to union + return partnerId == null; + } + @Override public Boolean field(Types.NestedField field, Integer partnerId, Boolean isFieldMissing) { return partnerId == null; @@ -160,6 +166,10 @@ private Type findFieldType(int fieldId) { } } + private Types.StructType findFieldsByName(int fieldId) { + return TypeUtil.asStructType(findFieldType(fieldId)); + } + private void addColumn(int parentId, Types.NestedField field) { String parentName = partnerSchema.findColumnName(parentId); String fullName = (parentName != null ? parentName + "." : "") + field.name(); @@ -230,7 +240,7 @@ public Integer fieldPartner(Integer partnerFieldId, int fieldId, String name) { if (partnerFieldId == -1) { struct = partnerSchema.asStruct(); } else { - struct = partnerSchema.findField(partnerFieldId).type().asStructType(); + struct = TypeUtil.asStructType(partnerSchema.findField(partnerFieldId).type()); } Types.NestedField field = diff --git a/core/src/main/java/org/apache/iceberg/types/FixupTypes.java b/core/src/main/java/org/apache/iceberg/types/FixupTypes.java index 1e4c0b597a6a..2b22f9a50341 100644 --- a/core/src/main/java/org/apache/iceberg/types/FixupTypes.java +++ b/core/src/main/java/org/apache/iceberg/types/FixupTypes.java @@ -79,6 +79,12 @@ public Type struct(Types.StructType struct, Iterable fieldTypes) { return struct; } + @Override + public Type file(Types.FileType file, Iterable fieldTypes) { + // the nested fields of a file are derived, so their types cannot be fixed up + return file; + } + @Override public Type field(Types.NestedField field, Supplier future) { Preconditions.checkArgument(sourceType.isStructType(), "Not a struct: %s", sourceType); diff --git a/core/src/test/java/org/apache/iceberg/TestFileTypeSchemaParser.java b/core/src/test/java/org/apache/iceberg/TestFileTypeSchemaParser.java new file mode 100644 index 000000000000..01487d0b38d5 --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/TestFileTypeSchemaParser.java @@ -0,0 +1,166 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg; + +import static org.apache.iceberg.types.Types.NestedField.optional; +import static org.apache.iceberg.types.Types.NestedField.required; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import org.apache.iceberg.types.Types; +import org.junit.jupiter.api.Test; + +class TestFileTypeSchemaParser { + @Test + void roundTripsAsATopLevelColumn() { + Schema schema = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "photo", Types.FileType.of(2))); + + String json = SchemaParser.toJson(schema); + assertThat(json).contains("\"name\":\"photo\",\"required\":false,\"type\":\"file\""); + + assertThat(SchemaParser.fromJson(json).asStruct()).isEqualTo(schema.asStruct()); + } + + @Test + void roundTripsNestedInAStruct() { + Schema schema = + new Schema( + optional( + 1, + "media", + Types.StructType.of( + optional(2, "photo", Types.FileType.of(2)), + optional(9, "caption", Types.StringType.get())))); + + Schema parsed = SchemaParser.fromJson(SchemaParser.toJson(schema)); + + assertThat(parsed.asStruct()).isEqualTo(schema.asStruct()); + assertThat(parsed.findField("media.photo").type()).isEqualTo(Types.FileType.of(2)); + assertThat(parsed.findField("media.photo.uri").fieldId()).isEqualTo(3); + } + + @Test + void roundTripsAsAListElement() { + Schema schema = + new Schema(optional(1, "photos", Types.ListType.ofOptional(2, Types.FileType.of(2)))); + + Schema parsed = SchemaParser.fromJson(SchemaParser.toJson(schema)); + + assertThat(parsed.asStruct()).isEqualTo(schema.asStruct()); + assertThat(parsed.findField("photos.element").type()).isEqualTo(Types.FileType.of(2)); + } + + @Test + void roundTripsAsAMapValue() { + Schema schema = + new Schema( + optional( + 1, + "byName", + Types.MapType.ofOptional(2, 3, Types.StringType.get(), Types.FileType.of(3)))); + + Schema parsed = SchemaParser.fromJson(SchemaParser.toJson(schema)); + + assertThat(parsed.asStruct()).isEqualTo(schema.asStruct()); + assertThat(parsed.findField("byName.value").type()).isEqualTo(Types.FileType.of(3)); + } + + @Test + void roundTripsAsAMapKey() { + Schema schema = + new Schema( + optional( + 1, + "byFile", + Types.MapType.ofOptional(2, 20, Types.FileType.of(2), Types.StringType.get()))); + + Schema parsed = SchemaParser.fromJson(SchemaParser.toJson(schema)); + + assertThat(parsed.asStruct()).isEqualTo(schema.asStruct()); + assertThat(parsed.findField("byFile.key").type()).isEqualTo(Types.FileType.of(2)); + } + + @Test + void acceptsAnyCaseAndWritesTheCanonicalName() { + String json = + "{\"type\":\"struct\",\"schema-id\":0,\"fields\":[" + + "{\"id\":5,\"name\":\"photo\",\"required\":false,\"type\":\"FILE\"}]}"; + + Schema parsed = SchemaParser.fromJson(json); + + assertThat(parsed.findField("photo").type()).isEqualTo(Types.FileType.of(5)); + assertThat(SchemaParser.toJson(parsed)).contains("\"type\":\"file\""); + } + + @Test + void rejectsAFileTypeWithoutAnEnclosingId() { + assertThatThrownBy(() -> SchemaParser.fromJson("\"file\"")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Cannot parse file type without an enclosing field ID"); + } + + @Test + void rejectsWritingUnderivedNestedIds() { + Schema schema = new Schema(optional(5, "photo", Types.FileType.of(9))); + + assertThatThrownBy(() -> SchemaParser.toJson(schema)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Invalid file type: nested field IDs are derived from 5, not 9"); + } + + @Test + void rejectsWritingUnderivedNestedIdsInAList() { + Schema schema = + new Schema(optional(1, "photos", Types.ListType.ofOptional(2, Types.FileType.of(9)))); + + assertThatThrownBy(() -> SchemaParser.toJson(schema)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Invalid file type: nested field IDs are derived from 2, not 9"); + } + + @Test + void rejectsWritingUnderivedNestedIdsInAMapKey() { + Schema schema = + new Schema( + optional( + 1, + "byFile", + Types.MapType.ofOptional(2, 20, Types.FileType.of(9), Types.StringType.get()))); + + assertThatThrownBy(() -> SchemaParser.toJson(schema)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Invalid file type: nested field IDs are derived from 2, not 9"); + } + + @Test + void rejectsWritingUnderivedNestedIdsInAMapValue() { + Schema schema = + new Schema( + optional( + 1, + "byName", + Types.MapType.ofOptional(2, 3, Types.StringType.get(), Types.FileType.of(9)))); + + assertThatThrownBy(() -> SchemaParser.toJson(schema)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Invalid file type: nested field IDs are derived from 3, not 9"); + } +} diff --git a/core/src/test/java/org/apache/iceberg/TestFileTypeTableMetadata.java b/core/src/test/java/org/apache/iceberg/TestFileTypeTableMetadata.java new file mode 100644 index 000000000000..820f8f145ff0 --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/TestFileTypeTableMetadata.java @@ -0,0 +1,62 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg; + +import static org.apache.iceberg.types.Types.NestedField.optional; +import static org.apache.iceberg.types.Types.NestedField.required; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.types.Types; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +class TestFileTypeTableMetadata { + private static final Schema SCHEMA = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "photo", Types.FileType.of(2))); + + @ParameterizedTest + @ValueSource(ints = {1, 2, 3}) + void rejectsAFileColumnBeforeFormatVersion4(int formatVersion) { + assertThatThrownBy(() -> newTableMetadata(formatVersion)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("Invalid type for photo: file is not supported until v4"); + } + + @Test + void keepsTheFileTypeThroughSerialization() { + TableMetadata metadata = newTableMetadata(4); + TableMetadata reparsed = TableMetadataParser.fromJson(TableMetadataParser.toJson(metadata)); + + assertThat(reparsed.schema().findField("photo").type()).isEqualTo(Types.FileType.of(2)); + assertThat(reparsed.lastColumnId()).isEqualTo(metadata.lastColumnId()); + assertThat(reparsed.schema().asStruct()).isEqualTo(SCHEMA.asStruct()); + } + + private static TableMetadata newTableMetadata(int formatVersion) { + return TableMetadata.newTableMetadata( + SCHEMA, + PartitionSpec.unpartitioned(), + "file:/tmp/table", + ImmutableMap.of(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion))); + } +} diff --git a/core/src/test/java/org/apache/iceberg/TestSchemaUpdate.java b/core/src/test/java/org/apache/iceberg/TestSchemaUpdate.java index 5325e4013c68..85dfdddb7f50 100644 --- a/core/src/test/java/org/apache/iceberg/TestSchemaUpdate.java +++ b/core/src/test/java/org/apache/iceberg/TestSchemaUpdate.java @@ -2590,4 +2590,136 @@ public void testCaseInsensitiveMoveAfterNewlyAddedField() { assertThat(actual.asStruct()).isEqualTo(expected.asStruct()); } + + private static final Schema FILE_SCHEMA = + new Schema( + required(1, "id", Types.LongType.get()), + optional(2, "photo", Types.FileType.of(2)), + optional(9, "data", Types.StringType.get())); + + private static SchemaUpdate fileUpdate() { + return new SchemaUpdate(FILE_SCHEMA, FILE_SCHEMA.highestFieldId()); + } + + @Test + void cannotAddColumnToFileColumn() { + assertThatThrownBy(() -> fileUpdate().addColumn("photo", "extra", Types.StringType.get())) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Cannot add to non-struct column: photo: file"); + } + + @Test + void cannotDeleteFileNestedField() { + assertThatThrownBy(() -> fileUpdate().deleteColumn("photo.checksum")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Cannot change a nested field of a file column: photo.checksum"); + } + + @Test + void cannotRenameFileNestedField() { + assertThatThrownBy(() -> fileUpdate().renameColumn("photo.uri", "location")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Cannot change a nested field of a file column: photo.uri"); + } + + @Test + void cannotPromoteFileNestedField() { + assertThatThrownBy(() -> fileUpdate().updateColumn("photo.size", Types.LongType.get())) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Cannot change a nested field of a file column: photo.size"); + } + + @Test + void cannotUpdateFileNestedFieldDoc() { + assertThatThrownBy(() -> fileUpdate().updateColumnDoc("photo.uri", "the location")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Cannot change a nested field of a file column: photo.uri"); + } + + @Test + void cannotUpdateFileNestedFieldDefault() { + assertThatThrownBy( + () -> fileUpdate().updateColumnDefault("photo.uri", Literal.of("s3://bucket/key"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Cannot change a nested field of a file column: photo.uri"); + } + + @Test + void cannotUpdateFileNestedFieldRequirement() { + assertThatThrownBy(() -> fileUpdate().requireColumn("photo.uri")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Cannot change a nested field of a file column: photo.uri"); + assertThatThrownBy(() -> fileUpdate().makeColumnOptional("photo.uri")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Cannot change a nested field of a file column: photo.uri"); + } + + @Test + void cannotMoveFileNestedField() { + assertThatThrownBy(() -> fileUpdate().moveFirst("photo.checksum")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Cannot move fields in non-struct type: file"); + assertThatThrownBy(() -> fileUpdate().moveBefore("photo.checksum", "photo.uri")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Cannot move fields in non-struct type: file"); + assertThatThrownBy(() -> fileUpdate().moveAfter("photo.uri", "photo.inline")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Cannot move fields in non-struct type: file"); + } + + @Test + void unionByNameCannotAddToFileColumn() { + Schema newSchema = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 2, + "photo", + Types.StructType.of( + optional(3, "uri", Types.StringType.get()), + optional(10, "extra", Types.StringType.get())))); + + assertThatThrownBy(() -> fileUpdate().unionByNameWith(newSchema)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Cannot add to non-struct column: photo: file"); + } + + @Test + void renameFileColumn() { + Schema renamed = fileUpdate().renameColumn("photo", "image").apply(); + + assertThat(renamed.findField("image").type()).isEqualTo(Types.FileType.of(2)); + assertThat(renamed.findField("image.uri").fieldId()).isEqualTo(3); + } + + @Test + void deleteFileColumn() { + Schema deleted = fileUpdate().deleteColumn("photo").apply(); + + assertThat(deleted.findField("photo")).isNull(); + assertThat(deleted.asStruct()) + .isEqualTo( + new Schema( + required(1, "id", Types.LongType.get()), + optional(9, "data", Types.StringType.get())) + .asStruct()); + } + + @Test + void addFileColumnReservesNestedIds() { + Schema schema = new Schema(required(1, "id", Types.LongType.get())); + + Schema updated = + new SchemaUpdate(schema, schema.highestFieldId()) + .addColumn("photo", Types.FileType.of(2)) + .addColumn("data", Types.StringType.get()) + .apply(); + + assertThat(updated.findField("photo").fieldId()).isEqualTo(2); + assertThat(updated.findField("photo").type()).isEqualTo(Types.FileType.of(2)); + assertThat(updated.findField("photo.uri").fieldId()).isEqualTo(3); + assertThat(updated.findField("photo.inline").fieldId()).isEqualTo(8); + assertThat(updated.findField("data").fieldId()).isEqualTo(9); + assertThat(updated.highestFieldId()).isEqualTo(9); + } } diff --git a/core/src/test/java/org/apache/iceberg/avro/TestFileTypeAvro.java b/core/src/test/java/org/apache/iceberg/avro/TestFileTypeAvro.java new file mode 100644 index 000000000000..435684d44389 --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/avro/TestFileTypeAvro.java @@ -0,0 +1,128 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg.avro; + +import static org.apache.iceberg.types.Types.NestedField.optional; +import static org.apache.iceberg.types.Types.NestedField.required; +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.IOException; +import java.nio.file.Path; +import java.util.List; +import org.apache.avro.generic.GenericData; +import org.apache.iceberg.Files; +import org.apache.iceberg.Schema; +import org.apache.iceberg.io.FileAppender; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.types.Types; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class TestFileTypeAvro { + private static final Schema SCHEMA = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "photo", Types.FileType.of(2))); + + @TempDir private Path temp; + + @Test + void visitsAFileColumnWithATypedAvroVisitor() { + org.apache.avro.Schema avroSchema = AvroSchemaUtil.convert(SCHEMA, "table"); + + assertThat(AvroSchemaWithTypeVisitor.visit(SCHEMA, avroSchema, new FieldNameCollector())) + .contains("uri", "offset", "size", "content_type", "checksum", "inline"); + } + + @Test + void roundTripsAFileColumnThroughAvro() throws IOException { + org.apache.avro.Schema avroSchema = AvroSchemaUtil.convert(SCHEMA, "table"); + org.apache.avro.Schema photoSchema = avroSchema.getField("photo").schema().getTypes().get(1); + + GenericData.Record photo = new GenericData.Record(photoSchema); + photo.put("uri", "s3://bucket/photo"); + photo.put("offset", 128L); + photo.put("size", 1024L); + photo.put("content_type", "image/png"); + photo.put("checksum", "abc123"); + photo.put("inline", null); + + GenericData.Record row = new GenericData.Record(avroSchema); + row.put("id", 1L); + row.put("photo", photo); + + OutputFile out = Files.localOutput(temp.resolve("file-type.avro").toFile()); + try (FileAppender writer = + Avro.write(out).schema(SCHEMA).named("table").build()) { + writer.add(row); + } + + List rows; + try (AvroIterable reader = + Avro.read(out.toInputFile()).project(SCHEMA).build()) { + rows = Lists.newArrayList(reader); + } + + assertThat(rows).hasSize(1); + GenericData.Record readPhoto = (GenericData.Record) rows.get(0).get("photo"); + assertThat(readPhoto.get("uri")).hasToString("s3://bucket/photo"); + assertThat(readPhoto.get("offset")).isEqualTo(128L); + assertThat(readPhoto.get("size")).isEqualTo(1024L); + } + + private static class FieldNameCollector extends AvroSchemaWithTypeVisitor> { + @Override + public List record( + Types.StructType iStruct, + org.apache.avro.Schema record, + List names, + List> fields) { + List all = Lists.newArrayList(names); + fields.stream().filter(java.util.Objects::nonNull).forEach(all::addAll); + return all; + } + + @Override + public List union( + org.apache.iceberg.types.Type iType, + org.apache.avro.Schema union, + List> options) { + List all = Lists.newArrayList(); + options.stream().filter(java.util.Objects::nonNull).forEach(all::addAll); + return all; + } + + @Override + public List array( + Types.ListType iList, org.apache.avro.Schema array, List element) { + return element; + } + + @Override + public List map(Types.MapType iMap, org.apache.avro.Schema map, List value) { + return value; + } + + @Override + public List primitive( + org.apache.iceberg.types.Type.PrimitiveType iPrimitive, org.apache.avro.Schema primitive) { + return Lists.newArrayList(); + } + } +} diff --git a/data/src/main/java/org/apache/iceberg/data/InternalRecordWrapper.java b/data/src/main/java/org/apache/iceberg/data/InternalRecordWrapper.java index 828bd58ec9c6..01fd1824f908 100644 --- a/data/src/main/java/org/apache/iceberg/data/InternalRecordWrapper.java +++ b/data/src/main/java/org/apache/iceberg/data/InternalRecordWrapper.java @@ -70,6 +70,9 @@ private static Function converter(Type type) { case STRUCT: InternalRecordWrapper wrapper = new InternalRecordWrapper(type.asStructType()); return struct -> wrapper.wrap((StructLike) struct); + case FILE: + InternalRecordWrapper fileWrapper = new InternalRecordWrapper(type.asFileType().asStruct()); + return file -> fileWrapper.wrap((StructLike) file); default: } return null; diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index f7cc6024d74b..01869e2b415b 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -81,6 +81,8 @@ nessie = "0.108.4" netty-buffer = "4.2.17.Final" object-client-bundle = "3.3.2" orc = "1.9.9" +# TODO: bump to a release that provides FileLogicalTypeAnnotation so that the Iceberg file type can +# be written with the Parquet FILE annotation (apache/parquet-java#3608) parquet = "1.17.1" roaringbitmap = "1.6.20" scala-collection-compat = "2.14.0" diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetTypeVisitor.java b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetTypeVisitor.java index 271d9e8bf819..2029ec15a43b 100644 --- a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetTypeVisitor.java +++ b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetTypeVisitor.java @@ -52,6 +52,10 @@ public static T visit(Type type, ParquetTypeVisitor visitor) { return visitVariant(group, visitor); } + // TODO: dispatch FILE-annotated groups to a file() hook once parquet is upgraded. This + // visitor has no Iceberg type to fall back on, so until FileLogicalTypeAnnotation exists a + // file group is indistinguishable from a struct here. Subclasses that rebuild the group + // (RemoveIds, ApplyNameMapping) will need to preserve the annotation, which struct() drops. return visitor.struct(group, visitFields(group, visitor)); } } diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/TypeToMessageType.java b/parquet/src/main/java/org/apache/iceberg/parquet/TypeToMessageType.java index f05001f5f43d..5f4f0f9d3b7a 100644 --- a/parquet/src/main/java/org/apache/iceberg/parquet/TypeToMessageType.java +++ b/parquet/src/main/java/org/apache/iceberg/parquet/TypeToMessageType.java @@ -36,6 +36,7 @@ import org.apache.iceberg.types.Type.TypeID; import org.apache.iceberg.types.TypeUtil; import org.apache.iceberg.types.Types.DecimalType; +import org.apache.iceberg.types.Types.FileType; import org.apache.iceberg.types.Types.FixedType; import org.apache.iceberg.types.Types.GeographyType; import org.apache.iceberg.types.Types.GeometryType; @@ -126,6 +127,9 @@ public Type field(NestedField field) { } else if (field.type().isVariantType()) { return variant(repetition, id, name); + } else if (field.type().isFileType()) { + return file(field.type().asFileType(), repetition, id, name); + } else { NestedType nested = field.type().asNestedType(); if (nested.isStructType()) { @@ -167,6 +171,14 @@ public GroupType map(MapType map, Type.Repetition repetition, int id, String nam .named(AvroSchemaUtil.makeCompatibleName(name)); } + public GroupType file(FileType file, Type.Repetition repetition, int id, String name) { + // TODO: annotate the group with the Parquet FILE logical type once parquet is upgraded. + // FileLogicalTypeAnnotation does not exist in parquet 1.17.1, so the group is written without + // an annotation. Iceberg readers resolve the nested fields by field ID, so they read these + // files correctly, but other readers see a plain group. + return struct(file.asStruct(), repetition, id, name); + } + public Type variant(Type.Repetition repetition, int id, String originalName) { String name = AvroSchemaUtil.makeCompatibleName(originalName); Type shreddedType; diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/TypeWithSchemaVisitor.java b/parquet/src/main/java/org/apache/iceberg/parquet/TypeWithSchemaVisitor.java index c5268bf51a26..9c4195a1043d 100644 --- a/parquet/src/main/java/org/apache/iceberg/parquet/TypeWithSchemaVisitor.java +++ b/parquet/src/main/java/org/apache/iceberg/parquet/TypeWithSchemaVisitor.java @@ -65,6 +65,9 @@ public static T visit( || (iType != null && iType.isVariantType())) { // when Parquet has a VARIANT logical type, use it here return visitVariant(iType != null ? iType.asVariantType() : null, group, visitor); + } else if (iType != null && iType.isFileType()) { + Types.FileType file = iType.asFileType(); + return visitor.file(file, group, visitFields(file.asStruct(), group, visitor)); } Types.StructType struct = iType != null ? iType.asStructType() : null; @@ -230,6 +233,16 @@ public T struct(Types.StructType iStruct, GroupType struct, List fields) { return null; } + /** + * Visits a file column, which is stored as a group of its nested fields. + * + *

The default handles the file as the struct of its nested fields. Override this to + * reconstruct a file column from those fields. + */ + public T file(Types.FileType iFile, GroupType file, List fields) { + return struct(iFile != null ? iFile.asStruct() : null, file, fields); + } + public T list(Types.ListType iList, GroupType array, T element) { return null; } diff --git a/parquet/src/test/java/org/apache/iceberg/parquet/TestFileTypeParquet.java b/parquet/src/test/java/org/apache/iceberg/parquet/TestFileTypeParquet.java new file mode 100644 index 000000000000..740257f9be3b --- /dev/null +++ b/parquet/src/test/java/org/apache/iceberg/parquet/TestFileTypeParquet.java @@ -0,0 +1,309 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg.parquet; + +import static org.apache.iceberg.parquet.ParquetWritingTestUtils.createTempFile; +import static org.apache.iceberg.types.Types.NestedField.optional; +import static org.apache.iceberg.types.Types.NestedField.required; +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.nio.file.Path; +import java.util.List; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.Files; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.data.GenericRecord; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.data.parquet.GenericParquetReaders; +import org.apache.iceberg.data.parquet.GenericParquetWriter; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.io.DataWriter; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.types.Types; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.MessageTypeParser; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class TestFileTypeParquet { + private static final Schema SCHEMA = + new Schema( + required(1, "id", Types.LongType.get()), + optional(2, "photo", Types.FileType.of(2)), + optional(9, "data", Types.StringType.get())); + + @TempDir private Path temp; + + @Test + void convertsToTheParquetFileGroup() { + MessageType expected = + MessageTypeParser.parseMessageType( + "message table {" + + " required int64 id = 1;" + + " optional group photo = 2 {" + + " optional binary uri (STRING) = 3;" + + " optional int64 offset = 4;" + + " optional int64 size = 5;" + + " optional binary content_type (STRING) = 6;" + + " optional binary checksum (STRING) = 7;" + + " optional binary inline = 8;" + + " }" + + " optional binary data (STRING) = 9;" + + "}"); + + assertThat(ParquetSchemaUtil.convert(SCHEMA, "table")).isEqualTo(expected); + } + + @Test + void convertsARequiredFileColumn() { + Schema schema = new Schema(required(2, "photo", Types.FileType.of(2))); + + MessageType expected = + MessageTypeParser.parseMessageType( + "message table {" + + " required group photo = 2 {" + + " optional binary uri (STRING) = 3;" + + " optional int64 offset = 4;" + + " optional int64 size = 5;" + + " optional binary content_type (STRING) = 6;" + + " optional binary checksum (STRING) = 7;" + + " optional binary inline = 8;" + + " }" + + "}"); + + assertThat(ParquetSchemaUtil.convert(schema, "table")).isEqualTo(expected); + } + + @Test + void convertsAFileListElement() { + Schema schema = + new Schema(optional(1, "photos", Types.ListType.ofOptional(2, Types.FileType.of(2)))); + + MessageType expected = + MessageTypeParser.parseMessageType( + "message table {" + + " optional group photos (LIST) = 1 {" + + " repeated group list {" + + " optional group element = 2 {" + + " optional binary uri (STRING) = 3;" + + " optional int64 offset = 4;" + + " optional int64 size = 5;" + + " optional binary content_type (STRING) = 6;" + + " optional binary checksum (STRING) = 7;" + + " optional binary inline = 8;" + + " }" + + " }" + + " }" + + "}"); + + assertThat(ParquetSchemaUtil.convert(schema, "table")).isEqualTo(expected); + } + + @Test + void convertsAFileMapValue() { + Schema schema = + new Schema( + optional( + 1, + "byName", + Types.MapType.ofOptional(2, 3, Types.StringType.get(), Types.FileType.of(3)))); + + MessageType expected = + MessageTypeParser.parseMessageType( + "message table {" + + " optional group byName (MAP) = 1 {" + + " repeated group key_value {" + + " required binary key (STRING) = 2;" + + " optional group value = 3 {" + + " optional binary uri (STRING) = 4;" + + " optional int64 offset = 5;" + + " optional int64 size = 6;" + + " optional binary content_type (STRING) = 7;" + + " optional binary checksum (STRING) = 8;" + + " optional binary inline = 9;" + + " }" + + " }" + + " }" + + "}"); + + assertThat(ParquetSchemaUtil.convert(schema, "table")).isEqualTo(expected); + } + + @Test + void convertsBackToAPlainStructWithoutTheFileAnnotation() { + Schema converted = ParquetSchemaUtil.convert(ParquetSchemaUtil.convert(SCHEMA, "table")); + + // parquet 1.17.1 has no FILE annotation, so the group is indistinguishable from a struct here. + // Readers recover the file type from the expected Iceberg schema instead. + assertThat(converted.findField("photo").type().isFileType()).isFalse(); + assertThat(converted.findField("photo").type()) + .isEqualTo(Types.StructType.of(Types.FileType.of(2).fields())); + } + + @Test + void prunesToASingleNestedField() { + MessageType pruned = + ParquetSchemaUtil.pruneColumns(ParquetSchemaUtil.convert(SCHEMA, "table"), uriProjection()); + + assertThat(pruned.getColumns()).hasSize(1); + assertThat(pruned.getColumns().get(0).getPath()).containsExactly("photo", "uri"); + } + + @Test + void roundTripsAllNestedFields() throws IOException { + List expected = records(); + OutputFile file = write(expected); + + List actual; + try (CloseableIterable reader = + Parquet.read(file.toInputFile()) + .project(SCHEMA) + .createReaderFunc(fileSchema -> GenericParquetReaders.buildReader(SCHEMA, fileSchema)) + .build()) { + actual = Lists.newArrayList(reader); + } + + assertThat(actual).hasSameSizeAs(expected); + assertThat(record(actual, 0)).isEqualTo(expected.get(0).getField("photo")); + assertThat(record(actual, 1).getField("uri")).isEqualTo("s3://bucket/partial"); + assertThat(record(actual, 1).getField("checksum")).isNull(); + assertThat(actual.get(2).getField("photo")).isNull(); + } + + @Test + void readsAProjectionOfASingleNestedField() throws IOException { + OutputFile file = write(records()); + Schema projection = uriProjection(); + + List actual; + try (CloseableIterable reader = + Parquet.read(file.toInputFile()) + .project(projection) + .createReaderFunc( + fileSchema -> GenericParquetReaders.buildReader(projection, fileSchema)) + .build()) { + actual = Lists.newArrayList(reader); + } + + assertThat(actual).hasSize(3); + Record photo = record(actual, 0); + assertThat(photo.struct().fields()).hasSize(1); + assertThat(photo.getField("uri")).isEqualTo("s3://bucket/full"); + } + + @Test + void collectsMetricsForNestedFieldsButNotTheContainer() throws IOException { + DataFile dataFile = writeDataFile(records()); + + assertThat(dataFile.nullValueCounts()).containsKeys(3, 4, 5, 6, 7, 8).doesNotContainKey(2); + assertThat(dataFile.lowerBounds()).containsKeys(3, 4, 5).doesNotContainKey(2); + assertThat(dataFile.nullValueCounts().get(3)).isEqualTo(1L); + assertThat(dataFile.nullValueCounts().get(7)).isEqualTo(2L); + } + + private static Schema uriProjection() { + return new Schema( + optional(2, "photo", Types.StructType.of(optional(3, "uri", Types.StringType.get())))); + } + + private static List records() { + GenericRecord row = GenericRecord.create(SCHEMA); + GenericRecord photo = GenericRecord.create(Types.FileType.of(2).asStruct()); + + return ImmutableList.of( + row.copy( + ImmutableMap.of( + "id", + 1L, + "photo", + photo.copy( + ImmutableMap.of( + "uri", + "s3://bucket/full", + "offset", + 128L, + "size", + 1024L, + "content_type", + "image/png", + "checksum", + "deadbeef", + "inline", + ByteBuffer.wrap("bytes".getBytes(StandardCharsets.UTF_8)))), + "data", + "a")), + row.copy( + ImmutableMap.of( + "id", + 2L, + "photo", + photo.copy(ImmutableMap.of("uri", "s3://bucket/partial", "size", 8L)), + "data", + "b")), + // the whole file column is null + row.copy(ImmutableMap.of("id", 3L, "data", "c"))); + } + + private static Record record(List rows, int position) { + return (Record) rows.get(position).getField("photo"); + } + + private OutputFile write(List rows) throws IOException { + OutputFile file = Files.localOutput(createTempFile(temp)); + DataWriter writer = + Parquet.writeData(file) + .schema(SCHEMA) + .createWriterFunc(GenericParquetWriter::create) + .overwrite() + .withSpec(PartitionSpec.unpartitioned()) + .build(); + try (DataWriter toClose = writer) { + for (Record row : rows) { + toClose.write(row); + } + } + + return file; + } + + private DataFile writeDataFile(List rows) throws IOException { + OutputFile file = Files.localOutput(createTempFile(temp)); + DataWriter writer = + Parquet.writeData(file) + .schema(SCHEMA) + .createWriterFunc(GenericParquetWriter::create) + .overwrite() + .withSpec(PartitionSpec.unpartitioned()) + .build(); + try (DataWriter toClose = writer) { + for (Record row : rows) { + toClose.write(row); + } + } + + return writer.toDataFile(); + } +}