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

@@ -17,43 +17,38 @@
  under the License.
 -->
 
-<div
-  *ngIf="!currentResult || currentResult.length === 0"
-  style="text-align: center">
+<div *ngIf="!currentResult || currentResult.length === 0" style="text-align: center">
   <h4>Empty result set</h4>
 </div>
-<div
-  [hidden]="!currentColumns"
-  class="result-table">
+<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
-      (nzQueryParams)="onTableQueryParamsChange($event)"
-      [nzData]="currentResult"
-      [nzFrontPagination]="isFrontPagination"
-      [nzLoading]="isLoadingResult"
-      [nzPageIndex]="currentPageIndex"
-      [nzPageSize]="pageSize"
-      [nzPaginationPosition]="'bottom'"
-      [nzScroll]="{ x: 'max-content'}"
-      [nzSize]="'small'"
-      [nzTableLayout]="'fixed'"
-      [nzTotal]="totalNumTuples"
-      nzBordered="true">
+    <nz-table #basicTable (nzQueryParams)="onTableQueryParamsChange($event)" [nzData]="currentResult"
+      [nzFrontPagination]="isFrontPagination" [nzLoading]="isLoadingResult" [nzPageIndex]="currentPageIndex"
+      [nzPageSize]="pageSize" [nzPaginationPosition]="'bottom'" [nzScroll]="{ x: 'max-content'}" [nzSize]="'small'"
+      [nzTableLayout]="'fixed'" [nzTotal]="totalNumTuples" nzBordered="true">
       <thead>
         <tr>
-          <th
-            *ngFor="let column of currentColumns; let i = index"
-            ngClass="header-size"
-            style="text-align: center"
+          <th *ngFor="let column of currentColumns; let i = index" ngClass="header-size" style="text-align: center"
             nzWidth="widthPercent">
             {{ column.header }}
           </th>
         </tr>
-        <tr
-          *ngIf="tableStats && prevTableStats"
-          #statsRow
-          class="custom-stats-row">
+        <tr *ngIf="tableStats && prevTableStats" #statsRow class="custom-stats-row">
           <th *ngFor="let column of currentColumns">
             <div class="statsHeader">
               <ng-container
@@ -118,28 +113,18 @@ <h5 class="rightAlign"><span [innerHTML]="compare(column.header, 'other')"></spa
         </tr>
       </thead>
       <tbody>
-        <tr
-          *ngFor="let row of basicTable.data; let i = index"
-          class="table-row-hover">
-          <td
-            *ngFor="let column of currentColumns; let columnIndex = index"
-            class="table-cell"
-            nzEllipsis
+        <tr *ngFor="let row of basicTable.data; let i = index" class="table-row-hover">
+          <td *ngFor="let column of currentColumns; let columnIndex = index" class="table-cell" nzEllipsis
             (click)="open(i, row)">
             <span class="cell-content">{{ column.getCell(row) }}</span>
             <button
               (click)="downloadData(currentResult[i][column.columnDef], i, columnIndex, column.columnDef); $event.stopPropagation()"
-              nz-button
-              nzType="link"
-              class="download-button"
-              title="Download data">
-              <i
-                nz-icon
-                nzType="cloud-download"></i>
+              nz-button nzType="link" class="download-button" title="Download data">
+              <i nz-icon nzType="cloud-download"></i>
             </button>
           </td>
         </tr>
       </tbody>
     </nz-table>
   </div>
-</div>
+</div>

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