diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/AttributeTypeSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/AttributeTypeSpec.scala new file mode 100644 index 00000000000..991fd2cc612 --- /dev/null +++ b/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)) + } +} diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/AttributeTypeUtilsSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/AttributeTypeUtilsSpec.scala index 446ce518faf..b6ec7e5c796 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/AttributeTypeUtilsSpec.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/AttributeTypeUtilsSpec.scala @@ -22,6 +22,7 @@ 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, @@ -29,6 +30,8 @@ import org.apache.texera.amber.core.tuple.AttributeTypeUtils.{ 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") + } } diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/TupleLikeSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/TupleLikeSpec.scala new file mode 100644 index 00000000000..1d259fc7bf1 --- /dev/null +++ b/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) + } +} diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/TupleSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/TupleSpec.scala index 4adb48fa0d7..aa882a23966 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/TupleSpec.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/TupleSpec.scala @@ -191,4 +191,126 @@ class TupleSpec extends AnyFlatSpec { .build() assert(inputTuple5.hashCode() == -2099556631) } + + it should "reject getField for an attribute name that is not in the tuple" in { + val schema = Schema().add(stringAttribute) + val tuple = Tuple.builder(schema).add(stringAttribute, "v").build() + val ex = intercept[RuntimeException] { + tuple.getField[String]("col-missing") + } + assert(ex.getMessage == "col-missing is not in the tuple") + } + + it should "get a field by Attribute" in { + val schema = Schema().add(stringAttribute) + val tuple = Tuple.builder(schema).add(stringAttribute, "v").build() + assert(tuple.getField[String](stringAttribute) == "v") + } + + it should "enforce its own schema and reject a different one" in { + val schema = Schema().add(stringAttribute) + val tuple = Tuple.builder(schema).add(stringAttribute, "v").build() + assert(tuple.enforceSchema(schema) eq tuple) + val ex = intercept[AssertionError] { + tuple.enforceSchema(Schema().add(integerAttribute)) + } + assert(ex.getMessage.contains("output tuple schema does not match the expected schema!")) + } + + it should "compare tuples by value, including binary contents" in { + val schema = Schema().add(stringAttribute).add(binaryAttribute) + def tupleWith(s: String): Tuple = + Tuple + .builder(schema) + .add(stringAttribute, s) + .add(binaryAttribute, Array[Byte](1, 2, 3)) + .build() + val tupleA = tupleWith("v") + val tupleB = tupleWith("v") + assert(tupleA == tupleB) // distinct Array[Byte] instances, same contents + val differentField = tupleWith("other") + assert(tupleA != differentField) + val differentSchema = + Tuple.builder(Schema().add(stringAttribute)).add(stringAttribute, "v").build() + assert(tupleA != differentSchema) + assert(!tupleA.equals("not a tuple")) + assert(!tupleA.equals(null)) + } + + it should "build a partial tuple following the requested attribute order" in { + val schema = Schema().add(stringAttribute).add(integerAttribute).add(boolAttribute) + val tuple = Tuple + .builder(schema) + .add(stringAttribute, "s") + .add(integerAttribute, 1) + .add(boolAttribute, true) + .build() + val partial = tuple.getPartialTuple(List("col-bool", "col-string")) + assert(partial.length == 2) + assert(partial.getField[Boolean]("col-bool")) + assert(partial.getField[String]("col-string") == "s") + assert(partial.getSchema == Schema().add(boolAttribute).add(stringAttribute)) + } + + it should "render toString with the schema and field values" in { + val schema = Schema().add(stringAttribute) + val tuple = Tuple.builder(schema).add(stringAttribute, "v").build() + assert(tuple.toString == s"Tuple [schema=$schema, fields=[v]]") + } + + it should "reject construction when schema and field sizes differ" in { + val schema = Schema().add(stringAttribute) + val ex = intercept[RuntimeException] { + Tuple(schema, Array[Any]("a", "b")) + } + assert(ex.getMessage == "Schema size (1) and field size (2) are different") + } + + it should "reject a field whose class does not match the attribute type" in { + val ex = intercept[RuntimeException] { + Tuple.builder(Schema().add(stringAttribute)).add(stringAttribute, Integer.valueOf(1)) + } + assert( + ex.getMessage == + "Attribute col-string's type (string) is different from field's type (integer)" + ) + } + + it should "accept any field class for an ANY-typed attribute" in { + val anyAttribute = new Attribute("col-any", AttributeType.ANY) + val tuple = + Tuple.builder(Schema().add(anyAttribute)).add(anyAttribute, List(1, 2, 3)).build() + assert(tuple.getField[List[Int]]("col-any") == List(1, 2, 3)) + } + + it should "support the three-argument builder add and reject null arguments" in { + val schema = Schema().add(stringAttribute) + val tuple = + Tuple.builder(schema).add("col-string", AttributeType.STRING, "v").build() + assert(tuple.getField[String]("col-string") == "v") + intercept[IllegalArgumentException] { + Tuple.builder(schema).add(null.asInstanceOf[String], null, "v") + } + intercept[IllegalArgumentException] { + Tuple.builder(schema).add(null.asInstanceOf[Attribute], "v") + } + intercept[IllegalArgumentException] { + Tuple.builder(schema).add(null.asInstanceOf[Tuple]) + } + intercept[IllegalArgumentException] { + Tuple.builder(schema).addSequentially(null) + } + } + + it should "expose the case-class members of Tuple" in { + val schema = Schema().add(stringAttribute) + val tuple = Tuple(schema, Array[Any]("hello")) + val copied = tuple.copy() + assert(copied == tuple) + assert(tuple.canEqual(copied)) + assert(tuple.productArity == 2) + assert(tuple.productElement(0) == schema) + val Tuple(unappliedSchema, _) = tuple + assert(unappliedSchema == schema) + } } diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/util/ArrowUtilsSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/util/ArrowUtilsSpec.scala index 4321008d46b..c91f09540ff 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/amber/util/ArrowUtilsSpec.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/util/ArrowUtilsSpec.scala @@ -20,15 +20,17 @@ package org.apache.texera.amber.util import org.apache.arrow.memory.RootAllocator -import org.apache.arrow.vector.VectorSchemaRoot +import org.apache.arrow.vector.{VarCharVector, VectorSchemaRoot} import org.apache.arrow.vector.types.{FloatingPointPrecision, TimeUnit} import org.apache.arrow.vector.types.pojo.{ArrowType, Field, FieldType} import org.apache.texera.amber.core.state.State -import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema} +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, LargeBinary, Schema, Tuple} import org.apache.texera.amber.core.tuple.AttributeTypeUtils.AttributeTypeException import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers +import java.nio.charset.StandardCharsets +import java.sql.Timestamp import java.util import scala.jdk.CollectionConverters.CollectionHasAsScala @@ -286,6 +288,112 @@ class ArrowUtilsSpec extends AnyFlatSpec with Matchers { Option(name.getMetadata).map(_.containsKey("texera_type")).getOrElse(false) shouldBe false } + // ----- setTexeraTuple / appendTexeraTuple / getTexeraTuple ----- + + private val tupleSchema = Schema( + List( + new Attribute("i", AttributeType.INTEGER), + new Attribute("l", AttributeType.LONG), + new Attribute("d", AttributeType.DOUBLE), + new Attribute("b", AttributeType.BOOLEAN), + new Attribute("t", AttributeType.TIMESTAMP), + new Attribute("s", AttributeType.STRING), + new Attribute("y", AttributeType.BINARY) + ) + ) + + private def withRoot(schema: Schema)(body: VectorSchemaRoot => Unit): Unit = { + val allocator = new RootAllocator() + val root = VectorSchemaRoot.create(ArrowUtils.fromTexeraSchema(schema), allocator) + try { + root.allocateNew() + body(root) + } finally { + root.close() + allocator.close() + } + } + + "appendTexeraTuple and getTexeraTuple" should "round-trip a tuple of every field type" in { + val tuple = Tuple + .builder(tupleSchema) + .addSequentially( + Array[Any]( + Int.box(42), + Long.box(7L), + Double.box(3.14), + Boolean.box(true), + new Timestamp(10000L), + "hello", + Array[Byte](1, 2, 3) + ) + ) + .build() + withRoot(tupleSchema) { root => + ArrowUtils.appendTexeraTuple(tuple, root) + root.getRowCount shouldBe 1 + val back = ArrowUtils.getTexeraTuple(0, root) + back.getField[Integer]("i") shouldBe 42 + back.getField[java.lang.Long]("l") shouldBe 7L + back.getField[java.lang.Double]("d") shouldBe 3.14 + back.getField[java.lang.Boolean]("b") shouldBe true + back.getField[Timestamp]("t") shouldBe new Timestamp(10000L) + back.getField[String]("s") shouldBe "hello" + back.getField[Array[Byte]]("y") shouldBe Array[Byte](1, 2, 3) + } + } + + it should "round-trip null values in every field type" in { + val nullTuple = Tuple + .builder(tupleSchema) + .addSequentially(Array[Any](null, null, null, null, null, null, null)) + .build() + withRoot(tupleSchema) { root => + ArrowUtils.appendTexeraTuple(nullTuple, root) + val back = ArrowUtils.getTexeraTuple(0, root) + tupleSchema.getAttributes.foreach { attribute => + back.getField[AnyRef](attribute.getName) shouldBe null + } + } + } + + it should "append consecutive tuples at increasing row indices" in { + val schema = Schema(List(new Attribute("s", AttributeType.STRING))) + val first = Tuple.builder(schema).addSequentially(Array[Any]("first")).build() + val second = Tuple.builder(schema).addSequentially(Array[Any]("second")).build() + withRoot(schema) { root => + ArrowUtils.appendTexeraTuple(first, root) + ArrowUtils.appendTexeraTuple(second, root) + root.getRowCount shouldBe 2 + ArrowUtils.getTexeraTuple(0, root).getField[String]("s") shouldBe "first" + ArrowUtils.getTexeraTuple(1, root).getField[String]("s") shouldBe "second" + } + } + + "getTexeraTuple" should "null out fields whose values fail to parse back" in { + // A Utf8 vector tagged LARGE_BINARY holds a value that is not a valid + // LargeBinary URI; parsing fails and the field falls back to null. + val metadata = new util.HashMap[String, String]() + metadata.put("texera_type", "LARGE_BINARY") + val arrowSchema = new org.apache.arrow.vector.types.pojo.Schema( + util.Arrays.asList(arrowField("blob", ArrowType.Utf8.INSTANCE, metadata)) + ) + val allocator = new RootAllocator() + val root = VectorSchemaRoot.create(arrowSchema, allocator) + try { + root.allocateNew() + root + .getVector(0) + .asInstanceOf[VarCharVector] + .setSafe(0, "not a uri".getBytes(StandardCharsets.UTF_8)) + root.setRowCount(1) + ArrowUtils.getTexeraTuple(0, root).getField[LargeBinary]("blob") shouldBe null + } finally { + root.close() + allocator.close() + } + } + // ----- Tuple <-> Arrow data round-trip (the State wire-hop contract) ----- "tuple round-trip through Arrow vectors" should "preserve every column of a multi-column State tuple" in { @@ -327,4 +435,30 @@ class ArrowUtilsSpec extends AnyFlatSpec with Matchers { allocator.close() } } + + // ----- fromAttributeType (null input) ----- + + "fromAttributeType" should "reject a null attribute type" in { + val ex = intercept[AttributeTypeException] { + ArrowUtils.fromAttributeType(null) + } + ex.getMessage shouldBe "Unexpected value: null" + } + + // ----- toTexeraSchema (unrecognized metadata) ----- + + "toTexeraSchema" should "fall back to the Arrow type for unrecognized texera_type metadata" in { + val unknownTag = new util.HashMap[String, String]() + unknownTag.put("texera_type", "SOMETHING_ELSE") + val otherKey = new util.HashMap[String, String]() + otherKey.put("other_key", "x") + val arrow = new org.apache.arrow.vector.types.pojo.Schema( + util.Arrays.asList( + arrowField("weird", ArrowType.Utf8.INSTANCE, unknownTag), + arrowField("plain", ArrowType.Utf8.INSTANCE, otherKey) + ) + ) + val attributes = ArrowUtils.toTexeraSchema(arrow).getAttributes + attributes.map(_.getType) shouldBe List(AttributeType.STRING, AttributeType.STRING) + } }