feat(python): give a port that carried no rows its declared columns - #8488
feat(python): give a port that carried no rows its declared columns#8488kz930 wants to merge 1 commit into
Conversation
A port can finish having carried no rows: an upstream filter that matches nothing still ends its channel, and DataProcessor calls on_finish either way. TableOperator then built its table out of no tuples, and since the column names are read off the tuples, the operator was handed a frame of no columns at all. Every table operator that names one of its own columns raised KeyError on it. The runtime now records what each input port was declared to carry, and the table falls back to it where the tuples cannot say. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Automated Reviewer SuggestionsBased on the
|
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #8488 +/- ##
=========================================
Coverage 93.69% 93.69%
Complexity 4826 4826
=========================================
Files 1209 1209
Lines 49871 49890 +19
Branches 6099 6101 +2
=========================================
+ Hits 46727 46746 +19
Misses 1652 1652
Partials 1492 1492
*This pull request uses carry forward flags. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
| config | throughput | MB/s | latency | max Δ latest / 7d | |
|---|---|---|---|---|---|
| 🔴 | bs=10 sw=10 sl=64 | 345 | 0.21 | 27,914/38,715/38,715 us | 🔴 +20.0% / 🔴 +140.0% |
| 🔴 | bs=100 sw=10 sl=64 | 774 | 0.472 | 128,505/153,463/153,463 us | 🟢 -7.5% / 🔴 +41.2% |
| ⚪ | bs=1000 sw=10 sl=64 | 920 | 0.561 | 1,091,359/1,141,539/1,141,539 us | ⚪ within ±5% / 🔴 +10.0% |
Baseline details
Latest main 1fbd346 from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 345 tuples/sec | 388 tuples/sec | 755.28 tuples/sec | -11.1% | -54.3% |
| bs=10 sw=10 sl=64 | MB/s | 0.21 MB/s | 0.237 MB/s | 0.461 MB/s | -11.4% | -54.4% |
| bs=10 sw=10 sl=64 | p50 | 27,914 us | 23,270 us | 12,957 us | +20.0% | +115.4% |
| bs=10 sw=10 sl=64 | p95 | 38,715 us | 39,438 us | 16,134 us | -1.8% | +140.0% |
| bs=10 sw=10 sl=64 | p99 | 38,715 us | 39,438 us | 20,333 us | -1.8% | +90.4% |
| bs=100 sw=10 sl=64 | throughput | 774 tuples/sec | 806 tuples/sec | 980.1 tuples/sec | -4.0% | -21.0% |
| bs=100 sw=10 sl=64 | MB/s | 0.472 MB/s | 0.492 MB/s | 0.598 MB/s | -4.1% | -21.1% |
| bs=100 sw=10 sl=64 | p50 | 128,505 us | 120,490 us | 101,894 us | +6.7% | +26.1% |
| bs=100 sw=10 sl=64 | p95 | 153,463 us | 165,976 us | 108,718 us | -7.5% | +41.2% |
| bs=100 sw=10 sl=64 | p99 | 153,463 us | 165,976 us | 122,482 us | -7.5% | +25.3% |
| bs=1000 sw=10 sl=64 | throughput | 920 tuples/sec | 915 tuples/sec | 1,011 tuples/sec | +0.5% | -9.0% |
| bs=1000 sw=10 sl=64 | MB/s | 0.561 MB/s | 0.558 MB/s | 0.617 MB/s | +0.5% | -9.1% |
| bs=1000 sw=10 sl=64 | p50 | 1,091,359 us | 1,090,198 us | 996,422 us | +0.1% | +9.5% |
| bs=1000 sw=10 sl=64 | p95 | 1,141,539 us | 1,152,241 us | 1,037,670 us | -0.9% | +10.0% |
| bs=1000 sw=10 sl=64 | p99 | 1,141,539 us | 1,152,241 us | 1,072,152 us | -0.9% | +6.5% |
Raw CSV
config_idx,batch_size,schema_width,string_len,num_batches,total_ms,total_tuples,total_bytes,tuples_per_sec,mb_per_sec,lat_p50_us,lat_p95_us,lat_p99_us
0,10,10,64,20,580.06,200,128000,345,0.210,27914.10,38714.64,38714.64
1,100,10,64,20,2583.98,2000,1280000,774,0.472,128504.72,153463.06,153463.06
2,1000,10,64,20,21742.96,20000,12800000,920,0.561,1091359.37,1141539.18,1141539.18
|
@carloea2 this one is a platform bug rather than an export change: a port that finishes having carried no rows hands a table operator a frame with no columns at all, so every operator naming one of its own columns raises KeyError. Would you take a look when you have a moment? |
| table. Building it through Arrow gives each column the dtype it would | ||
| have had with rows in it. | ||
| """ | ||
| return pa.Table.from_pylist([], schema=schema.as_arrow_schema()).to_pandas() |
There was a problem hiding this comment.
Could this return a Table rather than a plain DataFrame? A TableOperator UDF using yield from table.as_tuples() now raises AttributeError when its input is empty and has a declared schema. The same UDF still works with populated input and with the previous empty-input path. I reproduced this against 2004373 with three local pytest cases: those two controls pass, while the declared empty input fails. Wrapping the frame in Table would preserve the existing UDF API while keeping the columns.
3 participants
What changes were proposed in this PR?
The runtime now tells an executor what each input port was declared to carry, and
TableOperatorfalls back to that when a port finishes with no rows.Operatorgains aninput_schemasmapping, keyed by port index.DataProcessorwrites the finishing port's schema into it just before callingon_finish, reading it from the input manager's own port; a source has no input port to ask, so it is left alone.TableOperator.on_finishuses it only when there are no tuples to read column names off, which is the only case where the tuples do not already say the same thing.Table.empty_ofbuilds that frame through Arrow, so each column carries the dtype it would have had with rows in it rather than object.Any related issues, documentation, discussions?
Found while building #8325, but not part of it: the bug is in the engine, not in
the export. The export's verification is what surfaced it, by running each
operator on a table with no rows, but it reaches Sort, the visualization
operators and any user-written
UDFTableOperatorin an ordinary run.Closes #8487, the bug this change is the whole of.
How was this PR tested?
A new case in
TestTableOperator,test_on_finish_with_no_rows_keeps_the_declared_columns, gives a port an INTEGER and a STRING column, finishes it with no rows, and asserts the operator receives both columns, an empty frame, andint32for the integer one. Removing the fallback turns it red on the missing columns.The rest of
amber/src/test/python/corepasses unchanged, 1107 tests. The one failure istest_iceberg_rest_catalog_integration, which needs a running catalog and fails the same way without this change.Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5)
🤖 Generated with Claude Code