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 06:45:16 UTC
(texera) branch gh-readonly-queue/main/pr-6980-7ad71a1e7cfae801ae45792f435e590396334bcb created (now 9d55227ade)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-6980-7ad71a1e7cfae801ae45792f435e590396334bcb
in repository https://gitbox.apache.org/repos/asf/texera.git
at 9d55227ade fix(amber): store execution sizes as BIGINT to stop silent >2GiB truncation (#6980)
This branch includes the following new commits:
new 9d55227ade fix(amber): store execution sizes as BIGINT to stop silent >2GiB truncation (#6980)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(texera) 01/01: fix(amber): store execution sizes as BIGINT to stop silent >2GiB truncation (#6980)
Posted by gi...@apache.org.
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-7ad71a1e7cfae801ae45792f435e590396334bcb
in repository https://gitbox.apache.org/repos/asf/texera.git
commit 9d55227aded223d6210372ba96e26cf9a6f4108f
Author: Meng Wang <me...@uci.edu>
AuthorDate: Tue Jul 28 23:44:55 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 e97d64ac7f..d1946f8d72 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 5e8100c2f7..69035ce19a 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;