Support bucket table for Iceberg

[jerryshao]

Current Iceberg doesn't support "bucket" semantics both in read and write, so we cannot leverage this to do bucketed join. We should add such support in Iceberg.

[aokolnychyi]

This would be a great feature to have. I think Spark might have to be adapted as well.

[jerryshao]

@rdblue @aokolnychyi I wrote a simple design doc about bucketing support in Iceberg, would you please help to review, appreciate your time.

https://docs.google.com/document/d/1X3tpcJFz8Fd9m2SixHP4psBFHc39Y21TXev4i26ve-I/edit?usp=sharing

[rdblue]

@jerryshao, thanks for posting this! I'll take a look as soon as I can, but I'm going to be at a conference next week so it may not be quick.

[jerryshao]

Sure, no problem, take your time :).

[jerryshao]

I'm roughly dividing this issue into 3 ongoing PRs:

1. Add Bucket Spec support in Iceberg API and metadata.
2. Support bucketing write in Iceberg Spark data source.
3. Support bucketing read in Iceberg Spark data source.

Currently I'm working on the first task.

[rdblue]

Thanks @jerryshao! I had a look at the doc and made some comments.

The main thing is that Iceberg already supports bucketing and has solved many of the challenges you identified, like schema evolution. There are two remaining problems:

1. Writing requires users to cluster data into buckets using a UDF.
2. Bucketed joins can't take advantage of Iceberg bucket values.

For problem 1, we need to allow Iceberg to control the requiredChildDistribution and requiredChildOrdering returned by WriteToDataSourceV2Exec. Here's a gist that shows what we use to automatically insert distribution/ordering requirements that allow automatic bucketing writes. That also depends on #317.

We also need a FunctionCatalog that allows us to return Iceberg transforms as UDFs that Spark can use.

For problem 2, we are planning to add support for Spark to be able to use bucket values to speed up joins. We aren't quite sure how to do this yet, but we know that Spark will need to recognize that a table is bucketed (using the Table's partitioning), get the bucket function from the table's catalog (using FunctionCatalog) and use that function to prepare data for the other side of the join. If the other side of the join uses the same partition function, then we can avoid a shuffle for that side of the join as well.

Hopefully this short write-up and the comments I left on the doc give you an idea of the current status of bucketed joins. Thanks for working on this!

[jerryshao]

Thanks a lot @rdblue and sorry for late response. I see from your point that you want to reuse bucket transform, and the key thing is that current Spark doesn't aware of child distribution regarding to #274 , let me revamp the whole design. Thanks again for your comments.

[jerryshao]

Also for #274, IIUC most of the partition transformations requires data sorted by this transformation functions, otherwise data cannot be grouped together, and will throw exceptions in partitioned writer, am I right? @rdblue

[rdblue]

@jerryshao, yes that's correct.

That's why we need to expose the transformation functions to Spark via FunctionCatalog, and add the ability for DSv2 sources to set distribution and ordering requirements with those functions.

[aokolnychyi]

@dbtsai, FYI

[aokolnychyi]

I am preparing a few optimizations for metadata compaction and will work on sort spec next.

[yupbank]

is there any progress on this ?

[rdblue]

@yupbank, @sunchao has a design doc for this that he's planning on sharing with the Spark community soon. I think that there is also some work going on to enable bucketed joins in Trino.

[sunchao]

Hi @yupbank , yes like @rdblue said I'm in the process of wrapping up a design doc for this feature. Will cross link it here soon!

[yupbank]

Nice! let me know if you need extra eyes, would love to help, as we run into the issue of shuffling big records recently