feat(ui): add pagination and search for wide-column tables

amber/src/main/scala/org/apache/texera/web/model/websocket/request/ResultPaginationRequest.scala

@@ -23,5 +23,8 @@ case class ResultPaginationRequest(
     requestID: String,
     operatorID: String,
     pageIndex: Int,
-    pageSize: Int
+    pageSize: Int,
+    columnOffset: Int = 0,
+    columnLimit: Int = Int.MaxValue,
+    columnSearch: Option[String] = None
 ) extends TexeraWebSocketRequest

amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala

@@ -437,12 +437,25 @@ class ExecutionResultService(
 
     storageUriOption match {
       case Some(storageUri) =>
+        val (document, schemaOption) = DocumentFactory.openDocument(storageUri)
+        val virtualDocument = document.asInstanceOf[VirtualDocument[Tuple]]
+
+        val columns = {
+          val schema = schemaOption.get
+          val allColumns = schema.getAttributeNames
+          val filteredColumns = request.columnSearch match {
+            case Some(search) =>
+              allColumns.filter(col => col.toLowerCase.contains(search.toLowerCase))
+            case None => allColumns
+          }
+          Some(
+            filteredColumns.slice(request.columnOffset, request.columnOffset + request.columnLimit)
+          )
+        }
+
         val paginationIterable = {
-          DocumentFactory
-            .openDocument(storageUri)
-            ._1
-            .asInstanceOf[VirtualDocument[Tuple]]
-            .getRange(from, from + request.pageSize)
+          virtualDocument
+            .getRange(from, from + request.pageSize, columns)
             .to(Iterable)
         }
         val mappedResults = convertTuplesToJson(paginationIterable)

common/config/src/main/resources/gui.conf

@@ -108,5 +108,9 @@ gui {
     # whether AI copilot feature is enabled
     copilot-enabled = false
     copilot-enabled = ${?GUI_WORKFLOW_WORKSPACE_COPILOT_ENABLED}
+
+    # the limit of columns to be displayed in the result table
+    limit-columns = 15
+    limit-columns = ${?GUI_WORKFLOW_WORKSPACE_LIMIT_COLUMNS}
   }
 }

common/config/src/main/scala/org/apache/texera/config/GuiConfig.scala

@@ -69,4 +69,6 @@ object GuiConfig {
     conf.getInt("gui.workflow-workspace.active-time-in-minutes")
   val guiWorkflowWorkspaceCopilotEnabled: Boolean =
     conf.getBoolean("gui.workflow-workspace.copilot-enabled")
+  val guiWorkflowWorkspaceLimitColumns: Int =
+    conf.getInt("gui.workflow-workspace.limit-columns")
 }

common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/DocumentFactory.scala

@@ -87,8 +87,8 @@ object DocumentFactory {
               overrideIfExists = true
             )
             val serde: (IcebergSchema, Tuple) => Record = IcebergUtil.toGenericRecord
-            val deserde: (IcebergSchema, Record) => Tuple = (_, record) =>
-              IcebergUtil.fromRecord(record, schema)
+            val deserde: (IcebergSchema, Record) => Tuple = (schema, record) =>
+              IcebergUtil.fromRecord(record, IcebergUtil.fromIcebergSchema(schema))
 
             new IcebergDocument[Tuple](
               namespace,
@@ -144,8 +144,8 @@ object DocumentFactory {
 
             val amberSchema = IcebergUtil.fromIcebergSchema(table.schema())
             val serde: (IcebergSchema, Tuple) => Record = IcebergUtil.toGenericRecord
-            val deserde: (IcebergSchema, Record) => Tuple = (_, record) =>
-              IcebergUtil.fromRecord(record, amberSchema)
+            val deserde: (IcebergSchema, Record) => Tuple = (schema, record) =>
+              IcebergUtil.fromRecord(record, IcebergUtil.fromIcebergSchema(schema))
 
             (
               new IcebergDocument[Tuple](

common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/ReadonlyLocalFileDocument.scala

@@ -54,7 +54,11 @@ private[storage] class ReadonlyLocalFileDocument(uri: URI)
   override def get(): Iterator[Nothing] =
     throw new NotImplementedError("get is not supported for ReadonlyLocalFileDocument")
 
-  override def getRange(from: Int, until: Int): Iterator[Nothing] =
+  override def getRange(
+      from: Int,
+      until: Int,
+      columns: Option[Seq[String]] = None
+  ): Iterator[Nothing] =
     throw new NotImplementedError("getRange is not supported for ReadonlyLocalFileDocument")
 
   override def getAfter(offset: Int): Iterator[Nothing] =

common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/ReadonlyVirtualDocument.scala

@@ -52,9 +52,10 @@ trait ReadonlyVirtualDocument[T] {
     * Get an iterator of a sequence starting from index `from`, until index `until`.
     * @param from the starting index (inclusive)
     * @param until the ending index (exclusive)
+    * @param columns optional sequence of column names to retrieve. If None, retrieves all columns.
     * @return an iterator that returns data items of type T
     */
-  def getRange(from: Int, until: Int): Iterator[T]
+  def getRange(from: Int, until: Int, columns: Option[Seq[String]] = None): Iterator[T]
 
   /**
     * Get an iterator of all items after the specified index `offset`.

common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/VirtualDocument.scala

@@ -55,9 +55,10 @@ abstract class VirtualDocument[T] extends ReadonlyVirtualDocument[T] {
     * get an iterator of a sequence starting from index `from`, until index `until`
     * @param from the starting index (inclusive)
     * @param until the ending index (exclusive)
+    * @param columns the columns to be projected
     * @return an iterator that returns data items of type T
     */
-  def getRange(from: Int, until: Int): Iterator[T] =
+  def getRange(from: Int, until: Int, columns: Option[Seq[String]] = None): Iterator[T] =
     throw new NotImplementedError("getRange method is not implemented")
 
   /**

common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergDocument.scala

@@ -105,8 +105,8 @@ private[storage] class IcebergDocument[T >: Null <: AnyRef](
   /**
     * Get records within a specified range [from, until).
     */
-  override def getRange(from: Int, until: Int): Iterator[T] = {
-    getUsingFileSequenceOrder(from, Some(until))
+  override def getRange(from: Int, until: Int, columns: Option[Seq[String]] = None): Iterator[T] = {
+    getUsingFileSequenceOrder(from, Some(until), columns)
   }
 
   /**
@@ -150,8 +150,13 @@ private[storage] class IcebergDocument[T >: Null <: AnyRef](
     *
     * @param from  start from which record inclusively, if 0 means start from the first
     * @param until end at which record exclusively, if None means read to the table's EOF
+    * @param columns columns to be projected
     */
-  private def getUsingFileSequenceOrder(from: Int, until: Option[Int]): Iterator[T] =
+  private def getUsingFileSequenceOrder(
+      from: Int,
+      until: Option[Int],
+      columns: Option[Seq[String]] = None
+  ): Iterator[T] =
     withReadLock(lock) {
       new Iterator[T] {
         private val iteLock = new ReentrantLock()
@@ -259,9 +264,13 @@ private[storage] class IcebergDocument[T >: Null <: AnyRef](
 
           while (!currentRecordIterator.hasNext && usableFileIterator.hasNext) {
             val nextFile = usableFileIterator.next()
+            val schemaToUse = columns match {
+              case Some(cols) => tableSchema.select(cols.asJava)
+              case None       => tableSchema
+            }
             currentRecordIterator = IcebergUtil.readDataFileAsIterator(
               nextFile.file(),
-              tableSchema,
+              schemaToUse,
               table.get
             )
 
@@ -281,7 +290,11 @@ private[storage] class IcebergDocument[T >: Null <: AnyRef](
 
           val record = currentRecordIterator.next()
           numOfReturnedRecords += 1
-          deserde(tableSchema, record)
+          val schemaToUse = columns match {
+            case Some(cols) => tableSchema.select(cols.asJava)
+            case None       => tableSchema
+          }
+          deserde(schemaToUse, record)
         }
       }
     }

config-service/src/main/scala/org/apache/texera/service/resource/ConfigResource.scala

@@ -56,6 +56,7 @@ class ConfigResource {
       ),
       "activeTimeInMinutes" -> GuiConfig.guiWorkflowWorkspaceActiveTimeInMinutes,
       "copilotEnabled" -> GuiConfig.guiWorkflowWorkspaceCopilotEnabled,
+      "limitColumns" -> GuiConfig.guiWorkflowWorkspaceLimitColumns,
       // flags from the auth.conf if needed
       "expirationTimeInMinutes" -> AuthConfig.jwtExpirationMinutes
     )

frontend/src/app/common/service/gui-config.service.mock.ts

@@ -49,6 +49,7 @@ export class MockGuiConfigService {
     expirationTimeInMinutes: 2880,
     activeTimeInMinutes: 15,
     copilotEnabled: false,
+    limitColumns: 15,
   };
 
   get env(): GuiConfig {

frontend/src/app/common/type/gui-config.ts

@@ -40,6 +40,7 @@ export interface GuiConfig {
   expirationTimeInMinutes: number;
   activeTimeInMinutes: number;
   copilotEnabled: boolean;
+  limitColumns: number;
 }
 
 export interface SidebarTabs {

frontend/src/app/workspace/component/result-panel/result-panel.component.ts

@@ -52,7 +52,7 @@ import { CompilationState } from "../../types/workflow-compiling.interface";
 import { WorkflowFatalError } from "../../types/workflow-websocket.interface";
 
 export const DEFAULT_WIDTH = 800;
-export const DEFAULT_HEIGHT = 300;
+export const DEFAULT_HEIGHT = 500;
 /**
  * ResultPanelComponent is the bottom level area that displays the
  *  execution result of a workflow after the execution finishes.

frontend/src/app/workspace/component/result-panel/result-table-frame/result-table-frame.component.html

@@ -25,6 +25,40 @@ <h4>Empty result set</h4>
 <div
   [hidden]="!currentColumns"
   class="result-table">
+  <div
+    class="column-search"
+    style="margin-bottom: 8px; display: flex; justify-content: flex-end">
+    <input
+      nz-input
+      placeholder="Search Columns"
+      (input)="onColumnSearch($event)"
+      style="width: 200px" />
+  </div>
+  <div
+    class="column-navigation"
+    style="margin-bottom: 8px; display: flex; justify-content: flex-end; gap: 8px"
+    [hidden]="currentColumnOffset === 0 && (!currentColumns || currentColumns.length < columnLimit)">
+    <button
+      nz-button
+      nzType="default"
+      (click)="onColumnShiftLeft()"
+      [disabled]="currentColumnOffset === 0">
+      <i
+        nz-icon
+        nzType="left"></i>
+      Previous Columns
+    </button>
+    <button
+      nz-button
+      nzType="default"
+      (click)="onColumnShiftRight()"
+      [disabled]="!currentColumns || currentColumns.length < columnLimit">
+      Next Columns
+      <i
+        nz-icon
+        nzType="right"></i>
+    </button>
+  </div>
   <div class="table-container">
     <nz-table
       #basicTable

frontend/src/app/workspace/component/result-panel/result-table-frame/result-table-frame.component.spec.ts

@@ -25,6 +25,7 @@ import { StubOperatorMetadataService } from "../../../service/operator-metadata/
 import { HttpClientTestingModule } from "@angular/common/http/testing";
 import { NzModalModule } from "ng-zorro-antd/modal";
 import { commonTestProviders } from "../../../../common/testing/test-utils";
+import { GuiConfigService } from "../../../../common/service/gui-config.service";
 
 describe("ResultTableFrameComponent", () => {
   let component: ResultTableFrameComponent;
@@ -39,6 +40,14 @@ describe("ResultTableFrameComponent", () => {
           provide: OperatorMetadataService,
           useClass: StubOperatorMetadataService,
         },
+        {
+          provide: GuiConfigService,
+          useValue: {
+            env: {
+              limitColumns: 15,
+            },
+          },
+        },
         ...commonTestProviders,
       ],
     }).compileComponents();
@@ -60,4 +69,8 @@ describe("ResultTableFrameComponent", () => {
 
     expect(component.currentResult).toEqual([{ test: "property" }]);
   });
+
+  it("should set columnLimit from config", () => {
+    expect(component.columnLimit).toEqual(15);
+  });
 });

frontend/src/app/workspace/component/result-panel/result-table-frame/result-table-frame.component.ts

@@ -31,6 +31,7 @@ import { DomSanitizer, SafeHtml } from "@angular/platform-browser";
 import { ResultExportationComponent } from "../../result-exportation/result-exportation.component";
 import { SchemaAttribute } from "../../../types/workflow-compiling.interface";
 import { WorkflowStatusService } from "../../../service/workflow-status/workflow-status.service";
+import { GuiConfigService } from "../../../../common/service/gui-config.service";
 
 /**
  * The Component will display the result in an excel table format,
@@ -66,6 +67,9 @@ export class ResultTableFrameComponent implements OnInit, OnChanges {
   currentPageIndex: number = 1;
   totalNumTuples: number = 0;
   pageSize = 5;
+  currentColumnOffset = 0;
+  columnLimit = 25;
+  columnSearch = "";
   panelHeight = 0;
   tableStats: Record<string, Record<string, number>> = {};
   prevTableStats: Record<string, Record<string, number>> = {};
@@ -81,7 +85,8 @@ export class ResultTableFrameComponent implements OnInit, OnChanges {
     private resizeService: PanelResizeService,
     private changeDetectorRef: ChangeDetectorRef,
     private sanitizer: DomSanitizer,
-    private workflowStatusService: WorkflowStatusService
+    private workflowStatusService: WorkflowStatusService,
+    private guiConfigService: GuiConfigService
   ) {}
 
   ngOnChanges(changes: SimpleChanges): void {
@@ -114,6 +119,8 @@ export class ResultTableFrameComponent implements OnInit, OnChanges {
         }
       });
 
+    this.columnLimit = this.guiConfigService.env.limitColumns;
+
     this.workflowResultService
       .getResultUpdateStream()
       .pipe(untilDestroyed(this))
@@ -233,8 +240,37 @@ export class ResultTableFrameComponent implements OnInit, OnChanges {
     return this.sanitizer.bypassSecurityTrustHtml(styledValue);
   }
 
+  /**
+   * Adjusts the number of result rows displayed per page based on the
+   * available vertical space of the Texera results panel.
+   *
+   * The method accounts for fixed UI elements within the panel—such as
+   * headers, column navigation controls, pagination, and the search bar—
+   * to determine the remaining space available for rendering result rows.
+   * The page size is then recalculated using the height of a single table row.
+   *
+   * To maintain a stable user experience during panel resizes, the current
+   * page index is recomputed so that the previously visible results remain
+   * in view and the user does not experience an abrupt jump in the dataset.
+   *
+   * @param panelHeight - The total height (in pixels) of the results panel.
+   */
   private adjustPageSizeBasedOnPanelSize(panelHeight: number) {
-    const newPageSize = Math.max(1, Math.floor((panelHeight - 38.62 - 64.27 - 56.6 - 32.63) / 38.62));
+    const TABLE_HEADER_HEIGHT = 38.62;
+    const PANEL_HEADER_HEIGHT = 64.27; // Includes panel title and tab bar
+    const COLUMN_NAVIGATION_HEIGHT = 56.6; // Previous/Next columns controls
+    const PAGINATION_HEIGHT = 32.63;
+    const SEARCH_BAR_HEIGHT_WITH_MARGIN = 77; // Approximate height for search bar and margins
+    const ROW_HEIGHT = 38.62;
+
+    const usedHeight =
+      TABLE_HEADER_HEIGHT +
+      PANEL_HEADER_HEIGHT +
+      COLUMN_NAVIGATION_HEIGHT +
+      PAGINATION_HEIGHT +
+      SEARCH_BAR_HEIGHT_WITH_MARGIN;
+
+    const newPageSize = Math.max(1, Math.floor((panelHeight - usedHeight) / ROW_HEIGHT));
 
     const oldOffset = (this.currentPageIndex - 1) * this.pageSize;
 
@@ -329,7 +365,7 @@ export class ResultTableFrameComponent implements OnInit, OnChanges {
     }
     this.isLoadingResult = true;
     paginatedResultService
-      .selectPage(this.currentPageIndex, this.pageSize)
+      .selectPage(this.currentPageIndex, this.pageSize, this.currentColumnOffset, this.columnLimit, this.columnSearch)
       .pipe(untilDestroyed(this))
       .subscribe(pageData => {
         if (this.currentPageIndex === pageData.pageIndex) {
@@ -405,4 +441,25 @@ export class ResultTableFrameComponent implements OnInit, OnChanges {
       nzFooter: null,
     });
   }
+
+  onColumnShiftLeft(): void {
+    if (this.currentColumnOffset > 0) {
+      this.currentColumnOffset = Math.max(0, this.currentColumnOffset - this.columnLimit);
+      this.changePaginatedResultData();
+    }
+  }
+
+  onColumnShiftRight(): void {
+    if (this.currentColumns && this.currentColumns.length === this.columnLimit) {
+      this.currentColumnOffset += this.columnLimit;
+      this.changePaginatedResultData();
+    }
+  }
+
+  onColumnSearch(event: Event): void {
+    const input = event.target as HTMLInputElement;
+    this.columnSearch = input.value;
+    this.currentColumnOffset = 0;
+    this.changePaginatedResultData();
+  }
 }