feat: Support metadata table "Entries"

[rshkv]

Re #823. This adds support for the the Manifest Entries (docs) which lists entries in the current snapshot's manifest files.

The code is structured with nested builders of StructArray. The hierarchy is roughly as follow (new classes in bold):

• EntriesTable
  • status, snapshot_id, sequence_number, file_sequence_number
  • data_file built by the DataFileStructBuilder, which has:
    • file_path, file_format, record_count, etc.
    • partition is a struct of partition values built by PartitionValuesStructBuilder
      • has for each partition column an AnyArrayBuilder which I had to introduce to do dynamic ArrayBuilder casting based on the column type
  • readable_metrics is built by the ReadableMetricsStructBuilder
    • for each column, has a PerColumnReadableMetricsBuilder
      • contains column_size, value_count
      • has an upper_bound and lower_bound struct which use AnyArrayBuilder to preserve the type

This PR ended up being quite verbose because arrow-rs is strict about declaring generic types of array builders at compile time. Unlike Python, which supports entries in ~100 lines, we can't shove a dict into a StructBuilder. Ideally, we could build StructArray row-by-row and write logic to convert manifest entries to rows.

Reference implementations:

• Java
• PyIceberg

[Xuanwo]

Hi, @rshkv, sorry for making you rebase the PRs again. There are multiple open PRs here, and I spent some time figuring out the best API. I believe we are now heading in the right direction.

We can focus on merging one PR first and then updating the other PRs to avoid additional work.

[Xuanwo]

Hi, I think we can simply hold a Table here, allowing us to remove the duplicate APIs exposed at the MetadataTable level and make it a straightforward wrapper instead.

[rshkv]

I appreciate this is quite verbose and I wish we didn't have to do all the pattern matching below. If you think of a another way to do this let me know.

[rshkv]

The file_sequence_number relies on this value but there was no way to get at the ManifestEntry::file_sequence_number before.

[rshkv]

This is list is exhaustive based on the ArrowSchemaVisitor::primitive function above. I.e., every type produced there is covered here.

[rshkv]

I understand it's correct to ignore the timezone here because that's not captured in the builder.

[rshkv]

This isn't based on the test data but wanted to have that reflected in tests.

[rshkv]

Adding these so we cover those as types in the lower and upper bounds.

I'm trying to limit the changes I'm making because it's already a lot. My preference would be to cover all types as partition columns as well.

[rshkv]

Could add support but thought that might be for another PR.

[rshkv]

Addressing this comment #822 (comment). I prefer that but don't need to do here.

[rshkv]

This is populated with ManifestStatus enum values.

In Java, the status column is i32 (here) but in Python this is u8.

My preference would be u8 but treating the Java implementation as authoritative.

[rshkv]

The casting is slightly annoying given we're dealing with non-negative types. But the Python and Java implementation use i64.

[rshkv]

The manifests table merged in #861 prefers using PrimitiveBuilder::new() which works because Int64Builder is just PrimitiveBuilder<Int64Type>.

I prefer saying Int64Builder here to be explicit about the type but happy to change.

[rshkv]

Thank you, @Xuanwo. Rebased and ready for review.

[liurenjie1024]

Thanks @rshkv for this contribution. I have finished first round review and left some cocerns with current api design.

[liurenjie1024]

Why return arrow schema rather iceberg schema here?

[rshkv]

Following existing API but happy to update. I understand the idea in #822 (comment) was for engines to fetch the schema before having to fetch data.

[rshkv]

@liurenjie1024, would you mind saying more. I'm happy to go with either but not sure why.

Even if there's no consumer of schema() currently, I follow @xxchan's argument's of the reader likely wanting an Arrow schema. Another benefit is that we can use the schema ourselves when constructing scans. I'm not sure what a consumer would do with an Iceberg schema (except maybe convert to Arrow).

As alternative to having an Arrow or Iceberg schema, we could also not have a public schema()?

[liurenjie1024]

This would not be required if we use iceberg schema

[rshkv]

Can you say more? I suppose we still need to construct those StructArray instances?

[rshkv]

I think checking the rendered table is easier to read and confirm with eyeballs. We lose type information but we assert types separately.

[xxchan]

👍 This looks great.

[rshkv]

This looks much worse here than it does locally:

If there's a way to pretty-print schemas let me know - couldn't find one.

[xxchan]

I don't think there exist one. We can write a separate function (or newtype) for pretty print the schema.
I once did similar things in RisingWave:

The code:

• https://github.com/risingwavelabs/risingwave/blob/main/src/connector/codec/tests/integration_tests/utils.rs#L23-L66
• https://github.com/risingwavelabs/risingwave/blob/main/src/connector/codec/tests/integration_tests/utils.rs#L167-L214

Basically we ignore noises (e.g., unnecessary fields, field names) and make it more concise and readable.

[rshkv]

Yeah, that'd be nice. E.g., we don't need to render dict_id and dict_is_ordered for every field.

I'll not introduce that in an already large PR (unless you prefer I do). If you think I should, maybe we can do the updates to check_record_batches (pretty-printing batches, ignoring struct fields, and the new schema pretty-print) in a separate PR?

[rshkv]

Don't want to block this PR basically.

[xxchan]

maybe we can do the updates to check_record_batches (pretty-printing batches, ignoring struct fields, and the new schema pretty-print) in a separate PR?

+1

[rshkv]

I've rebased on #870 and #872 to address the follow:

• The entries table now lives in a separate entries.rs file.
• Batches for manifest files are now computed asynchronously.

I haven't address @liurenjie1024's point about schema() returning an Iceberg schema instead of an Arrow one. We have issue #868 and PR #871 but I'm not sure that's what we want generally and whether you want this to happen before this PR.

Otherwise this is ready for another review.

[liurenjie1024]

I've rebased on #870 and #872 to address the follow:

• The entries table now lives in a separate entries.rs file.
• Batches for manifest files are now computed asynchronously.

I haven't address @liurenjie1024's point about schema() returning an Iceberg schema instead of an Arrow one. We have issue #868 and PR #871 but I'm not sure that's what we want generally and whether you want this to happen before this PR.

Otherwise this is ready for another review.

Thanks @rshkv for the contribution, let's continue discussion about schema in #868

[rshkv]

I'm still working on this and will continue working on this. I find it quite difficult to get Arrow and Iceberg to agree on types because of field ids.

The Iceberg schema requires that all fields have a field id, which in the converted Arrow schema becomes type metadata. But when constructing StructArray or RecordBatch, Arrow checks that the schema matches that of the data (e.g.). And they tend to not match because the schema has field ids and the data does not.

E.g., with MapBuilder, we have with_values_field to pass in a field with a field id in metadata. However there is no with_keys_field equivalent. Yet, the key field in the Iceberg schema must have a field id.

@liurenjie1024, I'd like to see this finished and some clarity on designs you'd merge would help. Here are some questions:

• We can make schema() return an Iceberg schema with field id, but how important is it that a returned RecordBatch has field ids in metadata? Are we ok with not having field ids on RecordBatch#schema but only on MetadataTable#schema?
• If we do need record batches types to have those field ids, we might need changes in arrow-rs to express something like "a RecordBatch or StructArray may have fields with metadata, but the respective types of the underlying ArrayData don't need to match metadata".

Let me know what you think or if I'm not being clear.

[xxchan]

Just noticed that in the Java implementation, the schema has field ids, defined like this:

https://github.com/apache/iceberg/blob/ec269ee3ec0de4184eb536a6ef4f3523dc91332a/api/src/main/java/org/apache/iceberg/DataFile.java#L123-L146

(BTW personally I don't think that's very necessary for compute engines though

[xxchan]

But when constructing StructArray or RecordBatch, Arrow checks that the schema matches that of the data (e.g.). And they tend to not match because the schema has field ids and the data does not.

A little confused about this point. What resulted them mismatching? IIUC if we use the correct schema to construct Array/RecordBatch, there should be no mismatch.

[rshkv]

@xxchan, that's right. The problem is constructing Array/RecordBatch with the right schema in the first place. MapBuilder doesn't let you configure the key field though, so the keys array won't have the field id.

[xxchan]

MapBuilder doesn't let you configure the key field though, so the keys array won't have the field id.

Not sure if I understand it correctly, this sounds like just some limitations of arrow-rs and can be solved easily by adding some new APIs, instead of some fundamental difficulty.

BTW maybe it's easier to discuss if you can push some sample code.

[Fokko]

We can make schema() return an Iceberg schema with field id, but how important is it that a returned RecordBatch has field ids in metadata? Are we ok with not having field ids on RecordBatch#schema but only on MetadataTable#schema?

Yes, that's alright. The field-IDs are internal to Iceberg, and when it goes to the engine it is based on the environment, SQL is position-based, while you could also lookup by name in dataframes.

[Fokko]

Just noticed that in the Java implementation, the schema has field ids, defined like this: #863 (review)

We don't have to expose the field-IDs internally, as long as we resolve the fields using the ID. The name might change over time (see apache/iceberg#5338), but if you read the data as an Iceberg table, then we resolve the fields by ID then everything will be correct.

[rshkv]

BTW maybe it's easier to discuss if you can push some sample code.

Pushed my work-in-progress. Currently failing here:

Incorrect datatype for StructArray field \"column_sizes\", expected

Map(Field { 
  name: \"key_value\", 
  data_type: Struct([
    Field { name: \"key\", data_type: Int32, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {\"PARQUET:field_id\": \"117\"} }, 
    Field { name: \"value\", data_type: Int64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {\"PARQUET:field_id\": \"118\"} }]),
  nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, false)

got

Map(Field {
  name: \"key_value\",
  data_type: Struct([
    Field { name: \"key\", data_type: Int32, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} },
    Field { name: \"value\", data_type: Int64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }]),
  nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, false)

Not sure if I understand it correctly, this sounds like just some limitations of arrow-rs and can be solved easily by adding some new APIs, instead of some fundamental difficulty.

That's right. Going to open a PR with arrow-rs.