From 4dc9af2dd0acfb52d940832fe12501942c8f4c40 Mon Sep 17 00:00:00 2001
From: mengw15 <125719918+mengw15@users.noreply.github.com>
Date: Thu, 15 Oct 2026 11:40:42 -0700
Subject: [PATCH 1/2] feat(amber): record each result port's output mode and
 row count

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 #8994
---
 .../coordinator/ClientEvent.scala             |  4 +-
 .../scheduling/RegionExecutionManager.scala   | 10 +--
 .../workflow/WorkflowExecutionsResource.scala | 20 +++++
 .../web/service/ExecutionResultService.scala  | 19 +++++
 .../WorkflowExecutionsResourceSpec.scala      | 23 +++++-
 .../service/ExecutionResultServiceSpec.scala  | 82 ++++++++++++++++++-
 sql/changelog.xml                             |  5 ++
 sql/texera_ddl.sql                            |  5 ++
 sql/updates/52.sql                            | 40 +++++++++
 9 files changed, 196 insertions(+), 12 deletions(-)
 create mode 100644 sql/updates/52.sql

diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/ClientEvent.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/ClientEvent.scala
index e9e1e8ee6cf..42e5c14af26 100644
--- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/ClientEvent.scala
+++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/ClientEvent.scala
@@ -21,6 +21,7 @@ package org.apache.texera.amber.engine.architecture.coordinator
 
 import org.apache.texera.amber.core.virtualidentity.ActorVirtualIdentity
 import org.apache.texera.amber.core.workflow.GlobalPortIdentity
+import org.apache.texera.amber.core.workflow.OutputPort.OutputMode
 import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState
 import org.apache.texera.amber.engine.common.ambermessage.WorkflowFIFOMessagePayload
 import org.apache.texera.amber.engine.common.executionruntimestate.OperatorMetrics
@@ -33,7 +34,8 @@ case class ExecutionStateUpdate(state: WorkflowAggregatedState) extends ClientEv
 
 case class OperatorPortResultUriAvailable(
     globalPortId: GlobalPortIdentity,
-    uri: URI
+    uri: URI,
+    mode: OutputMode
 ) extends ClientEvent
 
 case class ExecutionStatsUpdate(operatorMetrics: Map[String, OperatorMetrics]) extends ClientEvent
diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala
index e884a6ba303..51910a5041e 100644
--- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala
+++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala
@@ -687,19 +687,15 @@ class RegionExecutionManager(
         // (The inner LoopEnd of a nested loop additionally drops its output
         // once per outer iteration on the Python worker side in
         // MainLoop._process_state_frame, which is orthogonal to this.)
-        val reuseStorage =
-          region
-            .getOperator(outputPortId.opId)
-            .outputPorts(outputPortId.portId)
-            ._1
-            .reuseStorage
+        val outputPort = region.getOperator(outputPortId.opId).outputPorts(outputPortId.portId)._1
+        val reuseStorage = outputPort.reuseStorage
         Seq((resultURI, schema), (stateURI, State.schema)).foreach {
           case (uri, sch) =>
             DocumentFactory.createOrReuseDocument(uri, sch, reuseStorage)
         }
         if (!isRestart) {
           asyncRPCClient.sendToClient(
-            OperatorPortResultUriAvailable(outputPortId, resultURI)
+            OperatorPortResultUriAvailable(outputPortId, resultURI, outputPort.mode)
           )
         }
     }
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 93a601d0de0..5f77899d384 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
@@ -413,6 +413,26 @@ object WorkflowExecutionsResource {
       .execute()
   }
 
+  /**
+    * Updates the row count of the corresponding Iceberg document in the database.
+    *
+    * @param eid          Execution ID associated with the result.
+    * @param globalPortId Global port identifier for the operator output.
+    * @param count        Number of rows in the result.
+    */
+  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()
+  }
+
   /**
     * Updates the size of the runtime statistics stored via Iceberg document.
     *
diff --git a/amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala b/amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala
index 30c845ed0fb..865364a2736 100644
--- a/amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala
+++ b/amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala
@@ -30,6 +30,7 @@ import org.apache.texera.amber.core.storage.{DocumentFactory, VFSURIFactory}
 import org.apache.texera.amber.core.tuple.{AttributeType, Tuple, TupleUtils}
 import org.apache.texera.amber.util.serde.GlobalPortIdentitySerde.SerdeOps
 import org.apache.texera.dao.SqlServer
+import org.apache.texera.dao.jooq.generated.enums.OutputModeEnum
 import org.apache.texera.dao.jooq.generated.tables.daos.OperatorPortExecutionsDao
 import org.apache.texera.dao.jooq.generated.tables.pojos.OperatorPortExecutions
 import org.apache.texera.amber.core.virtualidentity.{
@@ -92,11 +93,24 @@ object ExecutionResultService {
     pojo.setWorkflowExecutionId(executionId.id.toInt)
     pojo.setGlobalPortId(evt.globalPortId.serializeAsString)
     pojo.setResultUri(evt.uri.toString)
+    pojo.setOutputMode(toOutputModeEnum(evt.mode))
     new OperatorPortExecutionsDao(
       SqlServer.getInstance().createDSLContext().configuration()
     ).insert(pojo)
   }
 
+  /**
+    * The stored form of a port's output mode. Recorded with the URI so that a finished
+    * execution's results can be listed without the engine's physical plan (#8994).
+    */
+  private[service] def toOutputModeEnum(mode: OutputMode): OutputModeEnum =
+    mode match {
+      case OutputMode.SET_SNAPSHOT    => OutputModeEnum.SET_SNAPSHOT
+      case OutputMode.SET_DELTA       => OutputModeEnum.SET_DELTA
+      case OutputMode.SINGLE_SNAPSHOT => OutputModeEnum.SINGLE_SNAPSHOT
+      case OutputMode.Unrecognized(_) => null
+    }
+
   private val defaultPageSize: Int = 5
   private val binaryPreviewLeadingBits: Int = 10
   private val binaryPreviewTrailingBits: Int = 3
@@ -447,6 +461,11 @@ class ExecutionResultService(
                     globalPortIdOption.get,
                     opStorage.getTotalFileSize
                   )
+                  WorkflowExecutionsResource.updateResultCount(
+                    executionId,
+                    globalPortIdOption.get,
+                    info.tupleCount
+                  )
                   WorkflowExecutionsResource.updateRuntimeStatsSize(executionId)
                   WorkflowExecutionsResource.updateConsoleMessageSize(executionId, opId)
                 }
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 7cc549004d6..7db3e4c6d51 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
@@ -34,6 +34,7 @@ import org.apache.texera.amber.core.workflow.{GlobalPortIdentity, PortIdentity}
 import org.apache.texera.amber.util.serde.GlobalPortIdentitySerde.SerdeOps
 import org.apache.texera.auth.{JwtAuth, SessionUser}
 import org.apache.texera.dao.MockTexeraDB
+import org.apache.texera.amber.core.workflow.OutputPort.OutputMode
 import org.apache.texera.dao.jooq.generated.enums.UserWarehouseFlavorEnum
 import org.apache.texera.dao.jooq.generated.Tables._
 import org.apache.texera.dao.jooq.generated.enums.{
@@ -356,7 +357,7 @@ class WorkflowExecutionsResourceSpec
   ): Unit =
     ExecutionResultService.persistOperatorPortResultUri(
       eid,
-      OperatorPortResultUriAvailable(globalPortId, uri)
+      OperatorPortResultUriAvailable(globalPortId, uri, OutputMode.SET_SNAPSHOT)
     )
 
   // ─── existing tests (preserved) ───────────────────────────────────────────
@@ -725,6 +726,26 @@ class WorkflowExecutionsResourceSpec
     assert(row.getResultSize.longValue() == threeGiB)
   }
 
+  "updateResultCount" should "set RESULT_COUNT on the matching (eid, globalPortId) row" in {
+    val execution = insertExecution()
+    val eid = ExecutionIdentity(execution.getEid.longValue())
+    val globalPortId = GlobalPortIdentity(
+      PhysicalOpIdentity(OperatorIdentity("op-count"), "main"),
+      PortIdentity(),
+      input = false
+    )
+    insertOperatorPortResult(eid, globalPortId, URI.create("vfs:///c"))
+
+    WorkflowExecutionsResource.updateResultCount(eid, globalPortId, 25L)
+
+    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.getResultCount == 25L)
+  }
+
   // ─── new: updateRuntimeStatsSize / updateConsoleMessageSize ───────────────
 
   "updateRuntimeStatsSize" should "store a >2GiB size on the matching execution" in {
diff --git a/amber/src/test/scala/org/apache/texera/web/service/ExecutionResultServiceSpec.scala b/amber/src/test/scala/org/apache/texera/web/service/ExecutionResultServiceSpec.scala
index 5e4975d17f8..862448d8e54 100644
--- a/amber/src/test/scala/org/apache/texera/web/service/ExecutionResultServiceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/service/ExecutionResultServiceSpec.scala
@@ -62,7 +62,7 @@ import org.apache.texera.dao.jooq.generated.Tables.{
   WORKFLOW_EXECUTIONS,
   WORKFLOW_VERSION
 }
-import org.apache.texera.dao.jooq.generated.enums.WorkflowComputingUnitTypeEnum
+import org.apache.texera.dao.jooq.generated.enums.{OutputModeEnum, WorkflowComputingUnitTypeEnum}
 import org.apache.texera.dao.jooq.generated.tables.daos.{
   UserDao,
   WorkflowComputingUnitDao,
@@ -231,7 +231,7 @@ class ExecutionResultServiceSpec
 
     ExecutionResultService.persistOperatorPortResultUri(
       eid,
-      OperatorPortResultUriAvailable(globalPortId, uri)
+      OperatorPortResultUriAvailable(globalPortId, uri, OutputMode.SINGLE_SNAPSHOT)
     )
 
     val rows = getDSLContext
@@ -241,6 +241,22 @@ class ExecutionResultServiceSpec
       .fetch()
     rows.size() shouldBe 1
     rows.get(0).getResultUri shouldBe uri.toString
+    // The mode is recorded with the URI, so a finished execution's results can be
+    // listed without the engine's physical plan (#8994).
+    rows.get(0).getOutputMode shouldBe OutputModeEnum.SINGLE_SNAPSHOT
+  }
+
+  "toOutputModeEnum" should "map every declared output mode to its stored form" in {
+    ExecutionResultService.toOutputModeEnum(OutputMode.SET_SNAPSHOT) shouldBe
+      OutputModeEnum.SET_SNAPSHOT
+    ExecutionResultService.toOutputModeEnum(OutputMode.SET_DELTA) shouldBe
+      OutputModeEnum.SET_DELTA
+    ExecutionResultService.toOutputModeEnum(OutputMode.SINGLE_SNAPSHOT) shouldBe
+      OutputModeEnum.SINGLE_SNAPSHOT
+  }
+
+  it should "store no mode for an unrecognized one" in {
+    ExecutionResultService.toOutputModeEnum(OutputMode.Unrecognized(99)) shouldBe null
   }
 
   "convertTuplesToJson" should "convert tuples with various field types correctly" in {
@@ -1052,6 +1068,51 @@ class ExecutionResultServiceSpec
     }
   }
 
+  it should "record the row count of a table-mode result" in {
+    val executionId = newExecution()
+    val schema = new Schema(List(new Attribute("id", AttributeType.INTEGER)))
+    storeResult(
+      executionId,
+      "erss-snapshot-count",
+      schema,
+      (1 to 3).map(i => Tuple.builder(schema).add("id", AttributeType.INTEGER, i).build())
+    )
+    val plan = planOf(
+      physicalOp(executionId, "erss-snapshot-count", PortIdentity() -> OutputMode.SET_SNAPSHOT)
+    )
+
+    withAttached(plan, executionId) { fixture =>
+      new ResultEvents(fixture.workflowStateStore)
+      fixture.setResultCounts("erss-snapshot-count" -> 3)
+
+      // The recorded count is the one the result store tracks, so a reader can list a
+      // finished execution's results without opening each table (#8994).
+      storedResultCount(executionId, "erss-snapshot-count") shouldBe 3L
+    }
+  }
+
+  it should "record no row count for a SINGLE_SNAPSHOT port" in {
+    val executionId = newExecution()
+    val schema = new Schema(List(new Attribute("html", AttributeType.STRING)))
+    storeResult(
+      executionId,
+      "erss-single-count",
+      schema,
+      List(Tuple.builder(schema).add("html", AttributeType.STRING, "<p/>").build())
+    )
+    val plan = planOf(
+      physicalOp(executionId, "erss-single-count", PortIdentity() -> OutputMode.SINGLE_SNAPSHOT)
+    )
+
+    withAttached(plan, executionId) { fixture =>
+      new ResultEvents(fixture.workflowStateStore)
+      fixture.setResultCounts("erss-single-count" -> 1)
+
+      // A visualization is shown whole, never paged, so it has no count to record.
+      storedResultCount(executionId, "erss-single-count") shouldBe null
+    }
+  }
+
   it should "send the whole untruncated snapshot, and no statistics, for a SINGLE_SNAPSHOT port" in {
     val executionId = newExecution()
     val schema = new Schema(List(new Attribute("html", AttributeType.STRING)))
@@ -1195,7 +1256,9 @@ class ExecutionResultServiceSpec
       val globalPortId = globalPortIdOf("erss-callback")
       val uri = resultUriOf(executionId, "erss-callback")
 
-      fixture.client.fire(OperatorPortResultUriAvailable(globalPortId, uri))
+      fixture.client.fire(
+        OperatorPortResultUriAvailable(globalPortId, uri, OutputMode.SET_SNAPSHOT)
+      )
 
       val rows = getDSLContext
         .selectFrom(OPERATOR_PORT_EXECUTIONS)
@@ -1516,6 +1579,19 @@ class ExecutionResultServiceSpec
     uri
   }
 
+  private def storedResultCount(
+      executionId: ExecutionIdentity,
+      operatorId: String
+  ): java.lang.Long =
+    getDSLContext
+      .select(OPERATOR_PORT_EXECUTIONS.RESULT_COUNT)
+      .from(OPERATOR_PORT_EXECUTIONS)
+      .where(OPERATOR_PORT_EXECUTIONS.WORKFLOW_EXECUTION_ID.eq(executionId.id.toInt))
+      .and(
+        OPERATOR_PORT_EXECUTIONS.GLOBAL_PORT_ID.eq(globalPortIdOf(operatorId).serializeAsString)
+      )
+      .fetchOne(OPERATOR_PORT_EXECUTIONS.RESULT_COUNT)
+
   private def storedResultSize(executionId: ExecutionIdentity, operatorId: String): Long =
     getDSLContext
       .select(OPERATOR_PORT_EXECUTIONS.RESULT_SIZE)
diff --git a/sql/changelog.xml b/sql/changelog.xml
index 97724697c65..62dfe42d522 100644
--- a/sql/changelog.xml
+++ b/sql/changelog.xml
@@ -169,6 +169,11 @@
         <sqlFile path="sql/updates/51.sql"/>
     </changeSet>
 
+    <!-- Record each result port's output mode and row count -->
+    <changeSet id="52" author="mengw15">
+        <sqlFile path="sql/updates/52.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 dde189976a1..046982283b6 100644
--- a/sql/texera_ddl.sql
+++ b/sql/texera_ddl.sql
@@ -102,6 +102,7 @@ CREATE TYPE workflow_computing_unit_termination_reason_enum AS ENUM ('USER_REQUE
 CREATE TYPE provider_type_enum AS ENUM ('LOCAL', 'GOOGLE', 'ORCID', 'APPLE');
 CREATE TYPE user_warehouse_flavor_enum AS ENUM ('local', 'aws');
 CREATE TYPE default_view_enum AS ENUM ('CANVAS', 'FORM');
+CREATE TYPE output_mode_enum AS ENUM ('SET_SNAPSHOT', 'SET_DELTA', 'SINGLE_SNAPSHOT');
 
 -- ============================================
 -- 5. Create tables
@@ -551,6 +552,10 @@ CREATE TABLE operator_port_executions
     global_port_id        VARCHAR(200) NOT NULL,
     result_uri            TEXT,
     result_size           BIGINT DEFAULT 0,
+    -- The port's declared output mode and row count, so a finished execution's results
+    -- can be listed without the engine's physical plan or opening each table (#8994).
+    output_mode           output_mode_enum,
+    result_count          BIGINT,
     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/52.sql b/sql/updates/52.sql
new file mode 100644
index 00000000000..f60e00807f7
--- /dev/null
+++ b/sql/updates/52.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;
+
+DO $$
+BEGIN
+    IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'output_mode_enum') THEN
+        CREATE TYPE output_mode_enum AS ENUM ('SET_SNAPSHOT', 'SET_DELTA', 'SINGLE_SNAPSHOT');
+    END IF;
+END $$;
+
+-- The port's declared output mode and row count, so a finished execution's results
+-- can be listed without the engine's physical plan or opening each table (#8994).
+-- Rows written before this change keep both columns null.
+ALTER TABLE operator_port_executions
+    ADD COLUMN IF NOT EXISTS output_mode output_mode_enum DEFAULT NULL,
+    ADD COLUMN IF NOT EXISTS result_count BIGINT DEFAULT NULL;
+
+COMMIT;

From 9e316d640bcfd5a77f9ca2881a62dcbbfd397109 Mon Sep 17 00:00:00 2001
From: mengw15 <125719918+mengw15@users.noreply.github.com>
Date: Tue, 20 Oct 2026 15:29:22 -0700
Subject: [PATCH 2/2] fix: address Copilot review feedback

- 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.
---
 .../workflow/WorkflowExecutionsResource.scala   |  2 +-
 .../web/service/ExecutionResultService.scala    | 12 ++++++------
 .../service/ExecutionResultServiceSpec.scala    |  8 ++++----
 sql/updates/52.sql                              | 17 +++++++++++++----
 4 files changed, 24 insertions(+), 15 deletions(-)

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 5f77899d384..32ba68eef38 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
@@ -414,7 +414,7 @@ object WorkflowExecutionsResource {
   }
 
   /**
-    * Updates the row count of the corresponding Iceberg document in the database.
+    * Updates the row count stored for this port's result in operator_port_executions.
     *
     * @param eid          Execution ID associated with the result.
     * @param globalPortId Global port identifier for the operator output.
diff --git a/amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala b/amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala
index 865364a2736..88559c2194c 100644
--- a/amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala
+++ b/amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala
@@ -93,7 +93,7 @@ object ExecutionResultService {
     pojo.setWorkflowExecutionId(executionId.id.toInt)
     pojo.setGlobalPortId(evt.globalPortId.serializeAsString)
     pojo.setResultUri(evt.uri.toString)
-    pojo.setOutputMode(toOutputModeEnum(evt.mode))
+    pojo.setOutputMode(toOutputModeEnum(evt.mode).orNull)
     new OperatorPortExecutionsDao(
       SqlServer.getInstance().createDSLContext().configuration()
     ).insert(pojo)
@@ -103,12 +103,12 @@ object ExecutionResultService {
     * The stored form of a port's output mode. Recorded with the URI so that a finished
     * execution's results can be listed without the engine's physical plan (#8994).
     */
-  private[service] def toOutputModeEnum(mode: OutputMode): OutputModeEnum =
+  private[service] def toOutputModeEnum(mode: OutputMode): Option[OutputModeEnum] =
     mode match {
-      case OutputMode.SET_SNAPSHOT    => OutputModeEnum.SET_SNAPSHOT
-      case OutputMode.SET_DELTA       => OutputModeEnum.SET_DELTA
-      case OutputMode.SINGLE_SNAPSHOT => OutputModeEnum.SINGLE_SNAPSHOT
-      case OutputMode.Unrecognized(_) => null
+      case OutputMode.SET_SNAPSHOT    => Some(OutputModeEnum.SET_SNAPSHOT)
+      case OutputMode.SET_DELTA       => Some(OutputModeEnum.SET_DELTA)
+      case OutputMode.SINGLE_SNAPSHOT => Some(OutputModeEnum.SINGLE_SNAPSHOT)
+      case OutputMode.Unrecognized(_) => None
     }
 
   private val defaultPageSize: Int = 5
diff --git a/amber/src/test/scala/org/apache/texera/web/service/ExecutionResultServiceSpec.scala b/amber/src/test/scala/org/apache/texera/web/service/ExecutionResultServiceSpec.scala
index 862448d8e54..443d75bf277 100644
--- a/amber/src/test/scala/org/apache/texera/web/service/ExecutionResultServiceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/service/ExecutionResultServiceSpec.scala
@@ -248,15 +248,15 @@ class ExecutionResultServiceSpec
 
   "toOutputModeEnum" should "map every declared output mode to its stored form" in {
     ExecutionResultService.toOutputModeEnum(OutputMode.SET_SNAPSHOT) shouldBe
-      OutputModeEnum.SET_SNAPSHOT
+      Some(OutputModeEnum.SET_SNAPSHOT)
     ExecutionResultService.toOutputModeEnum(OutputMode.SET_DELTA) shouldBe
-      OutputModeEnum.SET_DELTA
+      Some(OutputModeEnum.SET_DELTA)
     ExecutionResultService.toOutputModeEnum(OutputMode.SINGLE_SNAPSHOT) shouldBe
-      OutputModeEnum.SINGLE_SNAPSHOT
+      Some(OutputModeEnum.SINGLE_SNAPSHOT)
   }
 
   it should "store no mode for an unrecognized one" in {
-    ExecutionResultService.toOutputModeEnum(OutputMode.Unrecognized(99)) shouldBe null
+    ExecutionResultService.toOutputModeEnum(OutputMode.Unrecognized(99)) shouldBe None
   }
 
   "convertTuplesToJson" should "convert tuples with various field types correctly" in {
diff --git a/sql/updates/52.sql b/sql/updates/52.sql
index f60e00807f7..5b36a491176 100644
--- a/sql/updates/52.sql
+++ b/sql/updates/52.sql
@@ -17,6 +17,10 @@
  * under the License.
  */
 
+-- Names are schema-qualified because the two runners disagree about the search path: the
+-- 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
 
 SET search_path TO texera_db;
@@ -25,16 +29,21 @@ BEGIN;
 
 DO $$
 BEGIN
-    IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'output_mode_enum') THEN
-        CREATE TYPE output_mode_enum AS ENUM ('SET_SNAPSHOT', 'SET_DELTA', 'SINGLE_SNAPSHOT');
+    IF NOT EXISTS (
+        SELECT 1
+        FROM pg_type t
+        JOIN pg_namespace n ON n.oid = t.typnamespace
+        WHERE t.typname = 'output_mode_enum' AND n.nspname = 'texera_db'
+    ) THEN
+        CREATE TYPE texera_db.output_mode_enum AS ENUM ('SET_SNAPSHOT', 'SET_DELTA', 'SINGLE_SNAPSHOT');
     END IF;
 END $$;
 
 -- The port's declared output mode and row count, so a finished execution's results
 -- can be listed without the engine's physical plan or opening each table (#8994).
 -- Rows written before this change keep both columns null.
-ALTER TABLE operator_port_executions
-    ADD COLUMN IF NOT EXISTS output_mode output_mode_enum DEFAULT NULL,
+ALTER TABLE texera_db.operator_port_executions
+    ADD COLUMN IF NOT EXISTS output_mode texera_db.output_mode_enum DEFAULT NULL,
     ADD COLUMN IF NOT EXISTS result_count BIGINT DEFAULT NULL;
 
 COMMIT;