Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
27 commits
Select commit Hold shift + click to select a range
76ac74a
API: Add file type as a struct-on-read schema type
RussellSpitzer Aug 21, 2026
fa9d5ab
API: Keep the file type intact through ID assignment
RussellSpitzer Aug 24, 2026
9d71f58
API, Core, Parquet: Cover the remaining file type cases in tests
RussellSpitzer Aug 24, 2026
32b42b7
API: Gate the file type through the min format version map
RussellSpitzer Aug 24, 2026
db1b17c
API, Core: Call the reserving ID overload only for the file type
RussellSpitzer Aug 24, 2026
1170af1
API, Parquet: Move file type tests to the classes that own the behavior
RussellSpitzer Aug 24, 2026
5423d25
API, Core: Name the file type's enclosing ID consistently
RussellSpitzer Aug 24, 2026
8549188
Core: Model the file logical type as its own nested type
RussellSpitzer Aug 26, 2026
824ef68
API, Core, Data: Handle the file type in typeId switches and struct v…
RussellSpitzer Aug 26, 2026
c01f92b
ORC: Support the file type in schema conversion and visitors
RussellSpitzer Aug 26, 2026
d3205ed
Hive: Convert a file column to a Hive struct
RussellSpitzer Aug 26, 2026
0a7893e
Kafka Connect: Handle the file type when converting records
RussellSpitzer Aug 26, 2026
8b0731e
AWS: Render a file column as a struct in Glue
RussellSpitzer Aug 26, 2026
d58d615
Spark: Implement the file type hook in schema visitors
RussellSpitzer Aug 26, 2026
b3b7e4d
Spark: Descend into a file column in InternalRowWrapper
RussellSpitzer Aug 26, 2026
60bb51d
Spark: Read a file column through the planned Avro reader
RussellSpitzer Aug 26, 2026
d80a432
Spark: Name the column when a write cannot express a file type
RussellSpitzer Aug 26, 2026
3db9007
Spark: Make the file type safe in Iceberg-typed casts
RussellSpitzer Aug 26, 2026
0942397
Flink: Convert a file column to a row of its nested fields
RussellSpitzer Aug 26, 2026
b62810e
Flink: Read and project a file column
RussellSpitzer Aug 26, 2026
c459d54
Flink: Compare a file column against an incoming struct
RussellSpitzer Aug 26, 2026
31ff0ad
Flink: Leave a file column alone when evolving a schema
RussellSpitzer Aug 26, 2026
ca62feb
Flink: Reject writes to a file column
RussellSpitzer Aug 26, 2026
e9df710
Spark: Wrap the file type scan test to spotless formatting
RussellSpitzer Aug 27, 2026
26ed301
Merge branch 'file_type_peripheral' into custom_file_type
RussellSpitzer Aug 27, 2026
60add5a
Merge branch 'file_type_flink' into custom_file_type
RussellSpitzer Aug 27, 2026
3af7d66
Merge branch 'file_type_spark' into custom_file_type
RussellSpitzer Aug 27, 2026
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 11 additions & 1 deletion api/src/main/java/org/apache/iceberg/Accessors.java
Original file line number Diff line number Diff line change
Expand Up @@ -213,8 +213,18 @@ public Map<Integer, Accessor<StructLike>> schema(
@Override
public Map<Integer, Accessor<StructLike>> struct(
Types.StructType struct, List<Map<Integer, Accessor<StructLike>>> fieldResults) {
return buildAccessors(struct.fields(), fieldResults);
}

@Override
public Map<Integer, Accessor<StructLike>> file(
Types.FileType file, List<Map<Integer, Accessor<StructLike>>> fieldResults) {
return buildAccessors(file.fields(), fieldResults);
}

private Map<Integer, Accessor<StructLike>> buildAccessors(
List<Types.NestedField> fields, List<Map<Integer, Accessor<StructLike>>> fieldResults) {
Map<Integer, Accessor<StructLike>> accessors = Maps.newHashMap();
List<Types.NestedField> fields = struct.fields();
for (int i = 0; i < fieldResults.size(); i += 1) {
Types.NestedField field = fields.get(i);
Map<Integer, Accessor<StructLike>> result = fieldResults.get(i);
Expand Down
38 changes: 26 additions & 12 deletions api/src/main/java/org/apache/iceberg/Schema.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -578,20 +579,33 @@ private List<NestedField> reassignIds(List<NestedField> 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.
*
Expand Down
44 changes: 32 additions & 12 deletions api/src/main/java/org/apache/iceberg/types/AssignFreshIds.java
Original file line number Diff line number Diff line change
Expand Up @@ -48,15 +48,24 @@ class AssignFreshIds extends TypeUtil.CustomOrderSchemaVisitor<Type> {
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) {
return field.fieldId();
}
}

return nextId.get();
return null;
}

private String name(int id) {
Expand All @@ -80,15 +89,17 @@ public Type struct(Types.StructType struct, Iterable<Type> futures) {
// assign IDs for this struct's fields first
List<Integer> 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<Types.NestedField> newFields = Lists.newArrayListWithExpectedSize(length);
Iterator<Type> 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);
Expand All @@ -101,22 +112,25 @@ public Type field(Types.NestedField field, Supplier<Type> future) {

@Override
public Type list(Types.ListType list, Supplier<Type> 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<Type> keyFuture, Supplier<Type> 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);
}
}

Expand All @@ -125,6 +139,12 @@ public Type variant(Types.VariantType variant) {
return variant;
}

@Override
public Type file(Types.FileType file, Iterable<Type> 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;
Expand Down
34 changes: 22 additions & 12 deletions api/src/main/java/org/apache/iceberg/types/AssignIds.java
Original file line number Diff line number Diff line change
Expand Up @@ -31,8 +31,8 @@ class AssignIds extends TypeUtil.CustomOrderSchemaVisitor<Type> {
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
Expand All @@ -48,15 +48,16 @@ public Type struct(Types.StructType struct, Iterable<Type> futures) {
// assign IDs for this struct's fields first
List<Integer> newIds = Lists.newArrayListWithExpectedSize(length);
for (Types.NestedField field : fields) {
newIds.add(idFor(field.fieldId()));
newIds.add(idFor(field.fieldId(), field.type()));
}

List<Types.NestedField> newFields = Lists.newArrayListWithExpectedSize(length);
Iterator<Type> 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);
Expand All @@ -69,22 +70,25 @@ public Type field(Types.NestedField field, Supplier<Type> future) {

@Override
public Type list(Types.ListType list, Supplier<Type> 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<Type> keyFuture, Supplier<Type> 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);
}
}

Expand All @@ -93,6 +97,12 @@ public Type variant(Types.VariantType variant) {
return variant;
}

@Override
public Type file(Types.FileType file, Iterable<Type> 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;
Expand Down
10 changes: 10 additions & 0 deletions api/src/main/java/org/apache/iceberg/types/CheckCompatibility.java
Original file line number Diff line number Diff line change
Expand Up @@ -260,6 +260,16 @@ public List<String> variant(Types.VariantType readVariant) {
return ImmutableList.of(String.format(": %s cannot be read as a %s", currentType, readVariant));
}

@Override
public List<String> file(Types.FileType readFile, Iterable<List<String>> 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<String> primitive(Type.PrimitiveType readPrimitive) {
if (currentType.equals(readPrimitive)) {
Expand Down
2 changes: 2 additions & 0 deletions api/src/main/java/org/apache/iceberg/types/Comparators.java
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,8 @@ private static <T> Comparator<T> internal(Type type) {
return forType(type.asPrimitiveType());
} else if (type.isStructType()) {
return (Comparator<T>) forType(type.asStructType());
} else if (type.isFileType()) {
return (Comparator<T>) forType(type.asFileType().asStruct());
} else if (type.isListType()) {
return (Comparator<T>) forType(type.asListType());
} else if (type.isMapType()) {
Expand Down
15 changes: 15 additions & 0 deletions api/src/main/java/org/apache/iceberg/types/FindTypeVisitor.java
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,21 @@ public Type variant(Types.VariantType variant) {
return null;
}

@Override
public Type file(Types.FileType file, List<Type> 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)) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -45,9 +45,14 @@ public Set<Integer> struct(Types.StructType struct, List<Set<Integer>> fieldResu
return fieldIds;
}

@Override
public Set<Integer> file(Types.FileType file, List<Set<Integer>> fieldResults) {
return fieldIds;
}

@Override
public Set<Integer> field(Types.NestedField field, Set<Integer> fieldResult) {
if ((includeStructIds && field.type().isStructType())
if ((includeStructIds && (field.type().isStructType() || field.type().isFileType()))
|| field.type().isPrimitiveType()
|| field.type().isVariantType()) {
fieldIds.add(field.fieldId());
Expand Down
6 changes: 6 additions & 0 deletions api/src/main/java/org/apache/iceberg/types/IndexById.java
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,12 @@ public Map<Integer, Types.NestedField> struct(
return index;
}

@Override
public Map<Integer, Types.NestedField> file(
Types.FileType file, List<Map<Integer, Types.NestedField>> fieldResults) {
return index;
}

@Override
public Map<Integer, Types.NestedField> field(
Types.NestedField field, Map<Integer, Types.NestedField> fieldResult) {
Expand Down
17 changes: 13 additions & 4 deletions api/src/main/java/org/apache/iceberg/types/IndexByName.java
Original file line number Diff line number Diff line change
Expand Up @@ -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());
}
}
Expand All @@ -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();
}
}
Expand All @@ -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());
}
}
Expand All @@ -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<String, Integer> schema(Schema schema, Map<String, Integer> structResult) {
return nameToId;
Expand Down Expand Up @@ -194,6 +198,11 @@ public Map<String, Integer> variant(Types.VariantType variant) {
return nameToId;
}

@Override
public Map<String, Integer> file(Types.FileType file, List<Map<String, Integer>> fieldResults) {
return nameToId;
}

@Override
public Map<String, Integer> primitive(Type.PrimitiveType primitive) {
return nameToId;
Expand Down
11 changes: 10 additions & 1 deletion api/src/main/java/org/apache/iceberg/types/IndexParents.java
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,16 @@ public Map<Integer, Integer> schema(Schema schema, Map<Integer, Integer> structR
@Override
public Map<Integer, Integer> struct(
Types.StructType struct, List<Map<Integer, Integer>> fieldResults) {
for (Types.NestedField field : struct.fields()) {
return indexFields(struct.fields());
}

@Override
public Map<Integer, Integer> file(Types.FileType file, List<Map<Integer, Integer>> fieldResults) {
return indexFields(file.fields());
}

private Map<Integer, Integer> indexFields(List<Types.NestedField> fields) {
for (Types.NestedField field : fields) {
Integer parentId = idStack.peek();
if (parentId != null) {
// fields in the root struct are not added
Expand Down
Loading
Loading