From dea2e58876f89b47628a48231265068290ece7b1 Mon Sep 17 00:00:00 2001
From: Xinyuan Lin <xinyual3@uci.edu>
Date: Fri, 24 Jul 2026 20:57:00 -0700
Subject: [PATCH] fix(amber): bound sync result reads so visualization fetches
 release their reader

SyncExecutionResource.collectOperatorResult read operator results
through an unbounded document.get() iterator; the visualization and
oversized-first-tuple branches returned after a single next(), and the
catch-all swallowed exceptions mid-drain, leaving the Iceberg Parquet
reader / S3 input stream open until GC finalization. Size each read to
what the method actually consumes: getRange(0, 1) for the first tuple,
getRange(1, totalCount) for the truncation loops (which always drain),
and drain the bounded remainder in the catch path. Bounded getRange
iterators release their reader inside the next() that serves their
last record (#6882), so no close API is needed.
---
 .../web/resource/SyncExecutionResource.scala  | 34 ++++++++++++++++---
 1 file changed, 30 insertions(+), 4 deletions(-)

diff --git a/amber/src/main/scala/org/apache/texera/web/resource/SyncExecutionResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/SyncExecutionResource.scala
index 3c9da603c86..ec5053694fb 100644
--- a/amber/src/main/scala/org/apache/texera/web/resource/SyncExecutionResource.scala
+++ b/amber/src/main/scala/org/apache/texera/web/resource/SyncExecutionResource.scala
@@ -524,6 +524,10 @@ class SyncExecutionResource extends LazyLogging {
   ): (String, Option[Any], Option[Int], Option[Int], Option[Boolean]) = {
     import com.fasterxml.jackson.databind.node.ObjectNode
 
+    // Hoisted so the catch below can drain a partially-consumed read: the
+    // iterator has no close API, but a bounded read (getRange) releases its
+    // underlying reader once drained to its limit.
+    var tupleIterator: Iterator[Tuple] = Iterator.empty
     try {
       val storageUriOption = WorkflowExecutionsResource.getResultUriByLogicalPortId(
         executionId,
@@ -540,9 +544,17 @@ class SyncExecutionResource extends LazyLogging {
 
           val totalCount = document.getCount.toInt
           val mapper = new ObjectMapper()
-          val tupleIterator = document.get()
-
-          if (totalCount == 0 || !tupleIterator.hasNext) {
+          // Bounded reads (getRange) release their Parquet/S3 reader inside the
+          // next() that serves their last record, so each read below is sized
+          // to what this method actually consumes. The unbounded get()
+          // previously used here outlived the early returns and was only
+          // reclaimed by the GC finalizer. The first tuple gets its own
+          // single-record read because the visualization and
+          // oversized-first-tuple branches return after consuming exactly one
+          // tuple.
+          val firstTupleIterator = document.getRange(0, 1)
+
+          if (totalCount == 0 || !firstTupleIterator.hasNext) {
             return (
               "table",
               Some(List.empty[ObjectNode].asJava),
@@ -554,7 +566,7 @@ class SyncExecutionResource extends LazyLogging {
 
           // A single tuple with html-content / json-content is a visualization payload —
           // the frontend renders it as an iframe rather than a table.
-          val firstTuple = tupleIterator.next()
+          val firstTuple = firstTupleIterator.next()
           if (totalCount == 1 && isVisualizationTuple(firstTuple)) {
             val jsonResults =
               ExecutionResultService.convertTuplesToJson(List(firstTuple), isVisualization = true)
@@ -588,6 +600,10 @@ class SyncExecutionResource extends LazyLogging {
             )
           }
 
+          // Records 1 until totalCount; the truncation loops below drain this
+          // iterator fully, releasing its reader at the bound.
+          tupleIterator = document.getRange(1, totalCount)
+
           val halfLimit = maxOperatorResultCharLimit / 2
           val truncationNoticeSize = 50 // reserved for the "...skipped..." marker
 
@@ -694,6 +710,16 @@ class SyncExecutionResource extends LazyLogging {
     } catch {
       case e: Exception =>
         logger.warn(s"Error collecting result for operator $opId: ${e.getMessage}", e)
+        // Drain what the truncation loops left behind so the bounded read
+        // reaches its limit and releases its reader; this costs at most what
+        // the successful path would have spent reading the same records.
+        try {
+          while (tupleIterator.hasNext) tupleIterator.next()
+        } catch {
+          // A failing reader ends the drain early; the GC finalizer still
+          // covers whatever the failure left open.
+          case _: Exception =>
+        }
         ("table", None, None, None, None)
     }
   }