Core: Add v4 manifest reader

[anoopj]

The reader supports:

• column projection
• partition pruning via a row filter, including multi-spec manifests

Content stats reading and metadata inheritance are not yet implemented.

[anoopj]

Not using the builder here so that we can set any status here.

[gaborkaszab]

I'm in the process of getting familiar with the details here. Asked early questions for my benefit.
Thanks @anoopj !

[stevenzwu]

this name V4ManifestReader would become stale when V5 rolls in. maybe TrackedFileManifestReader or just TrackedFileReader.

[anoopj]

TrackedFile is an abstraction that we want to keep internal for now. V4ManifestReader is the best name I could come up with. Open to other suggestions here. cc @rdblue for his thoughts.

[gaborkaszab]

This class implements and returns a CloseableIterable<TrackedFile> so we don't really keep TrackedFile internal in my opinion. Following this design we can name this TrackedFileReader.

[anoopj]

That is actually a good point. Leaving it open to hear from others.

[stevenzwu]

TrackedFile is internal in the sense that it is not a public interface (at least not yet). This class TrackedFileReader would probably only be package private.

[rdblue]

This is a good point, but I don't have a name suggestion that is better right now. Let's keep it in mind.

[gaborkaszab]

This class implements and returns a CloseableIterable<TrackedFile> so we don't really keep TrackedFile internal in my opinion. Following this design we can name this TrackedFileReader.

[gaborkaszab]

Thanks for the explanation! Projecting stats makes total sense.

I think most of my comment on this area are for a single reason: for me this readSchema() function seems a bit more complicated than what it is for. What I have in mind is like this:

if (fileProjection == null) {
  return TrackedFileStruct.READ_SCHEMA;
  // this could be prepared with the desired Tracking schema, omitting stats, including partition unconditionally. Preferable a static field in TrackedFileStruct.
}

// construct the projected schema including the mandatory fields similarly as now

[danielcweeks]

I'm not convinced this is how we'll be projecting stats from the read path. Ideally, we wouldn't project the whole stats field (and that's just the containing field id).

What we should have is the specific field ids for the individual status required to evaluate the filter.

[anoopj]

I do agree: we will be projecting only the field IDs we need. This is just a placeholder for now, as the interfaces for content stats are being revamped.

[rdblue]

I don't think using allActiveFieldIds is correct here.

That method will filter out partition fields when the source is no longer part of the table schema. In some cases (bucket joins), that is fine because the data is still partitioned by a subset of its partition fields.

However, for a partition type that can hold any partition tuple, it doesn't work. Equality deletes are matched using partition tuple equality. For example, say I have file_a.parquet in partition (id_bucket=7, is_test=false), and I also have a delete file deletes.parquet in partition (id_bucket=7, is_test=true). The delete file does not apply to the data file because they are in different partitions.

With the use of allActiveFieldIds here, dropping the is_test identity partition and then the is_test column from the table would result in a unified partition type that only has id_bucket. As a result, checking whether the partition matches (and the deletes should be applied) will pass for file_a.parquet and deletes.parquet.

This needs to union all partition fields so that we don't drop tuple values that define equality.

[anoopj]

Thanks for pointing this out - I didn't think about this scenario. We can fix it by dropping the filter on active fields, but there might be another problem: partition specs don't store the type, and we derive it based on the type of the source column. So if the source column doesn't exist in the schema anymore, we can't derive the partition type. I think this might be working in v3 because of the way the reader is implemented: the reader gets the schema from the Avro schema, and uses it instead of the schema from the metadata.

Maybe the Parquet reader can also work that way (can verify), but is this by design? Does the other Iceberg implementations actually work in this scenario in v3 if this detail is not in the spec?

[rdblue]

So if the source column doesn't exist in the schema anymore, we can't derive the partition type.

This is a good point to think about, but I think it will not affect what we do in practice. It's possible to remove old partition specs, but in practice this hardly ever happens. In addition, the only cases where we can't determine the output type of the transform are identity and truncate transforms. Truncation isn't very common and identity columns are unlikely to be removed from the table.

Because there are good reasons to think this case is unlikely, I think we should move forward assuming that the partition specs will be there. If we want to be more careful, we can require a metadata check to remove old specs to avoid breaking anything. Also, we can always go back and get the type from existing metadata files at read time if we can't determine the type.

I've also thought about storing the output type of transforms when we add column ranges for partition output values, like we do for UDF result types. We may just need to solve this by storing the output type.

[anoopj]

Fixed. partitionType now unions all partition fields across specs and the reader's builder no longer takes a table schema. I left the existing partitionType(Table) unchanged since fixing it has wider implications; we can revisit separately. Added a test that a partition field survives dropping its source column.

I agree the output type is only underivable for identity and truncate, but PartitionSpec.partitionType() currently returns unknown whenever the source column is missing so today even a bucket field degrades. If it makes sense, I can do a follow-up that only falls back to unknown for source-dependent transforms, which shrinks this to the rare case you described. I also did some code reading in Iceberg rust, which seems to have a hard failure, possibly because unknown type is not supported yet in Iceberg Rust.

BTW +1 to storing the output type in metadata. v4 seems like the right time to add it since we're already revising the spec. It seems like a good overall improvement. Happy to write that up.

[rdblue]

Isn't this tested elsewhere? I seem to remember it.

[anoopj]

i don't think it's covered elsewhere. We have testPartitionFilterKeepsFileWithUnknownSpec for a file whose spec ID isn't in the reader's specsById while this one has a null spec ID. they follow different code paths, so worth having a test.

[danielcweeks]

Minor oversight here: we only test for struct/list, but not a primitive field (it does work, but there's no test)

[anoopj]

Good point. Added testReplaceFieldTypesPrimitive.

[danielcweeks]

Non-blocker as result would be correct, but it's also overly restrictive. If there's a partition filter that has fields compatible with the tracked file's spec, then some amount of filtering could still be performed. This would trigger a lot of over-reads for schema evolution.

[anoopj]

I don't think there's an over-read here beyond the unavoidable case. Inclusive projection is per-predicate, so id = 1 AND data = 'z' against a spec that partitions on id (but not data) still prunes by id. So any field compatible with a spec does contribute to filtering for that spec's files.

The only files kept unfiltered are those whose spec partitions on none of the filter's columns, where there's no partition info to prune on - this is the same as v3 ManifestReader. Added testPartialFilterStillPrunesOnCompatibleField to verify this.

[nastra]

minor: I think it might be worth pulling this out into a separate PR in order to reduce complexity in this one and harden the impl + have more extensive tests (especially around default value handling)

[anoopj]

You are correct. That would have been a better choice. Keeping it here because the code has been reviewed. Will add the tests.

[nastra]

nit: project(null) would read everything. If that is a supported case, then we should have a test for it or add a newProjection != null check

[anoopj]

This is by design (there was a long thread with Ryan and Gabor) to be compatible with v3 reader. Added a test.

[nastra]

I think this can be shortened/simplified via

private void incrementSkipCount(FileContent content) {
    switch (content) {
      case DATA -> scanMetrics.skippedDataFiles().increment();
      case EQUALITY_DELETES -> scanMetrics.skippedDeleteFiles().increment();
      case DATA_MANIFEST -> scanMetrics.skippedDataManifests().increment();
      case DELETE_MANIFEST -> scanMetrics.skippedDeleteManifests().increment();
      default -> throw new UnsupportedOperationException("Unsupported content type: " + content);
    }
  }

[anoopj]

moved to the arrow form

[nastra]

I believe for new tests we usually told folks to omit the test prefix because it's not really adding any value and is a legacy thing from JUnit3. Also test methods can be easily identified, since they either have a @Test or another annotation

[anoopj]

You are correct. I will send a followup PR to clean this up. (too much churn in this PR)

[danielcweeks]

I'm guessing this behavior is to be consistent with previous readers, but it seems like not materializing the empty partition creates a special case we'll have to deal with when working with mixed partitions.

[anoopj]

yes, this is consistent with v2/v3 behavior. Good news is that it doesn't leak into mixed partitions. The reader builds its read schema from the union partition type across all specs in the manifest. The null partition case only occurs when that union is empty. With mixed spes, the union is non empty so even a file with an unpartitioned spec produces a PartitionData over the union type. So there is no special case scenario.

[danielcweeks]

Hmm, is this test actually testing anything? I guess technically it produces the correct result, but we're not actually evaluating the filter.

I guess this shields against breaking this behavior in the future, but it's not actually testing what we intend it to test just yet.

[anoopj]

The test is structured that way because we didn't incorporate content stats yet. We can change it when we do the followup cc @nastra