Repository navigation
Conversation
There was a problem hiding this comment.
Warning
Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.
Copilot review overview
3 open issues
What changed in this PR
Adds persistence of result-port metadata (declared output mode and row count) into operator_port_executions so completed workflow results can be listed without consulting the computing unit’s physical plan.
Changes:
- Introduces
output_mode_enum, plusoutput_modeandresult_countcolumns foroperator_port_executions(DDL + Liquibase update). - Extends
OperatorPortResultUriAvailableto carryOutputModeand persists it when storing result URIs. - Records table-result row counts into
RESULT_COUNTand adds/updates unit tests around mode/count behavior.
| File | Description |
|---|---|
| sql/updates/52.sql | Adds idempotent migration for enum + new columns |
| sql/texera_ddl.sql | Updates baseline schema with new enum + columns |
| sql/changelog.xml | Registers Liquibase changeSet for update 52 |
| amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala | Persists output mode; writes result_count via resource update |
| amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala | Adds updateResultCount DB update helper |
| amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala | Includes output mode in result-URI event |
| amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/ClientEvent.scala | Extends event payload to include output mode |
| amber/src/test/scala/org/apache/texera/web/service/ExecutionResultServiceSpec.scala | Adds coverage for output mode mapping and result_count persistence rules |
| amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala | Adds test for updateResultCount behavior |
Review effort: Lite
There was a problem hiding this comment.
Warning
Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.
Copilot review overview
2 open issues
3 resolved since last review
Review effort: Lite
| -- liquibase runner in sql/docker-compose.yml strips `SET search_path` out of these files before | ||
| -- applying them, while bin/local-dev.sh keeps it. | ||
|
|
||
| \c texera_db |
There was a problem hiding this comment.
Keeping it for consistency: 51 of the 52 scripts in sql/updates have it, and both runners strip it before executing (see #9049 (comment)).
| def updateResultCount( | ||
| eid: ExecutionIdentity, | ||
| globalPortId: GlobalPortIdentity, | ||
| count: Long | ||
| ): Unit = { | ||
| context | ||
| .update(OPERATOR_PORT_EXECUTIONS) | ||
| .set(OPERATOR_PORT_EXECUTIONS.RESULT_COUNT, java.lang.Long.valueOf(count)) | ||
| .where(OPERATOR_PORT_EXECUTIONS.WORKFLOW_EXECUTION_ID.eq(eid.id.toInt)) | ||
| .and(OPERATOR_PORT_EXECUTIONS.GLOBAL_PORT_ID.eq(globalPortId.serializeAsString)) | ||
| .execute() | ||
| } |
There was a problem hiding this comment.
Keeping Long, like updateResultSize beside it: the insert already leaves result_count null, and the only caller always has a count.
Automated Reviewer SuggestionsBased on the
|
operator_port_executions gains output_mode and result_count, so a finished execution's results can be listed without the computing unit: the mode lives only in the engine's physical plan, and the count otherwise needs each table opened. OperatorPortResultUriAvailable carries the port's declared mode, and the count is written where result_size already is, from the result store's own count. Closes apache#8994
- Schema-qualify the type and table in 52.sql: the liquibase runner strips `SET search_path` before applying update scripts. - Return Option from toOutputModeEnum instead of null. - Reword updateResultCount's Scaladoc to name the table it writes.
6305d59 to
9e316d6
Compare
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #9049 +/- ##
=========================================
Coverage 91.81% 91.81%
Complexity 5069 5069
=========================================
Files 1271 1271
Lines 55208 55221 +13
Branches 6987 6991 +4
=========================================
+ Hits 50688 50700 +12
Misses 2800 2800
- Partials 1720 1721 +1
*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 | 645 | 0.394 | 15,359/20,238/20,238 us | 🔴 +26.2% / 🔴 +65.5% |
| 🔴 | bs=100 sw=10 sl=64 | 1,281 | 0.782 | 78,832/92,925/92,925 us | 🔴 +12.5% / 🟢 -12.9% |
| ⚪ | bs=1000 sw=10 sl=64 | 1,543 | 0.942 | 646,062/725,515/725,515 us | ⚪ within ±5% / 🟢 -17.2% |
Baseline details
Latest main 1299992 from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 645 tuples/sec | 682 tuples/sec | 1,019 tuples/sec | -5.4% | -36.7% |
| bs=10 sw=10 sl=64 | MB/s | 0.394 MB/s | 0.416 MB/s | 0.622 MB/s | -5.3% | -36.7% |
| bs=10 sw=10 sl=64 | p50 | 15,359 us | 12,173 us | 10,144 us | +26.2% | +51.4% |
| bs=10 sw=10 sl=64 | p95 | 20,238 us | 23,786 us | 12,225 us | -14.9% | +65.5% |
| bs=10 sw=10 sl=64 | p99 | 20,238 us | 23,786 us | 15,527 us | -14.9% | +30.3% |
| bs=100 sw=10 sl=64 | throughput | 1,281 tuples/sec | 1,410 tuples/sec | 1,371 tuples/sec | -9.1% | -6.6% |
| bs=100 sw=10 sl=64 | MB/s | 0.782 MB/s | 0.861 MB/s | 0.837 MB/s | -9.2% | -6.6% |
| bs=100 sw=10 sl=64 | p50 | 78,832 us | 70,102 us | 79,910 us | +12.5% | -1.3% |
| bs=100 sw=10 sl=64 | p95 | 92,925 us | 98,341 us | 87,155 us | -5.5% | +6.6% |
| bs=100 sw=10 sl=64 | p99 | 92,925 us | 98,341 us | 106,643 us | -5.5% | -12.9% |
| bs=1000 sw=10 sl=64 | throughput | 1,543 tuples/sec | 1,515 tuples/sec | 1,415 tuples/sec | +1.8% | +9.0% |
| bs=1000 sw=10 sl=64 | MB/s | 0.942 MB/s | 0.925 MB/s | 0.864 MB/s | +1.8% | +9.0% |
| bs=1000 sw=10 sl=64 | p50 | 646,062 us | 647,810 us | 780,407 us | -0.3% | -17.2% |
| bs=1000 sw=10 sl=64 | p95 | 725,515 us | 741,454 us | 817,814 us | -2.1% | -11.3% |
| bs=1000 sw=10 sl=64 | p99 | 725,515 us | 741,454 us | 848,683 us | -2.1% | -14.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,309.99,200,128000,645,0.394,15358.89,20237.60,20237.60
1,100,10,64,20,1561.18,2000,1280000,1281,0.782,78832.38,92925.37,92925.37
2,1000,10,64,20,12959.21,20000,12800000,1543,0.942,646061.68,725515.37,725515.37
| WorkflowExecutionsResource.updateResultCount( | ||
| executionId, | ||
| globalPortIdOption.get, | ||
| info.tupleCount | ||
| ) |
There was a problem hiding this comment.
info.tupleCount belongs to the operator, not to this port. onResultUpdate builds the result store keyed by logical operator from every result URI, so when an operator has two result ports (Split, If), whichever URI comes last wins. That means this row (port 0) can get the other port's count, and its result_size, which comes from port 0's own document, then disagrees with it. It's also an Int (getCount.toInt) going into a BIGINT column. That matches the existing TODO, but since the description says "per result port", it may be worth either covering them or saying so.
| def updateResultCount( | ||
| eid: ExecutionIdentity, | ||
| globalPortId: GlobalPortIdentity, | ||
| count: Long | ||
| ): Unit = { | ||
| context | ||
| .update(OPERATOR_PORT_EXECUTIONS) | ||
| .set(OPERATOR_PORT_EXECUTIONS.RESULT_COUNT, java.lang.Long.valueOf(count)) | ||
| .where(OPERATOR_PORT_EXECUTIONS.WORKFLOW_EXECUTION_ID.eq(eid.id.toInt)) | ||
| .and(OPERATOR_PORT_EXECUTIONS.GLOBAL_PORT_ID.eq(globalPortId.serializeAsString)) | ||
| .execute() | ||
| } |
There was a problem hiding this comment.
This and updateResultSize update the same row and they run on every result-update tick. One method setting both columns would save a round trip per operator per tick and keep size and count written together.
What changes were proposed in this PR?
Records two facts per result port in
operator_port_executions, so a finished execution's results can be listed without the computing unit (#5363):output_mode(newoutput_mode_enum): whether the port's result is a paginated table or a visualization. Today only the engine's physical plan knows this, and it lives in the CU. The engine already looks up the output port when it creates the port's storage, soOperatorPortResultUriAvailablenow carries the port's declared mode, and the row is inserted with it.result_count: the number of rows, written whereresult_sizeis written today, from the count the result store already tracks — no extra table reads. Visualization (SINGLE_SNAPSHOT) ports skip that path and keep a null count; they are never paged.Schema change:
sql/updates/52.sql(idempotent),texera_ddl.sql, and achangelog.xmlentry. Existing rows keep both columns null. Nothing reads the columns yet; that is #8995.Any related issues, documentation, discussions?
Closes #8994. Part of #5363, proposed in #8945.
How was this PR tested?
ExecutionResultServiceSpec: the mode carried by the event is persisted; every output mode maps to its stored form and an unrecognized one to null; the result store's count is recorded for a table port, and no count for aSINGLE_SNAPSHOTport.WorkflowExecutionsResourceSpec:updateResultCountsets the column on the matching row.52.sqltwice to a local database (the second run is a no-op), and ran the fulltexera_ddl.sqlinto a scratch database.Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (claude-opus-5-5)