test(workflow-core): fill uncovered branches in tuple and type utilities

common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/AttributeTypeSpec.scala

@@ -0,0 +1,71 @@
+/*
+ * 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.core.tuple
+
+import org.scalatest.flatspec.AnyFlatSpec
+
+class AttributeTypeSpec extends AnyFlatSpec {
+
+  // ----- getName -----
+
+  "AttributeType.getName" should "return the lowercase wire name for each concrete type" in {
+    assert(AttributeType.STRING.getName == "string")
+    assert(AttributeType.INTEGER.getName == "integer")
+    assert(AttributeType.LONG.getName == "long")
+    assert(AttributeType.DOUBLE.getName == "double")
+    assert(AttributeType.BOOLEAN.getName == "boolean")
+    assert(AttributeType.TIMESTAMP.getName == "timestamp")
+    assert(AttributeType.BINARY.getName == "binary")
+    assert(AttributeType.LARGE_BINARY.getName == "large_binary")
+  }
+
+  it should "return an empty string for ANY (excluded from the JSON schema)" in {
+    assert(AttributeType.ANY.getName == "")
+    assert(AttributeType.ANY.toString == "")
+    assert(AttributeType.ANY.name() == "ANY") // the built-in enum name is unaffected
+  }
+
+  // ----- getAttributeType -----
+
+  "AttributeType.getAttributeType" should "map each supported field class to its type" in {
+    assert(AttributeType.getAttributeType(classOf[String]) == AttributeType.STRING)
+    assert(AttributeType.getAttributeType(classOf[java.lang.Integer]) == AttributeType.INTEGER)
+    assert(AttributeType.getAttributeType(classOf[java.lang.Long]) == AttributeType.LONG)
+    assert(AttributeType.getAttributeType(classOf[java.lang.Double]) == AttributeType.DOUBLE)
+    assert(AttributeType.getAttributeType(classOf[java.lang.Boolean]) == AttributeType.BOOLEAN)
+    assert(AttributeType.getAttributeType(classOf[java.sql.Timestamp]) == AttributeType.TIMESTAMP)
+    assert(AttributeType.getAttributeType(classOf[Array[Byte]]) == AttributeType.BINARY)
+    assert(AttributeType.getAttributeType(classOf[LargeBinary]) == AttributeType.LARGE_BINARY)
+  }
+
+  it should "fall back to ANY for unrecognized classes, including primitives" in {
+    assert(AttributeType.getAttributeType(classOf[Object]) == AttributeType.ANY)
+    // Scala's classOf[Int] is the primitive int.class, which the boxed-class
+    // checks do not match — it falls through to ANY.
+    assert(AttributeType.getAttributeType(classOf[Int]) == AttributeType.ANY)
+  }
+
+  it should "round-trip every concrete type through its field class" in {
+    AttributeType
+      .values()
+      .filter(_ != AttributeType.ANY)
+      .foreach(t => assert(AttributeType.getAttributeType(t.getFieldClass) == t))
+  }
+}

common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/AttributeTypeUtilsSpec.scala

@@ -22,13 +22,16 @@ package org.apache.texera.amber.core.tuple
 import org.apache.texera.amber.core.tuple.AttributeType._
 import org.apache.texera.amber.core.tuple.AttributeTypeUtils.{
   AttributeTypeException,
+  SchemaCasting,
   add,
   compare,
   inferField,
   inferSchemaFromRows,
   maxValue,
   minValue,
   parseField,
+  parseFields,
+  tupleCasting,
   zeroValue
 }
 import org.scalatest.funsuite.AnyFunSuite
@@ -379,4 +382,97 @@ class AttributeTypeUtilsSpec extends AnyFunSuite {
       minValue(STRING)
     }
   }
+
+  test("SchemaCasting casts the named attribute and retains the others") {
+    val schema = Schema(
+      List(new Attribute("a", STRING), new Attribute("b", STRING))
+    )
+    val casted = SchemaCasting(schema, "b", INTEGER)
+    assert(
+      casted.getAttributes == List(new Attribute("a", STRING), new Attribute("b", INTEGER))
+    )
+  }
+
+  test("SchemaCasting with an ANY result type retains the schema unchanged") {
+    val schema = Schema(
+      List(new Attribute("a", STRING), new Attribute("b", STRING))
+    )
+    assert(SchemaCasting(schema, "b", ANY) == schema)
+  }
+
+  test("tupleCasting parses targeted columns with forced number parsing") {
+    val schema = Schema(
+      List(new Attribute("a", STRING), new Attribute("b", STRING))
+    )
+    val tuple = Tuple.builder(schema).addSequentially(Array[Any]("1,234", "note")).build()
+    val result = tupleCasting(tuple, Map("a" -> INTEGER))
+    assert(result.getFields.sameElements(Array[Any](1234, "note")))
+  }
+
+  test("parseFields parses by attribute-type array and by schema") {
+    assert(
+      parseFields(Array[Any]("1", "2.5", "true"), Array(INTEGER, DOUBLE, BOOLEAN))
+        .sameElements(Array[Any](1, 2.5, true))
+    )
+    assert(
+      parseFields(Array[Any]("7"), Schema(List(new Attribute("x", INTEGER))))
+        .sameElements(Array[Any](7))
+    )
+    assertThrows[AttributeTypeException] {
+      parseFields(Array[Any]("abc"), Array(INTEGER))
+    }
+  }
+
+  test("inferField with a null value keeps the currently inferred type") {
+    assert(inferField(INTEGER, null) == INTEGER)
+    assert(inferField(LONG, null) == LONG)
+    assert(inferField(TIMESTAMP, null) == TIMESTAMP)
+    assert(inferField(DOUBLE, null) == DOUBLE)
+    assert(inferField(BOOLEAN, null) == BOOLEAN)
+  }
+
+  test("inferField falls back to STRING for BINARY and ANY bases") {
+    assert(inferField(BINARY, "anything") == STRING)
+    assert(inferField(ANY, "anything") == STRING)
+  }
+
+  test("compare orders LONG and DOUBLE values") {
+    assert(compare(1L, 2L, LONG) < 0)
+    assert(compare(2L, 1L, LONG) > 0)
+    assert(compare(5L, 5L, LONG) == 0)
+    assert(compare(1.5d, 2.5d, DOUBLE) < 0)
+    assert(compare(2.5d, 1.5d, DOUBLE) > 0)
+    assert(compare(3.14d, 3.14d, DOUBLE) == 0)
+  }
+
+  test("compare rejects unsupported types and orders binary unsigned") {
+    val ex = intercept[UnsupportedOperationException] {
+      compare("a", "b", LARGE_BINARY)
+    }
+    // the interpolated type renders via its lowercase wire name
+    assert(ex.getMessage == "Unsupported attribute type for compare: large_binary")
+    assert(compare(Array[Byte](0, 2, 0), Array[Byte](0, 1, 2), BINARY) > 0)
+    assert(compare(Array[Byte](1, 2), Array[Byte](1, 2), BINARY) == 0)
+    // bytes compare unsigned: 0xFF (-1) sorts above 0x01
+    assert(compare(Array[Byte](-1), Array[Byte](1), BINARY) > 0)
+  }
+
+  test("add rejects unsupported types and zero-fills all-null binary operands") {
+    val ex = intercept[UnsupportedOperationException] {
+      add("a", "b", STRING)
+    }
+    assert(ex.getMessage == "Unsupported attribute type for addition: string")
+    assert(add(null, null, BINARY).asInstanceOf[Array[Byte]].isEmpty)
+  }
+
+  test("parseField with force enabled raises AttributeTypeException on unparsable numbers") {
+    val intEx = intercept[AttributeTypeException] {
+      parseField("abc", INTEGER, force = true)
+    }
+    assert(intEx.getMessage == "Failed to parse type java.lang.String to Integer: abc")
+    val longEx = intercept[AttributeTypeException] {
+      parseField("abc", LONG, force = true)
+    }
+    assert(longEx.getMessage == "Failed to parse type java.lang.String to Long: abc")
+  }
 }

common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/TupleLikeSpec.scala

@@ -0,0 +1,132 @@
+/*
+ * 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.core.tuple
+
+import org.apache.texera.amber.core.workflow.PortIdentity
+import org.scalatest.flatspec.AnyFlatSpec
+
+class TupleLikeSpec extends AnyFlatSpec {
+
+  private val intAttr = new Attribute("col-int", AttributeType.INTEGER)
+  private val strAttr = new Attribute("col-string", AttributeType.STRING)
+  private val schema = Schema().add(intAttr).add(strAttr)
+
+  // ----- internal markers (case-class synthetics; construction covered by InternalMarkerSpec) -----
+
+  "FinalizePort" should "support copy, hashCode, toString, and product members" in {
+    val marker = FinalizePort(PortIdentity(1), input = true)
+    assert(marker.copy(input = false) == FinalizePort(PortIdentity(1), input = false))
+    assert(marker.hashCode == FinalizePort(PortIdentity(1), input = true).hashCode)
+    assert(marker.toString.startsWith("FinalizePort("))
+    assert(marker.productArity == 2)
+    assert(marker.productElement(0) == PortIdentity(1))
+    assert(marker.productElement(1) == true)
+    assert(marker.productPrefix == "FinalizePort")
+    assert(!marker.canEqual(FinalizeExecutor()))
+  }
+
+  "FinalizeExecutor" should "support copy, hashCode, toString, and product members" in {
+    val marker = FinalizeExecutor()
+    assert(marker.copy() == marker)
+    assert(marker.hashCode == FinalizeExecutor().hashCode)
+    assert(marker.toString == "FinalizeExecutor()")
+    assert(marker.productArity == 0)
+  }
+
+  // ----- SeqTupleLike -----
+
+  "SeqTupleLike" should "not implement inMemSize" in {
+    assertThrows[NotImplementedError](TupleLike(1, 2).inMemSize)
+  }
+
+  it should "enforce a schema by pairing fields positionally" in {
+    val tuple = TupleLike(42, "hello").enforceSchema(schema)
+    assert(tuple.getField[Int]("col-int") == 42)
+    assert(tuple.getField[String]("col-string") == "hello")
+  }
+
+  it should "fail schema enforcement when fields are missing or extra" in {
+    assertThrows[TupleBuildingException](TupleLike(42).enforceSchema(schema))
+    assertThrows[IndexOutOfBoundsException](TupleLike(42, "hello", true).enforceSchema(schema))
+  }
+
+  // ----- MapTupleLike -----
+
+  "MapTupleLike" should "expose its mapping values as fields and not implement inMemSize" in {
+    val mapLike = TupleLike(Map[String, Any]("a" -> 1, "b" -> "x"))
+    assert(mapLike.getFields.toSet == Set[Any](1, "x"))
+    assertThrows[NotImplementedError](mapLike.inMemSize)
+  }
+
+  it should "enforce a schema by attribute name, nulling missing keys and dropping extras" in {
+    val tuple =
+      TupleLike(Map[String, Any]("col-int" -> 7, "unrelated-key" -> true)).enforceSchema(schema)
+    assert(tuple.getField[Int]("col-int") == 7)
+    assert(tuple.getField[Any]("col-string") == null)
+  }
+
+  // ----- TupleLike factory overloads -----
+
+  "TupleLike" should "build from a Map of field mappings" in {
+    val mapLike = TupleLike(Map[String, Any]("k1" -> 1, "k2" -> "v"))
+    assert(mapLike.fieldMappings == Map[String, Any]("k1" -> 1, "k2" -> "v"))
+  }
+
+  it should "build from an Iterable of name-value pairs" in {
+    val pairs: Iterable[(String, Any)] = Seq("k1" -> 1, "k2" -> "v")
+    val mapLike = TupleLike(pairs)
+    assert(mapLike.fieldMappings == Map[String, Any]("k1" -> 1, "k2" -> "v"))
+  }
+
+  it should "build from name-value pair varargs with last-wins duplicate keys" in {
+    val mapLike = TupleLike("k1" -> 1, "k2" -> "v")
+    assert(mapLike.fieldMappings == Map[String, Any]("k1" -> 1, "k2" -> "v"))
+    assert(TupleLike("k" -> 1, "k" -> 2).fieldMappings == Map[String, Any]("k" -> 2))
+  }
+
+  it should "build from a java.util.List of fields" in {
+    val javaList = new java.util.ArrayList[Any]()
+    javaList.add(1)
+    javaList.add("x")
+    assert(TupleLike(javaList).getFields.toSeq == Seq(1, "x"))
+  }
+
+  it should "build from field varargs for non-iterable types" in {
+    assert(TupleLike(1, 2, 3).getFields.toSeq == Seq(1, 2, 3))
+    assert(TupleLike("a", "b").getFields.toSeq == Seq("a", "b"))
+    assert(TupleLike(1, "a").getFields.toSeq == Seq(1, "a"))
+  }
+
+  it should "provide no NotAnIterable evidence for iterable types" in {
+    // The guard implicit exists so iterable varargs cannot materialize
+    // evidence; invoking it directly pins the rejection.
+    val ex = intercept[RuntimeException](TupleLike.NotAnIterable.iterableIsNotAnIterable[List, Any])
+    assert(ex.getMessage == "Iterable types are not allowed")
+  }
+
+  it should "build from a single Iterable of fields" in {
+    assert(TupleLike(List(1, 2, 3)).getFields.toSeq == Seq(1, 2, 3))
+  }
+
+  it should "build from an Array of fields, keeping the array by reference" in {
+    val array: Array[Any] = Array(1, "x")
+    assert(TupleLike(array).getFields eq array)
+  }
+}