You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by vh...@apache.org on 2026/09/14 09:16:16 UTC
(hudi) branch master updated: refactor(spark): share parquet filter rebuild helper (#19891)
This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 8ea7aaa7b485 refactor(spark): share parquet filter rebuild helper (#19891)
8ea7aaa7b485 is described below
commit 8ea7aaa7b4857504fd0c0dc146dc4a2886598af4
Author: voonhous <vo...@gmail.com>
AuthorDate: Mon Sep 14 17:16:09 2026 +0800
refactor(spark): share parquet filter rebuild helper (#19891)
* refactor(spark): share parquet filter rebuild helper
The recursive walk that re-spells a pushed-down Spark filter onto the
column names an evolved parquet file carries existed five times: as a
private helper inside the ParquetSchemaEvolutionUtils class, and byte
for byte in the companion object of each of the Spark3, Spark40,
Spark41 and Spark42 LegacyHoodieParquetFileFormats.
Move the 3-arg helper into the ParquetSchemaEvolutionUtils companion
object as a public function and delete the four copies, which now call
it directly. The 1-arg instance method keeps its signature and
delegates to the companion, so the SparkNNParquetReader family is
unchanged. The body is verbatim, so no behaviour changes; every
version module already depended on hudi-spark-common.
Drop the imports the deletion leaves unused in the four formats and
add one TestParquetSchemaEvolutionUtils case for the now-public
function: a rename is re-spelled, a column absent from the file
collapses to AlwaysTrue, And/Or/Not recurse, and a null schema on
either side passes the filter through.
Exercise both the re-spelled and the absent-column arm for each leaf
filter type, and the AlwaysTrue/AlwaysFalse passthrough, in
rebuildFilterFromParquet.
---
.../parquet/ParquetSchemaEvolutionUtils.scala | 122 +++++++++++----------
.../parquet/TestParquetSchemaEvolutionUtils.scala | 63 +++++++++++
.../Spark3LegacyHoodieParquetFileFormat.scala | 64 +----------
.../Spark40LegacyHoodieParquetFileFormat.scala | 64 +----------
.../Spark41LegacyHoodieParquetFileFormat.scala | 64 +----------
.../Spark42LegacyHoodieParquetFileFormat.scala | 64 +----------
6 files changed, 139 insertions(+), 302 deletions(-)
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala
index ad67f5cd007c..9df771b115e6 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala
@@ -75,64 +75,7 @@ class ParquetSchemaEvolutionUtils(sharedConf: Configuration,
}
def rebuildFilterFromParquet(filter: Filter): Filter = {
- rebuildFilterFromParquetHelper(filter, fileSchema, querySchemaOption.orElse(null))
- }
-
- private def rebuildFilterFromParquetHelper(oldFilter: Filter, fileSchema: InternalSchema, querySchema: InternalSchema): Filter = {
- if (fileSchema == null || querySchema == null) {
- oldFilter
- } else {
- oldFilter match {
- case eq: EqualTo =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(eq.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else eq.copy(attribute = newAttribute)
- case eqs: EqualNullSafe =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(eqs.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else eqs.copy(attribute = newAttribute)
- case gt: GreaterThan =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(gt.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else gt.copy(attribute = newAttribute)
- case gtr: GreaterThanOrEqual =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(gtr.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else gtr.copy(attribute = newAttribute)
- case lt: LessThan =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(lt.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else lt.copy(attribute = newAttribute)
- case lte: LessThanOrEqual =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(lte.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else lte.copy(attribute = newAttribute)
- case i: In =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(i.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else i.copy(attribute = newAttribute)
- case isn: IsNull =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(isn.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else isn.copy(attribute = newAttribute)
- case isnn: IsNotNull =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(isnn.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else isnn.copy(attribute = newAttribute)
- case And(left, right) =>
- And(rebuildFilterFromParquetHelper(left, fileSchema, querySchema), rebuildFilterFromParquetHelper(right, fileSchema, querySchema))
- case Or(left, right) =>
- Or(rebuildFilterFromParquetHelper(left, fileSchema, querySchema), rebuildFilterFromParquetHelper(right, fileSchema, querySchema))
- case Not(child) =>
- Not(rebuildFilterFromParquetHelper(child, fileSchema, querySchema))
- case ssw: StringStartsWith =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(ssw.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else ssw.copy(attribute = newAttribute)
- case ses: StringEndsWith =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(ses.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else ses.copy(attribute = newAttribute)
- case sc: StringContains =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(sc.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else sc.copy(attribute = newAttribute)
- case AlwaysTrue =>
- AlwaysTrue
- case AlwaysFalse =>
- AlwaysFalse
- case _ =>
- AlwaysTrue
- }
- }
+ ParquetSchemaEvolutionUtils.rebuildFilterFromParquet(filter, fileSchema, querySchemaOption.orElse(null))
}
protected var typeChangeInfos: java.util.Map[Integer, Pair[DataType, DataType]] = null
@@ -215,6 +158,69 @@ object ParquetSchemaEvolutionUtils {
}
}
+ /**
+ * Maps a pushed-down query filter onto the names the file actually carries: a column renamed
+ * under schema-on-read is rewritten to its file-schema name, and one the file does not hold at
+ * all collapses to AlwaysTrue, since a filter on an absent column cannot skip any of its row
+ * groups. A table without an internal schema passes its filters through untouched.
+ */
+ def rebuildFilterFromParquet(oldFilter: Filter, fileSchema: InternalSchema, querySchema: InternalSchema): Filter = {
+ if (fileSchema == null || querySchema == null) {
+ oldFilter
+ } else {
+ oldFilter match {
+ case eq: EqualTo =>
+ val newAttribute = InternalSchemaUtils.reBuildFilterName(eq.attribute, fileSchema, querySchema)
+ if (newAttribute.isEmpty) AlwaysTrue else eq.copy(attribute = newAttribute)
+ case eqs: EqualNullSafe =>
+ val newAttribute = InternalSchemaUtils.reBuildFilterName(eqs.attribute, fileSchema, querySchema)
+ if (newAttribute.isEmpty) AlwaysTrue else eqs.copy(attribute = newAttribute)
+ case gt: GreaterThan =>
+ val newAttribute = InternalSchemaUtils.reBuildFilterName(gt.attribute, fileSchema, querySchema)
+ if (newAttribute.isEmpty) AlwaysTrue else gt.copy(attribute = newAttribute)
+ case gtr: GreaterThanOrEqual =>
+ val newAttribute = InternalSchemaUtils.reBuildFilterName(gtr.attribute, fileSchema, querySchema)
+ if (newAttribute.isEmpty) AlwaysTrue else gtr.copy(attribute = newAttribute)
+ case lt: LessThan =>
+ val newAttribute = InternalSchemaUtils.reBuildFilterName(lt.attribute, fileSchema, querySchema)
+ if (newAttribute.isEmpty) AlwaysTrue else lt.copy(attribute = newAttribute)
+ case lte: LessThanOrEqual =>
+ val newAttribute = InternalSchemaUtils.reBuildFilterName(lte.attribute, fileSchema, querySchema)
+ if (newAttribute.isEmpty) AlwaysTrue else lte.copy(attribute = newAttribute)
+ case i: In =>
+ val newAttribute = InternalSchemaUtils.reBuildFilterName(i.attribute, fileSchema, querySchema)
+ if (newAttribute.isEmpty) AlwaysTrue else i.copy(attribute = newAttribute)
+ case isn: IsNull =>
+ val newAttribute = InternalSchemaUtils.reBuildFilterName(isn.attribute, fileSchema, querySchema)
+ if (newAttribute.isEmpty) AlwaysTrue else isn.copy(attribute = newAttribute)
+ case isnn: IsNotNull =>
+ val newAttribute = InternalSchemaUtils.reBuildFilterName(isnn.attribute, fileSchema, querySchema)
+ if (newAttribute.isEmpty) AlwaysTrue else isnn.copy(attribute = newAttribute)
+ case And(left, right) =>
+ And(rebuildFilterFromParquet(left, fileSchema, querySchema), rebuildFilterFromParquet(right, fileSchema, querySchema))
+ case Or(left, right) =>
+ Or(rebuildFilterFromParquet(left, fileSchema, querySchema), rebuildFilterFromParquet(right, fileSchema, querySchema))
+ case Not(child) =>
+ Not(rebuildFilterFromParquet(child, fileSchema, querySchema))
+ case ssw: StringStartsWith =>
+ val newAttribute = InternalSchemaUtils.reBuildFilterName(ssw.attribute, fileSchema, querySchema)
+ if (newAttribute.isEmpty) AlwaysTrue else ssw.copy(attribute = newAttribute)
+ case ses: StringEndsWith =>
+ val newAttribute = InternalSchemaUtils.reBuildFilterName(ses.attribute, fileSchema, querySchema)
+ if (newAttribute.isEmpty) AlwaysTrue else ses.copy(attribute = newAttribute)
+ case sc: StringContains =>
+ val newAttribute = InternalSchemaUtils.reBuildFilterName(sc.attribute, fileSchema, querySchema)
+ if (newAttribute.isEmpty) AlwaysTrue else sc.copy(attribute = newAttribute)
+ case AlwaysTrue =>
+ AlwaysTrue
+ case AlwaysFalse =>
+ AlwaysFalse
+ case _ =>
+ AlwaysTrue
+ }
+ }
+ }
+
/**
* Fails fast when schema-on-read meets a shredded variant file. The internal schema models a
* variant as a two-field {metadata, value} record (with sentinel negative field ids, see
diff --git a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala
index 4321728c4cc3..0df5f28af986 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala
@@ -28,6 +28,7 @@ import org.apache.parquet.hadoop.metadata.FileMetaData
import org.apache.parquet.schema.{MessageType, Type, Types}
import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
import org.apache.spark.sql.execution.datasources.parquet.VariantParquetTestFixtures.{shreddedVariant, stringKeyMap, threeLevelList, twoLevelList, unshreddedVariant}
+import org.apache.spark.sql.sources.{AlwaysFalse, AlwaysTrue, And, EqualNullSafe, EqualTo, Filter, GreaterThan, GreaterThanOrEqual, In, IsNotNull, IsNull, LessThan, LessThanOrEqual, Not, Or, StringContains, StringEndsWith, StringStartsWith}
import org.apache.spark.sql.types.{ArrayType, BinaryType, IntegerType, MapType, MetadataBuilder, StringType, StructField, StructType}
import org.junit.jupiter.api.{Assertions, Test}
@@ -328,4 +329,66 @@ class TestParquetSchemaEvolutionUtils {
*/
private def nestedVariantElement: Type =
Types.optionalGroup().addField(shreddedVariant("inner")).named("e")
+
+ /**
+ * The filter rebuild the four legacy parquet formats share with the SparkNNParquetReader family.
+ * Field ids are positional, so the file's "original" carries the same id 1 the query spells
+ * "renamed", and "added" (id 2) has no field in the file at all.
+ */
+ @Test
+ def testRebuildFilterFromParquetRespellsFiltersOntoFileNames(): Unit = {
+ val intType = HoodieSchema.create(HoodieSchemaType.INT)
+ val stringType = HoodieSchema.create(HoodieSchemaType.STRING)
+ val querySchema = internalSchemaOf(("id", intType), ("renamed", stringType), ("added", intType))
+ val fileSchema = internalSchemaOf(("id", intType), ("original", stringType))
+
+ def rebuild(filter: Filter): Filter =
+ ParquetSchemaEvolutionUtils.rebuildFilterFromParquet(filter, fileSchema, querySchema)
+
+ // A renamed column is re-spelled to the name the file carries; an untouched one is left alone.
+ Assertions.assertEquals(EqualTo("original", "x"), rebuild(EqualTo("renamed", "x")))
+ Assertions.assertEquals(StringStartsWith("original", "a"), rebuild(StringStartsWith("renamed", "a")))
+ Assertions.assertEquals(GreaterThan("id", 1), rebuild(GreaterThan("id", 1)))
+
+ // A column added after the file was written has no field to filter on, and a filter that
+ // cannot be evaluated must not skip any of the file's row groups.
+ Assertions.assertEquals(AlwaysTrue, rebuild(IsNotNull("added")))
+
+ // Every leaf filter type takes both arms: re-spelled when the file holds the column under
+ // another name, AlwaysTrue when it does not hold it at all.
+ val leafFilters: Seq[String => Filter] = Seq(
+ EqualTo(_, "x"), EqualNullSafe(_, "x"), GreaterThan(_, "x"), GreaterThanOrEqual(_, "x"),
+ LessThan(_, "x"), LessThanOrEqual(_, "x"), In(_, Array[Any]("x", "y")), IsNull(_), IsNotNull(_),
+ StringStartsWith(_, "x"), StringEndsWith(_, "x"), StringContains(_, "x"))
+ leafFilters.foreach { leaf =>
+ Assertions.assertEquals(leaf("original"), rebuild(leaf("renamed")))
+ Assertions.assertEquals(AlwaysTrue, rebuild(leaf("added")), s"${leaf("added")} on an absent column")
+ }
+
+ // The constant filters reference no column and pass through as they are.
+ Assertions.assertEquals(AlwaysTrue, rebuild(AlwaysTrue))
+ Assertions.assertEquals(AlwaysFalse, rebuild(AlwaysFalse))
+
+ // And/Or/Not rebuild their children.
+ Assertions.assertEquals(
+ And(EqualTo("original", "x"), AlwaysTrue),
+ rebuild(And(EqualTo("renamed", "x"), IsNull("added"))))
+ Assertions.assertEquals(
+ Or(Not(EqualTo("original", "x")), GreaterThanOrEqual("id", 2)),
+ rebuild(Or(Not(EqualTo("renamed", "x")), GreaterThanOrEqual("id", 2))))
+
+ // A table with no internal schema on either side pushes its filters down untouched.
+ val untouched = EqualTo("renamed", "x")
+ Assertions.assertSame(untouched,
+ ParquetSchemaEvolutionUtils.rebuildFilterFromParquet(untouched, null, querySchema))
+ Assertions.assertSame(untouched,
+ ParquetSchemaEvolutionUtils.rebuildFilterFromParquet(untouched, fileSchema, null))
+ Assertions.assertSame(untouched,
+ ParquetSchemaEvolutionUtils.rebuildFilterFromParquet(untouched, null, null))
+ }
+
+ /** An internal schema over the given top-level columns, in order; field ids are positional. */
+ private def internalSchemaOf(fields: (String, HoodieSchema)*): InternalSchema =
+ InternalSchemaConverter.convert(HoodieSchema.createRecord("query", "org.apache.hudi.test", null,
+ Arrays.asList(fields.map { case (name, schema) => HoodieSchemaField.of(name, schema) }: _*)))
}
diff --git a/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark3LegacyHoodieParquetFileFormat.scala b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark3LegacyHoodieParquetFileFormat.scala
index e84201f1a33c..39833829227f 100644
--- a/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark3LegacyHoodieParquetFileFormat.scala
+++ b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark3LegacyHoodieParquetFileFormat.scala
@@ -19,9 +19,8 @@ package org.apache.spark.sql.execution.datasources.parquet
import org.apache.hudi.client.utils.SparkInternalSchemaConverter
import org.apache.hudi.common.fs.FSUtils
-import org.apache.hudi.common.schema.internal.InternalSchema
import org.apache.hudi.common.schema.internal.action.InternalSchemaMerger
-import org.apache.hudi.common.schema.internal.utils.{InternalSchemaUtils, SerDeHelper}
+import org.apache.hudi.common.schema.internal.utils.SerDeHelper
import org.apache.hudi.common.table.timeline.TimelineLayout
import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion
import org.apache.hudi.common.util.HoodieStorageUtils
@@ -48,7 +47,7 @@ import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.execution.datasources.{DataSourceUtils, PartitionedFile, RecordReaderIterator}
import org.apache.spark.sql.execution.datasources.parquet.Spark3LegacyHoodieParquetFileFormat._
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.sources._
+import org.apache.spark.sql.sources.Filter
import org.apache.spark.sql.types.{AtomicType, DataType, StructField, StructType}
import org.apache.spark.util.SerializableConfiguration
@@ -218,7 +217,7 @@ abstract class Spark3LegacyHoodieParquetFileFormat(shouldAppendPartitionValues:
pushDownInFilterThreshold,
isCaseSensitive,
datetimeRebaseSpec)
- filters.map(rebuildFilterFromParquet(_, fileSchema, querySchemaOption.orElse(null)))
+ filters.map(ParquetSchemaEvolutionUtils.rebuildFilterFromParquet(_, fileSchema, querySchemaOption.orElse(null)))
// Collects all converted Parquet filter predicates. Notice that not all predicates can be
// converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
// is used here.
@@ -434,61 +433,4 @@ object Spark3LegacyHoodieParquetFileFormat {
internalSchemaStr
}
}
-
- private def rebuildFilterFromParquet(oldFilter: Filter, fileSchema: InternalSchema, querySchema: InternalSchema): Filter = {
- if (fileSchema == null || querySchema == null) {
- oldFilter
- } else {
- oldFilter match {
- case eq: EqualTo =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(eq.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else eq.copy(attribute = newAttribute)
- case eqs: EqualNullSafe =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(eqs.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else eqs.copy(attribute = newAttribute)
- case gt: GreaterThan =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(gt.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else gt.copy(attribute = newAttribute)
- case gtr: GreaterThanOrEqual =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(gtr.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else gtr.copy(attribute = newAttribute)
- case lt: LessThan =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(lt.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else lt.copy(attribute = newAttribute)
- case lte: LessThanOrEqual =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(lte.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else lte.copy(attribute = newAttribute)
- case i: In =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(i.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else i.copy(attribute = newAttribute)
- case isn: IsNull =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(isn.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else isn.copy(attribute = newAttribute)
- case isnn: IsNotNull =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(isnn.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else isnn.copy(attribute = newAttribute)
- case And(left, right) =>
- And(rebuildFilterFromParquet(left, fileSchema, querySchema), rebuildFilterFromParquet(right, fileSchema, querySchema))
- case Or(left, right) =>
- Or(rebuildFilterFromParquet(left, fileSchema, querySchema), rebuildFilterFromParquet(right, fileSchema, querySchema))
- case Not(child) =>
- Not(rebuildFilterFromParquet(child, fileSchema, querySchema))
- case ssw: StringStartsWith =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(ssw.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else ssw.copy(attribute = newAttribute)
- case ses: StringEndsWith =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(ses.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else ses.copy(attribute = newAttribute)
- case sc: StringContains =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(sc.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else sc.copy(attribute = newAttribute)
- case AlwaysTrue =>
- AlwaysTrue
- case AlwaysFalse =>
- AlwaysFalse
- case _ =>
- AlwaysTrue
- }
- }
- }
}
diff --git a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark40LegacyHoodieParquetFileFormat.scala b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark40LegacyHoodieParquetFileFormat.scala
index fdf780255a85..a27c66224291 100644
--- a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark40LegacyHoodieParquetFileFormat.scala
+++ b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark40LegacyHoodieParquetFileFormat.scala
@@ -19,9 +19,8 @@ package org.apache.spark.sql.execution.datasources.parquet
import org.apache.hudi.client.utils.SparkInternalSchemaConverter
import org.apache.hudi.common.fs.FSUtils
-import org.apache.hudi.common.schema.internal.InternalSchema
import org.apache.hudi.common.schema.internal.action.InternalSchemaMerger
-import org.apache.hudi.common.schema.internal.utils.{InternalSchemaUtils, SerDeHelper}
+import org.apache.hudi.common.schema.internal.utils.SerDeHelper
import org.apache.hudi.common.table.HoodieTableMetaClient
import org.apache.hudi.common.table.timeline.TimelineLayout
import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion
@@ -50,7 +49,7 @@ import org.apache.spark.sql.execution.WholeStageCodegenExec
import org.apache.spark.sql.execution.datasources.{DataSourceUtils, PartitionedFile, RecordReaderIterator}
import org.apache.spark.sql.execution.datasources.parquet.Spark40LegacyHoodieParquetFileFormat._
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.sources._
+import org.apache.spark.sql.sources.Filter
import org.apache.spark.sql.types.{AtomicType, DataType, StructField, StructType}
import org.apache.spark.util.SerializableConfiguration
@@ -210,7 +209,7 @@ class Spark40LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
pushDownInFilterThreshold,
isCaseSensitive,
datetimeRebaseSpec)
- filters.map(rebuildFilterFromParquet(_, fileSchema, querySchemaOption.orElse(null)))
+ filters.map(ParquetSchemaEvolutionUtils.rebuildFilterFromParquet(_, fileSchema, querySchemaOption.orElse(null)))
// Collects all converted Parquet filter predicates. Notice that not all predicates can be
// converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
// is used here.
@@ -409,61 +408,4 @@ object Spark40LegacyHoodieParquetFileFormat {
internalSchemaStr
}
}
-
- private def rebuildFilterFromParquet(oldFilter: Filter, fileSchema: InternalSchema, querySchema: InternalSchema): Filter = {
- if (fileSchema == null || querySchema == null) {
- oldFilter
- } else {
- oldFilter match {
- case eq: EqualTo =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(eq.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else eq.copy(attribute = newAttribute)
- case eqs: EqualNullSafe =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(eqs.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else eqs.copy(attribute = newAttribute)
- case gt: GreaterThan =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(gt.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else gt.copy(attribute = newAttribute)
- case gtr: GreaterThanOrEqual =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(gtr.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else gtr.copy(attribute = newAttribute)
- case lt: LessThan =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(lt.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else lt.copy(attribute = newAttribute)
- case lte: LessThanOrEqual =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(lte.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else lte.copy(attribute = newAttribute)
- case i: In =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(i.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else i.copy(attribute = newAttribute)
- case isn: IsNull =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(isn.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else isn.copy(attribute = newAttribute)
- case isnn: IsNotNull =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(isnn.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else isnn.copy(attribute = newAttribute)
- case And(left, right) =>
- And(rebuildFilterFromParquet(left, fileSchema, querySchema), rebuildFilterFromParquet(right, fileSchema, querySchema))
- case Or(left, right) =>
- Or(rebuildFilterFromParquet(left, fileSchema, querySchema), rebuildFilterFromParquet(right, fileSchema, querySchema))
- case Not(child) =>
- Not(rebuildFilterFromParquet(child, fileSchema, querySchema))
- case ssw: StringStartsWith =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(ssw.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else ssw.copy(attribute = newAttribute)
- case ses: StringEndsWith =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(ses.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else ses.copy(attribute = newAttribute)
- case sc: StringContains =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(sc.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else sc.copy(attribute = newAttribute)
- case AlwaysTrue =>
- AlwaysTrue
- case AlwaysFalse =>
- AlwaysFalse
- case _ =>
- AlwaysTrue
- }
- }
- }
}
diff --git a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark41LegacyHoodieParquetFileFormat.scala b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark41LegacyHoodieParquetFileFormat.scala
index 9c25fddbb524..372a78b386b5 100644
--- a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark41LegacyHoodieParquetFileFormat.scala
+++ b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark41LegacyHoodieParquetFileFormat.scala
@@ -19,9 +19,8 @@ package org.apache.spark.sql.execution.datasources.parquet
import org.apache.hudi.client.utils.SparkInternalSchemaConverter
import org.apache.hudi.common.fs.FSUtils
-import org.apache.hudi.common.schema.internal.InternalSchema
import org.apache.hudi.common.schema.internal.action.InternalSchemaMerger
-import org.apache.hudi.common.schema.internal.utils.{InternalSchemaUtils, SerDeHelper}
+import org.apache.hudi.common.schema.internal.utils.SerDeHelper
import org.apache.hudi.common.table.timeline.TimelineLayout
import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion
import org.apache.hudi.common.util.HoodieStorageUtils
@@ -48,7 +47,7 @@ import org.apache.spark.sql.execution.WholeStageCodegenExec
import org.apache.spark.sql.execution.datasources.{DataSourceUtils, PartitionedFile, RecordReaderIterator}
import org.apache.spark.sql.execution.datasources.parquet.Spark41LegacyHoodieParquetFileFormat._
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.sources._
+import org.apache.spark.sql.sources.Filter
import org.apache.spark.sql.types.{AtomicType, DataType, StructField, StructType}
import org.apache.spark.util.{SerializableConfiguration, Utils}
@@ -213,7 +212,7 @@ class Spark41LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
pushDownInFilterThreshold,
isCaseSensitive,
datetimeRebaseSpec)
- filters.map(rebuildFilterFromParquet(_, fileSchema, querySchemaOption.orElse(null)))
+ filters.map(ParquetSchemaEvolutionUtils.rebuildFilterFromParquet(_, fileSchema, querySchemaOption.orElse(null)))
// Collects all converted Parquet filter predicates. Notice that not all predicates can be
// converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
// is used here.
@@ -421,61 +420,4 @@ object Spark41LegacyHoodieParquetFileFormat {
internalSchemaStr
}
}
-
- private def rebuildFilterFromParquet(oldFilter: Filter, fileSchema: InternalSchema, querySchema: InternalSchema): Filter = {
- if (fileSchema == null || querySchema == null) {
- oldFilter
- } else {
- oldFilter match {
- case eq: EqualTo =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(eq.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else eq.copy(attribute = newAttribute)
- case eqs: EqualNullSafe =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(eqs.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else eqs.copy(attribute = newAttribute)
- case gt: GreaterThan =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(gt.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else gt.copy(attribute = newAttribute)
- case gtr: GreaterThanOrEqual =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(gtr.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else gtr.copy(attribute = newAttribute)
- case lt: LessThan =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(lt.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else lt.copy(attribute = newAttribute)
- case lte: LessThanOrEqual =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(lte.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else lte.copy(attribute = newAttribute)
- case i: In =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(i.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else i.copy(attribute = newAttribute)
- case isn: IsNull =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(isn.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else isn.copy(attribute = newAttribute)
- case isnn: IsNotNull =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(isnn.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else isnn.copy(attribute = newAttribute)
- case And(left, right) =>
- And(rebuildFilterFromParquet(left, fileSchema, querySchema), rebuildFilterFromParquet(right, fileSchema, querySchema))
- case Or(left, right) =>
- Or(rebuildFilterFromParquet(left, fileSchema, querySchema), rebuildFilterFromParquet(right, fileSchema, querySchema))
- case Not(child) =>
- Not(rebuildFilterFromParquet(child, fileSchema, querySchema))
- case ssw: StringStartsWith =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(ssw.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else ssw.copy(attribute = newAttribute)
- case ses: StringEndsWith =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(ses.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else ses.copy(attribute = newAttribute)
- case sc: StringContains =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(sc.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else sc.copy(attribute = newAttribute)
- case AlwaysTrue =>
- AlwaysTrue
- case AlwaysFalse =>
- AlwaysFalse
- case _ =>
- AlwaysTrue
- }
- }
- }
}
diff --git a/hudi-spark-datasource/hudi-spark4.2.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark42LegacyHoodieParquetFileFormat.scala b/hudi-spark-datasource/hudi-spark4.2.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark42LegacyHoodieParquetFileFormat.scala
index 5eb276b1ac3f..b117d34450b8 100644
--- a/hudi-spark-datasource/hudi-spark4.2.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark42LegacyHoodieParquetFileFormat.scala
+++ b/hudi-spark-datasource/hudi-spark4.2.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark42LegacyHoodieParquetFileFormat.scala
@@ -19,9 +19,8 @@ package org.apache.spark.sql.execution.datasources.parquet
import org.apache.hudi.client.utils.SparkInternalSchemaConverter
import org.apache.hudi.common.fs.FSUtils
-import org.apache.hudi.common.schema.internal.InternalSchema
import org.apache.hudi.common.schema.internal.action.InternalSchemaMerger
-import org.apache.hudi.common.schema.internal.utils.{InternalSchemaUtils, SerDeHelper}
+import org.apache.hudi.common.schema.internal.utils.SerDeHelper
import org.apache.hudi.common.table.timeline.TimelineLayout
import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion
import org.apache.hudi.common.util.HoodieStorageUtils
@@ -48,7 +47,7 @@ import org.apache.spark.sql.execution.WholeStageCodegenExec
import org.apache.spark.sql.execution.datasources.{DataSourceUtils, PartitionedFile, RecordReaderIterator}
import org.apache.spark.sql.execution.datasources.parquet.Spark42LegacyHoodieParquetFileFormat._
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.sources._
+import org.apache.spark.sql.sources.Filter
import org.apache.spark.sql.types.{AtomicType, DataType, StructField, StructType}
import org.apache.spark.util.{SerializableConfiguration, Utils}
@@ -213,7 +212,7 @@ class Spark42LegacyHoodieParquetFileFormat(private val shouldAppendPartitionValu
pushDownInFilterThreshold,
isCaseSensitive,
datetimeRebaseSpec)
- filters.map(rebuildFilterFromParquet(_, fileSchema, querySchemaOption.orElse(null)))
+ filters.map(ParquetSchemaEvolutionUtils.rebuildFilterFromParquet(_, fileSchema, querySchemaOption.orElse(null)))
// Collects all converted Parquet filter predicates. Notice that not all predicates can be
// converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap`
// is used here.
@@ -421,61 +420,4 @@ object Spark42LegacyHoodieParquetFileFormat {
internalSchemaStr
}
}
-
- private def rebuildFilterFromParquet(oldFilter: Filter, fileSchema: InternalSchema, querySchema: InternalSchema): Filter = {
- if (fileSchema == null || querySchema == null) {
- oldFilter
- } else {
- oldFilter match {
- case eq: EqualTo =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(eq.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else eq.copy(attribute = newAttribute)
- case eqs: EqualNullSafe =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(eqs.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else eqs.copy(attribute = newAttribute)
- case gt: GreaterThan =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(gt.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else gt.copy(attribute = newAttribute)
- case gtr: GreaterThanOrEqual =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(gtr.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else gtr.copy(attribute = newAttribute)
- case lt: LessThan =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(lt.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else lt.copy(attribute = newAttribute)
- case lte: LessThanOrEqual =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(lte.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else lte.copy(attribute = newAttribute)
- case i: In =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(i.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else i.copy(attribute = newAttribute)
- case isn: IsNull =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(isn.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else isn.copy(attribute = newAttribute)
- case isnn: IsNotNull =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(isnn.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else isnn.copy(attribute = newAttribute)
- case And(left, right) =>
- And(rebuildFilterFromParquet(left, fileSchema, querySchema), rebuildFilterFromParquet(right, fileSchema, querySchema))
- case Or(left, right) =>
- Or(rebuildFilterFromParquet(left, fileSchema, querySchema), rebuildFilterFromParquet(right, fileSchema, querySchema))
- case Not(child) =>
- Not(rebuildFilterFromParquet(child, fileSchema, querySchema))
- case ssw: StringStartsWith =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(ssw.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else ssw.copy(attribute = newAttribute)
- case ses: StringEndsWith =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(ses.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else ses.copy(attribute = newAttribute)
- case sc: StringContains =>
- val newAttribute = InternalSchemaUtils.reBuildFilterName(sc.attribute, fileSchema, querySchema)
- if (newAttribute.isEmpty) AlwaysTrue else sc.copy(attribute = newAttribute)
- case AlwaysTrue =>
- AlwaysTrue
- case AlwaysFalse =>
- AlwaysFalse
- case _ =>
- AlwaysTrue
- }
- }
- }
}