[data][2/n] Ordered + FIFO bundle queue by iamjustinhsu · Pull Request #60228 · ray-project/ray

iamjustinhsu · 2026-01-16T18:22:31Z

Description

Moves OrderedBundleQueue to bundle_queue/ directory, and rebases off of the BaseBundleQueue class. Creates a FIFOBundleQueue for simple ref bundle queues.

Old PR: #59093

There are about ~400 lines of tests, so not actually that big

Related issues

Additional information

Signed-off-by: iamjustinhsu <jhsu@anyscale.com>

gemini-code-assist

Code Review

This pull request refactors the bundle queue logic by introducing OrderedBundleQueue and FIFOBundleQueue based on a common BaseBundleQueue. The changes are mostly clean and improve the structure of the code. However, I've found a few critical issues where methods are called on queue objects that are not implemented, which will lead to runtime errors. Please see the detailed comments for specifics.

python/ray/data/_internal/execution/operators/base_physical_operator.py

python/ray/data/_internal/execution/operators/map_operator.py

python/ray/data/_internal/execution/operators/output_splitter.py

Signed-off-by: iamjustinhsu <jhsu@anyscale.com>

cursor

Cursor Bugbot has reviewed your changes and found 1 potential issue.

python/ray/data/_internal/execution/operators/output_splitter.py

Signed-off-by: iamjustinhsu <jhsu@anyscale.com>

…/unify-internal-queue-abstraction-add-fifo-and-ordered2

Signed-off-by: iamjustinhsu <jhsu@anyscale.com>

iamjustinhsu · 2026-01-22T16:34:20Z

python/ray/data/_internal/execution/bundle_queue/ordered.py

+    from ray.data._internal.execution.interfaces import RefBundle
+
+
+class OrderedBundleQueue(BaseBundleQueue):


Moved from map_operator.py and renamed to _OrderedOutputQueue

Let's name it ReorderingBundleQueue (b/c all queues are ordered)

alexeykudinkin · 2026-01-23T00:50:19Z

python/ray/data/_internal/execution/bundle_queue/ordered.py

+    from ray.data._internal.execution.interfaces import RefBundle
+
+
+class OrderedBundleQueue(BaseBundleQueue):


Let's name it ReorderingBundleQueue (b/c all queues are ordered)

Signed-off-by: iamjustinhsu <jhsu@anyscale.com>

## Description Moves OrderedBundleQueue to `bundle_queue/` directory, and rebases off of the `BaseBundleQueue` class. Creates a `FIFOBundleQueue` for simple ref bundle queues. Old PR: ray-project#59093 There are about ~400 lines of tests, so not actually that big ## Related issues ## Additional information --------- Signed-off-by: iamjustinhsu <jhsu@anyscale.com>

…eue` (#60538) #60017 and #60228 refactored the `FIFOBundleQueue` interface and renamed `FIFOBundleQueue.popleft` with `FIFOBundleQueue.get_next`. However, this name change wasn't reflected in the `UnionOperator` implementation, and as a result the operator can error when it clears its output queue. This change also fixes the flaky `test_union.py`. Signed-off-by: Balaji Veeramani <bveeramani@berkeley.edu>

## Description Moves OrderedBundleQueue to `bundle_queue/` directory, and rebases off of the `BaseBundleQueue` class. Creates a `FIFOBundleQueue` for simple ref bundle queues. Old PR: ray-project#59093 There are about ~400 lines of tests, so not actually that big ## Related issues ## Additional information --------- Signed-off-by: iamjustinhsu <jhsu@anyscale.com> Signed-off-by: jinbum-kim <jinbum9958@gmail.com>

…eue` (ray-project#60538) ray-project#60017 and ray-project#60228 refactored the `FIFOBundleQueue` interface and renamed `FIFOBundleQueue.popleft` with `FIFOBundleQueue.get_next`. However, this name change wasn't reflected in the `UnionOperator` implementation, and as a result the operator can error when it clears its output queue. This change also fixes the flaky `test_union.py`. Signed-off-by: Balaji Veeramani <bveeramani@berkeley.edu> Signed-off-by: jinbum-kim <jinbum9958@gmail.com>

…eue` (ray-project#60538) ray-project#60017 and ray-project#60228 refactored the `FIFOBundleQueue` interface and renamed `FIFOBundleQueue.popleft` with `FIFOBundleQueue.get_next`. However, this name change wasn't reflected in the `UnionOperator` implementation, and as a result the operator can error when it clears its output queue. This change also fixes the flaky `test_union.py`. Signed-off-by: Balaji Veeramani <bveeramani@berkeley.edu>

## Description Moves OrderedBundleQueue to `bundle_queue/` directory, and rebases off of the `BaseBundleQueue` class. Creates a `FIFOBundleQueue` for simple ref bundle queues. Old PR: ray-project#59093 There are about ~400 lines of tests, so not actually that big ## Related issues ## Additional information --------- Signed-off-by: iamjustinhsu <jhsu@anyscale.com> Signed-off-by: 400Ping <jiekaichang@apache.org>

…eue` (ray-project#60538) ray-project#60017 and ray-project#60228 refactored the `FIFOBundleQueue` interface and renamed `FIFOBundleQueue.popleft` with `FIFOBundleQueue.get_next`. However, this name change wasn't reflected in the `UnionOperator` implementation, and as a result the operator can error when it clears its output queue. This change also fixes the flaky `test_union.py`. Signed-off-by: Balaji Veeramani <bveeramani@berkeley.edu> Signed-off-by: 400Ping <jiekaichang@apache.org>

…eue` (ray-project#60538) ray-project#60017 and ray-project#60228 refactored the `FIFOBundleQueue` interface and renamed `FIFOBundleQueue.popleft` with `FIFOBundleQueue.get_next`. However, this name change wasn't reflected in the `UnionOperator` implementation, and as a result the operator can error when it clears its output queue. This change also fixes the flaky `test_union.py`. Signed-off-by: Balaji Veeramani <bveeramani@berkeley.edu> Signed-off-by: Adel Nour <ans9868@nyu.edu>

## Description Moves OrderedBundleQueue to `bundle_queue/` directory, and rebases off of the `BaseBundleQueue` class. Creates a `FIFOBundleQueue` for simple ref bundle queues. Old PR: ray-project#59093 There are about ~400 lines of tests, so not actually that big ## Related issues ## Additional information --------- Signed-off-by: iamjustinhsu <jhsu@anyscale.com> Signed-off-by: peterxcli <peterxcli@gmail.com>

…eue` (ray-project#60538) ray-project#60017 and ray-project#60228 refactored the `FIFOBundleQueue` interface and renamed `FIFOBundleQueue.popleft` with `FIFOBundleQueue.get_next`. However, this name change wasn't reflected in the `UnionOperator` implementation, and as a result the operator can error when it clears its output queue. This change also fixes the flaky `test_union.py`. Signed-off-by: Balaji Veeramani <bveeramani@berkeley.edu> Signed-off-by: peterxcli <peterxcli@gmail.com>

## Description Moves OrderedBundleQueue to `bundle_queue/` directory, and rebases off of the `BaseBundleQueue` class. Creates a `FIFOBundleQueue` for simple ref bundle queues. Old PR: ray-project#59093 There are about ~400 lines of tests, so not actually that big ## Related issues ## Additional information --------- Signed-off-by: iamjustinhsu <jhsu@anyscale.com> Signed-off-by: peterxcli <peterxcli@gmail.com>

…eue` (ray-project#60538) ray-project#60017 and ray-project#60228 refactored the `FIFOBundleQueue` interface and renamed `FIFOBundleQueue.popleft` with `FIFOBundleQueue.get_next`. However, this name change wasn't reflected in the `UnionOperator` implementation, and as a result the operator can error when it clears its output queue. This change also fixes the flaky `test_union.py`. Signed-off-by: Balaji Veeramani <bveeramani@berkeley.edu> Signed-off-by: peterxcli <peterxcli@gmail.com>

[data] Ordered + FIFO bundle queue

15df785

Signed-off-by: iamjustinhsu <jhsu@anyscale.com>

iamjustinhsu requested a review from a team as a code owner January 16, 2026 18:22

gemini-code-assist bot reviewed Jan 16, 2026

View reviewed changes

remove dead code

234f23f

Signed-off-by: iamjustinhsu <jhsu@anyscale.com>

This comment was marked as outdated.

Sign in to view

ray-gardener bot added the data Ray Data-related issues label Jan 16, 2026

general fixes

f06b8af

Signed-off-by: iamjustinhsu <jhsu@anyscale.com>

iamjustinhsu changed the title ~~[data] Ordered + FIFO bundle queue~~ [data][2/n] Ordered + FIFO bundle queue Jan 16, 2026

This comment was marked as outdated.

Sign in to view

fix missing args

482c5d3

Signed-off-by: iamjustinhsu <jhsu@anyscale.com>

This comment was marked as outdated.

Sign in to view

remove python. imports

d61a08e

Signed-off-by: iamjustinhsu <jhsu@anyscale.com>

iamjustinhsu added go add ONLY when ready to merge, run all tests and removed go add ONLY when ready to merge, run all tests labels Jan 16, 2026

iamjustinhsu added 2 commits January 16, 2026 23:08

use override methods for repr

8049ba7

Signed-off-by: iamjustinhsu <jhsu@anyscale.com>

rebase

09a8eb6

Signed-off-by: iamjustinhsu <jhsu@anyscale.com>

cursor bot reviewed Jan 21, 2026

View reviewed changes

python/ray/data/_internal/execution/operators/output_splitter.py Show resolved Hide resolved

iamjustinhsu added 5 commits January 21, 2026 10:00

add ordered bundle queue tests

cf63dcf

Signed-off-by: iamjustinhsu <jhsu@anyscale.com>

add fifo bundle queue tests

f781812

Signed-off-by: iamjustinhsu <jhsu@anyscale.com>

Merge branch 'master' of https://github.com/ray-project/ray into jhsu…

0f958d8

…/unify-internal-queue-abstraction-add-fifo-and-ordered2

fix

be5efe9

Signed-off-by: iamjustinhsu <jhsu@anyscale.com>

lint

6eb4bc6

Signed-off-by: iamjustinhsu <jhsu@anyscale.com>

iamjustinhsu commented Jan 22, 2026

View reviewed changes

alexeykudinkin approved these changes Jan 23, 2026

View reviewed changes

rename to ReorderingBundleQueue

7f7167d

Signed-off-by: iamjustinhsu <jhsu@anyscale.com>

iamjustinhsu added the go add ONLY when ready to merge, run all tests label Jan 23, 2026

alexeykudinkin merged commit db06618 into ray-project:master Jan 24, 2026
7 checks passed

bveeramani mentioned this pull request Jan 27, 2026

[Data] Fix attribute error in UnionOperator.clear_internal_output_queue #60538

Merged

alexeykudinkin mentioned this pull request Jan 27, 2026

[Data] Fixing ReorderingBundleQueue handling of empty output sequences #60470

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[data][2/n] Ordered + FIFO bundle queue#60228

[data][2/n] Ordered + FIFO bundle queue#60228
alexeykudinkin merged 13 commits intoray-project:masterfrom
iamjustinhsu:jhsu/unify-internal-queue-abstraction-add-fifo-and-ordered2

iamjustinhsu commented Jan 16, 2026 •

edited

Loading

Uh oh!

gemini-code-assist bot left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

This comment was marked as outdated.

Uh oh!

This comment was marked as outdated.

Uh oh!

This comment was marked as outdated.

Uh oh!

cursor bot left a comment

Uh oh!

Uh oh!

iamjustinhsu Jan 22, 2026 •

edited

Loading

Uh oh!

alexeykudinkin Jan 23, 2026

Uh oh!

alexeykudinkin Jan 23, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

		from ray.data._internal.execution.interfaces import RefBundle


		class OrderedBundleQueue(BaseBundleQueue):

Conversation

iamjustinhsu commented Jan 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Related issues

Additional information

Uh oh!

gemini-code-assist bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

This comment was marked as outdated.

Uh oh!

This comment was marked as outdated.

Uh oh!

This comment was marked as outdated.

Uh oh!

cursor bot left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

iamjustinhsu Jan 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

alexeykudinkin Jan 23, 2026

Choose a reason for hiding this comment

Uh oh!

alexeykudinkin Jan 23, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

iamjustinhsu commented Jan 16, 2026 •

edited

Loading

iamjustinhsu Jan 22, 2026 •

edited

Loading