test(workflow-core): extend small-class coverage and add LargeBinary/OnIceberg specs

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")
+  }
+}

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)
+    }
+  }
 }