This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6980-a61702fd10ccde130f0a24a1298adff24c4a38a5 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 6d038ddefbe96c7055803a588a72031654374f7c Author: Meng Wang <[email protected]> AuthorDate: Tue Jul 28 23:11:38 2026 -0700 fix(amber): store execution sizes as BIGINT to stop silent >2GiB truncation (#6980) ### What changes were proposed in this PR? `updateResultSize` / `updateRuntimeStatsSize` / `updateConsoleMessageSize` (`WorkflowExecutionsResource`) stored `Long` byte counts into `INT` columns via `Integer.valueOf(size.toInt)`. Scala's `Long.toInt` keeps only the low 32 bits without raising, so a size ≥ 2 GiB wrapped silently — values in [2 GiB, 4 GiB) became **negative** — and `UserQuotaResource`, which sums `result_size` / `runtime_stats_size` / `console_messages_size` into a user's storage quota, reported corrupted totals. With BigObject (#4067) supporting >2 GB results, such sizes are reachable in practice. - **Widen the three columns to `BIGINT`** in `sql/texera_ddl.sql`, with migration `sql/updates/29.sql` (registered as changelog changeSet 29) for existing deployments — a lossless in-place `ALTER COLUMN ... TYPE BIGINT` for each. - **Store the `Long` directly** at the three write sites, dropping the `.toInt` narrowing (`java.lang.Long.valueOf(size)`; the jOOQ-generated fields become `Long` from the widened schema). - **Adapt the quota reads** in `UserQuotaResource`: the `getOrElse(0).asInstanceOf[Integer]` pattern would throw `ClassCastException` on the now-`Long` fields; simplified to `Option(...).map(_.toLong).getOrElse(0L)`. - **Split the size write out of `updateRuntimeStatsSize` / `updateConsoleMessageSize`** into `(eid, size)` overloads, mirroring the existing `updateResultSize` shape. The outer signatures are unchanged (callers untouched), but the DB write is now reachable without an Iceberg/LakeFS-backed document — so all three writes are directly testable. ### Any related issues, documentation, discussions? Closes #6978. Size columns introduced with the execution result/stats storage; >2 GB results enabled by #4067 (BigObject). ### How was this PR tested? - Added regression cases to `WorkflowExecutionsResourceSpec` (unit spec on embedded Postgres, no external infra): each of the three size writes stores a 3 GiB value and is asserted to round-trip untruncated, plus the two no-URI no-op branches. **Verified the truncation case fails before the fix** — `-1073741824 did not equal 3221225472` (the low-32-bit wrap) — and passes after. - Full spec run locally: 27/27 passed (jOOQ regenerated against the widened schema; the embedded test DB loads the updated `texera_ddl.sql`). - `sql/updates/29.sql` applied cleanly to a local Postgres 15 `texera_db` (three `ALTER TABLE`s in one transaction); columns verified `bigint` afterwards. - `WorkflowExecutionService/scalafmtCheck` (main + Test) passes. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (claude-opus-4-8) --------- Co-authored-by: Yicong Huang <[email protected]> Co-authored-by: Xinyuan Lin <[email protected]> --- .../dashboard/user/quota/UserQuotaResource.scala | 10 +-- .../user/workflow/WorkflowExecutionsResource.scala | 54 +++++++++---- .../workflow/WorkflowExecutionsResourceSpec.scala | 91 ++++++++++++++++++++++ sql/changelog.xml | 5 ++ sql/texera_ddl.sql | 6 +- sql/updates/29.sql | 40 ++++++++++ 6 files changed, 181 insertions(+), 25 deletions(-) diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/quota/UserQuotaResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/quota/UserQuotaResource.scala index 6f1bb79b15..7f89d3fe1c 100644 --- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/quota/UserQuotaResource.scala +++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/quota/UserQuotaResource.scala @@ -149,10 +149,7 @@ object UserQuotaResource { .where(OPERATOR_PORT_EXECUTIONS.WORKFLOW_EXECUTION_ID.eq(eid)) .fetch() .asScala - .map(r => - Option(r.get(OPERATOR_PORT_EXECUTIONS.RESULT_SIZE)).getOrElse(0).asInstanceOf[Integer] - ) - .map(_.toLong) + .map(r => Option(r.get(OPERATOR_PORT_EXECUTIONS.RESULT_SIZE)).map(_.toLong).getOrElse(0L)) .sum val logSize = context @@ -162,11 +159,8 @@ object UserQuotaResource { .fetch() .asScala .map(r => - Option(r.get(OPERATOR_EXECUTIONS.CONSOLE_MESSAGES_SIZE)) - .getOrElse(0) - .asInstanceOf[Integer] + Option(r.get(OPERATOR_EXECUTIONS.CONSOLE_MESSAGES_SIZE)).map(_.toLong).getOrElse(0L) ) - .map(_.toLong) .sum QuotaStorage( diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala index 784d678f37..bca17d9eed 100644 --- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala +++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala @@ -404,7 +404,7 @@ object WorkflowExecutionsResource { ): Unit = { context .update(OPERATOR_PORT_EXECUTIONS) - .set(OPERATOR_PORT_EXECUTIONS.RESULT_SIZE, Integer.valueOf(size.toInt)) + .set(OPERATOR_PORT_EXECUTIONS.RESULT_SIZE, java.lang.Long.valueOf(size)) .where(OPERATOR_PORT_EXECUTIONS.WORKFLOW_EXECUTION_ID.eq(eid.id.toInt)) .and(OPERATOR_PORT_EXECUTIONS.GLOBAL_PORT_ID.eq(globalPortId.serializeAsString)) .execute() @@ -424,15 +424,27 @@ object WorkflowExecutionsResource { .map(URI.create) if (statsUriOpt.isPresent) { - val size = DocumentFactory.openDocument(statsUriOpt.get)._1.getTotalFileSize - context - .update(WORKFLOW_EXECUTIONS) - .set(WORKFLOW_EXECUTIONS.RUNTIME_STATS_SIZE, Integer.valueOf(size.toInt)) - .where(WORKFLOW_EXECUTIONS.EID.eq(eid.id.toInt)) - .execute() + updateRuntimeStatsSize( + eid, + DocumentFactory.openDocument(statsUriOpt.get)._1.getTotalFileSize + ) } } + /** + * Stores an already-measured runtime statistics size, mirroring updateResultSize. + * + * @param eid Execution ID associated with the runtime statistics document. + * @param size Size of the runtime statistics in bytes. + */ + def updateRuntimeStatsSize(eid: ExecutionIdentity, size: Long): Unit = { + context + .update(WORKFLOW_EXECUTIONS) + .set(WORKFLOW_EXECUTIONS.RUNTIME_STATS_SIZE, java.lang.Long.valueOf(size)) + .where(WORKFLOW_EXECUTIONS.EID.eq(eid.id.toInt)) + .execute() + } + /** * Updates the size of the console message stored via Iceberg document. * @@ -449,16 +461,30 @@ object WorkflowExecutionsResource { .map(URI.create) if (uriOpt.isPresent) { - val size = DocumentFactory.openDocument(uriOpt.get)._1.getTotalFileSize - context - .update(OPERATOR_EXECUTIONS) - .set(OPERATOR_EXECUTIONS.CONSOLE_MESSAGES_SIZE, Integer.valueOf(size.toInt)) - .where(OPERATOR_EXECUTIONS.WORKFLOW_EXECUTION_ID.eq(eid.id.toInt)) - .and(OPERATOR_EXECUTIONS.OPERATOR_ID.eq(opId.id)) - .execute() + updateConsoleMessageSize( + eid, + opId, + DocumentFactory.openDocument(uriOpt.get)._1.getTotalFileSize + ) } } + /** + * Stores an already-measured console message size, mirroring updateResultSize. + * + * @param eid Execution ID associated with the console message. + * @param opId Operator ID of the corresponding operator. + * @param size Size of the console messages in bytes. + */ + def updateConsoleMessageSize(eid: ExecutionIdentity, opId: OperatorIdentity, size: Long): Unit = { + context + .update(OPERATOR_EXECUTIONS) + .set(OPERATOR_EXECUTIONS.CONSOLE_MESSAGES_SIZE, java.lang.Long.valueOf(size)) + .where(OPERATOR_EXECUTIONS.WORKFLOW_EXECUTION_ID.eq(eid.id.toInt)) + .and(OPERATOR_EXECUTIONS.OPERATOR_ID.eq(opId.id)) + .execute() + } + /** * This method is mainly used for frontend requests. Given a logicalOpId and an outputPortId of an execution, * this method finds the URI for a globalPortId that both: 1. matches the logicalOpId and outputPortId, and diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala index 04104fb065..984a10d0e1 100644 --- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala @@ -587,6 +587,97 @@ class WorkflowExecutionsResourceSpec assert(row.getResultSize == 4096) } + it should "store a >2GiB size without truncation (#6978)" in { + val execution = insertExecution() + val eid = ExecutionIdentity(execution.getEid.longValue()) + val globalPortId = GlobalPortIdentity( + PhysicalOpIdentity(OperatorIdentity("op-big-size"), "main"), + PortIdentity(), + input = false + ) + insertOperatorPortResult(eid, globalPortId, URI.create("vfs:///big")) + + // 3 GiB exceeds Int.MaxValue; a Long->Int narrowing would wrap it negative. + val threeGiB = 3L * 1024 * 1024 * 1024 + WorkflowExecutionsResource.updateResultSize(eid, globalPortId, threeGiB) + + val row = getDSLContext + .selectFrom(OPERATOR_PORT_EXECUTIONS) + .where(OPERATOR_PORT_EXECUTIONS.WORKFLOW_EXECUTION_ID.eq(execution.getEid)) + .and(OPERATOR_PORT_EXECUTIONS.GLOBAL_PORT_ID.eq(globalPortId.serializeAsString)) + .fetchOne() + assert(row.getResultSize.longValue() == threeGiB) + } + + // ─── new: updateRuntimeStatsSize / updateConsoleMessageSize ─────────────── + + "updateRuntimeStatsSize" should "store a >2GiB size on the matching execution" in { + val execution = insertExecution() + val eid = ExecutionIdentity(execution.getEid.longValue()) + val threeGiB = 3L * 1024 * 1024 * 1024 + + WorkflowExecutionsResource.updateRuntimeStatsSize(eid, threeGiB) + + val row = getDSLContext + .selectFrom(WORKFLOW_EXECUTIONS) + .where(WORKFLOW_EXECUTIONS.EID.eq(execution.getEid)) + .fetchOne() + assert(row.getRuntimeStatsSize.longValue() == threeGiB) + } + + it should "leave the size untouched when the execution has no runtime stats URI" in { + val execution = insertExecution(runtimeStatsUri = null) + + WorkflowExecutionsResource.updateRuntimeStatsSize( + ExecutionIdentity(execution.getEid.longValue()) + ) + + val row = getDSLContext + .selectFrom(WORKFLOW_EXECUTIONS) + .where(WORKFLOW_EXECUTIONS.EID.eq(execution.getEid)) + .fetchOne() + // The fixture never set a size, so a no-op leaves the column as inserted. + assert(row.getRuntimeStatsSize == null) + } + + "updateConsoleMessageSize" should "store a >2GiB size on the matching (eid, opId) row" in { + val execution = insertExecution() + val eid = ExecutionIdentity(execution.getEid.longValue()) + val opId = OperatorIdentity("op-console-size") + WorkflowExecutionsResource.insertOperatorExecutions( + execution.getEid.longValue(), + opId.id, + URI.create("vfs:///console-big") + ) + + val threeGiB = 3L * 1024 * 1024 * 1024 + WorkflowExecutionsResource.updateConsoleMessageSize(eid, opId, threeGiB) + + val row = getDSLContext + .selectFrom(OPERATOR_EXECUTIONS) + .where(OPERATOR_EXECUTIONS.WORKFLOW_EXECUTION_ID.eq(execution.getEid)) + .and(OPERATOR_EXECUTIONS.OPERATOR_ID.eq(opId.id)) + .fetchOne() + assert(row.getConsoleMessagesSize.longValue() == threeGiB) + } + + it should "leave the size untouched when the operator has no console messages URI" in { + val execution = insertExecution() + val opId = OperatorIdentity("op-no-console-uri") + + WorkflowExecutionsResource.updateConsoleMessageSize( + ExecutionIdentity(execution.getEid.longValue()), + opId + ) + + val row = getDSLContext + .selectFrom(OPERATOR_EXECUTIONS) + .where(OPERATOR_EXECUTIONS.WORKFLOW_EXECUTION_ID.eq(execution.getEid)) + .and(OPERATOR_EXECUTIONS.OPERATOR_ID.eq(opId.id)) + .fetchOne() + assert(row == null) + } + // ─── new: getResultUriByLogicalPortId ───────────────────────────────────── "getResultUriByLogicalPortId" should "match by logical operator id, port id, and resource type" in { diff --git a/sql/changelog.xml b/sql/changelog.xml index 469868f895..2cec53da2f 100644 --- a/sql/changelog.xml +++ b/sql/changelog.xml @@ -53,6 +53,11 @@ <sqlFile path="sql/updates/28.sql"/> </changeSet> + <!-- Widen execution size columns to BIGINT (#6978) --> + <changeSet id="29" author="mengw15"> + <sqlFile path="sql/updates/29.sql"/> + </changeSet> + <!-- example changeSet <changeSet id="1" author="author"> <sqlFile path="sql/updates/1.sql"/> diff --git a/sql/texera_ddl.sql b/sql/texera_ddl.sql index 8a1e8782cd..4132aba32c 100644 --- a/sql/texera_ddl.sql +++ b/sql/texera_ddl.sql @@ -260,7 +260,7 @@ CREATE TABLE IF NOT EXISTS workflow_executions environment_version VARCHAR(128) NOT NULL, log_location TEXT, runtime_stats_uri TEXT, - runtime_stats_size INT DEFAULT 0, + runtime_stats_size BIGINT DEFAULT 0, FOREIGN KEY (vid) REFERENCES workflow_version(vid) ON DELETE CASCADE, FOREIGN KEY (uid) REFERENCES "user"(uid) ON DELETE CASCADE, FOREIGN KEY (cuid) REFERENCES workflow_computing_unit(cuid) ON DELETE CASCADE @@ -365,7 +365,7 @@ CREATE TABLE IF NOT EXISTS operator_executions workflow_execution_id INT NOT NULL, operator_id VARCHAR(100) NOT NULL, console_messages_uri TEXT, - console_messages_size INT DEFAULT 0, + console_messages_size BIGINT DEFAULT 0, PRIMARY KEY (workflow_execution_id, operator_id), FOREIGN KEY (workflow_execution_id) REFERENCES workflow_executions(eid) ON DELETE CASCADE ); @@ -376,7 +376,7 @@ CREATE TABLE operator_port_executions workflow_execution_id INT NOT NULL, global_port_id VARCHAR(200) NOT NULL, result_uri TEXT, - result_size INT DEFAULT 0, + result_size BIGINT DEFAULT 0, PRIMARY KEY (workflow_execution_id, global_port_id), FOREIGN KEY (workflow_execution_id) REFERENCES workflow_executions(eid) ON DELETE CASCADE ); diff --git a/sql/updates/29.sql b/sql/updates/29.sql new file mode 100644 index 0000000000..8870179f00 --- /dev/null +++ b/sql/updates/29.sql @@ -0,0 +1,40 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +\c texera_db + +SET search_path TO texera_db; + +BEGIN; + +-- Execution size columns hold byte counts that can exceed 2 GiB (see #6978): +-- they were INT, so Long sizes were narrowed via Scala's silent `.toInt`, +-- wrapping sizes in [2 GiB, 4 GiB) to negative values and corrupting the +-- user-quota sums computed over these columns. Widen them to BIGINT to match +-- the Long byte counts the application writes. +ALTER TABLE workflow_executions + ALTER COLUMN runtime_stats_size TYPE BIGINT; + +ALTER TABLE operator_executions + ALTER COLUMN console_messages_size TYPE BIGINT; + +ALTER TABLE operator_port_executions + ALTER COLUMN result_size TYPE BIGINT; + +COMMIT;
