voonhous commented on code in PR #19405: URL: https://github.com/apache/hudi/pull/19405#discussion_r3690171860
########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestCatalystExpressionOrderPreserving.scala: ########## @@ -0,0 +1,94 @@ +/* + * 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.spark.sql.catalyst.expressions.{Add, AttributeReference, BitwiseOr, Cast, DateAdd, DateSub, Divide, Exp, Expression, Literal, Log, Lower, Multiply, ShiftLeft, Sqrt, Upper} +import org.apache.spark.sql.types.{DateType, DoubleType, IntegerType, LongType, StringType} +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test + +/** + * Branch coverage for the order-preserving transformation matcher in + * [[org.apache.spark.sql.BaseHoodieCatalystExpressionUtils]]. This is what lets data skipping map a + * transformed column reference back to its source attribute, so each case is pinned to the exact + * source [[AttributeReference]] it must recover, and non-order-preserving shapes must not match. + */ +class TestCatalystExpressionOrderPreserving extends SparkAdapterSupport { + + private val intAttr = AttributeReference("i", IntegerType)() + private val strAttr = AttributeReference("s", StringType)() + private val dblAttr = AttributeReference("d", DoubleType)() + private val dateAttr = AttributeReference("dt", DateType)() + + private def matched(expr: Expression): Option[AttributeReference] = + sparkAdapter.getCatalystExpressionUtils.tryMatchAttributeOrderingPreservingTransformation(expr) + + @Test + def testIdentityAttributeMatches(): Unit = { + assertEquals(Some(intAttr), matched(intAttr)) + } + + @Test + def testArithmeticTransformationsPreserveOrdering(): Unit = { + assertEquals(Some(intAttr), matched(Add(intAttr, Literal(1)))) + assertEquals(Some(intAttr), matched(Add(Literal(1), intAttr))) + assertEquals(Some(intAttr), matched(Multiply(intAttr, Literal(2)))) + assertEquals(Some(intAttr), matched(Multiply(Literal(2), intAttr))) + assertEquals(Some(intAttr), matched(Divide(intAttr, Literal(2)))) + assertEquals(Some(intAttr), matched(BitwiseOr(intAttr, Literal(1)))) + assertEquals(Some(intAttr), matched(BitwiseOr(Literal(1), intAttr))) + assertEquals(Some(intAttr), matched(ShiftLeft(intAttr, Literal(1)))) + } + + @Test + def testUnaryMathAndStringTransformationsPreserveOrdering(): Unit = { + assertEquals(Some(dblAttr), matched(Exp(dblAttr))) + assertEquals(Some(dblAttr), matched(Log(dblAttr))) + assertEquals(Some(strAttr), matched(Upper(strAttr))) + assertEquals(Some(strAttr), matched(Lower(strAttr))) + } + + @Test + def testDateTransformationsPreserveOrdering(): Unit = { + assertEquals(Some(dateAttr), matched(DateAdd(dateAttr, Literal(1)))) + assertEquals(Some(dateAttr), matched(DateSub(dateAttr, Literal(1)))) + } + + @Test + def testUpCastPreservesOrderingButNumericToStringDoesNot(): Unit = { + // Widening a numeric column preserves ordering, so the source attribute is recovered. + assertEquals(Some(intAttr), matched(Cast(intAttr, LongType))) + // Casting a numeric column to string can reorder values, so it must not match. + assertEquals(None, matched(Cast(intAttr, StringType))) Review Comment: Both `Cast` cases pin the safe directions, but the discriminating one is missing: `HoodieSparkTypeUtils.isCastPreservingOrdering` only rejects `String<->Numeric` and returns `true` for everything else, including narrowing numeric casts. Adding ```scala val longAttr = AttributeReference("l", LongType)() assertEquals(None, matched(Cast(longAttr, IntegerType))) ``` fails today: the matcher recovers the attribute even though non-ANSI narrowing wraps around, and `DataSkippingUtils.translateIntoColumnStatsIndexFilterExpr` then rewrites min/max through the cast. Concrete failure: bigint col with min=-2147483643, max=4294967301; `cast(a as int) < 5` gives cast(min)=5, cast(max)=5, so the file is pruned even though it contains a=4294967301 -> silently missing rows. Same family: the `Multiply`/`Divide` arms match any literal operand, including negative ones that reverse ordering. Since the missing assertion exposes a production bug, please file a GitHub issue on `isCastPreservingOrdering` (numeric-to-numeric should require `Cast.canUpCast`), and either fix it in this PR or pin current behavior here with a TODO referencing the issue so the hazard is documented rather than invisible. ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/TestFileFormatUtilsForFileGroupReader.scala: ########## @@ -0,0 +1,96 @@ +/* + * 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 + +import org.apache.spark.sql.catalyst.expressions.{And, Attribute, AttributeReference, Contains, EndsWith, EqualNullSafe, EqualTo, Expression, GreaterThan, In, IsNotNull, IsNull, LessThanOrEqual, Literal, Not, Or, StartsWith} +import org.apache.spark.sql.catalyst.plans.logical.{Filter, LocalRelation} +import org.apache.spark.sql.types.{BooleanType, IntegerType, StringType, StructType} +import org.junit.jupiter.api.Assertions.{assertEquals, assertSame} +import org.junit.jupiter.api.Test + +/** + * Coverage for [[FileFormatUtilsForFileGroupReader.applyFiltersToPlan]], which lowers pushed-down + * data-source [[org.apache.spark.sql.sources.Filter]]s into a Catalyst [[Filter]] on the plan. Each + * case pins the exact translated Catalyst expression so a wrong mapping would fail, and the + * no-filter case must return the input plan untouched. + */ +class TestFileFormatUtilsForFileGroupReader { Review Comment: Scope note: this file covers `applyFiltersToPlan` only, which is the low-risk member of `FileFormatUtilsForFileGroupReader`. The fgReader entry point `applyNewFileFormatChanges` (lines 34-48) keeps zero coverage and carries the sharp edges: unchecked `asInstanceOf[ParquetFileFormat with HoodieFormatTrait]`, mutable `ff.isProjected` whose double-apply guard lives in the callers, and a non-exhaustive match on `fs.location` (`MatchError` for any other `FileIndex`). No new test required for this PR, but please add one line to the PR description saying `applyNewFileFormatChanges` is out of scope, so the class's coverage number is not read as fgReader-path coverage. ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/TestFileFormatUtilsForFileGroupReader.scala: ########## @@ -0,0 +1,96 @@ +/* + * 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 + +import org.apache.spark.sql.catalyst.expressions.{And, Attribute, AttributeReference, Contains, EndsWith, EqualNullSafe, EqualTo, Expression, GreaterThan, In, IsNotNull, IsNull, LessThanOrEqual, Literal, Not, Or, StartsWith} +import org.apache.spark.sql.catalyst.plans.logical.{Filter, LocalRelation} +import org.apache.spark.sql.types.{BooleanType, IntegerType, StringType, StructType} +import org.junit.jupiter.api.Assertions.{assertEquals, assertSame} +import org.junit.jupiter.api.Test + +/** + * Coverage for [[FileFormatUtilsForFileGroupReader.applyFiltersToPlan]], which lowers pushed-down + * data-source [[org.apache.spark.sql.sources.Filter]]s into a Catalyst [[Filter]] on the plan. Each + * case pins the exact translated Catalyst expression so a wrong mapping would fail, and the + * no-filter case must return the input plan untouched. + */ +class TestFileFormatUtilsForFileGroupReader { + + private val a: Attribute = AttributeReference("a", IntegerType)() + private val b: Attribute = AttributeReference("b", StringType)() + private val tableSchema: StructType = new StructType().add("a", IntegerType).add("b", StringType) + private val resolved: Seq[Attribute] = Seq(a, b) + private val plan: LocalRelation = LocalRelation(a, b) + + private def condOf(filters: Seq[sources.Filter]): Expression = + FileFormatUtilsForFileGroupReader.applyFiltersToPlan(plan, tableSchema, resolved, filters) match { + case Filter(cond, child) => + assertSame(plan, child) + cond + case other => throw new AssertionError(s"expected a Filter, got $other") + } + + @Test + def testComparisonAndNullFilters(): Unit = { Review Comment: The one comparison production actually pushes is the one missing: the only caller with non-empty `requiredFilters` (`MergeOnReadIncrementalRelationV1.incrementalSpanRecordFilters`) emits `IsNotNull` + `GreaterThanOrEqual` (the common, non-archived-start case) or `GreaterThan`, plus `LessThanOrEqual` -- and the `sources.GreaterThanOrEqual` and `sources.LessThan` arms of `translate` are unreached (so is top-level `sources.And`). Also unreached is the method's only failure mode: `filters.map(...).get` (`FileFormatUtilsForFileGroupReader.scala:121`) throws a bare `NoSuchElementException: None.get` for a filter naming a column absent from `tableSchema` -- reachable via `readStream.schema(...)` omitting the meta columns while the relation injects `_hoodie_commit_time` filters. Action: add `sources.GreaterThanOrEqual` and `sources.LessThan` here, a `sources.And(...)` case to `testCompositeFilters`, and one failure pin: ```scala assertThrows(classOf[NoSuchElementException], () => FileFormatUtilsForFileGroupReader.applyFiltersToPlan( plan, tableSchema, resolved, Seq(sources.EqualTo("missing", 1)))) ``` ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestCatalystExpressionOrderPreserving.scala: ########## @@ -0,0 +1,94 @@ +/* + * 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.spark.sql.catalyst.expressions.{Add, AttributeReference, BitwiseOr, Cast, DateAdd, DateSub, Divide, Exp, Expression, Literal, Log, Lower, Multiply, ShiftLeft, Sqrt, Upper} +import org.apache.spark.sql.types.{DateType, DoubleType, IntegerType, LongType, StringType} +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test + +/** + * Branch coverage for the order-preserving transformation matcher in + * [[org.apache.spark.sql.BaseHoodieCatalystExpressionUtils]]. This is what lets data skipping map a + * transformed column reference back to its source attribute, so each case is pinned to the exact + * source [[AttributeReference]] it must recover, and non-order-preserving shapes must not match. + */ +class TestCatalystExpressionOrderPreserving extends SparkAdapterSupport { + + private val intAttr = AttributeReference("i", IntegerType)() + private val strAttr = AttributeReference("s", StringType)() + private val dblAttr = AttributeReference("d", DoubleType)() + private val dateAttr = AttributeReference("dt", DateType)() + + private def matched(expr: Expression): Option[AttributeReference] = + sparkAdapter.getCatalystExpressionUtils.tryMatchAttributeOrderingPreservingTransformation(expr) + + @Test + def testIdentityAttributeMatches(): Unit = { + assertEquals(Some(intAttr), matched(intAttr)) + } + + @Test + def testArithmeticTransformationsPreserveOrdering(): Unit = { + assertEquals(Some(intAttr), matched(Add(intAttr, Literal(1)))) + assertEquals(Some(intAttr), matched(Add(Literal(1), intAttr))) + assertEquals(Some(intAttr), matched(Multiply(intAttr, Literal(2)))) + assertEquals(Some(intAttr), matched(Multiply(Literal(2), intAttr))) + assertEquals(Some(intAttr), matched(Divide(intAttr, Literal(2)))) + assertEquals(Some(intAttr), matched(BitwiseOr(intAttr, Literal(1)))) + assertEquals(Some(intAttr), matched(BitwiseOr(Literal(1), intAttr))) + assertEquals(Some(intAttr), matched(ShiftLeft(intAttr, Literal(1)))) + } + + @Test + def testUnaryMathAndStringTransformationsPreserveOrdering(): Unit = { + assertEquals(Some(dblAttr), matched(Exp(dblAttr))) + assertEquals(Some(dblAttr), matched(Log(dblAttr))) + assertEquals(Some(strAttr), matched(Upper(strAttr))) + assertEquals(Some(strAttr), matched(Lower(strAttr))) + } + + @Test + def testDateTransformationsPreserveOrdering(): Unit = { Review Comment: The file covers only shared match arms, not the one branch of this matcher that differs per Spark version: `unapplyOrderPreservingDateParsing` (`BaseHoodieCatalystExpressionUtils.scala:107-110`), whose overrides pattern-match `ParseToDate`/`ParseToTimestamp` with different arities in each version module (3.3: `ParseToDate(child,_,_)`; 3.5/4.x: `ParseToDate(child,_,_,_)`), re-implemented in #19149. That is exactly the branch a per-profile unit test should pin. Related data point worth a follow-up: both nodes are `RuntimeReplaceable`, and Spark's first optimizer batch (`ReplaceExpressions`) rewrites them before filters reach data skipping, so this branch may never fire on real queries; the existing `to_timestamp` coverage in `TestDataSkippingUtils` (line 677) only survives because that harness applies `OptimizeIn` alone. Action: add one case built portably so it compiles on every profile, e.g. ```scala assertEquals(Some(strAttr), matched(sparkAdapter.getExpressionFromColumn( functions.to_date(sparkAdapter.createColumnFromExpression(strAttr))))) ``` (`SparkAdapter.scala:376,384` provide both directions), and consider a follow-up issue to confirm whether `to_date`/`to_timestamp` data skipping still works post-`ReplaceExpressions`. ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/TestFileFormatUtilsForFileGroupReader.scala: ########## @@ -0,0 +1,96 @@ +/* + * 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 + +import org.apache.spark.sql.catalyst.expressions.{And, Attribute, AttributeReference, Contains, EndsWith, EqualNullSafe, EqualTo, Expression, GreaterThan, In, IsNotNull, IsNull, LessThanOrEqual, Literal, Not, Or, StartsWith} +import org.apache.spark.sql.catalyst.plans.logical.{Filter, LocalRelation} +import org.apache.spark.sql.types.{BooleanType, IntegerType, StringType, StructType} +import org.junit.jupiter.api.Assertions.{assertEquals, assertSame} +import org.junit.jupiter.api.Test + +/** + * Coverage for [[FileFormatUtilsForFileGroupReader.applyFiltersToPlan]], which lowers pushed-down + * data-source [[org.apache.spark.sql.sources.Filter]]s into a Catalyst [[Filter]] on the plan. Each + * case pins the exact translated Catalyst expression so a wrong mapping would fail, and the + * no-filter case must return the input plan untouched. + */ +class TestFileFormatUtilsForFileGroupReader { + + private val a: Attribute = AttributeReference("a", IntegerType)() + private val b: Attribute = AttributeReference("b", StringType)() + private val tableSchema: StructType = new StructType().add("a", IntegerType).add("b", StringType) + private val resolved: Seq[Attribute] = Seq(a, b) + private val plan: LocalRelation = LocalRelation(a, b) + + private def condOf(filters: Seq[sources.Filter]): Expression = + FileFormatUtilsForFileGroupReader.applyFiltersToPlan(plan, tableSchema, resolved, filters) match { + case Filter(cond, child) => + assertSame(plan, child) + cond + case other => throw new AssertionError(s"expected a Filter, got $other") + } + + @Test + def testComparisonAndNullFilters(): Unit = { + assertEquals(EqualTo(a, Literal(1)), condOf(Seq(sources.EqualTo("a", 1)))) + assertEquals(EqualNullSafe(a, Literal(1)), condOf(Seq(sources.EqualNullSafe("a", 1)))) + assertEquals(GreaterThan(a, Literal(5)), condOf(Seq(sources.GreaterThan("a", 5)))) + assertEquals(LessThanOrEqual(a, Literal(5)), condOf(Seq(sources.LessThanOrEqual("a", 5)))) + assertEquals(IsNull(b), condOf(Seq(sources.IsNull("b")))) + assertEquals(IsNotNull(b), condOf(Seq(sources.IsNotNull("b")))) + } + + @Test + def testStringAndInFilters(): Unit = { + assertEquals(Contains(b, Literal("x")), condOf(Seq(sources.StringContains("b", "x")))) + assertEquals(StartsWith(b, Literal("x")), condOf(Seq(sources.StringStartsWith("b", "x")))) + assertEquals(EndsWith(b, Literal("x")), condOf(Seq(sources.StringEndsWith("b", "x")))) + assertEquals( + In(a, Seq(Literal(1), Literal(2), Literal(3))), + condOf(Seq(sources.In("a", Array[Any](1, 2, 3))))) + } + + @Test + def testConstantFilters(): Unit = { + assertEquals(Literal(true, BooleanType), condOf(Seq(sources.AlwaysTrue()))) + assertEquals(Literal(false, BooleanType), condOf(Seq(sources.AlwaysFalse()))) + } + + @Test + def testCompositeFilters(): Unit = { + // A nested and/or/not tree is lowered structurally. + assertEquals( + Or(EqualTo(a, Literal(1)), Not(IsNull(b))), + condOf(Seq(sources.Or(sources.EqualTo("a", 1), sources.Not(sources.IsNull("b")))))) + } + + @Test + def testMultipleFiltersAreAndedInOrder(): Unit = { + // Several top-level filters combine left-to-right via And. + assertEquals( + And(GreaterThan(a, Literal(5)), IsNotNull(b)), + condOf(Seq(sources.GreaterThan("a", 5), sources.IsNotNull("b")))) Review Comment: With two filters this cannot pin what the test name claims: `reduceLeft(And)` and `reduceRight(And)` produce the identical tree at n=2. Three filters -- also the exact shape the only production caller emits (`MergeOnReadIncrementalRelationV1.incrementalSpanRecordFilters`) -- make the left-nesting observable: ```suggestion assertEquals( And(And(IsNotNull(b), GreaterThan(a, Literal(5))), LessThanOrEqual(a, Literal(9))), condOf(Seq(sources.IsNotNull("b"), sources.GreaterThan("a", 5), sources.LessThanOrEqual("a", 9)))) ``` ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestCatalystExpressionOrderPreserving.scala: ########## @@ -0,0 +1,94 @@ +/* + * 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.spark.sql.catalyst.expressions.{Add, AttributeReference, BitwiseOr, Cast, DateAdd, DateSub, Divide, Exp, Expression, Literal, Log, Lower, Multiply, ShiftLeft, Sqrt, Upper} +import org.apache.spark.sql.types.{DateType, DoubleType, IntegerType, LongType, StringType} +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test + +/** + * Branch coverage for the order-preserving transformation matcher in + * [[org.apache.spark.sql.BaseHoodieCatalystExpressionUtils]]. This is what lets data skipping map a + * transformed column reference back to its source attribute, so each case is pinned to the exact + * source [[AttributeReference]] it must recover, and non-order-preserving shapes must not match. + */ +class TestCatalystExpressionOrderPreserving extends SparkAdapterSupport { Review Comment: nit, feel free to ignore: the other two files in this PR follow `Test<ClassUnderTest>`; this one names the behavior instead, so a duplicate-hunter grepping by class name will not land here by filename (the scaladoc link does cover content grep). `TestHoodieCatalystExpressionUtils` would keep the convention; no name collision exists. ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/avro/TestAvroUtils.scala: ########## @@ -0,0 +1,171 @@ +/* + * 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.avro + +import org.apache.avro.Schema +import org.apache.spark.sql.types._ +import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertThrows, assertTrue} +import org.junit.jupiter.api.Test + +/** + * Direct coverage for [[AvroUtils]] and its [[AvroUtils.AvroSchemaHelper]], which are otherwise only + * exercised indirectly through the Avro serializers. Focuses on the type-support predicate and the + * schema-matching/validation error paths (extra Catalyst fields, extra required Avro fields, + * positional vs by-name matching, ambiguous by-name lookup), pinning the raised exception messages. + */ +class TestAvroUtils { + + private def parse(json: String): Schema = new Schema.Parser().parse(json) + + @Test + def testSupportsDataType(): Unit = { + assertTrue(AvroUtils.supportsDataType(IntegerType)) + assertTrue(AvroUtils.supportsDataType(StringType)) + assertTrue(AvroUtils.supportsDataType(NullType)) + assertTrue(AvroUtils.supportsDataType(ArrayType(LongType))) + assertTrue(AvroUtils.supportsDataType(MapType(StringType, IntegerType))) + assertTrue(AvroUtils.supportsDataType( + new StructType().add("a", IntegerType).add("b", ArrayType(StringType)))) + // CalendarInterval is not representable in Avro, so every wrapper around it is unsupported too. + assertFalse(AvroUtils.supportsDataType(CalendarIntervalType)) + assertFalse(AvroUtils.supportsDataType(ArrayType(CalendarIntervalType))) + assertFalse(AvroUtils.supportsDataType(MapType(StringType, CalendarIntervalType))) + assertFalse(AvroUtils.supportsDataType(new StructType().add("a", CalendarIntervalType))) + } + + @Test + def testToFieldStr(): Unit = { + assertEquals("top-level record", AvroUtils.toFieldStr(Seq.empty)) + assertEquals("field 'foo'", AvroUtils.toFieldStr(Seq("foo"))) + assertEquals("field 'foo.bar'", AvroUtils.toFieldStr(Seq("foo", "bar"))) + } + + @Test + def testIsNullable(): Unit = { + val record = parse( + """{"type":"record","name":"r","fields":[ + | {"name":"req","type":"int"}, + | {"name":"opt","type":["null","int"]}, + | {"name":"uni","type":["int","long"]} + |]}""".stripMargin) + assertFalse(AvroUtils.isNullable(record.getField("req"))) + assertTrue(AvroUtils.isNullable(record.getField("opt"))) + // A union without a NULL branch is not nullable. + assertFalse(AvroUtils.isNullable(record.getField("uni"))) + } + + @Test + def testSchemaHelperRejectsNonRecordSchema(): Unit = { + val ex = assertThrows(classOf[IncompatibleSchemaException], + () => new AvroUtils.AvroSchemaHelper( + Schema.create(Schema.Type.INT), new StructType(), Seq.empty, Seq.empty, false)) + assertTrue(ex.getMessage.contains("as a RECORD")) + } + + @Test + def testMatchedFieldsAndGetAvroFieldByName(): Unit = { + val avro = parse( + """{"type":"record","name":"r","fields":[ + | {"name":"id","type":"int"}, + | {"name":"name","type":["null","string"]} + |]}""".stripMargin) Review Comment: With the Avro and Catalyst fields in the same order, all three assertions return identical results if `positionalFieldMatch` were flipped to `true` -- the by-name path is exercised but never discriminated. Make the orders differ so a positional lookup would return the wrong field: ```suggestion // Avro order differs from Catalyst so by-name lookup is discriminated from positional. val avro = parse( """{"type":"record","name":"r","fields":[ | {"name":"name","type":["null","string"]}, | {"name":"id","type":"int"} |]}""".stripMargin) ``` ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/avro/TestAvroUtils.scala: ########## @@ -0,0 +1,171 @@ +/* + * 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.avro + +import org.apache.avro.Schema +import org.apache.spark.sql.types._ +import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertThrows, assertTrue} +import org.junit.jupiter.api.Test + +/** + * Direct coverage for [[AvroUtils]] and its [[AvroUtils.AvroSchemaHelper]], which are otherwise only + * exercised indirectly through the Avro serializers. Focuses on the type-support predicate and the + * schema-matching/validation error paths (extra Catalyst fields, extra required Avro fields, + * positional vs by-name matching, ambiguous by-name lookup), pinning the raised exception messages. + */ +class TestAvroUtils { + + private def parse(json: String): Schema = new Schema.Parser().parse(json) + + @Test + def testSupportsDataType(): Unit = { Review Comment: nit, author's call: 13 assertions on a path with no live callers. `AvroUtils.supportsDataType` has zero production callers repo-wide (only self-recursion and this test; the `Spark4DefaultSource` hit is an unrelated `CreatableRelationProvider` override). Likewise `positionalFieldMatch = true` (tested below at lines 96 and 122) is hardcoded `false` at every Hudi construction site (`AvroSerializer`/`AvroDeserializer` 3-arg constructors only). The vendored file is byte-identical to upstream Spark 3.3.0 apart from the `RowReader` removal in #19147, so these cases pin upstream semantics on code Hudi cannot reach. Either drop `testSupportsDataType` (consistent with #19147 pruning unused members) or keep it with a one-line comment saying it guards the vendored copy against edit drift, so a future pruner does not read the test as evidence of live use. ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/avro/TestAvroUtils.scala: ########## @@ -0,0 +1,171 @@ +/* + * 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.avro + +import org.apache.avro.Schema +import org.apache.spark.sql.types._ +import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertThrows, assertTrue} +import org.junit.jupiter.api.Test + +/** + * Direct coverage for [[AvroUtils]] and its [[AvroUtils.AvroSchemaHelper]], which are otherwise only + * exercised indirectly through the Avro serializers. Focuses on the type-support predicate and the + * schema-matching/validation error paths (extra Catalyst fields, extra required Avro fields, + * positional vs by-name matching, ambiguous by-name lookup), pinning the raised exception messages. + */ +class TestAvroUtils { + + private def parse(json: String): Schema = new Schema.Parser().parse(json) + + @Test + def testSupportsDataType(): Unit = { + assertTrue(AvroUtils.supportsDataType(IntegerType)) + assertTrue(AvroUtils.supportsDataType(StringType)) + assertTrue(AvroUtils.supportsDataType(NullType)) + assertTrue(AvroUtils.supportsDataType(ArrayType(LongType))) + assertTrue(AvroUtils.supportsDataType(MapType(StringType, IntegerType))) + assertTrue(AvroUtils.supportsDataType( + new StructType().add("a", IntegerType).add("b", ArrayType(StringType)))) + // CalendarInterval is not representable in Avro, so every wrapper around it is unsupported too. + assertFalse(AvroUtils.supportsDataType(CalendarIntervalType)) + assertFalse(AvroUtils.supportsDataType(ArrayType(CalendarIntervalType))) + assertFalse(AvroUtils.supportsDataType(MapType(StringType, CalendarIntervalType))) + assertFalse(AvroUtils.supportsDataType(new StructType().add("a", CalendarIntervalType))) + } + + @Test + def testToFieldStr(): Unit = { + assertEquals("top-level record", AvroUtils.toFieldStr(Seq.empty)) + assertEquals("field 'foo'", AvroUtils.toFieldStr(Seq("foo"))) + assertEquals("field 'foo.bar'", AvroUtils.toFieldStr(Seq("foo", "bar"))) + } + + @Test + def testIsNullable(): Unit = { + val record = parse( + """{"type":"record","name":"r","fields":[ + | {"name":"req","type":"int"}, + | {"name":"opt","type":["null","int"]}, + | {"name":"uni","type":["int","long"]} + |]}""".stripMargin) + assertFalse(AvroUtils.isNullable(record.getField("req"))) + assertTrue(AvroUtils.isNullable(record.getField("opt"))) + // A union without a NULL branch is not nullable. + assertFalse(AvroUtils.isNullable(record.getField("uni"))) + } + + @Test + def testSchemaHelperRejectsNonRecordSchema(): Unit = { + val ex = assertThrows(classOf[IncompatibleSchemaException], + () => new AvroUtils.AvroSchemaHelper( + Schema.create(Schema.Type.INT), new StructType(), Seq.empty, Seq.empty, false)) + assertTrue(ex.getMessage.contains("as a RECORD")) + } + + @Test + def testMatchedFieldsAndGetAvroFieldByName(): Unit = { + val avro = parse( + """{"type":"record","name":"r","fields":[ + | {"name":"id","type":"int"}, + | {"name":"name","type":["null","string"]} + |]}""".stripMargin) + val catalyst = new StructType().add("id", IntegerType).add("name", StringType) + val helper = new AvroUtils.AvroSchemaHelper(avro, catalyst, Seq.empty, Seq.empty, false) + + assertEquals(2, helper.matchedFields.size) + assertEquals("id", helper.getAvroField("id", 0).get.name()) + assertTrue(helper.getAvroField("missing", 5).isEmpty) + } + + @Test + def testGetAvroFieldPositional(): Unit = { + val avro = parse( + """{"type":"record","name":"r","fields":[{"name":"only","type":"int"}]}""") + val helper = new AvroUtils.AvroSchemaHelper(avro, new StructType(), Seq.empty, Seq.empty, true) + // Positional matching ignores the name and selects by index. + assertEquals("only", helper.getAvroField("anything", 0).get.name()) + assertTrue(helper.getAvroField("anything", 1).isEmpty) + } + + @Test + def testValidateNoExtraCatalystFieldsByName(): Unit = { + val avro = parse( + """{"type":"record","name":"r","fields":[{"name":"id","type":"int"}]}""") + // A nullable Catalyst field with no Avro counterpart. + val catalyst = new StructType().add("id", IntegerType).add("extra", StringType, nullable = true) + val helper = new AvroUtils.AvroSchemaHelper(avro, catalyst, Seq.empty, Seq.empty, false) + + val ex = assertThrows(classOf[IncompatibleSchemaException], + () => helper.validateNoExtraCatalystFields(ignoreNullable = false)) + assertTrue(ex.getMessage.contains("Cannot find field 'extra' in Avro schema")) + + // When nullable Catalyst fields are ignored, the same extra field is tolerated. + helper.validateNoExtraCatalystFields(ignoreNullable = true) Review Comment: The only production call passes `ignoreNullable = true` (`AvroDeserializer.scala:410` in `hudi-spark3-common`; same in the 4.x copies), and in that mode the case that must still throw is a non-nullable extra Catalyst field. Both assertions here use a nullable extra, so a mutation dropping `|| !sqlField.nullable` from the condition in `AvroUtils.validateNoExtraCatalystFields` keeps this test green while the deserializer silently accepts a required Catalyst field with no Avro counterpart. Please add the production-shaped case: ```suggestion helper.validateNoExtraCatalystFields(ignoreNullable = true) // A non-nullable extra Catalyst field must still throw even when nullables are ignored. val strictCatalyst = new StructType().add("id", IntegerType).add("extra", StringType, nullable = false) val strictHelper = new AvroUtils.AvroSchemaHelper(avro, strictCatalyst, Seq.empty, Seq.empty, false) val ex2 = assertThrows(classOf[IncompatibleSchemaException], () => strictHelper.validateNoExtraCatalystFields(ignoreNullable = true)) assertTrue(ex2.getMessage.contains("Cannot find field 'extra' in Avro schema")) ``` ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/avro/TestAvroUtils.scala: ########## @@ -0,0 +1,171 @@ +/* + * 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.avro + +import org.apache.avro.Schema +import org.apache.spark.sql.types._ +import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertThrows, assertTrue} +import org.junit.jupiter.api.Test + +/** + * Direct coverage for [[AvroUtils]] and its [[AvroUtils.AvroSchemaHelper]], which are otherwise only + * exercised indirectly through the Avro serializers. Focuses on the type-support predicate and the + * schema-matching/validation error paths (extra Catalyst fields, extra required Avro fields, + * positional vs by-name matching, ambiguous by-name lookup), pinning the raised exception messages. + */ +class TestAvroUtils { + + private def parse(json: String): Schema = new Schema.Parser().parse(json) + + @Test + def testSupportsDataType(): Unit = { + assertTrue(AvroUtils.supportsDataType(IntegerType)) + assertTrue(AvroUtils.supportsDataType(StringType)) + assertTrue(AvroUtils.supportsDataType(NullType)) + assertTrue(AvroUtils.supportsDataType(ArrayType(LongType))) + assertTrue(AvroUtils.supportsDataType(MapType(StringType, IntegerType))) + assertTrue(AvroUtils.supportsDataType( + new StructType().add("a", IntegerType).add("b", ArrayType(StringType)))) + // CalendarInterval is not representable in Avro, so every wrapper around it is unsupported too. + assertFalse(AvroUtils.supportsDataType(CalendarIntervalType)) + assertFalse(AvroUtils.supportsDataType(ArrayType(CalendarIntervalType))) + assertFalse(AvroUtils.supportsDataType(MapType(StringType, CalendarIntervalType))) + assertFalse(AvroUtils.supportsDataType(new StructType().add("a", CalendarIntervalType))) + } + + @Test + def testToFieldStr(): Unit = { + assertEquals("top-level record", AvroUtils.toFieldStr(Seq.empty)) + assertEquals("field 'foo'", AvroUtils.toFieldStr(Seq("foo"))) + assertEquals("field 'foo.bar'", AvroUtils.toFieldStr(Seq("foo", "bar"))) + } + + @Test + def testIsNullable(): Unit = { + val record = parse( + """{"type":"record","name":"r","fields":[ + | {"name":"req","type":"int"}, + | {"name":"opt","type":["null","int"]}, + | {"name":"uni","type":["int","long"]} + |]}""".stripMargin) + assertFalse(AvroUtils.isNullable(record.getField("req"))) + assertTrue(AvroUtils.isNullable(record.getField("opt"))) + // A union without a NULL branch is not nullable. + assertFalse(AvroUtils.isNullable(record.getField("uni"))) + } + + @Test + def testSchemaHelperRejectsNonRecordSchema(): Unit = { + val ex = assertThrows(classOf[IncompatibleSchemaException], + () => new AvroUtils.AvroSchemaHelper( + Schema.create(Schema.Type.INT), new StructType(), Seq.empty, Seq.empty, false)) + assertTrue(ex.getMessage.contains("as a RECORD")) + } + + @Test + def testMatchedFieldsAndGetAvroFieldByName(): Unit = { + val avro = parse( + """{"type":"record","name":"r","fields":[ + | {"name":"id","type":"int"}, + | {"name":"name","type":["null","string"]} + |]}""".stripMargin) + val catalyst = new StructType().add("id", IntegerType).add("name", StringType) + val helper = new AvroUtils.AvroSchemaHelper(avro, catalyst, Seq.empty, Seq.empty, false) + + assertEquals(2, helper.matchedFields.size) + assertEquals("id", helper.getAvroField("id", 0).get.name()) + assertTrue(helper.getAvroField("missing", 5).isEmpty) + } + + @Test + def testGetAvroFieldPositional(): Unit = { + val avro = parse( + """{"type":"record","name":"r","fields":[{"name":"only","type":"int"}]}""") + val helper = new AvroUtils.AvroSchemaHelper(avro, new StructType(), Seq.empty, Seq.empty, true) + // Positional matching ignores the name and selects by index. + assertEquals("only", helper.getAvroField("anything", 0).get.name()) + assertTrue(helper.getAvroField("anything", 1).isEmpty) + } + + @Test + def testValidateNoExtraCatalystFieldsByName(): Unit = { + val avro = parse( + """{"type":"record","name":"r","fields":[{"name":"id","type":"int"}]}""") + // A nullable Catalyst field with no Avro counterpart. + val catalyst = new StructType().add("id", IntegerType).add("extra", StringType, nullable = true) + val helper = new AvroUtils.AvroSchemaHelper(avro, catalyst, Seq.empty, Seq.empty, false) + + val ex = assertThrows(classOf[IncompatibleSchemaException], + () => helper.validateNoExtraCatalystFields(ignoreNullable = false)) + assertTrue(ex.getMessage.contains("Cannot find field 'extra' in Avro schema")) + + // When nullable Catalyst fields are ignored, the same extra field is tolerated. + helper.validateNoExtraCatalystFields(ignoreNullable = true) + } + + @Test + def testValidateNoExtraCatalystFieldsPositional(): Unit = { + val avro = parse( + """{"type":"record","name":"r","fields":[{"name":"id","type":"int"}]}""") + val catalyst = new StructType().add("id", IntegerType).add("second", IntegerType) + val helper = new AvroUtils.AvroSchemaHelper(avro, catalyst, Seq.empty, Seq.empty, true) + + val ex = assertThrows(classOf[IncompatibleSchemaException], + () => helper.validateNoExtraCatalystFields(ignoreNullable = false)) + assertTrue(ex.getMessage.contains("Cannot find field at position 1")) + } + + @Test + def testValidateNoExtraRequiredAvroFields(): Unit = { + val avroWithRequiredGhost = parse( + """{"type":"record","name":"r","fields":[ + | {"name":"id","type":"int"}, + | {"name":"ghost","type":"int"} + |]}""".stripMargin) + val catalyst = new StructType().add("id", IntegerType) + val helper = new AvroUtils.AvroSchemaHelper( + avroWithRequiredGhost, catalyst, Seq.empty, Seq.empty, false) + val ex = assertThrows(classOf[IncompatibleSchemaException], + () => helper.validateNoExtraRequiredAvroFields()) + assertTrue(ex.getMessage.contains("Found field 'ghost'")) + + // A nullable extra Avro field is not required, so it is tolerated. + val avroWithOptionalGhost = parse( + """{"type":"record","name":"r","fields":[ + | {"name":"id","type":"int"}, + | {"name":"ghost","type":["null","int"]} + |]}""".stripMargin) + val helperWithOptionalGhost = new AvroUtils.AvroSchemaHelper( + avroWithOptionalGhost, catalyst, Seq.empty, Seq.empty, false) + helperWithOptionalGhost.validateNoExtraRequiredAvroFields() Review Comment: nit, optional: this bare call (and the one at line 118) is an implicit must-not-throw assertion. Wrap it so an intentional behavior change reads as a failed assertion instead of a test error. Repo pattern (the SAM overload is ambiguous in Scala 2.12, hence the explicit `Executable` -- see `TestIncrementalQueryWithArchivedInstants.scala:95`): ```scala assertDoesNotThrow(new Executable { def execute(): Unit = helperWithOptionalGhost.validateNoExtraRequiredAvroFields() }) ``` ########## hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/avro/TestAvroUtils.scala: ########## @@ -0,0 +1,171 @@ +/* + * 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.avro + +import org.apache.avro.Schema +import org.apache.spark.sql.types._ +import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertThrows, assertTrue} +import org.junit.jupiter.api.Test + +/** + * Direct coverage for [[AvroUtils]] and its [[AvroUtils.AvroSchemaHelper]], which are otherwise only + * exercised indirectly through the Avro serializers. Focuses on the type-support predicate and the + * schema-matching/validation error paths (extra Catalyst fields, extra required Avro fields, + * positional vs by-name matching, ambiguous by-name lookup), pinning the raised exception messages. + */ +class TestAvroUtils { + + private def parse(json: String): Schema = new Schema.Parser().parse(json) + + @Test + def testSupportsDataType(): Unit = { + assertTrue(AvroUtils.supportsDataType(IntegerType)) + assertTrue(AvroUtils.supportsDataType(StringType)) + assertTrue(AvroUtils.supportsDataType(NullType)) + assertTrue(AvroUtils.supportsDataType(ArrayType(LongType))) + assertTrue(AvroUtils.supportsDataType(MapType(StringType, IntegerType))) + assertTrue(AvroUtils.supportsDataType( + new StructType().add("a", IntegerType).add("b", ArrayType(StringType)))) + // CalendarInterval is not representable in Avro, so every wrapper around it is unsupported too. + assertFalse(AvroUtils.supportsDataType(CalendarIntervalType)) + assertFalse(AvroUtils.supportsDataType(ArrayType(CalendarIntervalType))) + assertFalse(AvroUtils.supportsDataType(MapType(StringType, CalendarIntervalType))) + assertFalse(AvroUtils.supportsDataType(new StructType().add("a", CalendarIntervalType))) + } + + @Test + def testToFieldStr(): Unit = { + assertEquals("top-level record", AvroUtils.toFieldStr(Seq.empty)) + assertEquals("field 'foo'", AvroUtils.toFieldStr(Seq("foo"))) + assertEquals("field 'foo.bar'", AvroUtils.toFieldStr(Seq("foo", "bar"))) + } + + @Test + def testIsNullable(): Unit = { + val record = parse( + """{"type":"record","name":"r","fields":[ + | {"name":"req","type":"int"}, + | {"name":"opt","type":["null","int"]}, + | {"name":"uni","type":["int","long"]} + |]}""".stripMargin) + assertFalse(AvroUtils.isNullable(record.getField("req"))) + assertTrue(AvroUtils.isNullable(record.getField("opt"))) + // A union without a NULL branch is not nullable. + assertFalse(AvroUtils.isNullable(record.getField("uni"))) + } + + @Test + def testSchemaHelperRejectsNonRecordSchema(): Unit = { + val ex = assertThrows(classOf[IncompatibleSchemaException], + () => new AvroUtils.AvroSchemaHelper( + Schema.create(Schema.Type.INT), new StructType(), Seq.empty, Seq.empty, false)) + assertTrue(ex.getMessage.contains("as a RECORD")) + } + + @Test + def testMatchedFieldsAndGetAvroFieldByName(): Unit = { + val avro = parse( + """{"type":"record","name":"r","fields":[ + | {"name":"id","type":"int"}, + | {"name":"name","type":["null","string"]} + |]}""".stripMargin) + val catalyst = new StructType().add("id", IntegerType).add("name", StringType) + val helper = new AvroUtils.AvroSchemaHelper(avro, catalyst, Seq.empty, Seq.empty, false) + + assertEquals(2, helper.matchedFields.size) + assertEquals("id", helper.getAvroField("id", 0).get.name()) + assertTrue(helper.getAvroField("missing", 5).isEmpty) + } + + @Test + def testGetAvroFieldPositional(): Unit = { + val avro = parse( + """{"type":"record","name":"r","fields":[{"name":"only","type":"int"}]}""") + val helper = new AvroUtils.AvroSchemaHelper(avro, new StructType(), Seq.empty, Seq.empty, true) + // Positional matching ignores the name and selects by index. + assertEquals("only", helper.getAvroField("anything", 0).get.name()) + assertTrue(helper.getAvroField("anything", 1).isEmpty) + } + + @Test + def testValidateNoExtraCatalystFieldsByName(): Unit = { + val avro = parse( + """{"type":"record","name":"r","fields":[{"name":"id","type":"int"}]}""") + // A nullable Catalyst field with no Avro counterpart. + val catalyst = new StructType().add("id", IntegerType).add("extra", StringType, nullable = true) + val helper = new AvroUtils.AvroSchemaHelper(avro, catalyst, Seq.empty, Seq.empty, false) + + val ex = assertThrows(classOf[IncompatibleSchemaException], + () => helper.validateNoExtraCatalystFields(ignoreNullable = false)) + assertTrue(ex.getMessage.contains("Cannot find field 'extra' in Avro schema")) + + // When nullable Catalyst fields are ignored, the same extra field is tolerated. + helper.validateNoExtraCatalystFields(ignoreNullable = true) + } + + @Test + def testValidateNoExtraCatalystFieldsPositional(): Unit = { + val avro = parse( + """{"type":"record","name":"r","fields":[{"name":"id","type":"int"}]}""") + val catalyst = new StructType().add("id", IntegerType).add("second", IntegerType) + val helper = new AvroUtils.AvroSchemaHelper(avro, catalyst, Seq.empty, Seq.empty, true) + + val ex = assertThrows(classOf[IncompatibleSchemaException], + () => helper.validateNoExtraCatalystFields(ignoreNullable = false)) + assertTrue(ex.getMessage.contains("Cannot find field at position 1")) + } + + @Test + def testValidateNoExtraRequiredAvroFields(): Unit = { + val avroWithRequiredGhost = parse( + """{"type":"record","name":"r","fields":[ + | {"name":"id","type":"int"}, + | {"name":"ghost","type":"int"} + |]}""".stripMargin) + val catalyst = new StructType().add("id", IntegerType) + val helper = new AvroUtils.AvroSchemaHelper( + avroWithRequiredGhost, catalyst, Seq.empty, Seq.empty, false) + val ex = assertThrows(classOf[IncompatibleSchemaException], + () => helper.validateNoExtraRequiredAvroFields()) + assertTrue(ex.getMessage.contains("Found field 'ghost'")) + + // A nullable extra Avro field is not required, so it is tolerated. + val avroWithOptionalGhost = parse( + """{"type":"record","name":"r","fields":[ + | {"name":"id","type":"int"}, + | {"name":"ghost","type":["null","int"]} + |]}""".stripMargin) + val helperWithOptionalGhost = new AvroUtils.AvroSchemaHelper( + avroWithOptionalGhost, catalyst, Seq.empty, Seq.empty, false) + helperWithOptionalGhost.validateNoExtraRequiredAvroFields() + } + + @Test + def testGetFieldByNameAmbiguousMatch(): Unit = { Review Comment: nit: the resolver comes from ambient `SQLConf.get` (`AvroUtils.getFieldByName`), and nothing pins `spark.sql.caseSensitive=false`, so this assertion depends on the JVM/thread default. The case-sensitive half of the branch (same two fields, exactly one match, no throw) is also untested. Pin the conf and close the branch in one move: ```scala val conf = SQLConf.get conf.setConfString("spark.sql.caseSensitive", "true") try { assertEquals("id", helper.getFieldByName("id").get.name()) } finally { conf.unsetConf("spark.sql.caseSensitive") } ``` -- 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]
