diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestCatalystExpressionOrderPreserving.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestCatalystExpressionOrderPreserving.scala
new file mode 100644
index 0000000000000..49cd4ec7c8f66
--- /dev/null
+++ b/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)))
+  }
+
+  @Test
+  def testNestedComposition(): Unit = {
+    assertEquals(Some(intAttr), matched(Add(Multiply(intAttr, Literal(2)), Literal(3))))
+  }
+
+  @Test
+  def testNonOrderPreservingExpressionsDoNotMatch(): Unit = {
+    // A bare literal carries no attribute.
+    assertEquals(None, matched(Literal(5)))
+    // No attribute on either operand.
+    assertEquals(None, matched(Add(Literal(1), Literal(2))))
+    // Sqrt is not one of the whitelisted order-preserving transformations.
+    assertEquals(None, matched(Sqrt(dblAttr)))
+  }
+}
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/TestFileFormatUtilsForFileGroupReader.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/TestFileFormatUtilsForFileGroupReader.scala
new file mode 100644
index 0000000000000..7de9e9a9f38f4
--- /dev/null
+++ b/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"))))
+  }
+
+  @Test
+  def testNoFiltersReturnsPlanUnchanged(): Unit = {
+    val result = FileFormatUtilsForFileGroupReader.applyFiltersToPlan(
+      plan, tableSchema, resolved, Seq.empty)
+    assertSame(plan, result)
+  }
+}
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/avro/TestAvroUtils.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/avro/TestAvroUtils.scala
new file mode 100644
index 0000000000000..deab23aec1cbf
--- /dev/null
+++ b/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 = {
+    // Two fields differing only by case collide under the default case-insensitive resolver.
+    val avro = parse(
+      """{"type":"record","name":"r","fields":[
+        |  {"name":"ID","type":"int"},
+        |  {"name":"id","type":"int"}
+        |]}""".stripMargin)
+    val helper = new AvroUtils.AvroSchemaHelper(avro, new StructType(), Seq.empty, Seq.empty, false)
+    val ex = assertThrows(classOf[IncompatibleSchemaException],
+      () => helper.getFieldByName("id"))
+    assertTrue(ex.getMessage.contains("gave 2 matches"))
+  }
+}