Error while executing "rewrite_position_delete_files"

[raphaeldouyere]

Apache Iceberg version

1.10.0 (latest release)

Query engine

Spark

Please describe the bug 🐞

We have an error when calling "rewrite_position_delete_files"

The call happens in a Spark (3.5.5) session:
spark.sql("CALL iceberg.system.rewrite_position_delete_files( table => 'lake.table', options => map('rewrite-all', 'true'))");

Starting with Iceberg 1.10.0 we have the following error:
Caused by: org.apache.iceberg.exceptions.ValidationException: Invalid partition field parent: list<string> at org.apache.iceberg.exceptions.ValidationException.check(ValidationException.java:49) at org.apache.iceberg.PartitionSpec.checkCompatibility(PartitionSpec.java:672) at org.apache.iceberg.PartitionSpec$Builder.build(PartitionSpec.java:628) at org.apache.iceberg.PartitionSpec$Builder.build(PartitionSpec.java:623) at org.apache.iceberg.expressions.ExpressionUtil.identitySpec(ExpressionUtil.java:656) at org.apache.iceberg.expressions.ExpressionUtil.extractByIdInclusive(ExpressionUtil.java:156) at org.apache.iceberg.spark.source.PositionDeletesRowReader.open(PositionDeletesRowReader.java:95) at org.apache.iceberg.spark.source.PositionDeletesRowReader.open(PositionDeletesRowReader.java:44)

Our tables had a partition layout change long ago and the partition spec currently looks like:
"default-spec-id" : 1, "partition-specs" : [ { "spec-id" : 0, "fields" : [ { "name" : "is_ok", "transform" : "identity", "source-id" : 38, "field-id" : 1000 }, { "name" : "timestamp_hour", "transform" : "hour", "source-id" : 2, "field-id" : 1001 } ] }, { "spec-id" : 1, "fields" : [ { "name" : "timestamp_hour", "transform" : "hour", "source-id" : 2, "field-id" : 1001 } ] } ]
The code seems to break on the old "is_ok" partition.

Willingness to contribute

• I can contribute a fix for this bug independently
• I would be willing to contribute a fix for this bug with guidance from the Iceberg community
• I cannot contribute a fix for this bug at this time

[chrigehr]

I have the same problem as described by the author of this issue. I tested this with Spark 3.5.7 as well as with Spark 4.0.1.

Calling rewrite_position_delete_files will fail on tables with lists or maps.

As far as I understand (from debugging and analyzing code) I see the following problem:

• PositionDeletesRowReader.open gets nonConstantFieldIds (in my understanding that are partition columns and similar constant fields that are know from metadata before opening the file. Thus this will contain essentially all data columns.)

  iceberg/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/PositionDeletesRowReader.java

  Line 93 in 15a72dc

  Set<Integer> nonConstantFieldIds = nonConstantFieldIds(idToConstant);

  iceberg/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/PositionDeletesRowReader.java

  Lines 113 to 119 in 15a72dc

  	private Set<Integer> nonConstantFieldIds(Map<Integer, ?> idToConstant) {
  	Set<Integer> fields = expectedSchema().idToName().keySet();
  	return fields.stream()
  	.filter(id -> expectedSchema().findField(id).type().isPrimitiveType())
  	.filter(id -> !idToConstant.containsKey(id))
  	.collect(Collectors.toSet());
  	}

• An expression is built with the method ExpressionUtil.extractByIdInclusive based on the field ids for all data columns
• This method creates a PartitionSpec based on all the given ids (remember, that are in this case the ids of basically all data fields of the table) and will try to build a projection based on these ids.

  iceberg/api/src/main/java/org/apache/iceberg/expressions/ExpressionUtil.java

  Lines 160 to 164 in 15a72dc

  	public static Expression extractByIdInclusive(
  	Expression expression, Schema schema, boolean caseSensitive, int... ids) {
  	PartitionSpec spec = identitySpec(schema, ids);
  	return Projections.inclusive(spec, caseSensitive).project(Expressions.rewriteNot(expression));
  	}

• Problem is, that Iceberg does not allow fields within lists or maps as partitioning columns. In older Iceberg versions this use of extractByIdInclusive was no problem, because the validations in PartitionSpec did not check for this. But Iceberg > 1.10.x now has this

  iceberg/api/src/main/java/org/apache/iceberg/PartitionSpec.java

  Lines 668 to 676 in 15a72dc

  	// The only valid parent types for a PartitionField are StructTypes. This must be checked
  	// recursively.
  	Integer parentId = parents.get(field.sourceId());
  	while (parentId != null) {
  	Type parentType = schema.findType(parentId);
  	ValidationException.check(
  	parentType.isStructType(), "Invalid partition field parent: %s", parentType);
  	parentId = parents.get(parentId);
  	}

  . Compare this with the code of Iceberg 1.9.x

  iceberg/api/src/main/java/org/apache/iceberg/PartitionSpec.java

  Lines 626 to 649 in 071d560

  	static void checkCompatibility(PartitionSpec spec, Schema schema) {
  	for (PartitionField field : spec.fields) {
  	Type sourceType = schema.findType(field.sourceId());
  	Transform<?, ?> transform = field.transform();
  	// In the case of a Version 1 partition-spec field gets deleted,
  	// it is replaced with a void transform, see:
  	// https://iceberg.apache.org/spec/#partition-transforms
  	// We don't care about the source type since a VoidTransform is always compatible and skip the
  	// checks
  	if (!transform.equals(Transforms.alwaysNull())) {
  	ValidationException.check(
  	sourceType != null, "Cannot find source column for partition field: %s", field);
  	ValidationException.check(
  	sourceType.isPrimitiveType(),
  	"Cannot partition by non-primitive source field: %s",
  	sourceType);
  	ValidationException.check(
  	transform.canTransform(sourceType),
  	"Invalid source type %s for transform: %s",
  	sourceType,
  	transform);
  	}
  	}
  	}

I'm not sure how to fix this. For an non-iceberg developer it seems, as if the ExpressionUtil.extractByIdInclusive is kind of "misused" for reading Position Delete Files. Possible ideas are:

• to remove the new validations in PartitionSpec, but actually the validations seems to be correct
• filter out fields within lists and maps within the method PositionDeletesRowReader.nonConstantFieldIds

To be honest: I do not get why rewrite_position_delete_files uses ExpressionUtil.extractByIdInclusive at all. Perhaps this is done because of schema evolution handling or makes sense for merging deletes? Seems to make only sense if the position delete files are written with additional columns values, but as far as I know this is not done at least by the spark implementation.

How can we proceed with this bug? At least for me this is a big problem because I rely on a working rewrite_position_delete_files procedure. Without this I see large performance degradation because multiple positition delete files per partition accumulate.

[turboFei]

Seems fixed by #15632 and #15743

[chrigehr]

Thank you very much @turboFei , looks promising.