This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7025-348fa884442990307b842fd7c0dd6ca005dd320c in repository https://gitbox.apache.org/repos/asf/texera.git
commit 2ff1e57d03df73b20139c299033398b5a8a59f72 Author: Xinyuan Lin <[email protected]> AuthorDate: Fri Jul 31 16:35:09 2026 -0700 test(workflow-core): extend small-class coverage and add LargeBinary/OnIceberg specs (#7025) ### What changes were proposed in this PR? Adds 42 tests across five `common/workflow-core` classes (49 -> 91 in these suites). No source changes; the three extended specs are **pure insertions**. - **`PartitionInfo`**: `satisfies`/`merge` sensitivity to hash-attribute contents *and order*, `RangePartition`'s always-Unknown merge override and the companion's collapse to `UnknownPartition` on an empty attribute list, plus the polymorphic-JSON contract (each subtype tagged with its registered type id, base-type round-trip preserving range bounds). - **`JGitVersionControl`**: `initRepo`'s branch extraction matching what JGit reports, `getRootFileNodeOfCommit` on an empty tree and on depth-two nested directories, symbolic refs (`HEAD`) on both read paths, per-commit content after re-staging a modified file, binary-safe copies through both stream APIs, and a staged deletion reported as uncommitted until committed. - **`Tuple`**: constructor null guards, positional `getField` and its out-of-range failure, `inMemSize` growth, and the Builder's missing-attribute / duplicate-add / `addSequentially` size-mismatch paths. - **`LargeBinary`** (new spec): URI validation with exact messages (including the case-sensitive `s3://` guard), bucket/object-key parsing with the empty-path branch, value semantics, the thread-local base-URI constructor and its unset failure, and Jackson `@JsonValue`/`@JsonCreator` behavior. - **`OnIceberg`** (new spec): `expireSnapshots` collapsing three snapshots to one while keeping every live row, idempotence, a no-op on a snapshot-less table, and the `NoSuchTableException` message — driven against a **local Hadoop catalog on a temp `file:/` warehouse**, so no REST/S3 backend is needed. A few defensive branches were left uncovered rather than contrived (unreachable `return null` guards in `initRepo`, and a parent-node creation branch the tree walk can't reach); they're listed in the review notes. ### Any related issues, documentation, discussions? Closes #7023. ### How was this PR tested? `sbt -java-home <jbr-17> "WorkflowCore/testOnly *PartitionInfoSpec *JGitVersionControlSpec *TupleSpec *LargeBinarySpec *OnIcebergSpec"` -> 91 succeeded, 0 failed. `Test/scalafmtCheck` + `Test/scalafix --check` clean. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 4.8 [1M context]) --------- Signed-off-by: Xinyuan Lin <[email protected]> Co-authored-by: Copilot Autofix powered by AI <[email protected]> --- .../storage/result/iceberg/OnIcebergSpec.scala | 181 +++++++++++++++++++++ .../util/dataset/JGitVersionControlSpec.scala | 171 +++++++++++++++++++ .../texera/amber/core/tuple/LargeBinarySpec.scala | 150 +++++++++++++++++ .../apache/texera/amber/core/tuple/TupleSpec.scala | 95 +++++++++++ .../amber/core/workflow/PartitionInfoSpec.scala | 112 ++++++++++++- 5 files changed, 708 insertions(+), 1 deletion(-) diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/result/iceberg/OnIcebergSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/result/iceberg/OnIcebergSpec.scala new file mode 100644 index 0000000000..24b1e08158 --- /dev/null +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/result/iceberg/OnIcebergSpec.scala @@ -0,0 +1,181 @@ +/* + * 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.storage.result.iceberg + +import org.apache.texera.amber.core.tuple.{AttributeType, Schema, Tuple} +import org.apache.texera.amber.util.IcebergUtil +import org.apache.iceberg.catalog.Catalog +import org.apache.iceberg.data.IcebergGenerics +import org.apache.iceberg.exceptions.NoSuchTableException +import org.apache.iceberg.{Schema => IcebergSchema, Table} +import org.scalatest.BeforeAndAfterAll +import org.scalatest.flatspec.AnyFlatSpec + +import java.nio.file.{Files, Path} +import java.util.UUID +import scala.jdk.CollectionConverters._ + +/** + * Unit tests for the [[OnIceberg]] trait's snapshot-expiration behavior, exercised + * through a minimal concrete implementation and a local Hadoop-backed catalog + * (temp `file:/` warehouse), so no REST catalog or S3 endpoint is needed. + */ +class OnIcebergSpec extends AnyFlatSpec with BeforeAndAfterAll { + + /** The smallest possible OnIceberg implementation: it only supplies the three abstract members. */ + private case class TestOnIceberg( + catalog: Catalog, + tableNamespace: String, + tableName: String + ) extends OnIceberg + + private val tableNamespace = "on_iceberg_spec" + private var warehouseDir: Path = _ + private var catalog: Catalog = _ + + private val amberSchema: Schema = Schema() + .add("id", AttributeType.INTEGER) + .add("name", AttributeType.STRING) + + private val icebergSchema: IcebergSchema = IcebergUtil.toIcebergSchema(amberSchema) + + override def beforeAll(): Unit = { + warehouseDir = Files.createTempDirectory("on-iceberg-spec") + catalog = IcebergUtil.createHadoopCatalog("on-iceberg-spec", warehouseDir) + } + + override def afterAll(): Unit = { + catalog match { + case closeable: AutoCloseable => closeable.close() + case _ => + } + } + + private def freshTableName(): String = + s"table_${UUID.randomUUID().toString.replace("-", "")}" + + private def createTable(tableName: String): Unit = { + IcebergUtil.createTable( + catalog, + tableNamespace, + tableName, + icebergSchema, + overrideIfExists = true + ) + } + + private def loadTable(tableName: String): Table = + IcebergUtil.loadTableMetadata(catalog, tableNamespace, tableName).get + + /** Commits one snapshot holding the given ids. */ + private def appendSnapshot(tableName: String, ids: Seq[Int]): Unit = { + val writer = new IcebergTableWriter[Tuple]( + s"writer_${UUID.randomUUID().toString.replace("-", "")}", + catalog, + tableNamespace, + tableName, + icebergSchema, + IcebergUtil.toGenericRecord + ) + writer.open() + ids.foreach(id => + writer.putOne( + Tuple.builder(amberSchema).addSequentially(Array(Int.box(id), s"name-$id")).build() + ) + ) + writer.close() + } + + private def snapshotCount(tableName: String): Int = + loadTable(tableName).snapshots().asScala.size + + private def readIds(tableName: String): List[Int] = { + val records = IcebergGenerics.read(loadTable(tableName)).build() + try { + records + .iterator() + .asScala + .map(IcebergUtil.fromRecord(_, amberSchema).getField[Int]("id")) + .toList + } finally { + records.close() + } + } + + "OnIceberg.expireSnapshots" should "retain only the most recent snapshot" in { + val tableName = freshTableName() + createTable(tableName) + appendSnapshot(tableName, Seq(1, 2)) + appendSnapshot(tableName, Seq(3)) + appendSnapshot(tableName, Seq(4)) + assert(snapshotCount(tableName) == 3) + + TestOnIceberg(catalog, tableNamespace, tableName).expireSnapshots() + + assert(snapshotCount(tableName) == 1) + } + + it should "keep every live row of the surviving snapshot" in { + val tableName = freshTableName() + createTable(tableName) + appendSnapshot(tableName, Seq(1, 2)) + appendSnapshot(tableName, Seq(3)) + + TestOnIceberg(catalog, tableNamespace, tableName).expireSnapshots() + + // expiring snapshots drops history, not data: rows appended by the expired + // snapshots are still referenced by the retained one + assert(readIds(tableName).sorted == List(1, 2, 3)) + assert(snapshotCount(tableName) == 1) + } + + it should "be idempotent when run twice" in { + val tableName = freshTableName() + createTable(tableName) + appendSnapshot(tableName, Seq(1)) + appendSnapshot(tableName, Seq(2)) + + val onIceberg = TestOnIceberg(catalog, tableNamespace, tableName) + onIceberg.expireSnapshots() + onIceberg.expireSnapshots() + + assert(snapshotCount(tableName) == 1) + assert(readIds(tableName).sorted == List(1, 2)) + } + + it should "be a no-op for a table that has never been written to" in { + val tableName = freshTableName() + createTable(tableName) + assert(snapshotCount(tableName) == 0) + + TestOnIceberg(catalog, tableNamespace, tableName).expireSnapshots() + + assert(snapshotCount(tableName) == 0) + assert(readIds(tableName).isEmpty) + } + + it should "throw NoSuchTableException naming the missing table" in { + val missing = freshTableName() + val ex = intercept[NoSuchTableException] { + TestOnIceberg(catalog, tableNamespace, missing).expireSnapshots() + } + assert(ex.getMessage == s"table $tableNamespace.$missing doesn't exist") + } +} diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/dataset/JGitVersionControlSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/dataset/JGitVersionControlSpec.scala index 36471b197c..9a9a619a89 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/dataset/JGitVersionControlSpec.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/dataset/JGitVersionControlSpec.scala @@ -382,4 +382,175 @@ class JGitVersionControlSpec extends AnyFlatSpec with Matchers { deleteIfExists(repo) } } + + it should "return the branch name that the new repository actually has checked out" in { + val repo = Files.createTempDirectory("texera-jgit-branch") + try { + val branch = JGitVersionControl.initRepo(repo) + // initRepo strips the "refs/heads/" prefix off the HEAD symref; the result must + // be the very branch JGit reports for the fresh repository. + val actual = Using.resource(Git.open(repo.toFile))(_.getRepository.getBranch) + branch shouldBe actual + branch should not startWith "refs/heads/" + } finally { + deleteIfExists(repo) + } + } + + "JGitVersionControl.getRootFileNodeOfCommit" should "return an empty set for a commit with no files" in { + val repo = Files.createTempDirectory("texera-jgit-empty-tree") + try { + JGitVersionControl.initRepo(repo) + setIdentity(repo) + // an initial commit with nothing staged has an empty tree, so the tree walk + // yields no entries at all + val hash = JGitVersionControl.commit(repo, "empty commit") + + JGitVersionControl.getRootFileNodeOfCommit(repo, hash).asScala shouldBe empty + } finally { + deleteIfExists(repo) + } + } + + it should "link nested directories at depth two down to the leaf file" in { + val repo = Files.createTempDirectory("texera-jgit-deep-tree") + try { + JGitVersionControl.initRepo(repo) + setIdentity(repo) + + val leafContent = "deep-content" + writeFile(repo.resolve("a").resolve("b").resolve("leaf.txt"), leafContent) + stageAll(repo) + val hash = JGitVersionControl.commit(repo, "add nested dirs") + + val rootNodes = JGitVersionControl.getRootFileNodeOfCommit(repo, hash).asScala.toSeq + // only the top-level directory "a" is a root node; everything else hangs off it + rootNodes.map(_.getRelativePath.toString) shouldBe Seq("a") + + val aNode = rootNodes.head + aNode.isDirectory shouldBe true + aNode.getChildren.size shouldBe 1 + + val bNode = aNode.getChildren.asScala.head + bNode.getRelativePath.toString should (be("a/b") or be("a\\b")) + bNode.isDirectory shouldBe true + bNode.getChildren.size shouldBe 1 + + val leafNode = bNode.getChildren.asScala.head + leafNode.getRelativePath.toString should (be("a/b/leaf.txt") or be("a\\b\\leaf.txt")) + leafNode.isDirectory shouldBe false + leafNode.getSize shouldBe leafContent.getBytes(StandardCharsets.UTF_8).length.toLong + } finally { + deleteIfExists(repo) + } + } + + "JGitVersionControl commit-hash resolution" should "accept a symbolic ref such as HEAD" in { + val repo = Files.createTempDirectory("texera-jgit-symref") + try { + JGitVersionControl.initRepo(repo) + setIdentity(repo) + val file = repo.resolve("data.txt") + writeFile(file, "head-content") + JGitVersionControl.add(repo, file) + JGitVersionControl.commit(repo, "add") + + // the commitHash argument is resolved by JGit, so any revision expression works + val out = new ByteArrayOutputStream() + JGitVersionControl.readFileContentOfCommitAsOutputStream(repo, "HEAD", file, out) + new String(out.toByteArray, StandardCharsets.UTF_8) shouldBe "head-content" + + Using.resource(JGitVersionControl.readFileContentOfCommitAsInputStream(repo, "HEAD", file)) { + in => new String(in.readAllBytes(), StandardCharsets.UTF_8) shouldBe "head-content" + } + } finally { + deleteIfExists(repo) + } + } + + it should "read the version stored in each commit after a tracked file is modified" in { + val repo = Files.createTempDirectory("texera-jgit-history") + try { + JGitVersionControl.initRepo(repo) + setIdentity(repo) + + val file = repo.resolve("data.txt") + writeFile(file, "v1") + JGitVersionControl.add(repo, file) + val firstHash = JGitVersionControl.commit(repo, "v1") + + // re-staging a tracked file picks up the modification + writeFile(file, "v2-longer") + JGitVersionControl.add(repo, file) + val secondHash = JGitVersionControl.commit(repo, "v2") + + secondHash should not be firstHash + + def contentAt(hash: String): String = { + val out = new ByteArrayOutputStream() + JGitVersionControl.readFileContentOfCommitAsOutputStream(repo, hash, file, out) + new String(out.toByteArray, StandardCharsets.UTF_8) + } + contentAt(firstHash) shouldBe "v1" + contentAt(secondHash) shouldBe "v2-longer" + + // the node size reported for the latest commit follows the new content + val node = JGitVersionControl + .getRootFileNodeOfCommit(repo, secondHash) + .asScala + .find(_.getRelativePath.toString == "data.txt") + .get + node.getSize shouldBe "v2-longer".length.toLong + } finally { + deleteIfExists(repo) + } + } + + "JGitVersionControl.readFileContentOfCommitAsOutputStream" should "copy binary content byte-for-byte" in { + val repo = Files.createTempDirectory("texera-jgit-binary") + try { + JGitVersionControl.initRepo(repo) + setIdentity(repo) + + val bytes = Array[Byte](0, 1, 127, -128, -1, 13, 10, 0) + val file = repo.resolve("blob.bin") + Files.write(file, bytes) + JGitVersionControl.add(repo, file) + val hash = JGitVersionControl.commit(repo, "add binary") + + val out = new ByteArrayOutputStream() + JGitVersionControl.readFileContentOfCommitAsOutputStream(repo, hash, file, out) + out.toByteArray shouldBe bytes + + Using.resource(JGitVersionControl.readFileContentOfCommitAsInputStream(repo, hash, file)) { + in => in.readAllBytes() shouldBe bytes + } + } finally { + deleteIfExists(repo) + } + } + + "JGitVersionControl.hasUncommittedChanges" should "report a staged deletion as uncommitted" in { + val repo = Files.createTempDirectory("texera-jgit-staged-rm") + try { + JGitVersionControl.initRepo(repo) + setIdentity(repo) + + val file = repo.resolve("data.txt") + writeFile(file, "content") + JGitVersionControl.add(repo, file) + JGitVersionControl.commit(repo, "add") + JGitVersionControl.hasUncommittedChanges(repo) shouldBe false + + // rm only stages the removal; the index now differs from HEAD + JGitVersionControl.rm(repo, file) + JGitVersionControl.hasUncommittedChanges(repo) shouldBe true + Files.exists(file) shouldBe false + + JGitVersionControl.commit(repo, "remove") + JGitVersionControl.hasUncommittedChanges(repo) shouldBe false + } finally { + deleteIfExists(repo) + } + } } diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/LargeBinarySpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/LargeBinarySpec.scala new file mode 100644 index 0000000000..eb9af987b8 --- /dev/null +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/tuple/LargeBinarySpec.scala @@ -0,0 +1,150 @@ +/* + * 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.util.JSONUtils.objectMapper +import org.apache.texera.service.util.LargeBinaryManager +import org.scalatest.flatspec.AnyFlatSpec + +/** + * Unit tests for [[LargeBinary]], the S3-URI reference stored in tuples. + * + * No S3 endpoint is involved: the URI-string constructor is pure, and the no-arg + * constructor only needs a base URI on the calling thread (see [[LargeBinaryManager]]), + * which is set on a dedicated thread so the thread-local does not leak between tests. + */ +class LargeBinarySpec extends AnyFlatSpec { + + /** Runs `body` on a fresh thread, optionally seeding that thread's base URI. */ + private def onFreshThread[T](baseUri: Option[String])(body: => T): Either[Throwable, T] = { + @volatile var result: Either[Throwable, T] = Left(new IllegalStateException("did not run")) + val thread = new Thread(() => { + result = + try { + baseUri.foreach(LargeBinaryManager.setCurrentBaseUri) + Right(body) + } catch { case e: Throwable => Left(e) } + }) + thread.start() + thread.join() + result + } + + "LargeBinary" should "accept an s3 URI and expose it verbatim" in { + val binary = new LargeBinary("s3://my-bucket/objects/42/abc") + assert(binary.getUri == "s3://my-bucket/objects/42/abc") + assert(binary.toString == "s3://my-bucket/objects/42/abc") + } + + it should "reject a null URI" in { + val ex = intercept[IllegalArgumentException] { + new LargeBinary(null) + } + assert(ex.getMessage == "LargeBinary URI cannot be null") + } + + it should "reject any URI that does not start with the s3 scheme" in { + // the prefix check is a literal, case-sensitive "s3://" match + List("http://bucket/key";, "s3:/bucket/key", "S3://bucket/key", "/local/path", "", "bucket/key") + .foreach { bad => + val ex = intercept[IllegalArgumentException] { + new LargeBinary(bad) + } + assert( + ex.getMessage == s"LargeBinary URI must start with 's3://', got: $bad", + s"unexpected message for '$bad'" + ) + } + } + + it should "split an s3 URI into bucket name and object key" in { + val binary = new LargeBinary("s3://texera-large-binaries/objects/7/some-uuid.bin") + assert(binary.getBucketName == "texera-large-binaries") + // the leading slash of the URI path is stripped from the object key + assert(binary.getObjectKey == "objects/7/some-uuid.bin") + } + + it should "return an empty object key when the URI carries no path" in { + val binary = new LargeBinary("s3://bucket-only") + assert(binary.getBucketName == "bucket-only") + assert(binary.getObjectKey == "") + } + + it should "keep a single-segment object key" in { + val binary = new LargeBinary("s3://bucket/key") + assert(binary.getBucketName == "bucket") + assert(binary.getObjectKey == "key") + } + + it should "compare by URI value" in { + val binary = new LargeBinary("s3://bucket/a") + val same = new LargeBinary("s3://bucket/a") + val other = new LargeBinary("s3://bucket/b") + + assert(binary == binary) // identity short-circuit + assert(binary == same) + assert(binary.hashCode() == same.hashCode()) + assert(binary != other) + assert(!binary.equals("s3://bucket/a")) // a bare String is not a LargeBinary + assert(!binary.equals(null)) + } + + "LargeBinary()" should "derive a unique URI from the current thread's base URI" in { + val base = "s3://unit-test-bucket/objects/999/" + val uris = onFreshThread(Some(base)) { + val first = new LargeBinary() + val second = new LargeBinary() + (first, second) + }.fold(e => throw e, identity) + + val (first, second) = uris + assert(first.getUri.startsWith(base)) + assert(first.getUri.stripPrefix(base).nonEmpty) + assert(first.getBucketName == "unit-test-bucket") + assert(first.getObjectKey.startsWith("objects/999/")) + // each instance gets its own generated suffix + assert(first != second) + } + + it should "propagate the failure when the thread has no base URI" in { + // a fresh thread starts with no base URI, so the generating constructor fails fast + val outcome = onFreshThread(None)(new LargeBinary()) + assert(outcome.isLeft) + assert(outcome.swap.toOption.exists(_.isInstanceOf[IllegalStateException])) + } + + "LargeBinary JSON" should "serialize to the bare URI string via @JsonValue" in { + val binary = new LargeBinary("s3://bucket/objects/1/blob") + assert(objectMapper.writeValueAsString(binary) == "\"s3://bucket/objects/1/blob\"") + } + + it should "deserialize from the uri property declared by its @JsonCreator" in { + val restored = + objectMapper.readValue("{\"uri\":\"s3://bucket/objects/1/blob\"}", classOf[LargeBinary]) + assert(restored == new LargeBinary("s3://bucket/objects/1/blob")) + } + + it should "reject a payload whose uri is not an s3 URI" in { + // the constructor guard runs during deserialization, so bad data fails loudly + intercept[Exception] { + objectMapper.readValue("{\"uri\":\"http://bucket/key\"}";, classOf[LargeBinary]) + } + } +} 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 aa882a2396..73ff029dd6 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 @@ -313,4 +313,99 @@ class TupleSpec extends AnyFlatSpec { val Tuple(unappliedSchema, _) = tuple assert(unappliedSchema == schema) } + + it should "reject construction with a null schema or a null field array" in { + // both arguments are null-checked before any schema/field matching happens + intercept[NullPointerException] { + Tuple(null, Array[Any]("a")) + } + intercept[NullPointerException] { + Tuple(Schema().add(stringAttribute), null) + } + } + + it should "read fields by positional index and fail outside the field array" in { + val schema = Schema().add(stringAttribute).add(integerAttribute) + val tuple = Tuple(schema, Array[Any]("s", Integer.valueOf(7))) + assert(tuple.getField[String](0) == "s") + assert(tuple.getField[Int](1) == 7) + assert(tuple.length == 2) + intercept[ArrayIndexOutOfBoundsException] { + tuple.getField[String](2) + } + } + + it should "compute an in-memory size that grows with the payload" in { + val schema = Schema().add(stringAttribute) + val small = Tuple(schema, Array[Any]("a")) + val large = Tuple(schema, Array[Any]("a" * 10000)) + assert(small.inMemSize > 0) + assert(large.inMemSize > small.inMemSize) + } + + it should "name the missing attributes when the builder is incomplete" in { + val schema = Schema().add(stringAttribute).add(integerAttribute).add(boolAttribute) + val ex = intercept[TupleBuildingException] { + Tuple.builder(schema).add(stringAttribute, "s").build() + } + assert(ex.getMessage.startsWith("Tuple does not have the same number of attributes as schema.")) + assert(ex.getMessage.contains("col-int")) + assert(ex.getMessage.contains("col-bool")) + assert(!ex.getMessage.contains("col-string")) + } + + it should "let a later add overwrite the value of the same attribute" in { + val schema = Schema().add(stringAttribute) + val tuple = Tuple + .builder(schema) + .add(stringAttribute, "first") + .add(stringAttribute, "second") + .build() + assert(tuple.length == 1) + assert(tuple.getField[String]("col-string") == "second") + } + + it should "reject addSequentially when the field count differs from the schema" in { + val schema = Schema().add(stringAttribute).add(integerAttribute) + val tooFew = intercept[RuntimeException] { + Tuple.builder(schema).addSequentially(Array[Any]("only-one")) + } + assert(tooFew.getMessage == "Schema size (2) and field size (1) are different") + val tooMany = intercept[RuntimeException] { + Tuple.builder(schema).addSequentially(Array[Any]("a", Integer.valueOf(1), "extra")) + } + assert(tooMany.getMessage == "Schema size (2) and field size (3) are different") + } + + it should "still require every schema attribute when adding a tuple non-strictly" in { + // non-strict mode only skips *extra* source attributes; it does not fill gaps + val sourceSchema = Schema().add(stringAttribute) + val source = Tuple.builder(sourceSchema).add(stringAttribute, "s").build() + val targetSchema = Schema().add(stringAttribute).add(integerAttribute) + val ex = intercept[TupleBuildingException] { + Tuple.builder(targetSchema).add(source, false).build() + } + assert(ex.getMessage.contains("col-int")) + } + + it should "distinguish binary fields whose contents differ" in { + val schema = Schema().add(binaryAttribute) + def withBytes(bytes: Array[Byte]): Tuple = + Tuple.builder(schema).add(binaryAttribute, bytes).build() + + val base = withBytes(Array[Byte](1, 2, 3)) + assert(base == withBytes(Array[Byte](1, 2, 3))) + assert(base.hashCode() == withBytes(Array[Byte](1, 2, 3)).hashCode()) + assert(base != withBytes(Array[Byte](1, 2, 4))) + assert(base != withBytes(Array[Byte](1, 2))) + assert(base != withBytes(Array[Byte]())) + } + + it should "reject a partial tuple for an attribute outside the schema" in { + val schema = Schema().add(stringAttribute) + val tuple = Tuple.builder(schema).add(stringAttribute, "s").build() + intercept[RuntimeException] { + tuple.getPartialTuple(List("col-missing")) + } + } } diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/workflow/PartitionInfoSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/workflow/PartitionInfoSpec.scala index 3a8240c378..a105d9d592 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/workflow/PartitionInfoSpec.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/workflow/PartitionInfoSpec.scala @@ -19,7 +19,8 @@ package org.apache.texera.amber.core.workflow -import com.fasterxml.jackson.annotation.JsonSubTypes +import com.fasterxml.jackson.annotation.{JsonSubTypes, JsonTypeInfo} +import org.apache.texera.amber.util.JSONUtils.objectMapper import org.scalatest.flatspec.AnyFlatSpec class PartitionInfoSpec extends AnyFlatSpec { @@ -196,4 +197,113 @@ class PartitionInfoSpec extends AnyFlatSpec { assert(SinglePartition() == SinglePartition()) assert(UnknownPartition() == UnknownPartition()) } + + // --------------------------------------------------------------------------- + // Attribute-list sensitivity of the HashPartition algebra + // --------------------------------------------------------------------------- + + "HashPartition" should "treat the empty attribute list as its own partitioning" in { + val hashAll: PartitionInfo = HashPartition() + // hash-on-all-attributes is not interchangeable with hash-on-"a" + assert(!hashAll.satisfies(hashA)) + assert(!hashA.satisfies(hashAll)) + assert(hashAll.merge(hashA) == unknown) + assert(hashA.merge(hashAll) == unknown) + // but it still merges with itself to itself, and satisfies Unknown + assert(hashAll.merge(HashPartition()) == hashAll) + assert(hashAll.satisfies(unknown)) + } + + it should "be order-sensitive in its hash attribute list" in { + val ab: PartitionInfo = HashPartition(List("a", "b")) + val ba: PartitionInfo = HashPartition(List("b", "a")) + assert(ab != ba) + assert(!ab.satisfies(ba)) + assert(ab.merge(ba) == unknown) + assert(ab.merge(HashPartition(List("a", "b"))) == ab) + } + + // --------------------------------------------------------------------------- + // Range bounds participate in identity, but never in merge + // --------------------------------------------------------------------------- + + "RangePartition" should "distinguish two ranges that differ only in their bounds" in { + val narrow: PartitionInfo = new RangePartition(List("a"), 0L, 10L) + val wide: PartitionInfo = new RangePartition(List("a"), 0L, 20L) + assert(narrow != wide) + assert(!narrow.satisfies(wide)) + // identical bounds are the same partition, so satisfies holds ... + assert(narrow.satisfies(new RangePartition(List("a"), 0L, 10L))) + // ... yet merging still forfeits the sort order (override always returns Unknown) + assert(narrow.merge(new RangePartition(List("a"), 0L, 10L)) == unknown) + assert(narrow.merge(wide) == unknown) + } + + it should "keep whatever bounds it is given, including inverted and negative ones" in { + // the factory only inspects the attribute list; the bounds are stored verbatim + val inverted = RangePartition(List("a"), 10L, -5L).asInstanceOf[RangePartition] + assert(inverted.rangeMin == 10L) + assert(inverted.rangeMax == -5L) + val extremes = RangePartition(List("a", "b"), Long.MinValue, Long.MaxValue) + .asInstanceOf[RangePartition] + assert(extremes.rangeAttributeNames == List("a", "b")) + assert(extremes.rangeMin == Long.MinValue) + assert(extremes.rangeMax == Long.MaxValue) + } + + it should "collapse to UnknownPartition on an empty attribute list regardless of bounds" in { + // the empty-attribute short-circuit ignores the bounds entirely + assert(RangePartition(List.empty, Long.MinValue, Long.MaxValue) == UnknownPartition()) + assert(RangePartition(List.empty, 5L, 5L) == UnknownPartition()) + } + + // --------------------------------------------------------------------------- + // Polymorphic JSON serialization (the @JsonTypeInfo / @JsonSubTypes contract) + // --------------------------------------------------------------------------- + + private val jsonCases: List[(String, PartitionInfo)] = List( + "hash" -> HashPartition(List("a", "b")), + "hash" -> HashPartition(), + "range" -> new RangePartition(List("a"), -1L, 42L), + "single" -> SinglePartition(), + "oneToOne" -> OneToOnePartition(), + "broadcast" -> BroadcastPartition(), + "none" -> UnknownPartition() + ) + + "PartitionInfo JSON" should "tag every subtype with its registered type name" in { + jsonCases.foreach { + case (typeName, partition) => + val node = objectMapper.readTree(objectMapper.writeValueAsString(partition)) + assert(node.has("type"), s"$partition must carry a type discriminator") + assert(node.get("type").asText() == typeName, s"$partition must be tagged '$typeName'") + } + } + + it should "round-trip every subtype back to an equal value through the base type" in { + jsonCases.foreach { + case (_, partition) => + val restored = + objectMapper.readValue(objectMapper.writeValueAsString(partition), classOf[PartitionInfo]) + assert(restored == partition, s"$partition must survive a JSON round-trip") + assert(restored.getClass == partition.getClass) + } + } + + it should "preserve the range bounds and attribute names of a RangePartition" in { + val original = new RangePartition(List("a", "b"), -7L, 99L) + val restored = objectMapper + .readValue(objectMapper.writeValueAsString(original), classOf[PartitionInfo]) + .asInstanceOf[RangePartition] + assert(restored.rangeAttributeNames == List("a", "b")) + assert(restored.rangeMin == -7L) + assert(restored.rangeMax == 99L) + } + + "PartitionInfo @JsonTypeInfo" should "use a NAME discriminator on a 'type' property" in { + val annotation = classOf[PartitionInfo].getAnnotation(classOf[JsonTypeInfo]) + assert(annotation.use() == JsonTypeInfo.Id.NAME) + assert(annotation.include() == JsonTypeInfo.As.PROPERTY) + assert(annotation.property() == "type") + } }
