…-project#59848)

# Fix StreamingRepartition hang with empty upstream results

## Summary
Fix a bug where `StreamingRepartitionRefBundler` would hang when
processing empty datasets (0 rows).

## Problem
When upstream operations (e.g., `filter`, `map`, etc.) produce an empty
result (0 rows), resulting empty `RefBundle` gets added to
`_pending_bundles` but never gets flushed because:
1. `add_bundle()` adds empty bundles (0 rows) to `_pending_bundles`
2. `_total_pending_rows` remains 0
3. `done_adding_bundles()` checks `len(_pending_bundles) > 0` and calls
`flush_remaining=True`
4. `_try_build_ready_bundle(flush_remaining=True)` checks
`_total_pending_rows > 0` → False, so no flush happens
5. Empty bundles remain in `_pending_bundles` forever (memory leak)

## Reproduction
```python
import ray
ray.init()
ds = ray.data.range(5).filter(lambda row: row['id'] > 100)
ds = ds.repartition(target_num_rows_per_block=8)
ds.count()
```

## Solution
Changed flush condition in `_try_build_ready_bundle()` from checking
`_total_pending_rows > 0` to `len(self._pending_bundles) > 0`:

```python
# Before:
if flush_remaining and self._total_pending_rows > 0:

# After:
if flush_remaining and len(self._pending_bundles) > 0:
```

This ensures empty bundles never enter the bundler state, preventing
both hangs and memory leaks.

Signed-off-by: dragongu <andrewgu@vip.qq.com>
Signed-off-by: jasonwrwang <jasonwrwang@tencent.com>

…-project#59848)

# Fix StreamingRepartition hang with empty upstream results

## Summary
Fix a bug where `StreamingRepartitionRefBundler` would hang when
processing empty datasets (0 rows).

## Problem
When upstream operations (e.g., `filter`, `map`, etc.) produce an empty
result (0 rows), resulting empty `RefBundle` gets added to
`_pending_bundles` but never gets flushed because:
1. `add_bundle()` adds empty bundles (0 rows) to `_pending_bundles`
2. `_total_pending_rows` remains 0
3. `done_adding_bundles()` checks `len(_pending_bundles) > 0` and calls
`flush_remaining=True`
4. `_try_build_ready_bundle(flush_remaining=True)` checks
`_total_pending_rows > 0` → False, so no flush happens
5. Empty bundles remain in `_pending_bundles` forever (memory leak)

## Reproduction
```python
import ray
ray.init()
ds = ray.data.range(5).filter(lambda row: row['id'] > 100)
ds = ds.repartition(target_num_rows_per_block=8)
ds.count()
```

## Solution
Changed flush condition in `_try_build_ready_bundle()` from checking
`_total_pending_rows > 0` to `len(self._pending_bundles) > 0`:

```python
# Before:
if flush_remaining and self._total_pending_rows > 0:

# After:
if flush_remaining and len(self._pending_bundles) > 0:
```

This ensures empty bundles never enter the bundler state, preventing
both hangs and memory leaks.

Signed-off-by: dragongu <andrewgu@vip.qq.com>

…-project#59848)

# Fix StreamingRepartition hang with empty upstream results

## Summary
Fix a bug where `StreamingRepartitionRefBundler` would hang when
processing empty datasets (0 rows).

## Problem
When upstream operations (e.g., `filter`, `map`, etc.) produce an empty
result (0 rows), resulting empty `RefBundle` gets added to
`_pending_bundles` but never gets flushed because:
1. `add_bundle()` adds empty bundles (0 rows) to `_pending_bundles`
2. `_total_pending_rows` remains 0
3. `done_adding_bundles()` checks `len(_pending_bundles) > 0` and calls
`flush_remaining=True`
4. `_try_build_ready_bundle(flush_remaining=True)` checks
`_total_pending_rows > 0` → False, so no flush happens
5. Empty bundles remain in `_pending_bundles` forever (memory leak)

## Reproduction
```python
import ray
ray.init()
ds = ray.data.range(5).filter(lambda row: row['id'] > 100)
ds = ds.repartition(target_num_rows_per_block=8)
ds.count()
```

## Solution
Changed flush condition in `_try_build_ready_bundle()` from checking
`_total_pending_rows > 0` to `len(self._pending_bundles) > 0`:

```python
# Before:
if flush_remaining and self._total_pending_rows > 0:

# After:
if flush_remaining and len(self._pending_bundles) > 0:
```

This ensures empty bundles never enter the bundler state, preventing
both hangs and memory leaks.

Signed-off-by: dragongu <andrewgu@vip.qq.com>

…-project#59848)

# Fix StreamingRepartition hang with empty upstream results

## Summary
Fix a bug where `StreamingRepartitionRefBundler` would hang when
processing empty datasets (0 rows).

## Problem
When upstream operations (e.g., `filter`, `map`, etc.) produce an empty
result (0 rows), resulting empty `RefBundle` gets added to
`_pending_bundles` but never gets flushed because:
1. `add_bundle()` adds empty bundles (0 rows) to `_pending_bundles`
2. `_total_pending_rows` remains 0
3. `done_adding_bundles()` checks `len(_pending_bundles) > 0` and calls
`flush_remaining=True`
4. `_try_build_ready_bundle(flush_remaining=True)` checks
`_total_pending_rows > 0` → False, so no flush happens
5. Empty bundles remain in `_pending_bundles` forever (memory leak)

## Reproduction
```python
import ray
ray.init()
ds = ray.data.range(5).filter(lambda row: row['id'] > 100)
ds = ds.repartition(target_num_rows_per_block=8)
ds.count()
```

## Solution
Changed flush condition in `_try_build_ready_bundle()` from checking
`_total_pending_rows > 0` to `len(self._pending_bundles) > 0`:

```python
# Before:
if flush_remaining and self._total_pending_rows > 0:

# After:
if flush_remaining and len(self._pending_bundles) > 0:
```

This ensures empty bundles never enter the bundler state, preventing
both hangs and memory leaks.

Signed-off-by: dragongu <andrewgu@vip.qq.com>
Signed-off-by: peterxcli <peterxcli@gmail.com>

…-project#59848)

# Fix StreamingRepartition hang with empty upstream results

## Summary
Fix a bug where `StreamingRepartitionRefBundler` would hang when
processing empty datasets (0 rows).

## Problem
When upstream operations (e.g., `filter`, `map`, etc.) produce an empty
result (0 rows), resulting empty `RefBundle` gets added to
`_pending_bundles` but never gets flushed because:
1. `add_bundle()` adds empty bundles (0 rows) to `_pending_bundles`
2. `_total_pending_rows` remains 0
3. `done_adding_bundles()` checks `len(_pending_bundles) > 0` and calls
`flush_remaining=True`
4. `_try_build_ready_bundle(flush_remaining=True)` checks
`_total_pending_rows > 0` → False, so no flush happens
5. Empty bundles remain in `_pending_bundles` forever (memory leak)

## Reproduction
```python
import ray
ray.init()
ds = ray.data.range(5).filter(lambda row: row['id'] > 100)
ds = ds.repartition(target_num_rows_per_block=8)
ds.count()
```

## Solution
Changed flush condition in `_try_build_ready_bundle()` from checking
`_total_pending_rows > 0` to `len(self._pending_bundles) > 0`:

```python
# Before:
if flush_remaining and self._total_pending_rows > 0:

# After:
if flush_remaining and len(self._pending_bundles) > 0:
```

This ensures empty bundles never enter the bundler state, preventing
both hangs and memory leaks.

Signed-off-by: dragongu <andrewgu@vip.qq.com>
Signed-off-by: peterxcli <peterxcli@gmail.com>