voonhous commented on code in PR #19164: URL: https://github.com/apache/hudi/pull/19164#discussion_r3563958281
########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/command/procedures/TestProcedureParameter.scala: ########## @@ -0,0 +1,63 @@ +/* + * 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: `equals` is the one method on `ProcedureParameterImpl` this file does not assert, and it is exactly the method fixed 7 days ago in #19167 (`88d34dc77146`: self-recursive `equals` -> StackOverflow, plus a pre-guard ClassCastException on wrong-type args). That fix shipped with no test, so this PR is its natural home, and `testHashCodeIsContentBased` already builds the two equal-valued instances needed. Please add the regression assertions before merge: ```suggestion } @Test def testEqualsGuardsAgainstRegressions(): Unit = { val a = ProcedureParameter.optional(1, "col", DataTypes.StringType, "def") val b = ProcedureParameter.optional(1, "col", DataTypes.StringType, "def") // Equal-valued params are equal; guards the self-recursive `equals` (StackOverflow) fixed in #19167. assertEquals(a, b) // A wrong-type argument returns false instead of throwing ClassCastException (also #19167). assertFalse(a.equals("not a param")) } ``` ########## hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/hudi/TestSparkDatasourceSmallClasses.scala: ########## @@ -0,0 +1,168 @@ +/* + * 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.common.table.read.BufferedRecord +import org.apache.hudi.exception.{HoodieDuplicateKeyException, 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 testDuplicateKeyMergerThrowsOnMerge(): Unit = { + val merger = new HoodieSparkValidateDuplicateKeyRecordMerger + val older = new BufferedRecord[AnyRef]("dup_key_1", null, null, null, null) + val ex = assertThrows(classOf[HoodieDuplicateKeyException], + () => merger.merge(older, null, null, null)) + assertTrue(ex.getMessage.contains("dup_key_1")) + } + + @Test + def testDuplicateKeyMergerMetadata(): Unit = { + val merger = new HoodieSparkValidateDuplicateKeyRecordMerger + assertEquals(HoodieSparkValidateDuplicateKeyRecordMerger.STRATEGY_ID, merger.getMergingStrategy) + assertEquals("fb092649-0fdc-4c14-9113-acde3034a6c4", merger.getMergingStrategy) + // Pre-combining falls back to the default Spark record merger. + assertTrue(merger.asPreCombiningMode().isInstanceOf[DefaultSparkRecordMerger]) + } +} + +/** + * Concrete [[CachingIterator]] backed by a plain iterator, used to exercise the trait. + */ +private class SeqCachingIterator(source: Iterator[String]) extends CachingIterator[String] { Review Comment: nit (optional): heads-up that the Scala trait `org.apache.hudi.util.CachingIterator` under test looks orphaned -- the reader path (`SparkFileFormatInternalRowReaderContext.scala:33,273`) imports and uses the Java `org.apache.hudi.common.util.collection.CachingIterator`, and no production code extends the Scala trait. The test is fine as coverage-tail work, but the trait itself is a deletion candidate for a separate cleanup. Feel free to ignore. ########## 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. + */ +public class TestSchemaProvider { + + private static final String SCHEMA_STR = + "{\"type\":\"record\",\"name\":\"r\",\"fields\":[{\"name\":\"id\",\"type\":\"int\"}]}"; + + private static Schema parseSchema() { + return new Schema.Parser().parse(SCHEMA_STR); + } + + /** Provider that only implements the required source schema. */ + private static class SourceOnlyProvider extends SchemaProvider { + private final Schema schema; + + SourceOnlyProvider(Schema schema) { + super(new TypedProperties()); + this.schema = schema; + } + + @Override + public Schema getSourceSchema() { + return schema; + } + } + + /** Provider whose legacy target schema is unavailable, exercising the fallback path. */ + private static class TargetUnsupportedProvider extends SchemaProvider { + private final Schema schema; + + TargetUnsupportedProvider(Schema schema) { + super(new TypedProperties()); + this.schema = schema; + } + + @Override + public Schema getSourceSchema() { + return schema; + } + + @Override + public Schema getTargetSchema() { + throw new UnsupportedOperationException("no target schema"); + } + } + + @Test + void testTargetSchemaDefaultsToSource() { + Schema schema = parseSchema(); + SchemaProvider provider = new SourceOnlyProvider(schema); + // By default the target schema is the source schema. + assertSame(schema, provider.getTargetSchema()); + } + + @Test + void testHoodieSchemaWrapsAvroSchema() { + Schema schema = parseSchema(); + SchemaProvider provider = new SourceOnlyProvider(schema); + HoodieSchema sourceHoodieSchema = provider.getSourceHoodieSchema(); + assertNotNull(sourceHoodieSchema); + assertEquals(schema, sourceHoodieSchema.getAvroSchema()); + // Target Hoodie schema defaults to the source when target schema is not overridden. + assertEquals(schema, provider.getTargetHoodieSchema().getAvroSchema()); + } + + @Test + void testNullSchemaYieldsNullHoodieSchema() { + SchemaProvider provider = new SourceOnlyProvider(null); + assertNull(provider.getSourceHoodieSchema()); + assertNull(provider.getTargetHoodieSchema()); + } + + @Test + void testTargetHoodieSchemaFallsBackToSourceOnUnsupported() { Review Comment: nit (optional): this exercises the "target throws, source works" path, which is the inverse of what the production comment documents (`SchemaProvider.java:76-77`: "if `getTargetSchema()` calls `getSourceSchema()` which is not implemented"). That documented scenario would actually re-throw, since the `catch` calls `getSourceHoodieSchema()` -> `getSourceSchema()` again. It is really a stale production comment, and the branch is unreachable via the only caller (`BootstrapExecutorUtils`, which uses the legacy `getTargetSchema()`). If you touch it, fix the production comment to match what the fallback actually handles; otherwise ignore. ########## hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/hudi/TestSparkDatasourceSmallClasses.scala: ########## @@ -0,0 +1,168 @@ +/* + * 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.common.table.read.BufferedRecord +import org.apache.hudi.exception.{HoodieDuplicateKeyException, 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 testDuplicateKeyMergerThrowsOnMerge(): Unit = { Review Comment: nit (optional): the duplicate-key throw is already pinned end-to-end -- `TestInsertTable5.scala:274` asserts the verbatim `Duplicate key found for insert statement, key is: ...` message via SQL insert (also `assertThrows[HoodieDuplicateKeyException]` in `TestInsertTable.scala:447,517` and `TestInsertTable4.scala:726`). This direct unit test is not strictly redundant since it localizes the failure to the merger, but if you want to trim, drop this method and keep `testDuplicateKeyMergerMetadata` (STRATEGY_ID + `asPreCombiningMode`), which is the genuinely-new coverage. Feel free to ignore. -- 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]
