[Data] Normalize block types before internal multi-block operations #43764
Conversation
Signed-off-by: Scott Lee <sjl@anyscale.com>
Signed-off-by: Scott Lee <sjl@anyscale.com>
scottjlee
requested review from
ericl,
scv119,
c21,
amogkam,
bveeramani,
raulchen,
stephanie-wang and
omatthew98
as code owners
| # If block types are different, but still both of TableBlock type, try | ||
| # converting both to default block type before zipping. | ||
| self_default, other_default = self.to_default(), acc.to_default() |
There was a problem hiding this comment.
can we call normalize_block_types instead to keep code to be consistent?
| self_default, other_default = self.to_default(), acc.to_default() | ||
| return BlockAccessor.for_block(self_default).zip(other_default) | ||
| else: | ||
| raise ValueError( |
There was a problem hiding this comment.
In whic case, this ValueError will be triggered?
There was a problem hiding this comment.
for Blocks which do not extend TableBlock class, i think this will be the case. since both ArrowBlock and PandasBlock are TableBlocks themselves, this isn't an issue for these classes, but this would cover any case in which we have other types of Blocks in the future. i can also remove this if we think it's not useful
| seen_types = set() | ||
| for block in blocks: | ||
| acc = BlockAccessor.for_block(block) | ||
| assert isinstance(acc, TableBlockAccessor), type(acc) |
There was a problem hiding this comment.
better to throw an actionable error message instead of assert
There was a problem hiding this comment.
@c21 do we still use non-table blocks anywhere?
There was a problem hiding this comment.
do we still use non-table blocks anywhere?
Actually I don't think we use non-table blocks anywhere.
| else: | ||
| results = [BlockAccessor.for_block(block).to_default() for block in blocks] | ||
|
|
||
| assert all(isinstance(block, type(results[0])) for block in results) |
| seen_types = set() | ||
| for block in blocks: | ||
| acc = BlockAccessor.for_block(block) | ||
| assert isinstance(acc, TableBlockAccessor), type(acc) |
There was a problem hiding this comment.
@c21 do we still use non-table blocks anywhere?
| def test_zip_multiple_block_types(ray_start_regular_shared): | ||
| df = pd.DataFrame({"spam": [0]}) | ||
| ds_pd = ray.data.from_pandas(df) | ||
| ds2_arrow = ray.data.from_items([{"ham": [0]}]) | ||
| assert ds_pd.zip(ds2_arrow).take_all() == [{"spam": 0, "ham": [0]}] |
There was a problem hiding this comment.
Should we move this to a different test module? I don't think zip is an all-to-all operation?
Signed-off-by: Scott Lee <sjl@anyscale.com>
Signed-off-by: Scott Lee <sjl@anyscale.com>
Why are these changes needed?
Applying grouping operations on
Datasets with different underlyingBlocktypes can cause exceptions (e.g. variousAttributeErrors) due toBlockAccessors assuming that all input block types are of the same type.We handle this case by normalizing the blocks (either
ArrowBlockorPandasBlock) to the first block type before applying the rest of the grouping/aggregation logic.Related issue number
Closes #31550
Closes #39206
Closes #39155
Closes #39291
Inspired by #39960
Checks
git commit -s) in this PR.scripts/format.shto lint the changes in this PR.method in Tune, I've added it in
doc/source/tune/api/under thecorresponding
.rstfile.