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 <[email protected]>
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
- }
- }
- }
}
