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 00000000000..24b1e081589
--- /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 36471b197ce..9a9a619a899 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 00000000000..eb9af987b8b
--- /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 aa882a23966..73ff029dd60 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 3a8240c3781..a105d9d5922 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")
+  }
 }