test(workflow-operator): add unit test coverage for operator support utilities

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/SpecialPhysicalOpFactorySpec.scala

@@ -0,0 +1,100 @@
+/*
+ * 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
+
+import org.apache.texera.amber.core.storage.VFSURIFactory
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema}
+import org.apache.texera.amber.core.virtualidentity.{
+  ExecutionIdentity,
+  OperatorIdentity,
+  PhysicalOpIdentity,
+  WorkflowIdentity
+}
+import org.apache.texera.amber.core.workflow.{GlobalPortIdentity, PortIdentity, PreferController}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class SpecialPhysicalOpFactorySpec extends AnyFlatSpec with Matchers {
+
+  private val workflowId = WorkflowIdentity(42L)
+  private val executionId = ExecutionIdentity(7L)
+  private val schema = Schema().add(new Attribute("col", AttributeType.STRING))
+
+  private def portUri(opName: String, layer: String, portId: Int) = {
+    val gpid = GlobalPortIdentity(
+      PhysicalOpIdentity(OperatorIdentity(opName), layer),
+      PortIdentity(portId),
+      input = false
+    )
+    VFSURIFactory.resultURI(VFSURIFactory.createPortBaseURI(workflowId, executionId, gpid))
+  }
+
+  "SpecialPhysicalOpFactory.newSourcePhysicalOp" should
+    "derive a source op that wires the decoded port identity, ports, and schema" in {
+    val uri = portUri("srcOp", "layerA", 3)
+    val downstream = PhysicalOpIdentity(OperatorIdentity("down-stream"), "main")
+    val op = SpecialPhysicalOpFactory.newSourcePhysicalOp(
+      workflowId,
+      executionId,
+      uri,
+      downstream,
+      PortIdentity(5),
+      schema
+    )
+    op.id.logicalOpId shouldBe OperatorIdentity("srcOp")
+    // layerName: "${layerName}_source_${portId.id}_${downstreamLogicalId'}_${downstreamPort.id}"
+    // with '-' in the downstream logical id replaced by '_'
+    op.id.layerName shouldBe "layerA_source_3_down_stream_5"
+    op.workflowId shouldBe workflowId
+    op.executionId shouldBe executionId
+    op.isSourceOperator shouldBe true
+    op.inputPorts shouldBe empty
+    op.outputPorts.keySet shouldBe Set(PortIdentity(0))
+    op.locationPreference shouldBe Some(PreferController)
+    op.parallelizable shouldBe false
+    op.outputPorts(PortIdentity(0))._3 shouldBe Right(schema)
+  }
+
+  it should "replace every dash in the downstream operator id when building the layer name" in {
+    val op = SpecialPhysicalOpFactory.newSourcePhysicalOp(
+      workflowId,
+      executionId,
+      portUri("srcOp", "layerA", 3),
+      PhysicalOpIdentity(OperatorIdentity("a-b-c"), "main"),
+      PortIdentity(9),
+      schema
+    )
+    op.id.layerName shouldBe "layerA_source_3_a_b_c_9"
+  }
+
+  it should "fail when the URI carries no global port identity" in {
+    val noPortUri = VFSURIFactory.createRuntimeStatisticsURI(workflowId, executionId)
+    intercept[NoSuchElementException] {
+      SpecialPhysicalOpFactory.newSourcePhysicalOp(
+        workflowId,
+        executionId,
+        noPortUri,
+        PhysicalOpIdentity(OperatorIdentity("down"), "main"),
+        PortIdentity(0),
+        schema
+      )
+    }
+  }
+}

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/TestOperatorsSpec.scala

@@ -0,0 +1,111 @@
+/*
+ * 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
+
+import org.apache.texera.amber.core.tuple.AttributeType
+import org.apache.texera.amber.operator.aggregate.AggregationFunction
+import org.scalatest.flatspec.AnyFlatSpec
+
+class TestOperatorsSpec extends AnyFlatSpec {
+
+  "TestOperators.getCsvScanOpDesc" should "build a resolved CSV desc honoring the header flag" in {
+    val op = TestOperators.getCsvScanOpDesc(TestOperators.CountrySalesSmallCsvPath, header = true)
+    assert(op.hasHeader)
+    assert(op.customDelimiter.contains(","))
+    assert(op.fileName.isDefined)
+    assert(op.fileResolved())
+    val headerless =
+      TestOperators.getCsvScanOpDesc(
+        TestOperators.CountrySalesHeaderlessSmallCsvPath,
+        header = false,
+        multiLine = true
+      )
+    assert(!headerless.hasHeader)
+  }
+
+  "TestOperators CSV factories" should "resolve their backing files" in {
+    assert(!TestOperators.headerlessSmallCsvScanOpDesc().hasHeader)
+    assert(!TestOperators.headerlessSmallMultiLineDataCsvScanOpDesc().hasHeader)
+    assert(TestOperators.smallCsvScanOpDesc().hasHeader)
+    assert(TestOperators.mediumCsvScanOpDesc().hasHeader)
+    assert(TestOperators.smallCsvScanOpDesc().fileResolved())
+  }
+
+  "TestOperators.getJSONLScanOpDesc" should "build a resolved JSONL desc honoring the flatten flag" in {
+    val op = TestOperators.getJSONLScanOpDesc(TestOperators.smallJsonLPath)
+    assert(!op.flatten)
+    assert(op.fileResolved())
+    val flattened = TestOperators.getJSONLScanOpDesc(TestOperators.mediumJsonLPath, flatten = true)
+    assert(flattened.flatten)
+  }
+
+  "TestOperators JSONL factories" should "resolve their backing files" in {
+    assert(!TestOperators.smallJSONLScanOpDesc().flatten)
+    assert(TestOperators.mediumFlattenJSONLScanOpDesc().flatten)
+    assert(TestOperators.smallJSONLScanOpDesc().fileResolved())
+  }
+
+  "TestOperators.joinOpDesc" should "set the build and probe attribute names" in {
+    val op = TestOperators.joinOpDesc("build_col", "probe_col")
+    assert(op.buildAttributeName == "build_col")
+    assert(op.probeAttributeName == "probe_col")
+  }
+
+  "TestOperators.keywordSearchOpDesc" should "set the attribute and keyword" in {
+    val op = TestOperators.keywordSearchOpDesc("text", "hello")
+    assert(op.attribute == "text")
+    assert(op.keyword == "hello")
+  }
+
+  "TestOperators.aggregateAndGroupByDesc" should "wire a single aggregation with group-by keys" in {
+    val op = TestOperators.aggregateAndGroupByDesc("price", AggregationFunction.SUM, List("region"))
+    assert(op.aggregations.length == 1)
+    assert(op.aggregations.head.aggFunction == AggregationFunction.SUM)
+    assert(op.aggregations.head.attribute == "price")
+    assert(op.aggregations.head.resultAttribute == "aggregate-result")
+    assert(op.groupByKeys == List("region"))
+  }
+
+  "TestOperators.asterixDBSourceOpDesc" should "build the AsterixDB desc without contacting a server" in {
+    val op = TestOperators.asterixDBSourceOpDesc()
+    assert(op.host == "ipubmed4.ics.uci.edu")
+    assert(op.port == "default")
+    assert(op.database == "twitter")
+    assert(op.table == "ds_tweet")
+    assert(op.limit.contains(1000L))
+  }
+
+  "TestOperators.pythonOpDesc" should "build a python UDF desc" in {
+    val op = TestOperators.pythonOpDesc()
+    assert(op.workers == 1)
+    assert(op.retainInputColumns)
+    assert(op.code.contains("ProcessTupleOperator"))
+    assert(op.code.contains("process_tuple"))
+  }
+
+  "TestOperators.pythonSourceOpDesc" should "build a python source UDF desc interpolating the tuple count" in {
+    val op = TestOperators.pythonSourceOpDesc(5)
+    assert(op.workers == 1)
+    assert(op.columns.length == 1)
+    assert(op.columns.head.getName == "field_1")
+    assert(op.columns.head.getType == AttributeType.STRING)
+    assert(op.code.contains("range(5)"))
+  }
+}

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/file/FileScanUtilsSpec.scala

@@ -0,0 +1,104 @@
+/*
+ * 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.scan.file
+
+import org.apache.texera.amber.operator.source.scan.{FileAttributeType, FileDecodingMethod}
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.flatspec.AnyFlatSpec
+
+import java.io.{BufferedOutputStream, FileOutputStream}
+import java.nio.file.{Files, Path}
+import java.util.zip.{ZipEntry, ZipOutputStream}
+
+class FileScanUtilsSpec extends AnyFlatSpec with BeforeAndAfterAll {
+
+  private val zips = scala.collection.mutable.ArrayBuffer.empty[Path]
+
+  private def makeZip(entries: (String, String)*): String = {
+    val path = Files.createTempFile("filescanutils-", ".zip")
+    zips += path
+    val zipOut = new ZipOutputStream(new BufferedOutputStream(new FileOutputStream(path.toFile)))
+    try {
+      entries.foreach {
+        case (name, content) =>
+          zipOut.putNextEntry(new ZipEntry(name))
+          zipOut.write(content.getBytes("UTF-8"))
+          zipOut.closeEntry()
+      }
+    } finally {
+      zipOut.close()
+    }
+    path.toFile.toURI.toString
+  }
+
+  override def afterAll(): Unit = {
+    zips.foreach(Files.deleteIfExists)
+    super.afterAll()
+  }
+
+  "FileScanUtils.createTuplesFromFile" should
+    "extract every zip entry as a single-string tuple" in {
+    val tuples = FileScanUtils
+      .createTuplesFromFile(
+        fileName = makeZip("a.txt" -> "Content A", "b.txt" -> "Content B"),
+        displayFileName = "ignored-when-extracting",
+        attributeType = FileAttributeType.SINGLE_STRING,
+        fileEncoding = FileDecodingMethod.UTF_8,
+        extract = true,
+        outputFileName = false,
+        fileScanOffset = None,
+        fileScanLimit = None
+      )
+      .toSeq
+    assert(tuples.size == 2)
+  }
+
+  it should "drop __MACOSX metadata entries when extracting" in {
+    val tuples = FileScanUtils
+      .createTuplesFromFile(
+        fileName = makeZip("real.txt" -> "keep me", "__MACOSX/._real.txt" -> "junk"),
+        displayFileName = "d",
+        attributeType = FileAttributeType.SINGLE_STRING,
+        fileEncoding = FileDecodingMethod.UTF_8,
+        extract = true,
+        outputFileName = false,
+        fileScanOffset = None,
+        fileScanLimit = None
+      )
+      .toSeq
+    assert(tuples.size == 1)
+  }
+
+  it should "flat-map each line of an extracted entry for a per-line attribute type" in {
+    val tuples = FileScanUtils
+      .createTuplesFromFile(
+        fileName = makeZip("lines.txt" -> "l1\nl2\nl3"),
+        displayFileName = "d",
+        attributeType = FileAttributeType.STRING,
+        fileEncoding = FileDecodingMethod.UTF_8,
+        extract = true,
+        outputFileName = false,
+        fileScanOffset = None,
+        fileScanLimit = None
+      )
+      .toSeq
+    assert(tuples.size == 3)
+  }
+}