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-6823-826eb2183d1da8b82852818e0b9f819c872691aa in repository https://gitbox.apache.org/repos/asf/texera.git
commit 8096cf0284be55eda051eadffe783efe0bf3e488 Author: Xinyuan Lin <[email protected]> AuthorDate: Thu Jul 23 20:47:25 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 + } +}
