[Managed Iceberg] Allow updating partition specs during pipeline runtime

[ahmedabu98]

Users may update the table's spec while a write pipeline is running (e.g. streaming). Sometimes, this update can happen after serializing DataFiles. The partition spec is not serializable so we don't preserve it, but we do keep its ID. We can use this information to fetch the correct partition spec and recreate the DataFile before appending to the table.

Fixes #32862

[ahmedabu98]

R: @DanielMorales9

[github-actions]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers

[DanielMorales9]

I've looked into the Iceberg codebase to check for any violations of snapshot consistency. Interestingly, it seems possible to commit data files produced with an old partitioning scheme (and therefore from a previous snapshot). Could this actually be a feature of partition evolution?! 🤔
Any thoughts?

[ahmedabu98]

Interestingly, it seems possible to commit data files produced with an old partitioning scheme

Yep this is what I found empirically as well. The docs mention that each partition spec's metadata is kept separate from each other. This allows Iceberg to carry out a separate query plan for each partition spec, then combine all files afterwards.

Old data written with the old partition spec doesn't get affected. What I'm seeing is new data written with the old spec is also not affected. This is convenient for streaming writes because after updating the table's partition spec, there may be some inflight data that is still using the old partition spec. Iceberg will still accept that data when it gets committed and manage the metadata accordingly.

[DanielMorales9]

Yep, makes sense! 👍
I checked once again and I can confirm that the manifest files are stored in the manifest list along with their Partition Spec Id and the writing process groups Data Files by their Partition Spec Id. We can merge ✅

Example Manifest list:

{
  "manifest_path": "gs://<bucket>/<schema>/<table>/metadata/c4aa6dc6-5e0b-454c-b126-1d97d2f223f8-m1.avro",
  "manifest_length": 8363467,
  "partition_spec_id": 0,
  "content": 0,
  "sequence_number": 4814,
  "min_sequence_number": 4093,
  "added_snapshot_id": 481579842725173797,
  "added_data_files_count": 0,
  "existing_data_files_count": 24356,
  "deleted_data_files_count": 0,
  "added_rows_count": 0,
  "existing_rows_count": 785677,
  "deleted_rows_count": 0,
  "partitions": {
    "array": []
  }
}
{
  "manifest_path": "gs://<bucket>/<schema>/<table>//metadata/9753c692-b48e-49bb-9c27-713c8a41842e-m1.avro",
  "manifest_length": 8372685,
  "partition_spec_id": 0,
  "content": 0,
  "sequence_number": 4093,
  "min_sequence_number": 3557,
  "added_snapshot_id": 1992827298894205835,
  "added_data_files_count": 0,
  "existing_data_files_count": 19878,
  "deleted_data_files_count": 0,
  "added_rows_count": 0,
  "existing_rows_count": 1263244,
  "deleted_rows_count": 0,
  "partitions": {
    "array": []
  }
}

[DanielMorales9]

lgtm

[ahmedabu98]

@DanielMorales9 sorry but I did some experimenting and found it needed some more work, namely what was missing:

• Refreshing our cached tables, so that we can use the updated partition spec when writing
• Handling batches that contain files using old and new spec. Performing a single commit on all data files is no longer an option because commits are limited to one partition spec. However, performing multiple commits is not safe (if a bundle fails between commits, we may end up with data duplication). Instead, we can create a manifest file for each spec and append them all using one commit

I fixed this up in the previous commit baba789, can you take another look?

[DanielMorales9]

This is a long method 😲

[ahmedabu98]

@DanielMorales9 any more thoughts on this one?

[DanielMorales9]

Nope, LGTM 👍