You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@texera.apache.org by gi...@apache.org on 2026/07/29 05:50:14 UTC
(texera) 01/01: fix(amber): store execution sizes as BIGINT to stop silent >2GiB truncation (#6980)
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-234270b32857e70f61e0dcc4701c1053bf13095e
in repository https://gitbox.apache.org/repos/asf/texera.git
commit f7de907d059a775b04ddb4ff67eb15f00955d739
Author: Meng Wang <me...@uci.edu>
AuthorDate: Tue Jul 28 22:49:52 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 <17...@users.noreply.github.com>
Co-authored-by: Xinyuan Lin <xi...@uci.edu>
---
.../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;