feat(amber): record each result port's output mode and row count

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

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)
           )
         }
     }

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 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.
+    * @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.
     *

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).orNull)
     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): Option[OutputModeEnum] =
+    mode match {
+      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
   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)
                 }

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 {

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
+      Some(OutputModeEnum.SET_SNAPSHOT)
+    ExecutionResultService.toOutputModeEnum(OutputMode.SET_DELTA) shouldBe
+      Some(OutputModeEnum.SET_DELTA)
+    ExecutionResultService.toOutputModeEnum(OutputMode.SINGLE_SNAPSHOT) shouldBe
+      Some(OutputModeEnum.SINGLE_SNAPSHOT)
+  }
+
+  it should "store no mode for an unrecognized one" in {
+    ExecutionResultService.toOutputModeEnum(OutputMode.Unrecognized(99)) shouldBe None
   }
 
   "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)

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"/>

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
 );

sql/updates/52.sql

@@ -0,0 +1,49 @@
+/*
+ * 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.
+ */
+
+-- 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;
+
+BEGIN;
+
+DO $$
+BEGIN
+    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 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;