[C++] Scanning datasets does not enforce back pressure

[asfimport]

I have a simple test case where I scan the batches of a 4GB dataset and print out the currently used memory:

import pyarrow as pa
import pyarrow.dataset as ds

dataset = ds.dataset('/home/pace/dev/data/dataset/csv/5_big', format='csv')
num_rows = 0
for batch in dataset.to_batches():
 print(pa.total_allocated_bytes())
 num_rows += batch.num_rows

print(num_rows)

In pyarrow 3.0.0 this consumes just over 5MB. In pyarrow 4.0.0 and 5.0.0 this consumes multiple GB of RAM.

Reporter: Weston Pace / @westonpace
Assignee: Weston Pace / @westonpace

Related issues:

• [C++] Ensure dataset writing applies back pressure (is duplicated by)
• [C++][Dataset] Dataset writes should respect backpressure (is depended upon by)

PRs and other links:

• GitHub Pull Request #11285

_{Note: This issue was originally created as ARROW-13611. Please see the migration documentation for further details.}

[asfimport]

Weston Pace / @westonpace:
It appears to make no difference if use_async is True or False.

[asfimport]

Weston Pace / @westonpace:
Additional information.

The synchronous scanner is properly applying back pressure when the source has multiple scan tasks (e.g. parquet). However it is not applying back pressure when sources have a single scan task (e.g. CSV / IPC). The root cause is here:

arrow/cpp/src/arrow/dataset/scanner.cc

Line 207 in f959141

for (auto maybe_batch : batch_it) {

The asynchronous scanner is not applying back pressure because it is now connected to an ExecPlan and ExecPlan does not have back pressure implemented yet.

I'm not sure how much effort we want to put into fixing the synchronous scanner given that it is slated for removal.

[asfimport]

Carl Boettiger / @cboettig:
Any news on this? I believe this bug is still the cause of the system crashes I see when trying to access large parquet files in arrow, e.g. in R:

library(arrow)
library(dplyr)
file <- "part-0.parquet"download.file("https://minio.cirrus.carlboettiger.info/shared-data/birddb/parquet/part-0.parquet", file)
ds <- open_dataset(file, format = "parquet")
ds %>% filter(COUNTRY == "Mexico", `COMMON NAME`=="Wood thrush") %>% compute()

[asfimport]

Weston Pace / @westonpace:
That is actually what I've been working on today (shame on me for not toggling Start Progress). Right now I have a PR up that adds backpressure back in for unordered scans. I expect to have a PR ready that will add backpressure for ordered scans soon (I am hoping tomorrow). If you are interested I can probably get you a test wheel (are you installing with conda or pypi?) sometime next week and would appreciate if you could let me know if it solves your issue.

[asfimport]

Weston Pace / @westonpace:
Issue resolved by pull request 11285
#11285