(tempdir)
| 4340 | @pytest.mark.parquet |
| 4341 | @pytest.mark.threading |
| 4342 | def test_write_dataset_with_backpressure(tempdir): |
| 4343 | consumer_gate = threading.Event() |
| 4344 | |
| 4345 | # A filesystem that blocks all writes so that we can build |
| 4346 | # up backpressure. The writes are released at the end of |
| 4347 | # the test. |
| 4348 | class GatingFs(ProxyHandler): |
| 4349 | def open_output_stream(self, path, metadata): |
| 4350 | # Block until the end of the test |
| 4351 | consumer_gate.wait() |
| 4352 | return self._fs.open_output_stream(path, metadata=metadata) |
| 4353 | gating_fs = fs.PyFileSystem(GatingFs(fs.LocalFileSystem())) |
| 4354 | |
| 4355 | schema = pa.schema([pa.field('data', pa.int32())]) |
| 4356 | # The scanner should queue ~ 8Mi rows (~8 batches) but due to ARROW-16258 |
| 4357 | # it always queues 32 batches. |
| 4358 | batch = pa.record_batch([pa.array(list(range(1_000_000)))], schema=schema) |
| 4359 | batches_read = 0 |
| 4360 | min_backpressure = 32 |
| 4361 | end = 200 |
| 4362 | keep_going = True |
| 4363 | |
| 4364 | def counting_generator(): |
| 4365 | nonlocal batches_read |
| 4366 | while batches_read < end: |
| 4367 | if not keep_going: |
| 4368 | return |
| 4369 | time.sleep(0.01) |
| 4370 | batches_read += 1 |
| 4371 | yield batch |
| 4372 | |
| 4373 | scanner = ds.Scanner.from_batches( |
| 4374 | counting_generator(), schema=schema, use_threads=True) |
| 4375 | |
| 4376 | write_thread = threading.Thread( |
| 4377 | target=lambda: ds.write_dataset( |
| 4378 | scanner, str(tempdir), format='parquet', filesystem=gating_fs)) |
| 4379 | write_thread.start() |
| 4380 | |
| 4381 | try: |
| 4382 | start = time.time() |
| 4383 | |
| 4384 | def duration(): |
| 4385 | return time.time() - start |
| 4386 | |
| 4387 | # This test is timing dependent. There is no signal from the C++ |
| 4388 | # when backpressure has been hit. We don't know exactly when |
| 4389 | # backpressure will be hit because it may take some time for the |
| 4390 | # signal to get from the sink to the scanner. |
| 4391 | # |
| 4392 | # The test may emit false positives on slow systems. It could |
| 4393 | # theoretically emit a false negative if the scanner managed to read |
| 4394 | # and emit all 200 batches before the backpressure signal had a chance |
| 4395 | # to propagate but the 0.01s delay in the generator should make that |
| 4396 | # scenario unlikely. |
| 4397 | last_value = 0 |
| 4398 | backpressure_probably_hit = False |
| 4399 | while duration() < 10: |
nothing calls this directly
no test coverage detected