Copilot commented on code in PR #6823: URL: https://github.com/apache/texera/pull/6823#discussion_r3636042321
########## common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergDocumentSpec.scala: ########## @@ -0,0 +1,329 @@ +/* + * 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 either outcome pins the + // existing-table branch. + try { + doc.getURI.toString should include(doc.tableName) + } catch { + case _: IllegalArgumentException => succeed + } + } Review Comment: The test currently treats any IllegalArgumentException from getURI as success on all platforms, which can hide real regressions on non-Windows hosts. Consider only allowing the IllegalArgumentException escape hatch on Windows (where Iceberg may return a raw `C:\...` path) and otherwise asserting the returned URI contains the table name. ########## common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/LocalHadoopIcebergCatalog.scala: ########## @@ -0,0 +1,56 @@ +/* + * 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 + +/** + * 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") + IcebergCatalogInstance.replaceInstance( + IcebergUtil.createHadoopCatalog("wfcore-test", warehouse) + ) Review Comment: LocalHadoopIcebergCatalog creates a temp warehouse directory but never cleans it up, which can leave `wfcore-iceberg-shared*` directories behind on developer machines/CI runners. Consider registering a JVM shutdown hook to delete the temp warehouse recursively once the test JVM exits. ########## common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergDocumentSpec.scala: ########## @@ -0,0 +1,329 @@ +/* + * 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 either outcome pins the + // existing-table branch. + try { + doc.getURI.toString should include(doc.tableName) + } catch { + case _: IllegalArgumentException => 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 after an offset via getAfter" in { Review Comment: Test description says "after an offset" but the assertion expects the returned iterator to start at the offset (inclusive). Updating the wording avoids confusion about getAfter’s semantics. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
