wombatu-kun commented on code in PR #19164: URL: https://github.com/apache/hudi/pull/19164#discussion_r3564093910
########## hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/util/CachingIterator.scala: ########## @@ -26,6 +26,10 @@ package org.apache.hudi.util * * NOTE: [[hasNext]] and [[next]] are purposefully marked as final, requiring iteration * semantic to be implemented t/h overriding of a single [[doHasNext]] method + * + * TODO: this Scala trait has no production users -- the reader path uses the Java Review Comment: The Impact section still says "Test-only change. No production code is modified", but the head commit edits two main sources (this file and `SchemaProvider`). The Summary also still lists "merge throws HoodieDuplicateKeyException with the offending key" for `HoodieSparkValidateDuplicateKeyRecordMerger`, which is the test that was dropped. Both sections need a refresh. Separately, an actionable removal TODO is better tracked as `TODO(HUDI-xxxx)` against a filed ticket than left bare, and this one sits inside the trait's Scaladoc block, so it renders into the published API docs. ########## hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/cli/TestSchemaProvider.java: ########## @@ -0,0 +1,111 @@ +/* + * 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.hudi.cli; + +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.schema.HoodieSchema; + +import org.apache.avro.Schema; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; + +/** + * Tests default behavior of the abstract {@link SchemaProvider} used by Hudi Streamer. Review Comment: `org.apache.hudi.cli.SchemaProvider` is not the Hudi Streamer one - Streamer uses `org.apache.hudi.utilities.schema.SchemaProvider`. This copy's only production consumer is `BootstrapExecutorUtils`, the CLI bootstrap path, and it calls only the legacy `getTargetSchema()`. The javadoc repeats a stale claim from the production class instead of describing what is actually under test. Worth renaming the class to `TestCliSchemaProvider` as well: `TestSchemaProvider` is already the name of stub provider implementations in the `hudi-utilities` and `hudi-kafka-connect` test sources, so `-Dtest=TestSchemaProvider` becomes ambiguous. ########## hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/hudi/TestSparkDatasourceSmallClasses.scala: ########## @@ -0,0 +1,159 @@ +/* + * 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.hudi + +import org.apache.hudi.exception.HoodieException +import org.apache.hudi.util.CachingIterator + +import org.apache.spark.sql.HoodieSparkCatalogUtils +import org.apache.spark.sql.connector.catalog.{Identifier, Table, TableCapability, TableCatalog} +import org.apache.spark.sql.connector.expressions.{Expressions, Transform} +import org.apache.spark.sql.connector.write.LogicalWriteInfo +import org.apache.spark.sql.hudi.catalog.BasicStagedTable +import org.apache.spark.sql.hudi.command.HoodieSparkValidateDuplicateKeyRecordMerger +import org.apache.spark.sql.types.{DataTypes, StructType} +import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertThrows, assertTrue, fail} +import org.junit.jupiter.api.Test +import org.mockito.Mockito.{mock, verify, when} + +/** + * Coverage for a handful of small, otherwise-untested classes in the Spark datasource: + * [[CachingIterator]], [[HoodieSparkCatalogUtils]], [[BasicStagedTable]] and + * [[HoodieSparkValidateDuplicateKeyRecordMerger]]. + */ +class TestSparkDatasourceSmallClasses { + + @Test + def testCachingIteratorYieldsAllElements(): Unit = { + val it = new SeqCachingIterator(Seq("a", "b", "c")) + val collected = scala.collection.mutable.ArrayBuffer[String]() + while (it.hasNext) { + collected += it.next + } + assertEquals(Seq("a", "b", "c"), collected.toSeq) + assertFalse(it.hasNext) + } + + @Test + def testCachingIteratorHasNextIsIdempotent(): Unit = { + val source = new CountingIterator(Seq("x", "y")) + val it = new SeqCachingIterator(source) + // Repeated hasNext without next must not advance the underlying iterator. + assertTrue(it.hasNext) + assertTrue(it.hasNext) + assertTrue(it.hasNext) + assertEquals(1, source.advances) + assertEquals("x", it.next) + assertEquals(1, source.advances) + assertTrue(it.hasNext) + assertEquals(2, source.advances) + assertEquals("y", it.next) + assertFalse(it.hasNext) + } + + @Test + def testMatchBucketTransformMatches(): Unit = { + val transform: Transform = Expressions.bucket(8, "id") + HoodieSparkCatalogUtils.MatchBucketTransform.unapply(transform) match { + case Some((numBuckets, refs, _)) => + assertEquals(8, numBuckets) + assertEquals("id", refs.head.fieldNames()(0)) + case None => fail("expected a bucket transform to match") + } + } + + @Test + def testMatchBucketTransformDoesNotMatchOthers(): Unit = { + val transform: Transform = Expressions.identity("id") + assertTrue(HoodieSparkCatalogUtils.MatchBucketTransform.unapply(transform).isEmpty) + } + + @Test + def testBasicStagedTableDelegatesToUnderlyingTable(): Unit = { + val ident = Identifier.of(Array("db"), "tbl") + val table = mock(classOf[Table]) + val catalog = mock(classOf[TableCatalog]) + val schema = new StructType().add("id", DataTypes.IntegerType) + when(table.schema()).thenReturn(schema) + when(table.partitioning()).thenReturn(Array.empty[Transform]) + when(table.capabilities()).thenReturn(java.util.EnumSet.of(TableCapability.BATCH_WRITE)) + when(table.properties()).thenReturn(java.util.Collections.singletonMap("k", "v")) + + val staged = BasicStagedTable(ident, table, catalog) + assertEquals("tbl", staged.name()) + assertEquals(schema, staged.schema()) + assertEquals(0, staged.partitioning().length) + assertTrue(staged.capabilities().contains(TableCapability.BATCH_WRITE)) + assertEquals("v", staged.properties().get("k")) + + // commit is a no-op; abort must drop the staged table through the catalog. + staged.commitStagedChanges() + staged.abortStagedChanges() + verify(catalog).dropTable(ident) + } + + @Test + def testBasicStagedTableRejectsNonWritableInfo(): Unit = { + val ident = Identifier.of(Array("db"), "tbl") + val staged = BasicStagedTable(ident, mock(classOf[Table]), mock(classOf[TableCatalog])) + val info = mock(classOf[LogicalWriteInfo]) + val ex = assertThrows(classOf[HoodieException], () => staged.newWriteBuilder(info)) Review Comment: This pins a production bug as intended behavior. `BasicStagedTable.newWriteBuilder` matches on `info`, but `SupportsWrite` is a `Table` mixin (`SupportsWrite extends Table`), so a `LogicalWriteInfo` can never satisfy `case supportsWrite: SupportsWrite`. The delegating branch is dead and `newWriteBuilder` throws for every input, including one whose underlying table is perfectly writable. The match was meant to be on `table`, as in Spark's own `BasicStagedTable`, which delegates via `table.asWritable`. Suggest fixing the match in `BasicStagedTable` and then asserting both branches here: delegation when the underlying `Table` is `SupportsWrite`, and the throw when it is not. If the production fix is out of scope for a test-only PR, at least drop the `NonWritableInfo` framing from the test name and file a JIRA, so the bug is not recorded as the contract. ########## hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/hudi/TestSparkDatasourceSmallClasses.scala: ########## @@ -0,0 +1,159 @@ +/* + * 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.hudi + +import org.apache.hudi.exception.HoodieException +import org.apache.hudi.util.CachingIterator + +import org.apache.spark.sql.HoodieSparkCatalogUtils +import org.apache.spark.sql.connector.catalog.{Identifier, Table, TableCapability, TableCatalog} +import org.apache.spark.sql.connector.expressions.{Expressions, Transform} +import org.apache.spark.sql.connector.write.LogicalWriteInfo +import org.apache.spark.sql.hudi.catalog.BasicStagedTable +import org.apache.spark.sql.hudi.command.HoodieSparkValidateDuplicateKeyRecordMerger +import org.apache.spark.sql.types.{DataTypes, StructType} +import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertThrows, assertTrue, fail} +import org.junit.jupiter.api.Test +import org.mockito.Mockito.{mock, verify, when} + +/** + * Coverage for a handful of small, otherwise-untested classes in the Spark datasource: + * [[CachingIterator]], [[HoodieSparkCatalogUtils]], [[BasicStagedTable]] and + * [[HoodieSparkValidateDuplicateKeyRecordMerger]]. + */ +class TestSparkDatasourceSmallClasses { Review Comment: `TestSparkDatasourceSmallClasses` bundles four unrelated production classes into one suite named after the coverage campaign rather than a subject, and three of them (`HoodieSparkCatalogUtils`, `BasicStagedTable`, `HoodieSparkValidateDuplicateKeyRecordMerger`) live outside `org.apache.hudi`. Every other test class in this module mirrors a single production class in that class's own package (`TestDataSourceOptions`, `TestProvidesHoodieConfig`, `TestRecordLevelIndexSupport`), and there is no precedent in the repo for a bundled `Test*Classes` name. The same objection on #19218 ended with that suite dissolved into the natural per-class tests. Suggest splitting into `TestCachingIterator`, `TestHoodieSparkCatalogUtils`, `TestBasicStagedTable` and `TestHoodieSparkValidateDuplicateKeyRecordMerger`, each in the package of the class it covers. ########## hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/hudi/TestSparkDatasourceSmallClasses.scala: ########## @@ -0,0 +1,159 @@ +/* + * 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.hudi + +import org.apache.hudi.exception.HoodieException +import org.apache.hudi.util.CachingIterator + +import org.apache.spark.sql.HoodieSparkCatalogUtils +import org.apache.spark.sql.connector.catalog.{Identifier, Table, TableCapability, TableCatalog} +import org.apache.spark.sql.connector.expressions.{Expressions, Transform} +import org.apache.spark.sql.connector.write.LogicalWriteInfo +import org.apache.spark.sql.hudi.catalog.BasicStagedTable +import org.apache.spark.sql.hudi.command.HoodieSparkValidateDuplicateKeyRecordMerger +import org.apache.spark.sql.types.{DataTypes, StructType} +import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertThrows, assertTrue, fail} +import org.junit.jupiter.api.Test +import org.mockito.Mockito.{mock, verify, when} + +/** + * Coverage for a handful of small, otherwise-untested classes in the Spark datasource: + * [[CachingIterator]], [[HoodieSparkCatalogUtils]], [[BasicStagedTable]] and + * [[HoodieSparkValidateDuplicateKeyRecordMerger]]. + */ +class TestSparkDatasourceSmallClasses { + + @Test + def testCachingIteratorYieldsAllElements(): Unit = { + val it = new SeqCachingIterator(Seq("a", "b", "c")) + val collected = scala.collection.mutable.ArrayBuffer[String]() + while (it.hasNext) { + collected += it.next + } + assertEquals(Seq("a", "b", "c"), collected.toSeq) + assertFalse(it.hasNext) + } + + @Test + def testCachingIteratorHasNextIsIdempotent(): Unit = { + val source = new CountingIterator(Seq("x", "y")) + val it = new SeqCachingIterator(source) + // Repeated hasNext without next must not advance the underlying iterator. + assertTrue(it.hasNext) + assertTrue(it.hasNext) + assertTrue(it.hasNext) + assertEquals(1, source.advances) + assertEquals("x", it.next) + assertEquals(1, source.advances) + assertTrue(it.hasNext) + assertEquals(2, source.advances) + assertEquals("y", it.next) + assertFalse(it.hasNext) + } + + @Test + def testMatchBucketTransformMatches(): Unit = { + val transform: Transform = Expressions.bucket(8, "id") + HoodieSparkCatalogUtils.MatchBucketTransform.unapply(transform) match { + case Some((numBuckets, refs, _)) => + assertEquals(8, numBuckets) + assertEquals("id", refs.head.fieldNames()(0)) + case None => fail("expected a bucket transform to match") + } + } + + @Test + def testMatchBucketTransformDoesNotMatchOthers(): Unit = { + val transform: Transform = Expressions.identity("id") + assertTrue(HoodieSparkCatalogUtils.MatchBucketTransform.unapply(transform).isEmpty) + } + + @Test + def testBasicStagedTableDelegatesToUnderlyingTable(): Unit = { + val ident = Identifier.of(Array("db"), "tbl") + val table = mock(classOf[Table]) + val catalog = mock(classOf[TableCatalog]) + val schema = new StructType().add("id", DataTypes.IntegerType) + when(table.schema()).thenReturn(schema) + when(table.partitioning()).thenReturn(Array.empty[Transform]) + when(table.capabilities()).thenReturn(java.util.EnumSet.of(TableCapability.BATCH_WRITE)) + when(table.properties()).thenReturn(java.util.Collections.singletonMap("k", "v")) + + val staged = BasicStagedTable(ident, table, catalog) + assertEquals("tbl", staged.name()) + assertEquals(schema, staged.schema()) + assertEquals(0, staged.partitioning().length) + assertTrue(staged.capabilities().contains(TableCapability.BATCH_WRITE)) + assertEquals("v", staged.properties().get("k")) + + // commit is a no-op; abort must drop the staged table through the catalog. + staged.commitStagedChanges() + staged.abortStagedChanges() + verify(catalog).dropTable(ident) + } + + @Test + def testBasicStagedTableRejectsNonWritableInfo(): Unit = { + val ident = Identifier.of(Array("db"), "tbl") + val staged = BasicStagedTable(ident, mock(classOf[Table]), mock(classOf[TableCatalog])) + val info = mock(classOf[LogicalWriteInfo]) + val ex = assertThrows(classOf[HoodieException], () => staged.newWriteBuilder(info)) + assertTrue(ex.getMessage.contains("does not support writes")) + } + + @Test + def testDuplicateKeyMergerMetadata(): Unit = { + val merger = new HoodieSparkValidateDuplicateKeyRecordMerger + assertEquals(HoodieSparkValidateDuplicateKeyRecordMerger.STRATEGY_ID, merger.getMergingStrategy) Review Comment: `getMergingStrategy` returns `STRATEGY_ID` directly, so this assertion cannot fail unless the literal-UUID assertion below it also fails. Only the raw-UUID pin carries information here. Suggest dropping this line. ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/command/procedures/TestProcedureParameter.scala: ########## @@ -0,0 +1,73 @@ +/* + * 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.spark.sql.hudi.command.procedures + +import org.apache.spark.sql.types.DataTypes +import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertNull, assertTrue} +import org.junit.jupiter.api.Test + +class TestProcedureParameter { + + @Test + def testRequiredParameter(): Unit = { + val p = ProcedureParameter.required(0, "path", DataTypes.StringType) + assertEquals(0, p.index) + assertEquals("path", p.name) + assertEquals(DataTypes.StringType, p.dataType) + assertTrue(p.required) + assertNull(p.default) + } + + @Test + def testOptionalParameterWithDefault(): Unit = { + val p = ProcedureParameter.optional(2, "limit", DataTypes.IntegerType, 10) + assertEquals(2, p.index) + assertEquals("limit", p.name) + assertEquals(DataTypes.IntegerType, p.dataType) + assertFalse(p.required) + assertEquals(10, p.default.asInstanceOf[Int]) + } + + @Test + def testHashCodeIsContentBased(): Unit = { + val a = ProcedureParameter.optional(1, "col", DataTypes.StringType, "def") + val b = ProcedureParameter.optional(1, "col", DataTypes.StringType, "def") + // hashCode is derived from all fields, so equal-valued parameters share a hash. + assertEquals(a.hashCode, b.hashCode) Review Comment: This assertion holds for any `hashCode`, including one that returns a constant, so it does not actually establish that the hash is content-based. `testEqualsGuardsAgainstRegressions` below has the same gap on the equality side: both instances are identical, so a degenerate `equals` comparing only the class would still pass. Suggest adding an inequality case to each (two params differing only in `index`, say): unequal hashes and `assertNotEquals`. That is what makes the field-by-field comparison the tests claim to cover actually load-bearing. -- 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]
