voonhous commented on code in PR #19164: URL: https://github.com/apache/hudi/pull/19164#discussion_r3564046109
########## 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: Done in 9f97920 -- added `testEqualsGuardsAgainstRegressions`: `assertEquals(a, b)` catches the self-recursion StackOverflow and `assertFalse(a.equals("not a param"))` catches the pre-guard ClassCastException, both from #19167. ########## 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: Done in 9f97920 -- dropped `testDuplicateKeyMergerThrowsOnMerge` and its now-unused imports; kept `testDuplicateKeyMergerMetadata`. ########## 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: Kept the test; added a TODO on the trait flagging it as a removal candidate (9f97920). Actual deletion left for a separate cleanup. ########## 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: Done in 9f97920 -- rewrote the comment in SchemaProvider.java to describe the case the catch actually handles (a subclass overriding getTargetSchema() to throw, source intact). -- 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]
