feat: add File Lister operator

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/LogicalOp.scala

@@ -75,6 +75,7 @@ import org.apache.texera.amber.operator.source.apis.twitter.v2.{
   TwitterFullArchiveSearchSourceOpDesc,
   TwitterSearchSourceOpDesc
 }
+import org.apache.texera.amber.operator.source.dataset.FileListerSourceOpDesc
 import org.apache.texera.amber.operator.source.fetcher.URLFetcherOpDesc
 import org.apache.texera.amber.operator.source.scan.FileScanSourceOpDesc
 import org.apache.texera.amber.operator.source.scan.arrow.ArrowSourceOpDesc
@@ -158,6 +159,7 @@ trait StateTransferFunc
     new Type(value = classOf[IfOpDesc], name = "If"),
     new Type(value = classOf[SankeyDiagramOpDesc], name = "SankeyDiagram"),
     new Type(value = classOf[IcicleChartOpDesc], name = "IcicleChart"),
+    new Type(value = classOf[FileListerSourceOpDesc], name = "FileLister"),
     new Type(value = classOf[CSVScanSourceOpDesc], name = "CSVFileScan"),
     // disabled the ParallelCSVScanSourceOpDesc so that it does not confuse user. it can be re-enabled when doing experiments.
     // new Type(value = classOf[ParallelCSVScanSourceOpDesc], name = "ParallelCSVFileScan"),

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/dataset/FileListerSourceOpDesc.scala

@@ -0,0 +1,68 @@
+/*
+ * 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.
+ */
+
+package org.apache.texera.amber.operator.source.dataset
+
+import com.fasterxml.jackson.annotation.JsonProperty
+import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle
+import org.apache.texera.amber.core.executor.OpExecWithClassName
+import org.apache.texera.amber.core.tuple.{AttributeType, Schema}
+import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
+import org.apache.texera.amber.core.workflow.{OutputPort, PhysicalOp, SchemaPropagationFunc}
+import org.apache.texera.amber.operator.LogicalOp
+import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
+import org.apache.texera.amber.util.JSONUtils.objectMapper
+
+class FileListerSourceOpDesc extends LogicalOp {
+
+  @JsonProperty(required = true)
+  @JsonSchemaTitle("Dataset")
+  var datasetVersionPath: String = _
+
+  override def getPhysicalOp(
+      workflowId: WorkflowIdentity,
+      executionId: ExecutionIdentity
+  ): PhysicalOp =
+    PhysicalOp
+      .sourcePhysicalOp(
+        workflowId,
+        executionId,
+        operatorIdentifier,
+        OpExecWithClassName(
+          "org.apache.texera.amber.operator.source.dataset.FileListerSourceOpExec",
+          objectMapper.writeValueAsString(this)
+        )
+      )
+      .withInputPorts(operatorInfo.inputPorts)
+      .withOutputPorts(operatorInfo.outputPorts)
+      .withPropagateSchema(
+        SchemaPropagationFunc(_ =>
+          Map(operatorInfo.outputPorts.head.id -> Schema().add("filename", AttributeType.STRING))
+        )
+      )
+
+  override def operatorInfo: OperatorInfo =
+    OperatorInfo(
+      userFriendlyName = "File Lister",
+      operatorDescription = "Select a dataset version and output one filename tuple per file",
+      operatorGroupName = OperatorGroupConstants.INPUT_GROUP,
+      inputPorts = List.empty,
+      outputPorts = List(OutputPort())
+    )
+}

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/dataset/FileListerSourceOpExec.scala

@@ -0,0 +1,59 @@
+/*
+ * 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.
+ */
+
+package org.apache.texera.amber.operator.source.dataset
+
+import org.apache.texera.amber.core.executor.SourceOperatorExecutor
+import org.apache.texera.amber.core.storage.util.LakeFSStorageClient
+import org.apache.texera.amber.core.tuple.TupleLike
+import org.apache.texera.amber.util.JSONUtils.objectMapper
+import org.apache.texera.dao.SqlServer
+import org.apache.texera.dao.jooq.generated.tables.Dataset.DATASET
+import org.apache.texera.dao.jooq.generated.tables.DatasetVersion.DATASET_VERSION
+import org.apache.texera.dao.jooq.generated.tables.User.USER
+
+class FileListerSourceOpExec private[dataset] (descString: String) extends SourceOperatorExecutor {
+  private val desc: FileListerSourceOpDesc =
+    objectMapper.readValue(descString, classOf[FileListerSourceOpDesc])
+
+  override def produceTuple(): Iterator[TupleLike] = {
+    val Seq(_, ownerEmail, datasetName, versionName) =
+      desc.datasetVersionPath.split("/").toSeq
+
+    val (repositoryName, versionHash) =
+      SqlServer
+        .getInstance()
+        .createDSLContext()
+        .select(DATASET.REPOSITORY_NAME, DATASET_VERSION.VERSION_HASH)
+        .from(DATASET)
+        .join(USER)
+        .on(USER.UID.eq(DATASET.OWNER_UID))
+        .join(DATASET_VERSION)
+        .on(DATASET_VERSION.DID.eq(DATASET.DID))
+        .where(USER.EMAIL.eq(ownerEmail))
+        .and(DATASET.NAME.eq(datasetName))
+        .and(DATASET_VERSION.NAME.eq(versionName))
+        .fetchOne(r => (r.value1(), r.value2()))
+
+    LakeFSStorageClient
+      .retrieveObjectsOfVersion(repositoryName, versionHash)
+      .map(obj => TupleLike("filename" -> s"${desc.datasetVersionPath}/${obj.getPath}"))
+      .iterator
+  }
+}

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/dataset/FileListerSourceOpDescSpec.scala

@@ -0,0 +1,43 @@
+/*
+ * 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.
+ */
+
+package org.apache.texera.amber.operator.source.dataset
+
+import org.apache.texera.amber.core.tuple.AttributeType
+import org.scalatest.flatspec.AnyFlatSpec
+
+class FileListerSourceOpDescSpec extends AnyFlatSpec {
+
+  "FileListerSourceOpDesc" should "expose a filename output column" in {
+    val opDesc = new FileListerSourceOpDesc()
+
+    val outputSchema = opDesc.getExternalOutputSchemas(Map.empty).values.head
+
+    assert(outputSchema.getAttributes.length == 1)
+    assert(outputSchema.getAttribute("filename").getType == AttributeType.STRING)
+  }
+
+  it should "use the expected operator metadata" in {
+    val opDesc = new FileListerSourceOpDesc()
+
+    assert(opDesc.operatorInfo.userFriendlyName == "File Lister")
+    assert(opDesc.operatorInfo.inputPorts.isEmpty)
+    assert(opDesc.operatorInfo.outputPorts.length == 1)
+  }
+}

frontend/src/app/app.module.ts

@@ -103,7 +103,9 @@ import { CoeditorUserIconComponent } from "./workspace/component/menu/coeditor-u
 import { AgentPanelComponent } from "./workspace/component/agent-panel/agent-panel.component";
 import { AgentChatComponent } from "./workspace/component/agent-panel/agent-chat/agent-chat.component";
 import { AgentRegistrationComponent } from "./workspace/component/agent-panel/agent-registration/agent-registration.component";
-import { InputAutoCompleteComponent } from "./workspace/component/input-autocomplete/input-autocomplete.component";
+import { DatasetFileSelectorComponent } from "./workspace/component/dataset-file-selector/dataset-file-selector.component";
+import { DatasetVersionSelectorComponent } from "./workspace/component/dataset-version-selector/dataset-version-selector.component";
+import { DatasetSelectionModalComponent } from "./workspace/component/dataset-selection-modal/dataset-selection-modal.component";
 import { CollabWrapperComponent } from "./common/formly/collab-wrapper/collab-wrapper/collab-wrapper.component";
 import { TexeraCopilot } from "./workspace/service/copilot/texera-copilot";
 import { NzSwitchModule } from "ng-zorro-antd/switch";
@@ -152,7 +154,6 @@ import { NzTreeModule } from "ng-zorro-antd/tree";
 import { NzTreeViewModule } from "ng-zorro-antd/tree-view";
 import { NzNoAnimationModule } from "ng-zorro-antd/core/no-animation";
 import { TreeModule } from "@ali-hm/angular-tree-component";
-import { FileSelectionComponent } from "./workspace/component/file-selection/file-selection.component";
 import { ResultExportationComponent } from "./workspace/component/result-exportation/result-exportation.component";
 import { ReportGenerationService } from "./workspace/service/report-generation/report-generation.service";
 import { SearchBarComponent } from "./dashboard/component/user/search-bar/search-bar.component";
@@ -257,8 +258,9 @@ registerLocaleData(en);
     AgentPanelComponent,
     AgentChatComponent,
     AgentRegistrationComponent,
-    InputAutoCompleteComponent,
-    FileSelectionComponent,
+    DatasetFileSelectorComponent,
+    DatasetVersionSelectorComponent,
+    DatasetSelectionModalComponent,
     CollabWrapperComponent,
     AboutComponent,
     UserWorkflowListItemComponent,

frontend/src/app/common/formly/formly-config.ts

@@ -24,9 +24,10 @@ import { MultiSchemaTypeComponent } from "./multischema.type";
 import { FormlyFieldConfig } from "@ngx-formly/core";
 import { CodeareaCustomTemplateComponent } from "../../workspace/component/codearea-custom-template/codearea-custom-template.component";
 import { PresetWrapperComponent } from "./preset-wrapper/preset-wrapper.component";
-import { InputAutoCompleteComponent } from "../../workspace/component/input-autocomplete/input-autocomplete.component";
+import { DatasetFileSelectorComponent } from "../../workspace/component/dataset-file-selector/dataset-file-selector.component";
 import { CollabWrapperComponent } from "./collab-wrapper/collab-wrapper/collab-wrapper.component";
 import { FormlyRepeatDndComponent } from "./repeat-dnd/repeat-dnd.component";
+import { DatasetVersionSelectorComponent } from "../../workspace/component/dataset-version-selector/dataset-version-selector.component";
 
 /**
  * Configuration for using Json Schema with Formly.
@@ -76,7 +77,8 @@ export const TEXERA_FORMLY_CONFIG = {
     { name: "object", component: ObjectTypeComponent },
     { name: "multischema", component: MultiSchemaTypeComponent },
     { name: "codearea", component: CodeareaCustomTemplateComponent },
-    { name: "inputautocomplete", component: InputAutoCompleteComponent, wrappers: ["form-field"] },
+    { name: "inputautocomplete", component: DatasetFileSelectorComponent, wrappers: ["form-field"] },
+    { name: "datasetversionselector", component: DatasetVersionSelectorComponent, wrappers: ["form-field"] },
     { name: "repeat-section-dnd", component: FormlyRepeatDndComponent },
   ],
   wrappers: [

frontend/src/app/workspace/component/input-autocomplete/input-autocomplete.component.html → frontend/src/app/workspace/component/dataset-file-selector/dataset-file-selector.component.html

@@ -16,27 +16,16 @@
  specific language governing permissions and limitations
  under the License.
 -->
-
-<div [ngClass]="{'input-autocomplete-container': isFileSelectionEnabled}">
-  <input
-    *ngIf="selectedFilePath || !isFileSelectionEnabled"
-    nz-input
-    [readOnly]="isFileSelectionEnabled"
-    required
-    [formControl]="formControl"
-    [formlyAttributes]="field" />
-  <button
-    *ngIf="isFileSelectionEnabled"
-    nz-button
-    class="file-select-button"
-    nzSize="small"
-    (click)="isFileSelectionEnabled && onClickOpenFileSelectionModal()">
-    {{ selectedFilePath ? 'Reselect File' : 'Select File' }}
-  </button>
-</div>
-<div
-  class="alert alert-danger"
-  role="alert"
-  *ngIf="props.showError && formControl.errors">
-  <formly-validation-message [field]="field"></formly-validation-message>
-</div>
+<input
+  *ngIf="formControl.value || !isFileSelectionEnabled"
+  nz-input
+  required
+  [readOnly]="isFileSelectionEnabled"
+  [formControl]="formControl" />
+<button
+  *ngIf="isFileSelectionEnabled"
+  nz-button
+  nzSize="small"
+  (click)="isFileSelectionEnabled && onClickOpenFileSelectionModal()">
+  Select File
+</button>

frontend/src/app/workspace/component/input-autocomplete/input-autocomplete.component.ts → frontend/src/app/workspace/component/dataset-file-selector/dataset-file-selector.component.ts

@@ -22,17 +22,14 @@ import { FieldType, FieldTypeConfig } from "@ngx-formly/core";
 import { UntilDestroy, untilDestroyed } from "@ngneat/until-destroy";
 import { WorkflowActionService } from "../../service/workflow-graph/model/workflow-action.service";
 import { NzModalService } from "ng-zorro-antd/modal";
-import { FileSelectionComponent } from "../file-selection/file-selection.component";
-import { DatasetFileNode, getFullPathFromDatasetFileNode } from "../../../common/type/datasetVersionFileTree";
+import { DatasetSelectionModalComponent } from "../dataset-selection-modal/dataset-selection-modal.component";
 import { GuiConfigService } from "../../../common/service/gui-config.service";
 
 @UntilDestroy()
 @Component({
-  selector: "texera-input-autocomplete-template",
-  templateUrl: "./input-autocomplete.component.html",
-  styleUrls: ["input-autocomplete.component.scss"],
+  templateUrl: "dataset-file-selector.component.html",
 })
-export class InputAutoCompleteComponent extends FieldType<FieldTypeConfig> {
+export class DatasetFileSelectorComponent extends FieldType<FieldTypeConfig> {
   constructor(
     private modalService: NzModalService,
     public workflowActionService: WorkflowActionService,
@@ -43,14 +40,13 @@ export class InputAutoCompleteComponent extends FieldType<FieldTypeConfig> {
 
   onClickOpenFileSelectionModal(): void {
     const modal = this.modalService.create({
-      nzTitle: "Please select one file from datasets",
-      nzContent: FileSelectionComponent,
+      nzContent: DatasetSelectionModalComponent,
       nzFooter: null,
       nzData: {
-        selectedFilePath: this.formControl.getRawValue(),
+        fileMode: true,
+        selectedPath: this.formControl.getRawValue(),
       },
       nzBodyStyle: {
-        // Enables the file selection window to be resizable
         resize: "both",
         overflow: "auto",
         minHeight: "200px",
@@ -60,22 +56,14 @@ export class InputAutoCompleteComponent extends FieldType<FieldTypeConfig> {
       },
       nzWidth: "fit-content",
     });
-    // Handle the selection from the modal
-    modal.afterClose.pipe(untilDestroyed(this)).subscribe(fileNode => {
-      const node: DatasetFileNode = fileNode as DatasetFileNode;
-      this.formControl.setValue(getFullPathFromDatasetFileNode(node));
+    modal.afterClose.pipe(untilDestroyed(this)).subscribe(selectedPath => {
+      if (selectedPath) {
+        this.formControl.setValue(selectedPath);
+      }
     });
   }
 
-  get enableDatasetSource(): boolean {
-    return this.config.env.selectingFilesFromDatasetsEnabled;
-  }
-
   get isFileSelectionEnabled(): boolean {
-    return this.enableDatasetSource;
-  }
-
-  get selectedFilePath(): string | null {
-    return this.formControl.value;
+    return this.config.env.selectingFilesFromDatasetsEnabled;
   }
 }