This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new 9ca9c57eb8 test(workflow-core): add storage unit test coverage
(DocumentFactory, DatasetFileDocument, IcebergDocument) (#6823)
9ca9c57eb8 is described below
commit 9ca9c57eb8b7bf4b11c23269102ca737ce6f0181
Author: Xinyuan Lin <[email protected]>
AuthorDate: Thu Jul 23 22:52:01 2026 -0700
test(workflow-core): add storage unit test coverage (DocumentFactory,
DatasetFileDocument, IcebergDocument) (#6823)
### What changes were proposed in this PR?
Adds unit coverage for three workflow-core storage classes (all verified
with the JBR-17 sbt toolchain; no source changes).
- `DocumentFactorySpec` (extended): the full scheme-dispatch surface of
`openReadonlyDocument`/`openDocument`/`createDocument`/`documentExists`
— dataset/file/vfs arms, the unsupported-scheme guards (each asserting
the operation-specific message), and the vfs create/open/exists happy
paths across the RESULT/STATE/CONSOLE_MESSAGES/RUNTIME_STATISTICS
namespaces, against a real local Iceberg catalog.
- `DatasetFileDocumentSpec` (extended): the companion-object
env-fallback lazy vals (presign-URL endpoint default, empty JWT
default). The LakeFS/network-backed read paths are left uncovered as
they need a live LakeFS.
- `IcebergDocumentSpec` (new, unit-level — not `@IntegrationTest`):
`getURI`, `getCount`, `get`/`getRange`/`getAfter` (with column
projection + multi-file ordering + iterator exhaustion),
`getTableStatistics`, `getTotalFileSize`, `asInputStream` (ZIP with
parquet entries), and `clear` — driven against a real local Hadoop
Iceberg catalog via a new `LocalHadoopIcebergCatalog` test helper.
### Any related issues, documentation, discussions?
Closes #6821.
### How was this PR tested?
`sbt -java-home <jbr-17> "WorkflowCore/testOnly *DocumentFactorySpec
*DatasetFileDocumentSpec *IcebergDocumentSpec"` -> 49 succeeded, 0
failed. scalafmtCheckAll + scalafixAll --check clean.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8 [1M context])
---
.../amber/core/storage/DocumentFactorySpec.scala | 199 ++++++++++++-
.../core/storage/LocalHadoopIcebergCatalog.scala | 66 ++++
.../storage/model/DatasetFileDocumentSpec.scala | 24 ++
.../result/iceberg/IcebergDocumentSpec.scala | 331 +++++++++++++++++++++
4 files changed, 611 insertions(+), 9 deletions(-)
diff --git
a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/DocumentFactorySpec.scala
b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/DocumentFactorySpec.scala
index d67b90f363..ad8b5580c5 100644
---
a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/DocumentFactorySpec.scala
+++
b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/DocumentFactorySpec.scala
@@ -19,27 +19,55 @@
package org.apache.texera.amber.core.storage
-import org.apache.texera.amber.core.storage.model.VirtualDocument
-import org.apache.texera.amber.core.tuple.{Schema, Tuple}
+import org.apache.texera.amber.core.storage.model.{
+ DatasetFileDocument,
+ ReadonlyLocalFileDocument,
+ VirtualDocument
+}
+import org.apache.texera.amber.core.storage.result.iceberg.IcebergDocument
+import org.apache.texera.amber.core.tuple.{AttributeType, Schema, Tuple}
+import org.apache.texera.amber.core.virtualidentity.{
+ ExecutionIdentity,
+ OperatorIdentity,
+ PhysicalOpIdentity,
+ WorkflowIdentity
+}
+import org.apache.texera.amber.core.workflow.{GlobalPortIdentity, PortIdentity}
+import org.scalatest.BeforeAndAfterAll
import org.scalatest.flatspec.AnyFlatSpec
import org.scalatest.matchers.should.Matchers
import java.net.URI
+import java.nio.file.Files
+import java.util.UUID
/**
- * Unit tests for `DocumentFactory.createOrReuseDocument`, the create-or-reuse
- * decision behind output-port storage provisioning. It always returns the
- * document (opened when reused, created otherwise) so the call site doesn't
- * branch.
+ * Unit tests for `DocumentFactory`.
*
- * `exists` / `open` / `create` are injected so the decision can be pinned
with
- * trivial document stubs -- no iceberg backend, no live region.
+ * The first group pins `createOrReuseDocument`, the create-or-reuse decision
+ * behind output-port storage provisioning, using injected
`exists`/`open`/`create`
+ * spies -- no iceberg backend, no live region.
+ *
+ * The remaining groups exercise the scheme-dispatch surface of
+ * `openReadonlyDocument` / `openDocument` / `createDocument` /
`documentExists`:
+ * the dataset / file / vfs scheme arms, the unsupported-scheme guards, and
the
+ * vfs create/open/exists happy paths against a local Hadoop-backed Iceberg
+ * catalog installed via [[LocalHadoopIcebergCatalog]].
*/
-class DocumentFactorySpec extends AnyFlatSpec with Matchers {
+class DocumentFactorySpec extends AnyFlatSpec with Matchers with
BeforeAndAfterAll {
private val uri = new URI("vfs:///wf/result/loop-end")
private val schema = Schema()
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ LocalHadoopIcebergCatalog.ensure()
+ }
+
+ //
---------------------------------------------------------------------------
+ // createOrReuseDocument
+ //
---------------------------------------------------------------------------
+
private def stubDoc: VirtualDocument[Tuple] =
new VirtualDocument[Tuple] {
override def getURI: URI = uri
@@ -93,4 +121,157 @@ class DocumentFactorySpec extends AnyFlatSpec with
Matchers {
)
probed shouldBe false
}
+
+ //
---------------------------------------------------------------------------
+ // openReadonlyDocument -- scheme dispatch
+ //
---------------------------------------------------------------------------
+
+ private val versionHash = "97fd4c2a755b69b7c66d322eab40b7e5c2ad5d10"
+
+ "openReadonlyDocument" should "return a DatasetFileDocument for the dataset
scheme" in {
+ val datasetUri = new URI(s"dataset:///repo/$versionHash/file.txt")
+ val doc = DocumentFactory.openReadonlyDocument(datasetUri)
+ doc shouldBe a[DatasetFileDocument]
+ doc.getURI shouldBe datasetUri
+ }
+
+ it should "return a ReadonlyLocalFileDocument for the file scheme" in {
+ val tempFile = Files.createTempFile("doc-factory-spec", ".txt")
+ try {
+ val doc = DocumentFactory.openReadonlyDocument(tempFile.toUri)
+ doc shouldBe a[ReadonlyLocalFileDocument]
+ doc.getURI shouldBe tempFile.toUri
+ } finally {
+ Files.deleteIfExists(tempFile)
+ }
+ }
+
+ it should "reject an unsupported scheme with a descriptive message" in {
+ val thrown = intercept[UnsupportedOperationException] {
+ DocumentFactory.openReadonlyDocument(new URI("ftp://host/path";))
+ }
+ thrown.getMessage should include("ftp")
+ thrown.getMessage should include("ReadonlyDocument")
+ }
+
+ //
---------------------------------------------------------------------------
+ // openDocument / createDocument / documentExists -- unsupported schemes
+ //
---------------------------------------------------------------------------
+
+ "openDocument" should "return a DatasetFileDocument and no schema for the
dataset scheme" in {
+ val datasetUri = new URI(s"dataset:///repo/$versionHash/file.txt")
+ val (doc, schemaOpt) = DocumentFactory.openDocument(datasetUri)
+ doc shouldBe a[DatasetFileDocument]
+ schemaOpt shouldBe None
+ }
+
+ it should "reject an unsupported scheme" in {
+ val thrown = intercept[UnsupportedOperationException] {
+ DocumentFactory.openDocument(new URI("ftp://host/path";))
+ }
+ thrown.getMessage should include("opening the document")
+ }
+
+ "createDocument" should "reject an unsupported scheme" in {
+ val thrown = intercept[UnsupportedOperationException] {
+ DocumentFactory.createDocument(
+ new URI("file:///tmp/x"),
+ Schema().add("v", AttributeType.INTEGER)
+ )
+ }
+ thrown.getMessage should include("creating the document")
+ }
+
+ "documentExists" should "reject an unsupported scheme" in {
+ val thrown = intercept[UnsupportedOperationException] {
+ DocumentFactory.documentExists(new URI("file:///tmp/x"))
+ }
+ thrown.getMessage should include("checking document existence")
+ }
+
+ //
---------------------------------------------------------------------------
+ // vfs scheme -- create / open / exists against a local Iceberg catalog
+ //
---------------------------------------------------------------------------
+
+ private val vfsSchema: Schema = Schema().add("v", AttributeType.INTEGER)
+
+ private def freshResultURI(): URI = {
+ val base = VFSURIFactory.createPortBaseURI(
+ WorkflowIdentity(0),
+ ExecutionIdentity(0),
+ GlobalPortIdentity(
+ PhysicalOpIdentity(
+ logicalOpId =
OperatorIdentity(s"op-${UUID.randomUUID().toString.replace("-", "")}"),
+ layerName = "main"
+ ),
+ PortIdentity()
+ )
+ )
+ VFSURIFactory.resultURI(base)
+ }
+
+ "createDocument" should "create an IcebergDocument for a vfs result URI" in {
+ val vfsUri = freshResultURI()
+ val doc = DocumentFactory.createDocument(vfsUri, vfsSchema)
+ doc shouldBe an[IcebergDocument[_]]
+ DocumentFactory.documentExists(vfsUri) shouldBe true
+ }
+
+ it should "create a state document for a vfs state URI (STATE namespace
arm)" in {
+ val base = VFSURIFactory.createPortBaseURI(
+ WorkflowIdentity(0),
+ ExecutionIdentity(0),
+ GlobalPortIdentity(
+ PhysicalOpIdentity(
+ logicalOpId =
OperatorIdentity(s"op-${UUID.randomUUID().toString.replace("-", "")}"),
+ layerName = "main"
+ ),
+ PortIdentity()
+ )
+ )
+ val stateUri = VFSURIFactory.stateURI(base)
+ val doc = DocumentFactory.createDocument(stateUri, vfsSchema)
+ doc shouldBe an[IcebergDocument[_]]
+ DocumentFactory.documentExists(stateUri) shouldBe true
+ }
+
+ "documentExists" should "report false before creation and true after for a
vfs URI" in {
+ val vfsUri = freshResultURI()
+ DocumentFactory.documentExists(vfsUri) shouldBe false
+ DocumentFactory.createDocument(vfsUri, vfsSchema)
+ DocumentFactory.documentExists(vfsUri) shouldBe true
+ }
+
+ it should "resolve the CONSOLE_MESSAGES namespace" in {
+ val consoleUri = VFSURIFactory.createConsoleMessagesURI(
+ WorkflowIdentity(0),
+ ExecutionIdentity(0),
+ OperatorIdentity(s"op-${UUID.randomUUID().toString.replace("-", "")}")
+ )
+ DocumentFactory.documentExists(consoleUri) shouldBe false
+ }
+
+ it should "resolve the RUNTIME_STATISTICS namespace" in {
+ val statsUri = VFSURIFactory.createRuntimeStatisticsURI(
+ WorkflowIdentity(0),
+ ExecutionIdentity(0)
+ )
+ DocumentFactory.documentExists(statsUri) shouldBe false
+ }
+
+ "openDocument" should "open a created vfs document and return its schema" in
{
+ val vfsUri = freshResultURI()
+ DocumentFactory.createDocument(vfsUri, vfsSchema)
+ val (doc, schemaOpt) = DocumentFactory.openDocument(vfsUri)
+ doc shouldBe an[IcebergDocument[_]]
+ schemaOpt shouldBe defined
+ schemaOpt.get.getAttributes.map(_.getName) should contain("v")
+ }
+
+ it should "throw IllegalArgumentException when opening a vfs URI with no
backing storage" in {
+ val thrown = intercept[IllegalArgumentException] {
+ DocumentFactory.openDocument(freshResultURI())
+ }
+ thrown.getMessage should include("No storage is found")
+ }
}
diff --git
a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/LocalHadoopIcebergCatalog.scala
b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/LocalHadoopIcebergCatalog.scala
new file mode 100644
index 0000000000..3b92b763a3
--- /dev/null
+++
b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/LocalHadoopIcebergCatalog.scala
@@ -0,0 +1,66 @@
+/*
+ * 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
+
+import org.apache.texera.amber.util.IcebergUtil
+
+import java.nio.file.{Files, Path}
+import java.util.Comparator
+
+/**
+ * Test-only helper that installs a local Hadoop-backed Iceberg catalog into
the
+ * shared `IcebergCatalogInstance` singleton exactly once per JVM.
+ *
+ * The default configured catalog type is `rest`, which needs a live REST
server;
+ * unit tests instead exercise Iceberg against a temp-directory `file:/`
warehouse
+ * (via the merged winutils-free local filesystem on Windows).
+ *
+ * `IcebergCatalogInstance` is a JVM-wide mutable singleton and ScalaTest runs
+ * suites in the module in parallel within the same JVM, so every suite that
needs
+ * an Iceberg backend calls [[ensure]]. Initialization is idempotent and
guarded by
+ * a single flag on this shared object, so the very first caller wins and all
+ * suites end up sharing the same catalog + warehouse instead of racing on
+ * `replaceInstance`. Suites keep their tables disjoint via distinct
namespaces and
+ * unique (UUID) table names.
+ */
+object LocalHadoopIcebergCatalog {
+
+ private var initialized = false
+
+ def ensure(): Unit =
+ synchronized {
+ if (!initialized) {
+ val warehouse = Files.createTempDirectory("wfcore-iceberg-shared")
+ // Best-effort recursive cleanup of the temp warehouse on JVM exit so
test runs
+ // don't leave wfcore-iceberg-shared* directories behind on dev
machines / CI.
+ sys.addShutdownHook {
+ try Files
+ .walk(warehouse)
+ .sorted(Comparator.reverseOrder[Path]())
+ .forEach((p: Path) => Files.deleteIfExists(p))
+ catch { case _: Throwable => () }
+ }
+ IcebergCatalogInstance.replaceInstance(
+ IcebergUtil.createHadoopCatalog("wfcore-test", warehouse)
+ )
+ initialized = true
+ }
+ }
+}
diff --git
a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/model/DatasetFileDocumentSpec.scala
b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/model/DatasetFileDocumentSpec.scala
index da3b799b63..c16d417a73 100644
---
a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/model/DatasetFileDocumentSpec.scala
+++
b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/model/DatasetFileDocumentSpec.scala
@@ -19,6 +19,7 @@
package org.apache.texera.amber.core.storage.model
+import org.apache.texera.common.config.EnvironmentalVariable
import org.scalatest.flatspec.AnyFlatSpec
import org.scalatest.matchers.should.Matchers
@@ -140,4 +141,27 @@ class DatasetFileDocumentSpec extends AnyFlatSpec with
Matchers {
thrown.getMessage shouldBe "URI format is incorrect"
}
}
+
+ // The companion object resolves the file-service endpoint and the user JWT
+ // token from environment variables, falling back to a trimmed default. These
+ // lazy vals are read whenever asInputStream needs to fetch a file; assert
their
+ // fallback behavior without requiring a live FileService or LakeFS. The
checks
+ // are guarded so they hold regardless of whether the env overrides are
present.
+ "DatasetFileDocument companion" should
+ "expose the default presigned-URL endpoint when the env override is
absent" in {
+ val expected =
+ sys.env
+ .getOrElse(
+ EnvironmentalVariable.ENV_FILE_SERVICE_GET_PRESIGNED_URL_ENDPOINT,
+ "http://localhost:9092/api/dataset/presign-download";
+ )
+ .trim
+ DatasetFileDocument.fileServiceGetPresignURLEndpoint shouldBe expected
+ }
+
+ it should "expose a trimmed user JWT token defaulting to empty" in {
+ val expected =
+ sys.env.getOrElse(EnvironmentalVariable.ENV_USER_JWT_TOKEN, "").trim
+ DatasetFileDocument.userJwtToken shouldBe expected
+ }
}
diff --git
a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergDocumentSpec.scala
b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergDocumentSpec.scala
new file mode 100644
index 0000000000..a1f4ea3023
--- /dev/null
+++
b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergDocumentSpec.scala
@@ -0,0 +1,331 @@
+/*
+ * 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.storage.IcebergCatalogInstance
+import org.apache.texera.amber.core.storage.LocalHadoopIcebergCatalog
+import org.apache.texera.amber.core.storage.model.VirtualDocument
+import org.apache.texera.amber.core.tuple.{AttributeType, Schema, Tuple}
+import org.apache.texera.amber.util.IcebergUtil
+import org.apache.iceberg.catalog.TableIdentifier
+import org.apache.iceberg.data.Record
+import org.apache.iceberg.exceptions.NoSuchTableException
+import org.apache.iceberg.{Schema => IcebergSchema}
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.sql.Timestamp
+import java.util.UUID
+import java.util.zip.ZipInputStream
+
+/**
+ * Unit-level tests for [[IcebergDocument]] running against a local
Hadoop-backed
+ * Iceberg catalog (temp `file:/` warehouse) installed into the shared
+ * `IcebergCatalogInstance` singleton via [[LocalHadoopIcebergCatalog]].
+ *
+ * `IcebergDocument` reads its catalog from
`IcebergCatalogInstance.getInstance()`,
+ * so the catalog must be installed before any document access. Each test
creates a
+ * fresh, uniquely-named table so the read/write/count/clear paths are
isolated.
+ */
+class IcebergDocumentSpec extends AnyFlatSpec with Matchers with
BeforeAndAfterAll {
+
+ private val tableNamespace = "iceberg_doc_spec"
+
+ private val amberSchema: Schema = Schema()
+ .add("id", AttributeType.INTEGER)
+ .add("amount", AttributeType.LONG)
+ .add("score", AttributeType.DOUBLE)
+ .add("name", AttributeType.STRING)
+ .add("ts", AttributeType.TIMESTAMP)
+
+ private val icebergSchema: IcebergSchema =
IcebergUtil.toIcebergSchema(amberSchema)
+
+ private val serde: (IcebergSchema, Tuple) => Record =
IcebergUtil.toGenericRecord
+ private val deserde: (IcebergSchema, Record) => Tuple =
+ (schema, record) => IcebergUtil.fromRecord(record,
IcebergUtil.fromIcebergSchema(schema))
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ LocalHadoopIcebergCatalog.ensure()
+ }
+
+ private def freshTableName(): String =
+ s"tbl_${UUID.randomUUID().toString.replace("-", "")}"
+
+ /** Create the backing table and return a document handle for it. */
+ private def newDocument(tableName: String = freshTableName()):
IcebergDocument[Tuple] = {
+ IcebergUtil.createTable(
+ IcebergCatalogInstance.getInstance(),
+ tableNamespace,
+ tableName,
+ icebergSchema,
+ overrideIfExists = true
+ )
+ new IcebergDocument[Tuple](tableNamespace, tableName, icebergSchema,
serde, deserde)
+ }
+
+ private def tuple(id: Int): Tuple =
+ Tuple
+ .builder(amberSchema)
+ .add("id", AttributeType.INTEGER, Int.box(id))
+ .add("amount", AttributeType.LONG, Long.box(id.toLong * 100L))
+ .add("score", AttributeType.DOUBLE, Double.box(id.toDouble + 0.5))
+ .add("name", AttributeType.STRING, s"name-$id")
+ .add("ts", AttributeType.TIMESTAMP, new Timestamp(1_600_000_000_000L +
id))
+ .build()
+
+ /** Write the given tuples through a single writer session (one committed
file). */
+ private def write(doc: IcebergDocument[Tuple], tuples: Seq[Tuple]): Unit = {
+ val writer = doc.writer(UUID.randomUUID().toString)
+ writer.open()
+ tuples.foreach(writer.putOne)
+ writer.close()
+ }
+
+ "IcebergDocument" should "resolve the table location through getURI for an
existing table" in {
+ val doc = newDocument()
+ // getURI loads the table metadata and wraps table.location() in
URI.create.
+ // On a local `file:/` warehouse the location string carries the table
name.
+ // On Windows the warehouse resolves to a raw `C:\...` path that URI.create
+ // rejects; in that case the method still executes its whole body (metadata
+ // load + location() + URI.create) before throwing, so it still pins the
+ // existing-table branch. The IllegalArgumentException escape is therefore
+ // allowed ONLY on Windows — elsewhere getURI must return a valid URI.
+ val isWindows = System.getProperty("os.name").toLowerCase.contains("win")
+ try {
+ doc.getURI.toString should include(doc.tableName)
+ } catch {
+ case _: IllegalArgumentException if isWindows => succeed
+ }
+ }
+
+ it should "throw NoSuchTableException from getURI when the table does not
exist" in {
+ val doc = new IcebergDocument[Tuple](
+ tableNamespace,
+ freshTableName(),
+ icebergSchema,
+ serde,
+ deserde
+ )
+ intercept[NoSuchTableException] {
+ doc.getURI
+ }
+ }
+
+ it should "return count 0 for a freshly created, empty table" in {
+ val doc = newDocument()
+ doc.getCount shouldBe 0L
+ doc.get().hasNext shouldBe false
+ }
+
+ it should "return count 0 for a table that was never created" in {
+ val doc = new IcebergDocument[Tuple](
+ tableNamespace,
+ freshTableName(),
+ icebergSchema,
+ serde,
+ deserde
+ )
+ doc.getCount shouldBe 0L
+ }
+
+ it should "count and read back all written records" in {
+ val doc = newDocument()
+ val tuples = (0 until 10).map(tuple)
+ write(doc, tuples)
+
+ doc.getCount shouldBe 10L
+ val read = doc.get().toList
+ read should have size 10
+ read.map(_.getField[Int]("id")).toSet shouldBe (0 until 10).toSet
+ }
+
+ it should "read records in file-sequence order within a single committed
file" in {
+ val doc = newDocument()
+ val tuples = (0 until 6).map(tuple)
+ write(doc, tuples)
+
+ doc.get().toList.map(_.getField[Int]("id")) shouldBe (0 until 6).toList
+ }
+
+ it should "return only the requested range via getRange" in {
+ val doc = newDocument()
+ write(doc, (0 until 10).map(tuple))
+
+ doc.getRange(2, 5).toList.map(_.getField[Int]("id")) shouldBe List(2, 3, 4)
+ }
+
+ it should "return records from an offset (inclusive) via getAfter" in {
+ val doc = newDocument()
+ write(doc, (0 until 5).map(tuple))
+
+ doc.getAfter(3).toList.map(_.getField[Int]("id")) shouldBe List(3, 4)
+ }
+
+ it should "project only the requested columns via getRange" in {
+ val doc = newDocument()
+ write(doc, (0 until 4).map(tuple))
+
+ val projected = doc.getRange(0, 4, Some(Seq("id"))).toList
+ projected.map(_.getField[Int]("id")) shouldBe List(0, 1, 2, 3)
+ }
+
+ it should "read incrementally across multiple committed files" in {
+ val doc = newDocument()
+ write(doc, (0 until 5).map(tuple))
+ write(doc, (5 until 10).map(tuple))
+
+ doc.getCount shouldBe 10L
+ doc.get().toList.map(_.getField[Int]("id")).toSet shouldBe (0 until
10).toSet
+ }
+
+ it should "throw NoSuchElementException when advancing an exhausted
iterator" in {
+ val doc = newDocument()
+ write(doc, Seq(tuple(1)))
+ val it = doc.get()
+ it.next()
+ it.hasNext shouldBe false
+ intercept[NoSuchElementException] {
+ it.next()
+ }
+ }
+
+ it should "compute per-field statistics for numeric, string and timestamp
columns" in {
+ val doc = newDocument()
+ write(doc, (1 to 5).map(tuple))
+
+ val stats = doc.getTableStatistics
+ stats.keySet should contain allOf ("id", "amount", "score", "name", "ts")
+
+ // Numeric fields expose min/max plus a not-null count.
+ stats("id")("min").asInstanceOf[Double] shouldBe 1.0
+ stats("id")("max").asInstanceOf[Double] shouldBe 5.0
+ stats("id")("not_null_count").asInstanceOf[Long] shouldBe 5L
+ stats("amount")("min").asInstanceOf[Double] shouldBe 100.0
+ stats("amount")("max").asInstanceOf[Double] shouldBe 500.0
+ stats("score")("max").asInstanceOf[Double] shouldBe 5.5
+
+ // String field only carries the not-null count (no min/max keys).
+ stats("name")("not_null_count").asInstanceOf[Long] shouldBe 5L
+ stats("name").contains("min") shouldBe false
+
+ // Timestamp field exposes min/max as ISO local-date strings.
+ stats("ts") should contain key "min"
+ stats("ts") should contain key "max"
+ }
+
+ it should "throw NoSuchTableException from getTableStatistics when the table
is missing" in {
+ val doc = new IcebergDocument[Tuple](
+ tableNamespace,
+ freshTableName(),
+ icebergSchema,
+ serde,
+ deserde
+ )
+ intercept[NoSuchTableException] {
+ doc.getTableStatistics
+ }
+ }
+
+ it should "report a positive total file size after writes and throw when
missing" in {
+ val doc = newDocument()
+ write(doc, (0 until 8).map(tuple))
+ doc.getTotalFileSize should be > 0L
+
+ val missing = new IcebergDocument[Tuple](
+ tableNamespace,
+ freshTableName(),
+ icebergSchema,
+ serde,
+ deserde
+ )
+ intercept[NoSuchTableException] {
+ missing.getTotalFileSize
+ }
+ }
+
+ it should "expose an empty stream for an empty table via asInputStream" in {
+ val doc = newDocument()
+ val stream = doc.asInputStream()
+ try {
+ stream.readAllBytes() shouldBe empty
+ } finally {
+ stream.close()
+ }
+ }
+
+ it should "expose written rows as a non-empty ZIP archive via asInputStream"
in {
+ val doc = newDocument()
+ write(doc, (0 until 3).map(tuple))
+
+ val stream = doc.asInputStream()
+ try {
+ val bytes = stream.readAllBytes()
+ bytes.length should be > 0
+ // ZIP local-file-header magic bytes: 0x50 0x4B ("PK").
+ bytes(0) shouldBe 0x50.toByte
+ bytes(1) shouldBe 0x4b.toByte
+ } finally {
+ stream.close()
+ }
+
+ // The archive should carry at least one .parquet entry.
+ val zip = new ZipInputStream(doc.asInputStream())
+ try {
+ val entry = zip.getNextEntry
+ entry should not be null
+ entry.getName should endWith(".parquet")
+ } finally {
+ zip.close()
+ }
+ }
+
+ it should "drop the table on clear so it no longer exists and counts as
empty" in {
+ val doc = newDocument()
+ write(doc, (0 until 4).map(tuple))
+ doc.getCount shouldBe 4L
+
+ doc.clear()
+
+ IcebergCatalogInstance
+ .getInstance()
+ .tableExists(TableIdentifier.of(tableNamespace, doc.tableName)) shouldBe
false
+ doc.getCount shouldBe 0L
+ }
+
+ it should "be a no-op when clearing a table that does not exist" in {
+ val doc = new IcebergDocument[Tuple](
+ tableNamespace,
+ freshTableName(),
+ icebergSchema,
+ serde,
+ deserde
+ )
+ noException should be thrownBy doc.clear()
+ }
+
+ it should "expose the constructed table identity and behave as a
VirtualDocument" in {
+ val name = freshTableName()
+ val doc: VirtualDocument[Tuple] = newDocument(name)
+ doc.asInstanceOf[IcebergDocument[Tuple]].tableNamespace shouldBe
tableNamespace
+ doc.asInstanceOf[IcebergDocument[Tuple]].tableName shouldBe name
+ }
+}
