This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 0ba508d0573776610c0f033054b2075bc01bc2c7 Author: Y Ethan Guo <[email protected]> AuthorDate: Thu Jul 2 22:49:33 2026 -0700 refactor(spark): dedup catalyst utils and analysis rules across Spark version modules (#19149) * refactor(spark): pull shared HoodieCatalystExpressionUtils bodies into a common base class Introduce BaseHoodieCatalystExpressionUtils in hudi-spark-common carrying the method bodies that are byte-identical across Spark 3.3-4.2 (normalizeExprs, extractPredicatesWithinOutputSet, canUpCast, and the order-preserving transformation matcher). The ParseToDate/ParseToTimestamp patterns, whose case-class shapes differ across Spark versions, are kept per version behind the unapplyOrderPreservingDateParsing hook. HoodieSpark4CatalystExpressionUtils in hudi-spark4-common now carries the bodies uniform within the Spark 4.x family (getEncoder, matchCast, unapplyCastExpression, date-parsing hook), turning the Spark 4.x objects into empty declarations. The Spark 3.x objects keep only genuine per-version deltas (RowEncoder vs ExpressionEncoder, AnsiCast, EvalMode). The empty HoodieSpark3CatalystExpressionUtils shim is removed; adapter-facing object names are unchanged. * refactor(spark): implement shared HoodieCatalystPlansUtils bodies in base and family classes Move the method bodies that are byte-identical across all six Spark version modules into the BaseHoodieCatalystPlanUtils trait in hudi-spark-common (unapplyResolvedTable, projectOverSchema, isRepairTable, getRepairTableChildren, the four index-plan matchers, unapplyInsertIntoStatement, and createProjectForByNameQuery, replacing the dead 'None' default that every version overrode identically). Add family-level bases for bodies identical only within a Spark major version: HoodieSpark3CatalystPlanUtils in hudi-spark3-common (unapplyUpdateAction with the 2-field UpdateAction, extractJsonFromSerializedOffset with the pre-4.1 package) and HoodieSpark4CatalystPlanUtils in hudi-spark4-common (7-field MergeIntoTable, 5-field LogicalRelation scan, error-class based failAnalysisForMIT and failTableNotFound). Per-version objects keep only genuine Spark API deltas: pattern arities of MergeIntoTable/ScanOperation/LogicalRelation/UpdateAction, the SerializedOffset package move in Spark 4.1, pre-error-class failure messages on 3.3, the _LEGACY_ERROR_TEMP_2309 error class on 3.4, and the Spark 3.4-only default-columns workaround in unapplyInsertIntoStatement. Adapter-facing object names are unchanged. * refactor(spark): pull shared HoodieSchemaUtils bodies into family base classes Nothing in this family is identical across all six Spark versions, so the bodies are shared at the Spark-major-version level: HoodieSpark3SchemaUtils in hudi-spark3-common carries getSchema (pre-4.0 JdbcUtils.getSchema signature without the Connection argument), and HoodieSpark4SchemaUtils in hudi-spark4-common carries all three methods, which are uniform across 4.0-4.2, turning those objects into empty declarations. Spark 3.x objects keep the per-version deltas: the 3-arg SchemaUtils.checkColumnNameDuplication on 3.3 (colType parameter removed in Spark 3.4) and StructType#toAttributes on 3.3/3.4 (moved to DataTypeUtils in Spark 3.5, SPARK-44353). * refactor(spark): share ResolveColumnsForInsertInto preprocessing across Spark 4.x The three-level preprocess logic (partition-spec normalization, user-specified column projection, output-column resolution with default-value support, and the arity-mismatch error rewrite) is byte-identical across Spark 4.0-4.2, so it moves into the abstract HoodieSpark4ResolveColumnsForInsertInto base class in hudi-spark4-common. The per-version case classes keep only apply(), whose InsertIntoStatement pattern arity differs in Spark 4.2 (two fields added). The DataSourceV2ToV1Fallback rules stay fully per-version: every adjacent pair differs in InsertIntoStatement and/or DataSourceV2Relation case-class shapes, so no clean seam exists within the family. Spark 3.5 keeps its full copy of ResolveColumnsForInsertInto since hudi-spark3-common also compiles against Spark 3.3/3.4, which lack the required APIs. Rule class names are unchanged (HoodieAnalysis instantiates them reflectively by name). (cherry picked from commit b933fab87559614c3548ee6909dfc1999b3b6b80) --- .../sql/BaseHoodieCatalystExpressionUtils.scala} | 49 ++++---- .../spark/sql/BaseHoodieCatalystPlanUtils.scala | 82 +++++++++++++- .../sql/HoodieSpark3CatalystExpressionUtils.scala | 37 ------- .../spark/sql/HoodieSpark3CatalystPlanUtils.scala | 44 ++++++++ .../spark/sql/HoodieSpark3SchemaUtils.scala} | 16 +-- .../sql/HoodieSpark33CatalystExpressionUtils.scala | 81 ++------------ .../spark/sql/HoodieSpark33CatalystPlanUtils.scala | 103 +---------------- .../spark/sql/HoodieSpark33SchemaUtils.scala | 14 +-- .../sql/HoodieSpark34CatalystExpressionUtils.scala | 81 ++------------ .../spark/sql/HoodieSpark34CatalystPlanUtils.scala | 94 +--------------- .../spark/sql/HoodieSpark34SchemaUtils.scala | 14 +-- .../sql/HoodieSpark35CatalystExpressionUtils.scala | 81 ++------------ .../spark/sql/HoodieSpark35CatalystPlanUtils.scala | 103 +---------------- .../spark/sql/HoodieSpark35SchemaUtils.scala | 16 +-- .../sql/HoodieSpark4CatalystExpressionUtils.scala | 45 +++++--- .../spark/sql/HoodieSpark4CatalystPlanUtils.scala | 66 +++++++++++ .../spark/sql/HoodieSpark4SchemaUtils.scala} | 4 +- .../sql/hudi/analysis/HoodieSpark4Analysis.scala} | 110 ++++-------------- .../sql/HoodieSpark40CatalystExpressionUtils.scala | 99 +---------------- .../spark/sql/HoodieSpark40CatalystPlanUtils.scala | 123 +-------------------- .../spark/sql/HoodieSpark40SchemaUtils.scala | 31 +----- .../sql/hudi/analysis/HoodieSpark40Analysis.scala | 122 ++------------------ .../sql/HoodieSpark41CatalystExpressionUtils.scala | 99 +---------------- .../spark/sql/HoodieSpark41CatalystPlanUtils.scala | 123 +-------------------- .../spark/sql/HoodieSpark41SchemaUtils.scala | 31 +----- .../sql/hudi/analysis/HoodieSpark41Analysis.scala | 121 ++------------------ 26 files changed, 327 insertions(+), 1462 deletions(-) diff --git a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/HoodieSpark40CatalystExpressionUtils.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/BaseHoodieCatalystExpressionUtils.scala similarity index 74% copy from hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/HoodieSpark40CatalystExpressionUtils.scala copy to hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/BaseHoodieCatalystExpressionUtils.scala index a183f754483e..821eebcce230 100644 --- a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/HoodieSpark40CatalystExpressionUtils.scala +++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/BaseHoodieCatalystExpressionUtils.scala @@ -18,32 +18,27 @@ package org.apache.spark.sql import org.apache.spark.sql.HoodieSparkTypeUtils.isCastPreservingOrdering -import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder -import org.apache.spark.sql.catalyst.expressions.{Add, Attribute, AttributeReference, AttributeSet, BitwiseOr, Cast, DateAdd, DateDiff, DateFormatClass, DateSub, Divide, EvalMode, Exp, Expm1, Expression, FromUnixTime, FromUTCTimestamp, Log, Log10, Log1p, Log2, Lower, Multiply, ParseToDate, ParseToTimestamp, PredicateHelper, ShiftLeft, ShiftRight, ToUnixTimestamp, ToUTCTimestamp, Upper} +import org.apache.spark.sql.catalyst.expressions.{Add, Attribute, AttributeReference, AttributeSet, BitwiseOr, Cast, DateAdd, DateDiff, DateFormatClass, DateSub, Divide, Exp, Expm1, Expression, FromUnixTime, FromUTCTimestamp, Log, Log10, Log1p, Log2, Lower, Multiply, PredicateHelper, ShiftLeft, ShiftRight, ToUnixTimestamp, ToUTCTimestamp, Upper} import org.apache.spark.sql.execution.datasources.DataSourceStrategy -import org.apache.spark.sql.types.{DataType, StructType} +import org.apache.spark.sql.types.DataType -object HoodieSpark40CatalystExpressionUtils extends HoodieSpark4CatalystExpressionUtils with PredicateHelper { - - override def getEncoder(schema: StructType): ExpressionEncoder[Row] = { - ExpressionEncoder.apply(schema).resolveAndBind() - } +/** + * Base implementation of [[HoodieCatalystExpressionUtils]] carrying the method bodies that are + * identical across all supported Spark versions. Methods relying on Spark APIs that changed + * across versions are implemented in the per-version `HoodieSparkXXCatalystExpressionUtils` + * objects (or in [[HoodieSpark4CatalystExpressionUtils]] when shared within a Spark major version). + */ +abstract class BaseHoodieCatalystExpressionUtils extends HoodieCatalystExpressionUtils with PredicateHelper { override def normalizeExprs(exprs: Seq[Expression], attributes: Seq[Attribute]): Seq[Expression] = { DataSourceStrategy.normalizeExprs(exprs, attributes) } - override def extractPredicatesWithinOutputSet(condition: Expression, outputSet: AttributeSet): Option[Expression] = { + override def extractPredicatesWithinOutputSet(condition: Expression, + outputSet: AttributeSet): Option[Expression] = { super[PredicateHelper].extractPredicatesWithinOutputSet(condition, outputSet) } - override def matchCast(expr: Expression): Option[(Expression, DataType, Option[String])] = { - expr match { - case Cast(child, dataType, timeZoneId, _) => Some((child, dataType, timeZoneId)) - case _ => None - } - } - override def tryMatchAttributeOrderingPreservingTransformation(expr: Expression): Option[AttributeReference] = { expr match { case OrderPreservingTransformation(attrRef) => Some(attrRef) @@ -54,12 +49,13 @@ object HoodieSpark40CatalystExpressionUtils extends HoodieSpark4CatalystExpressi def canUpCast(fromType: DataType, toType: DataType): Boolean = Cast.canUpCast(fromType, toType) - override def unapplyCastExpression(expr: Expression): Option[(Expression, DataType, Option[String], Boolean)] = - expr match { - case Cast(castedExpr, dataType, timeZoneId, ansiEnabled) => - Some((castedExpr, dataType, timeZoneId, if (ansiEnabled == EvalMode.ANSI) true else false)) - case _ => None - } + /** + * Matches order-preserving date/time parsing expressions whose case-class shapes differ across + * Spark versions (currently [[org.apache.spark.sql.catalyst.expressions.ParseToDate]] and + * [[org.apache.spark.sql.catalyst.expressions.ParseToTimestamp]]), returning the source child + * expression that the order-preserving transformation matching should recurse into + */ + protected def unapplyOrderPreservingDateParsing(expr: Expression): Option[Expression] private object OrderPreservingTransformation { def unapply(expr: Expression): Option[AttributeReference] = { @@ -72,8 +68,6 @@ object HoodieSpark40CatalystExpressionUtils extends HoodieSpark4CatalystExpressi case DateDiff(_, OrderPreservingTransformation(attrRef)) => Some(attrRef) case FromUnixTime(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) case FromUTCTimestamp(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case ParseToDate(OrderPreservingTransformation(attrRef), _, _, _) => Some(attrRef) - case ParseToTimestamp(OrderPreservingTransformation(attrRef), _, _, _, _) => Some(attrRef) case ToUnixTimestamp(OrderPreservingTransformation(attrRef), _, _, _) => Some(attrRef) case ToUTCTimestamp(OrderPreservingTransformation(attrRef), _) => Some(attrRef) @@ -109,8 +103,11 @@ object HoodieSpark40CatalystExpressionUtils extends HoodieSpark4CatalystExpressi // Identity transformation case attrRef: AttributeReference => Some(attrRef) - // No match - case _ => None + // Date/time parsing expressions whose shapes are Spark-version-specific + case _ => unapplyOrderPreservingDateParsing(expr) match { + case Some(child) => unapply(child) + case None => None + } } } } diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/BaseHoodieCatalystPlanUtils.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/BaseHoodieCatalystPlanUtils.scala index 498f3f276dc0..c8ceb713a6a3 100644 --- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/BaseHoodieCatalystPlanUtils.scala +++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/BaseHoodieCatalystPlanUtils.scala @@ -20,14 +20,14 @@ package org.apache.spark.sql import org.apache.hudi.SparkAdapterSupport import org.apache.spark.sql.catalyst.TableIdentifier -import org.apache.spark.sql.catalyst.analysis.TableOutputResolver +import org.apache.spark.sql.catalyst.analysis.{ResolvedTable, TableOutputResolver} import org.apache.spark.sql.catalyst.catalog.CatalogStorageFormat import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeSet, Expression, ProjectionOverSchema} import org.apache.spark.sql.catalyst.plans.JoinType -import org.apache.spark.sql.catalyst.plans.logical.{InsertIntoStatement, Join, JoinHint, LogicalPlan} +import org.apache.spark.sql.catalyst.plans.logical.{CreateIndex, DropIndex, HoodieShowIndexes, InsertIntoStatement, Join, JoinHint, LogicalPlan, RefreshIndex} import org.apache.spark.sql.connector.catalog.{Identifier, Table, TableCatalog} import org.apache.spark.sql.execution.{ExtendedMode, SimpleMode} -import org.apache.spark.sql.execution.command.{CreateTableLikeCommand, ExplainCommand} +import org.apache.spark.sql.execution.command.{CreateTableLikeCommand, ExplainCommand, RepairTableCommand} import org.apache.spark.sql.execution.datasources.LogicalRelation import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types.StructType @@ -37,12 +37,17 @@ trait BaseHoodieCatalystPlanUtils extends HoodieCatalystPlansUtils { /** * Instantiates [[ProjectionOverSchema]] utility */ - def projectOverSchema(schema: StructType, output: AttributeSet): ProjectionOverSchema + def projectOverSchema(schema: StructType, output: AttributeSet): ProjectionOverSchema = + ProjectionOverSchema(schema, output) /** * Un-applies [[ResolvedTable]] that had its signature changed in Spark 3.2 */ - def unapplyResolvedTable(plan: LogicalPlan): Option[(TableCatalog, Identifier, Table)] + def unapplyResolvedTable(plan: LogicalPlan): Option[(TableCatalog, Identifier, Table)] = + plan match { + case ResolvedTable(catalog, identifier, table, _) => Some((catalog, identifier, table)) + case _ => None + } def resolveOutputColumns(tableName: String, expected: Seq[Attribute], @@ -77,7 +82,72 @@ trait BaseHoodieCatalystPlanUtils extends HoodieCatalystPlansUtils { a.sameOutput(b) } - override def createProjectForByNameQuery(lr: LogicalRelation, plan: LogicalPlan): Option[LogicalPlan] = None + override def isRepairTable(plan: LogicalPlan): Boolean = { + plan.isInstanceOf[RepairTableCommand] + } + + override def getRepairTableChildren(plan: LogicalPlan): Option[(TableIdentifier, Boolean, Boolean, String)] = { + plan match { + case rtc: RepairTableCommand => + Some((rtc.tableName, rtc.enableAddPartitions, rtc.enableDropPartitions, rtc.cmd)) + case _ => + None + } + } + + override def unapplyCreateIndex(plan: LogicalPlan): Option[(LogicalPlan, String, String, Boolean, Seq[(Seq[String], Map[String, String])], Map[String, String])] = { + plan match { + case ci@CreateIndex(table, indexName, indexType, ignoreIfExists, columns, properties) => + Some((table, indexName, indexType, ignoreIfExists, columns.map(col => (col._1.name, col._2)), properties)) + case _ => + None + } + } + + override def unapplyDropIndex(plan: LogicalPlan): Option[(LogicalPlan, String, Boolean)] = { + plan match { + case ci@DropIndex(table, indexName, ignoreIfNotExists) => + Some((table, indexName, ignoreIfNotExists)) + case _ => + None + } + } + + override def unapplyShowIndexes(plan: LogicalPlan): Option[(LogicalPlan, Seq[Attribute])] = { + plan match { + case ci@HoodieShowIndexes(table, output) => + Some((table, output)) + case _ => + None + } + } + + override def unapplyRefreshIndex(plan: LogicalPlan): Option[(LogicalPlan, String)] = { + plan match { + case ci@RefreshIndex(table, indexName) => + Some((table, indexName)) + case _ => + None + } + } + + override def unapplyInsertIntoStatement(plan: LogicalPlan): Option[(LogicalPlan, Seq[String], Map[String, Option[String]], LogicalPlan, Boolean, Boolean)] = { + plan match { + case insert: InsertIntoStatement => + Some((insert.table, insert.userSpecifiedCols, insert.partitionSpec, insert.query, insert.overwrite, insert.ifPartitionNotExists)) + case _ => + None + } + } + + override def createProjectForByNameQuery(lr: LogicalRelation, plan: LogicalPlan): Option[LogicalPlan] = { + plan match { + case insert: InsertIntoStatement => + Some(ResolveInsertionBase.createProjectForByNameQuery(lr.catalogTable.get.qualifiedName, insert)) + case _ => + None + } + } } object BaseHoodieCatalystPlanUtils extends SparkAdapterSupport { diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/HoodieSpark3CatalystExpressionUtils.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/HoodieSpark3CatalystExpressionUtils.scala deleted file mode 100644 index cf63383ef35a..000000000000 --- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/HoodieSpark3CatalystExpressionUtils.scala +++ /dev/null @@ -1,37 +0,0 @@ -/* - * 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.{Attribute, AttributeSet, Expression} -abstract class HoodieSpark3CatalystExpressionUtils extends HoodieCatalystExpressionUtils { - - /** - * The attribute name may differ from the one in the schema if the query analyzer - * is case insensitive. We should change attribute names to match the ones in the schema, - * so we do not need to worry about case sensitivity anymore - */ - def normalizeExprs(exprs: Seq[Expression], attributes: Seq[Attribute]): Seq[Expression] - - /** - * Returns a filter that its reference is a subset of `outputSet` and it contains the maximum - * constraints from `condition`. This is used for predicate push-down - * When there is no such filter, `None` is returned. - */ - def extractPredicatesWithinOutputSet(condition: Expression, - outputSet: AttributeSet): Option[Expression] -} diff --git a/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/HoodieSpark3CatalystPlanUtils.scala b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/HoodieSpark3CatalystPlanUtils.scala new file mode 100644 index 000000000000..eb22674d9de7 --- /dev/null +++ b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/HoodieSpark3CatalystPlanUtils.scala @@ -0,0 +1,44 @@ +/* + * 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.Expression +import org.apache.spark.sql.catalyst.plans.logical.{Assignment, UpdateAction} +import org.apache.spark.sql.execution.streaming.SerializedOffset + +/** + * Implementation of [[HoodieCatalystPlansUtils]] carrying the method bodies shared by all + * supported Spark 3.x versions + */ +abstract class HoodieSpark3CatalystPlanUtils extends BaseHoodieCatalystPlanUtils { + + override def unapplyUpdateAction(mergeAction: Any): Option[(Option[Expression], Seq[Assignment])] = { + mergeAction match { + case UpdateAction(condition, assignments) => Some((condition, assignments)) + case _ => None + } + } + + override def extractJsonFromSerializedOffset(offset: Any): Option[String] = { + offset match { + case SerializedOffset(json) => Some(json) + case _ => None + } + } +} diff --git a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/HoodieSpark34SchemaUtils.scala b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/HoodieSpark3SchemaUtils.scala similarity index 70% copy from hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/HoodieSpark34SchemaUtils.scala copy to hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/HoodieSpark3SchemaUtils.scala index c5fb1beded17..6adad02c1d91 100644 --- a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/HoodieSpark34SchemaUtils.scala +++ b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/HoodieSpark3SchemaUtils.scala @@ -19,28 +19,16 @@ package org.apache.spark.sql -import org.apache.spark.sql.catalyst.expressions.Attribute import org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils import org.apache.spark.sql.jdbc.JdbcDialect import org.apache.spark.sql.types.StructType -import org.apache.spark.sql.util.SchemaUtils import java.sql.{Connection, ResultSet} /** - * Utils on schema for Spark 3.4. + * Utils on schema shared by all supported Spark 3.x versions. */ -object HoodieSpark34SchemaUtils extends HoodieSchemaUtils { - override def checkColumnNameDuplication(columnNames: Seq[String], - colType: String, - caseSensitiveAnalysis: Boolean): Unit = { - SchemaUtils.checkColumnNameDuplication(columnNames, caseSensitiveAnalysis) - } - - override def toAttributes(struct: StructType): Seq[Attribute] = { - struct.toAttributes - } - +abstract class HoodieSpark3SchemaUtils extends HoodieSchemaUtils { override def getSchema(conn: Connection, resultSet: ResultSet, dialect: JdbcDialect, diff --git a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/HoodieSpark33CatalystExpressionUtils.scala b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/HoodieSpark33CatalystExpressionUtils.scala index e08331374240..8ebfce7785fc 100644 --- a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/HoodieSpark33CatalystExpressionUtils.scala +++ b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/HoodieSpark33CatalystExpressionUtils.scala @@ -17,26 +17,16 @@ package org.apache.spark.sql -import org.apache.spark.sql.HoodieSparkTypeUtils.isCastPreservingOrdering import org.apache.spark.sql.catalyst.encoders.{ExpressionEncoder, RowEncoder} -import org.apache.spark.sql.catalyst.expressions.{Add, AnsiCast, Attribute, AttributeReference, AttributeSet, BitwiseOr, Cast, DateAdd, DateDiff, DateFormatClass, DateSub, Divide, Exp, Expm1, Expression, FromUnixTime, FromUTCTimestamp, Log, Log10, Log1p, Log2, Lower, Multiply, ParseToDate, ParseToTimestamp, PredicateHelper, ShiftLeft, ShiftRight, ToUnixTimestamp, ToUTCTimestamp, Upper} -import org.apache.spark.sql.execution.datasources.DataSourceStrategy +import org.apache.spark.sql.catalyst.expressions.{AnsiCast, Cast, Expression, ParseToDate, ParseToTimestamp} import org.apache.spark.sql.types.{DataType, StructType} -object HoodieSpark33CatalystExpressionUtils extends HoodieSpark3CatalystExpressionUtils with PredicateHelper { +object HoodieSpark33CatalystExpressionUtils extends BaseHoodieCatalystExpressionUtils { override def getEncoder(schema: StructType): ExpressionEncoder[Row] = { RowEncoder.apply(schema).resolveAndBind() } - override def normalizeExprs(exprs: Seq[Expression], attributes: Seq[Attribute]): Seq[Expression] = - DataSourceStrategy.normalizeExprs(exprs, attributes) - - override def extractPredicatesWithinOutputSet(condition: Expression, - outputSet: AttributeSet): Option[Expression] = { - super[PredicateHelper].extractPredicatesWithinOutputSet(condition, outputSet) - } - override def matchCast(expr: Expression): Option[(Expression, DataType, Option[String])] = expr match { case Cast(child, dataType, timeZoneId, _) => Some((child, dataType, timeZoneId)) @@ -44,16 +34,6 @@ object HoodieSpark33CatalystExpressionUtils extends HoodieSpark3CatalystExpressi case _ => None } - override def tryMatchAttributeOrderingPreservingTransformation(expr: Expression): Option[AttributeReference] = { - expr match { - case OrderPreservingTransformation(attrRef) => Some(attrRef) - case _ => None - } - } - - def canUpCast(fromType: DataType, toType: DataType): Boolean = - Cast.canUpCast(fromType, toType) - override def unapplyCastExpression(expr: Expression): Option[(Expression, DataType, Option[String], Boolean)] = expr match { case Cast(castedExpr, dataType, timeZoneId, ansiEnabled) => @@ -63,57 +43,10 @@ object HoodieSpark33CatalystExpressionUtils extends HoodieSpark3CatalystExpressi case _ => None } - private object OrderPreservingTransformation { - def unapply(expr: Expression): Option[AttributeReference] = { - expr match { - // Date/Time Expressions - case DateFormatClass(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case DateAdd(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case DateSub(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case DateDiff(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case DateDiff(_, OrderPreservingTransformation(attrRef)) => Some(attrRef) - case FromUnixTime(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case FromUTCTimestamp(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case ParseToDate(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case ParseToTimestamp(OrderPreservingTransformation(attrRef), _, _, _) => Some(attrRef) - case ToUnixTimestamp(OrderPreservingTransformation(attrRef), _, _, _) => Some(attrRef) - case ToUTCTimestamp(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - - // String Expressions - case Lower(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Upper(OrderPreservingTransformation(attrRef)) => Some(attrRef) - // Left API change: Improve RuntimeReplaceable - // https://issues.apache.org/jira/browse/SPARK-38240 - case org.apache.spark.sql.catalyst.expressions.Left(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - - // Math Expressions - // Binary - case Add(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case Add(_, OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case Multiply(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case Multiply(_, OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case Divide(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case BitwiseOr(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case BitwiseOr(_, OrderPreservingTransformation(attrRef)) => Some(attrRef) - // Unary - case Exp(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Expm1(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Log(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Log10(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Log1p(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Log2(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case ShiftLeft(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case ShiftRight(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - - // Other - case cast @ Cast(OrderPreservingTransformation(attrRef), _, _, _) - if isCastPreservingOrdering(cast.child.dataType, cast.dataType) => Some(attrRef) - - // Identity transformation - case attrRef: AttributeReference => Some(attrRef) - // No match - case _ => None - } + override protected def unapplyOrderPreservingDateParsing(expr: Expression): Option[Expression] = + expr match { + case ParseToDate(child, _, _) => Some(child) + case ParseToTimestamp(child, _, _, _) => Some(child) + case _ => None } - } } diff --git a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/HoodieSpark33CatalystPlanUtils.scala b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/HoodieSpark33CatalystPlanUtils.scala index 114bc958c172..1a81ab046866 100644 --- a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/HoodieSpark33CatalystPlanUtils.scala +++ b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/HoodieSpark33CatalystPlanUtils.scala @@ -18,25 +18,14 @@ package org.apache.spark.sql -import org.apache.spark.sql.catalyst.TableIdentifier -import org.apache.spark.sql.catalyst.analysis.{AnalysisErrorAt, ResolvedTable} -import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeSet, Expression, ProjectionOverSchema} +import org.apache.spark.sql.catalyst.analysis.AnalysisErrorAt +import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression} import org.apache.spark.sql.catalyst.planning.ScanOperation -import org.apache.spark.sql.catalyst.plans.logical._ -import org.apache.spark.sql.connector.catalog.{Identifier, Table, TableCatalog} -import org.apache.spark.sql.execution.command.RepairTableCommand +import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, MergeIntoTable} import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation} import org.apache.spark.sql.execution.datasources.parquet.{HoodieFormatTrait, ParquetFileFormat} -import org.apache.spark.sql.execution.streaming.SerializedOffset -import org.apache.spark.sql.types.StructType -object HoodieSpark33CatalystPlanUtils extends BaseHoodieCatalystPlanUtils { - - def unapplyResolvedTable(plan: LogicalPlan): Option[(TableCatalog, Identifier, Table)] = - plan match { - case ResolvedTable(catalog, identifier, table, _) => Some((catalog, identifier, table)) - case _ => None - } +object HoodieSpark33CatalystPlanUtils extends HoodieSpark3CatalystPlanUtils { override def unapplyMergeIntoTable(plan: LogicalPlan): Option[(LogicalPlan, LogicalPlan, Expression)] = { plan match { @@ -57,22 +46,6 @@ object HoodieSpark33CatalystPlanUtils extends BaseHoodieCatalystPlanUtils { } } - override def projectOverSchema(schema: StructType, output: AttributeSet): ProjectionOverSchema = - ProjectionOverSchema(schema, output) - - override def isRepairTable(plan: LogicalPlan): Boolean = { - plan.isInstanceOf[RepairTableCommand] - } - - override def getRepairTableChildren(plan: LogicalPlan): Option[(TableIdentifier, Boolean, Boolean, String)] = { - plan match { - case rtc: RepairTableCommand => - Some((rtc.tableName, rtc.enableAddPartitions, rtc.enableDropPartitions, rtc.cmd)) - case _ => - None - } - } - override def failAnalysisForMIT(a: Attribute, cols: String): Unit = { a.failAnalysis(s"cannot resolve ${a.sql} in MERGE command given columns [$cols]") } @@ -80,72 +53,4 @@ object HoodieSpark33CatalystPlanUtils extends BaseHoodieCatalystPlanUtils { override def failTableNotFound(tableName: String): Unit = { throw new AnalysisException(s"Table or view not found: $tableName") } - - override def unapplyCreateIndex(plan: LogicalPlan): Option[(LogicalPlan, String, String, Boolean, Seq[(Seq[String], Map[String, String])], Map[String, String])] = { - plan match { - case ci @ CreateIndex(table, indexName, indexType, ignoreIfExists, columns, properties) => - Some((table, indexName, indexType, ignoreIfExists, columns.map(col => (col._1.name, col._2)), properties)) - case _ => - None - } - } - - override def unapplyDropIndex(plan: LogicalPlan): Option[(LogicalPlan, String, Boolean)] = { - plan match { - case ci @ DropIndex(table, indexName, ignoreIfNotExists) => - Some((table, indexName, ignoreIfNotExists)) - case _ => - None - } - } - - override def unapplyShowIndexes(plan: LogicalPlan): Option[(LogicalPlan, Seq[Attribute])] = { - plan match { - case ci @ HoodieShowIndexes(table, output) => - Some((table, output)) - case _ => - None - } - } - - override def unapplyRefreshIndex(plan: LogicalPlan): Option[(LogicalPlan, String)] = { - plan match { - case ci @ RefreshIndex(table, indexName) => - Some((table, indexName)) - case _ => - None - } - } - - override def unapplyInsertIntoStatement(plan: LogicalPlan): Option[(LogicalPlan, Seq[String], Map[String, Option[String]], LogicalPlan, Boolean, Boolean)] = { - plan match { - case insert: InsertIntoStatement => - Some((insert.table, insert.userSpecifiedCols, insert.partitionSpec, insert.query, insert.overwrite, insert.ifPartitionNotExists)) - case _ => - None - } - } - - override def createProjectForByNameQuery(lr: LogicalRelation, plan: LogicalPlan): Option[LogicalPlan] = { - plan match { - case insert: InsertIntoStatement => - Some(ResolveInsertionBase.createProjectForByNameQuery(lr.catalogTable.get.qualifiedName, insert)) - case _ => - None - } - } - - override def unapplyUpdateAction(mergeAction: Any): Option[(Option[Expression], Seq[Assignment])] = { - mergeAction match { - case UpdateAction(condition, assignments) => Some((condition, assignments)) - case _ => None - } - } - - override def extractJsonFromSerializedOffset(offset: Any): Option[String] = { - offset match { - case SerializedOffset(json) => Some(json) - case _ => None - } - } } diff --git a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/HoodieSpark33SchemaUtils.scala b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/HoodieSpark33SchemaUtils.scala index 41748ac15553..7b968bde8064 100644 --- a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/HoodieSpark33SchemaUtils.scala +++ b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/HoodieSpark33SchemaUtils.scala @@ -20,17 +20,13 @@ package org.apache.spark.sql import org.apache.spark.sql.catalyst.expressions.Attribute -import org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils -import org.apache.spark.sql.jdbc.JdbcDialect import org.apache.spark.sql.types.StructType import org.apache.spark.sql.util.SchemaUtils -import java.sql.{Connection, ResultSet} - /** * Utils on schema for Spark 3.3. */ -object HoodieSpark33SchemaUtils extends HoodieSchemaUtils { +object HoodieSpark33SchemaUtils extends HoodieSpark3SchemaUtils { override def checkColumnNameDuplication(columnNames: Seq[String], colType: String, caseSensitiveAnalysis: Boolean): Unit = { @@ -40,12 +36,4 @@ object HoodieSpark33SchemaUtils extends HoodieSchemaUtils { override def toAttributes(struct: StructType): Seq[Attribute] = { struct.toAttributes } - - override def getSchema(conn: Connection, - resultSet: ResultSet, - dialect: JdbcDialect, - alwaysNullable: Boolean = false, - isTimestampNTZ: Boolean = false): StructType = { - JdbcUtils.getSchema(resultSet, dialect, alwaysNullable) - } } diff --git a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/HoodieSpark34CatalystExpressionUtils.scala b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/HoodieSpark34CatalystExpressionUtils.scala index 03c7d0412f8a..f5767eb82a76 100644 --- a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/HoodieSpark34CatalystExpressionUtils.scala +++ b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/HoodieSpark34CatalystExpressionUtils.scala @@ -17,26 +17,16 @@ package org.apache.spark.sql -import org.apache.spark.sql.HoodieSparkTypeUtils.isCastPreservingOrdering import org.apache.spark.sql.catalyst.encoders.{ExpressionEncoder, RowEncoder} -import org.apache.spark.sql.catalyst.expressions.{Add, Attribute, AttributeReference, AttributeSet, BitwiseOr, Cast, DateAdd, DateDiff, DateFormatClass, DateSub, Divide, EvalMode, Exp, Expm1, Expression, FromUnixTime, FromUTCTimestamp, Log, Log10, Log1p, Log2, Lower, Multiply, ParseToDate, ParseToTimestamp, PredicateHelper, ShiftLeft, ShiftRight, ToUnixTimestamp, ToUTCTimestamp, Upper} -import org.apache.spark.sql.execution.datasources.DataSourceStrategy +import org.apache.spark.sql.catalyst.expressions.{Cast, EvalMode, Expression, ParseToDate, ParseToTimestamp} import org.apache.spark.sql.types.{DataType, StructType} -object HoodieSpark34CatalystExpressionUtils extends HoodieSpark3CatalystExpressionUtils with PredicateHelper { +object HoodieSpark34CatalystExpressionUtils extends BaseHoodieCatalystExpressionUtils { override def getEncoder(schema: StructType): ExpressionEncoder[Row] = { RowEncoder.apply(schema).resolveAndBind() } - override def normalizeExprs(exprs: Seq[Expression], attributes: Seq[Attribute]): Seq[Expression] = { - DataSourceStrategy.normalizeExprs(exprs, attributes) - } - - override def extractPredicatesWithinOutputSet(condition: Expression, outputSet: AttributeSet): Option[Expression] = { - super[PredicateHelper].extractPredicatesWithinOutputSet(condition, outputSet) - } - override def matchCast(expr: Expression): Option[(Expression, DataType, Option[String])] = { expr match { case Cast(child, dataType, timeZoneId, _) => Some((child, dataType, timeZoneId)) @@ -44,16 +34,6 @@ object HoodieSpark34CatalystExpressionUtils extends HoodieSpark3CatalystExpressi } } - override def tryMatchAttributeOrderingPreservingTransformation(expr: Expression): Option[AttributeReference] = { - expr match { - case OrderPreservingTransformation(attrRef) => Some(attrRef) - case _ => None - } - } - - def canUpCast(fromType: DataType, toType: DataType): Boolean = - Cast.canUpCast(fromType, toType) - override def unapplyCastExpression(expr: Expression): Option[(Expression, DataType, Option[String], Boolean)] = expr match { case Cast(castedExpr, dataType, timeZoneId, ansiEnabled) => @@ -61,57 +41,10 @@ object HoodieSpark34CatalystExpressionUtils extends HoodieSpark3CatalystExpressi case _ => None } - private object OrderPreservingTransformation { - def unapply(expr: Expression): Option[AttributeReference] = { - expr match { - // Date/Time Expressions - case DateFormatClass(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case DateAdd(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case DateSub(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case DateDiff(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case DateDiff(_, OrderPreservingTransformation(attrRef)) => Some(attrRef) - case FromUnixTime(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case FromUTCTimestamp(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case ParseToDate(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case ParseToTimestamp(OrderPreservingTransformation(attrRef), _, _, _, _) => Some(attrRef) - case ToUnixTimestamp(OrderPreservingTransformation(attrRef), _, _, _) => Some(attrRef) - case ToUTCTimestamp(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - - // String Expressions - case Lower(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Upper(OrderPreservingTransformation(attrRef)) => Some(attrRef) - // Left API change: Improve RuntimeReplaceable - // https://issues.apache.org/jira/browse/SPARK-38240 - case org.apache.spark.sql.catalyst.expressions.Left(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - - // Math Expressions - // Binary - case Add(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case Add(_, OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case Multiply(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case Multiply(_, OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case Divide(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case BitwiseOr(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case BitwiseOr(_, OrderPreservingTransformation(attrRef)) => Some(attrRef) - // Unary - case Exp(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Expm1(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Log(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Log10(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Log1p(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Log2(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case ShiftLeft(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case ShiftRight(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - - // Other - case cast @ Cast(OrderPreservingTransformation(attrRef), _, _, _) - if isCastPreservingOrdering(cast.child.dataType, cast.dataType) => Some(attrRef) - - // Identity transformation - case attrRef: AttributeReference => Some(attrRef) - // No match - case _ => None - } + override protected def unapplyOrderPreservingDateParsing(expr: Expression): Option[Expression] = + expr match { + case ParseToDate(child, _, _) => Some(child) + case ParseToTimestamp(child, _, _, _, _) => Some(child) + case _ => None } - } } diff --git a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/HoodieSpark34CatalystPlanUtils.scala b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/HoodieSpark34CatalystPlanUtils.scala index 78c918f325a1..97b922ba4cb6 100644 --- a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/HoodieSpark34CatalystPlanUtils.scala +++ b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/HoodieSpark34CatalystPlanUtils.scala @@ -18,26 +18,15 @@ package org.apache.spark.sql -import org.apache.spark.sql.catalyst.TableIdentifier -import org.apache.spark.sql.catalyst.analysis.{AnalysisErrorAt, ResolvedTable} -import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeSet, Expression, ProjectionOverSchema} +import org.apache.spark.sql.catalyst.analysis.AnalysisErrorAt +import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression} import org.apache.spark.sql.catalyst.planning.ScanOperation -import org.apache.spark.sql.catalyst.plans.logical._ -import org.apache.spark.sql.connector.catalog.{Identifier, Table, TableCatalog} -import org.apache.spark.sql.execution.command.RepairTableCommand +import org.apache.spark.sql.catalyst.plans.logical.{InsertIntoStatement, LogicalPlan, MergeIntoTable} import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation} import org.apache.spark.sql.execution.datasources.parquet.{HoodieFormatTrait, ParquetFileFormat} -import org.apache.spark.sql.execution.streaming.SerializedOffset import org.apache.spark.sql.internal.SQLConf -import org.apache.spark.sql.types.StructType -object HoodieSpark34CatalystPlanUtils extends BaseHoodieCatalystPlanUtils { - - def unapplyResolvedTable(plan: LogicalPlan): Option[(TableCatalog, Identifier, Table)] = - plan match { - case ResolvedTable(catalog, identifier, table, _) => Some((catalog, identifier, table)) - case _ => None - } +object HoodieSpark34CatalystPlanUtils extends HoodieSpark3CatalystPlanUtils { override def unapplyMergeIntoTable(plan: LogicalPlan): Option[(LogicalPlan, LogicalPlan, Expression)] = { plan match { @@ -58,22 +47,6 @@ object HoodieSpark34CatalystPlanUtils extends BaseHoodieCatalystPlanUtils { } } - override def projectOverSchema(schema: StructType, output: AttributeSet): ProjectionOverSchema = - ProjectionOverSchema(schema, output) - - override def isRepairTable(plan: LogicalPlan): Boolean = { - plan.isInstanceOf[RepairTableCommand] - } - - override def getRepairTableChildren(plan: LogicalPlan): Option[(TableIdentifier, Boolean, Boolean, String)] = { - plan match { - case rtc: RepairTableCommand => - Some((rtc.tableName, rtc.enableAddPartitions, rtc.enableDropPartitions, rtc.cmd)) - case _ => - None - } - } - override def failAnalysisForMIT(a: Attribute, cols: String): Unit = { a.failAnalysis( errorClass = "_LEGACY_ERROR_TEMP_2309", @@ -88,42 +61,6 @@ object HoodieSpark34CatalystPlanUtils extends BaseHoodieCatalystPlanUtils { messageParameters = Map("relationName" -> s"`$tableName`")) } - override def unapplyCreateIndex(plan: LogicalPlan): Option[(LogicalPlan, String, String, Boolean, Seq[(Seq[String], Map[String, String])], Map[String, String])] = { - plan match { - case ci@CreateIndex(table, indexName, indexType, ignoreIfExists, columns, properties) => - Some((table, indexName, indexType, ignoreIfExists, columns.map(col => (col._1.name, col._2)), properties)) - case _ => - None - } - } - - override def unapplyDropIndex(plan: LogicalPlan): Option[(LogicalPlan, String, Boolean)] = { - plan match { - case ci@DropIndex(table, indexName, ignoreIfNotExists) => - Some((table, indexName, ignoreIfNotExists)) - case _ => - None - } - } - - override def unapplyShowIndexes(plan: LogicalPlan): Option[(LogicalPlan, Seq[Attribute])] = { - plan match { - case ci@HoodieShowIndexes(table, output) => - Some((table, output)) - case _ => - None - } - } - - override def unapplyRefreshIndex(plan: LogicalPlan): Option[(LogicalPlan, String)] = { - plan match { - case ci@RefreshIndex(table, indexName) => - Some((table, indexName)) - case _ => - None - } - } - override def unapplyInsertIntoStatement(plan: LogicalPlan): Option[(LogicalPlan, Seq[String], Map[String, Option[String]], LogicalPlan, Boolean, Boolean)] = { plan match { case insert: InsertIntoStatement => @@ -145,27 +82,4 @@ object HoodieSpark34CatalystPlanUtils extends BaseHoodieCatalystPlanUtils { None } } - - override def createProjectForByNameQuery(lr: LogicalRelation, plan: LogicalPlan): Option[LogicalPlan] = { - plan match { - case insert: InsertIntoStatement => - Some(ResolveInsertionBase.createProjectForByNameQuery(lr.catalogTable.get.qualifiedName, insert)) - case _ => - None - } - } - - override def unapplyUpdateAction(mergeAction: Any): Option[(Option[Expression], Seq[Assignment])] = { - mergeAction match { - case UpdateAction(condition, assignments) => Some((condition, assignments)) - case _ => None - } - } - - override def extractJsonFromSerializedOffset(offset: Any): Option[String] = { - offset match { - case SerializedOffset(json) => Some(json) - case _ => None - } - } } diff --git a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/HoodieSpark34SchemaUtils.scala b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/HoodieSpark34SchemaUtils.scala index c5fb1beded17..c272208f7b8a 100644 --- a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/HoodieSpark34SchemaUtils.scala +++ b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/HoodieSpark34SchemaUtils.scala @@ -20,17 +20,13 @@ package org.apache.spark.sql import org.apache.spark.sql.catalyst.expressions.Attribute -import org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils -import org.apache.spark.sql.jdbc.JdbcDialect import org.apache.spark.sql.types.StructType import org.apache.spark.sql.util.SchemaUtils -import java.sql.{Connection, ResultSet} - /** * Utils on schema for Spark 3.4. */ -object HoodieSpark34SchemaUtils extends HoodieSchemaUtils { +object HoodieSpark34SchemaUtils extends HoodieSpark3SchemaUtils { override def checkColumnNameDuplication(columnNames: Seq[String], colType: String, caseSensitiveAnalysis: Boolean): Unit = { @@ -40,12 +36,4 @@ object HoodieSpark34SchemaUtils extends HoodieSchemaUtils { override def toAttributes(struct: StructType): Seq[Attribute] = { struct.toAttributes } - - override def getSchema(conn: Connection, - resultSet: ResultSet, - dialect: JdbcDialect, - alwaysNullable: Boolean = false, - isTimestampNTZ: Boolean = false): StructType = { - JdbcUtils.getSchema(resultSet, dialect, alwaysNullable) - } } diff --git a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/HoodieSpark35CatalystExpressionUtils.scala b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/HoodieSpark35CatalystExpressionUtils.scala index 6f1456972a52..8039e6b96815 100644 --- a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/HoodieSpark35CatalystExpressionUtils.scala +++ b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/HoodieSpark35CatalystExpressionUtils.scala @@ -17,26 +17,16 @@ package org.apache.spark.sql -import org.apache.spark.sql.HoodieSparkTypeUtils.isCastPreservingOrdering import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder -import org.apache.spark.sql.catalyst.expressions.{Add, Attribute, AttributeReference, AttributeSet, BitwiseOr, Cast, DateAdd, DateDiff, DateFormatClass, DateSub, Divide, EvalMode, Exp, Expm1, Expression, FromUnixTime, FromUTCTimestamp, Log, Log10, Log1p, Log2, Lower, Multiply, ParseToDate, ParseToTimestamp, PredicateHelper, ShiftLeft, ShiftRight, ToUnixTimestamp, ToUTCTimestamp, Upper} -import org.apache.spark.sql.execution.datasources.DataSourceStrategy +import org.apache.spark.sql.catalyst.expressions.{Cast, EvalMode, Expression, ParseToDate, ParseToTimestamp} import org.apache.spark.sql.types.{DataType, StructType} -object HoodieSpark35CatalystExpressionUtils extends HoodieSpark3CatalystExpressionUtils with PredicateHelper { +object HoodieSpark35CatalystExpressionUtils extends BaseHoodieCatalystExpressionUtils { override def getEncoder(schema: StructType): ExpressionEncoder[Row] = { ExpressionEncoder.apply(schema).resolveAndBind() } - override def normalizeExprs(exprs: Seq[Expression], attributes: Seq[Attribute]): Seq[Expression] = { - DataSourceStrategy.normalizeExprs(exprs, attributes) - } - - override def extractPredicatesWithinOutputSet(condition: Expression, outputSet: AttributeSet): Option[Expression] = { - super[PredicateHelper].extractPredicatesWithinOutputSet(condition, outputSet) - } - override def matchCast(expr: Expression): Option[(Expression, DataType, Option[String])] = { expr match { case Cast(child, dataType, timeZoneId, _) => Some((child, dataType, timeZoneId)) @@ -44,16 +34,6 @@ object HoodieSpark35CatalystExpressionUtils extends HoodieSpark3CatalystExpressi } } - override def tryMatchAttributeOrderingPreservingTransformation(expr: Expression): Option[AttributeReference] = { - expr match { - case OrderPreservingTransformation(attrRef) => Some(attrRef) - case _ => None - } - } - - def canUpCast(fromType: DataType, toType: DataType): Boolean = - Cast.canUpCast(fromType, toType) - override def unapplyCastExpression(expr: Expression): Option[(Expression, DataType, Option[String], Boolean)] = expr match { case Cast(castedExpr, dataType, timeZoneId, ansiEnabled) => @@ -61,57 +41,10 @@ object HoodieSpark35CatalystExpressionUtils extends HoodieSpark3CatalystExpressi case _ => None } - private object OrderPreservingTransformation { - def unapply(expr: Expression): Option[AttributeReference] = { - expr match { - // Date/Time Expressions - case DateFormatClass(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case DateAdd(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case DateSub(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case DateDiff(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case DateDiff(_, OrderPreservingTransformation(attrRef)) => Some(attrRef) - case FromUnixTime(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case FromUTCTimestamp(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case ParseToDate(OrderPreservingTransformation(attrRef), _, _, _) => Some(attrRef) - case ParseToTimestamp(OrderPreservingTransformation(attrRef), _, _, _, _) => Some(attrRef) - case ToUnixTimestamp(OrderPreservingTransformation(attrRef), _, _, _) => Some(attrRef) - case ToUTCTimestamp(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - - // String Expressions - case Lower(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Upper(OrderPreservingTransformation(attrRef)) => Some(attrRef) - // Left API change: Improve RuntimeReplaceable - // https://issues.apache.org/jira/browse/SPARK-38240 - case org.apache.spark.sql.catalyst.expressions.Left(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - - // Math Expressions - // Binary - case Add(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case Add(_, OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case Multiply(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case Multiply(_, OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case Divide(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case BitwiseOr(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case BitwiseOr(_, OrderPreservingTransformation(attrRef)) => Some(attrRef) - // Unary - case Exp(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Expm1(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Log(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Log10(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Log1p(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Log2(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case ShiftLeft(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case ShiftRight(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - - // Other - case cast @ Cast(OrderPreservingTransformation(attrRef), _, _, _) - if isCastPreservingOrdering(cast.child.dataType, cast.dataType) => Some(attrRef) - - // Identity transformation - case attrRef: AttributeReference => Some(attrRef) - // No match - case _ => None - } + override protected def unapplyOrderPreservingDateParsing(expr: Expression): Option[Expression] = + expr match { + case ParseToDate(child, _, _, _) => Some(child) + case ParseToTimestamp(child, _, _, _, _) => Some(child) + case _ => None } - } } diff --git a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/HoodieSpark35CatalystPlanUtils.scala b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/HoodieSpark35CatalystPlanUtils.scala index db480f4fb42e..534d8696b2c4 100644 --- a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/HoodieSpark35CatalystPlanUtils.scala +++ b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/HoodieSpark35CatalystPlanUtils.scala @@ -18,25 +18,14 @@ package org.apache.spark.sql -import org.apache.spark.sql.catalyst.TableIdentifier -import org.apache.spark.sql.catalyst.analysis.{AnalysisErrorAt, ResolvedTable} -import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeSet, Expression, ProjectionOverSchema} +import org.apache.spark.sql.catalyst.analysis.AnalysisErrorAt +import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression} import org.apache.spark.sql.catalyst.planning.ScanOperation -import org.apache.spark.sql.catalyst.plans.logical._ -import org.apache.spark.sql.connector.catalog.{Identifier, Table, TableCatalog} -import org.apache.spark.sql.execution.command.RepairTableCommand +import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, MergeIntoTable} import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation} import org.apache.spark.sql.execution.datasources.parquet.{HoodieFormatTrait, ParquetFileFormat} -import org.apache.spark.sql.execution.streaming.SerializedOffset -import org.apache.spark.sql.types.StructType -object HoodieSpark35CatalystPlanUtils extends BaseHoodieCatalystPlanUtils { - - def unapplyResolvedTable(plan: LogicalPlan): Option[(TableCatalog, Identifier, Table)] = - plan match { - case ResolvedTable(catalog, identifier, table, _) => Some((catalog, identifier, table)) - case _ => None - } +object HoodieSpark35CatalystPlanUtils extends HoodieSpark3CatalystPlanUtils { override def unapplyMergeIntoTable(plan: LogicalPlan): Option[(LogicalPlan, LogicalPlan, Expression)] = { plan match { @@ -57,22 +46,6 @@ object HoodieSpark35CatalystPlanUtils extends BaseHoodieCatalystPlanUtils { } } - override def projectOverSchema(schema: StructType, output: AttributeSet): ProjectionOverSchema = - ProjectionOverSchema(schema, output) - - override def isRepairTable(plan: LogicalPlan): Boolean = { - plan.isInstanceOf[RepairTableCommand] - } - - override def getRepairTableChildren(plan: LogicalPlan): Option[(TableIdentifier, Boolean, Boolean, String)] = { - plan match { - case rtc: RepairTableCommand => - Some((rtc.tableName, rtc.enableAddPartitions, rtc.enableDropPartitions, rtc.cmd)) - case _ => - None - } - } - override def failAnalysisForMIT(a: Attribute, cols: String): Unit = { a.failAnalysis( errorClass = "UNRESOLVED_COLUMN.WITH_SUGGESTION", @@ -86,72 +59,4 @@ object HoodieSpark35CatalystPlanUtils extends BaseHoodieCatalystPlanUtils { errorClass = "TABLE_OR_VIEW_NOT_FOUND", messageParameters = Map("relationName" -> s"`$tableName`")) } - - override def unapplyCreateIndex(plan: LogicalPlan): Option[(LogicalPlan, String, String, Boolean, Seq[(Seq[String], Map[String, String])], Map[String, String])] = { - plan match { - case ci@CreateIndex(table, indexName, indexType, ignoreIfExists, columns, properties) => - Some((table, indexName, indexType, ignoreIfExists, columns.map(col => (col._1.name, col._2)), properties)) - case _ => - None - } - } - - override def unapplyDropIndex(plan: LogicalPlan): Option[(LogicalPlan, String, Boolean)] = { - plan match { - case ci@DropIndex(table, indexName, ignoreIfNotExists) => - Some((table, indexName, ignoreIfNotExists)) - case _ => - None - } - } - - override def unapplyShowIndexes(plan: LogicalPlan): Option[(LogicalPlan, Seq[Attribute])] = { - plan match { - case ci@HoodieShowIndexes(table, output) => - Some((table, output)) - case _ => - None - } - } - - override def unapplyRefreshIndex(plan: LogicalPlan): Option[(LogicalPlan, String)] = { - plan match { - case ci@RefreshIndex(table, indexName) => - Some((table, indexName)) - case _ => - None - } - } - - override def unapplyInsertIntoStatement(plan: LogicalPlan): Option[(LogicalPlan, Seq[String], Map[String, Option[String]], LogicalPlan, Boolean, Boolean)] = { - plan match { - case insert: InsertIntoStatement => - Some((insert.table, insert.userSpecifiedCols, insert.partitionSpec, insert.query, insert.overwrite, insert.ifPartitionNotExists)) - case _ => - None - } - } - - override def createProjectForByNameQuery(lr: LogicalRelation, plan: LogicalPlan): Option[LogicalPlan] = { - plan match { - case insert: InsertIntoStatement => - Some(ResolveInsertionBase.createProjectForByNameQuery(lr.catalogTable.get.qualifiedName, insert)) - case _ => - None - } - } - - override def unapplyUpdateAction(mergeAction: Any): Option[(Option[Expression], Seq[Assignment])] = { - mergeAction match { - case UpdateAction(condition, assignments) => Some((condition, assignments)) - case _ => None - } - } - - override def extractJsonFromSerializedOffset(offset: Any): Option[String] = { - offset match { - case SerializedOffset(json) => Some(json) - case _ => None - } - } } diff --git a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/HoodieSpark35SchemaUtils.scala b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/HoodieSpark35SchemaUtils.scala index 708abd69446b..519345059960 100644 --- a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/HoodieSpark35SchemaUtils.scala +++ b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/HoodieSpark35SchemaUtils.scala @@ -21,17 +21,13 @@ package org.apache.spark.sql import org.apache.spark.sql.catalyst.expressions.Attribute import org.apache.spark.sql.catalyst.types.DataTypeUtils -import org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils -import org.apache.spark.sql.jdbc.JdbcDialect import org.apache.spark.sql.types.StructType import org.apache.spark.sql.util.SchemaUtils -import java.sql.{Connection, ResultSet} - /** - * Utils on schema for Spark 3.4+. + * Utils on schema for Spark 3.5. */ -object HoodieSpark35SchemaUtils extends HoodieSchemaUtils { +object HoodieSpark35SchemaUtils extends HoodieSpark3SchemaUtils { override def checkColumnNameDuplication(columnNames: Seq[String], colType: String, caseSensitiveAnalysis: Boolean): Unit = { @@ -41,12 +37,4 @@ object HoodieSpark35SchemaUtils extends HoodieSchemaUtils { override def toAttributes(struct: StructType): Seq[Attribute] = { DataTypeUtils.toAttributes(struct) } - - override def getSchema(conn: Connection, - resultSet: ResultSet, - dialect: JdbcDialect, - alwaysNullable: Boolean = false, - isTimestampNTZ: Boolean = false): StructType = { - JdbcUtils.getSchema(resultSet, dialect, alwaysNullable) - } } diff --git a/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/HoodieSpark4CatalystExpressionUtils.scala b/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/HoodieSpark4CatalystExpressionUtils.scala index 3da62104db73..22272b2b031c 100644 --- a/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/HoodieSpark4CatalystExpressionUtils.scala +++ b/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/HoodieSpark4CatalystExpressionUtils.scala @@ -17,22 +17,37 @@ package org.apache.spark.sql -import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeSet, Expression} +import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder +import org.apache.spark.sql.catalyst.expressions.{Cast, EvalMode, Expression, ParseToDate, ParseToTimestamp} +import org.apache.spark.sql.types.{DataType, StructType} -abstract class HoodieSpark4CatalystExpressionUtils extends HoodieCatalystExpressionUtils { +/** + * Implementation of [[HoodieCatalystExpressionUtils]] shared by all supported Spark 4.x versions + */ +abstract class HoodieSpark4CatalystExpressionUtils extends BaseHoodieCatalystExpressionUtils { + + override def getEncoder(schema: StructType): ExpressionEncoder[Row] = { + ExpressionEncoder.apply(schema).resolveAndBind() + } + + override def matchCast(expr: Expression): Option[(Expression, DataType, Option[String])] = { + expr match { + case Cast(child, dataType, timeZoneId, _) => Some((child, dataType, timeZoneId)) + case _ => None + } + } - /** - * The attribute name may differ from the one in the schema if the query analyzer - * is case insensitive. We should change attribute names to match the ones in the schema, - * so we do not need to worry about case sensitivity anymore - */ - def normalizeExprs(exprs: Seq[Expression], attributes: Seq[Attribute]): Seq[Expression] + override def unapplyCastExpression(expr: Expression): Option[(Expression, DataType, Option[String], Boolean)] = + expr match { + case Cast(castedExpr, dataType, timeZoneId, ansiEnabled) => + Some((castedExpr, dataType, timeZoneId, if (ansiEnabled == EvalMode.ANSI) true else false)) + case _ => None + } - /** - * Returns a filter that its reference is a subset of `outputSet` and it contains the maximum - * constraints from `condition`. This is used for predicate push-down - * When there is no such filter, `None` is returned. - */ - def extractPredicatesWithinOutputSet(condition: Expression, - outputSet: AttributeSet): Option[Expression] + override protected def unapplyOrderPreservingDateParsing(expr: Expression): Option[Expression] = + expr match { + case ParseToDate(child, _, _, _) => Some(child) + case ParseToTimestamp(child, _, _, _, _) => Some(child) + case _ => None + } } diff --git a/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/HoodieSpark4CatalystPlanUtils.scala b/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/HoodieSpark4CatalystPlanUtils.scala new file mode 100644 index 000000000000..7af2089ae2d2 --- /dev/null +++ b/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/HoodieSpark4CatalystPlanUtils.scala @@ -0,0 +1,66 @@ +/* + * 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.analysis.AnalysisErrorAt +import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression} +import org.apache.spark.sql.catalyst.planning.ScanOperation +import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, MergeIntoTable} +import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation} +import org.apache.spark.sql.execution.datasources.parquet.{HoodieFormatTrait, ParquetFileFormat} + +/** + * Implementation of [[HoodieCatalystPlansUtils]] carrying the method bodies shared by all + * supported Spark 4.x versions + */ +abstract class HoodieSpark4CatalystPlanUtils extends BaseHoodieCatalystPlanUtils { + + override def unapplyMergeIntoTable(plan: LogicalPlan): Option[(LogicalPlan, LogicalPlan, Expression)] = { + plan match { + case MergeIntoTable(targetTable, sourceTable, mergeCondition, _, _, _, _) => + Some((targetTable, sourceTable, mergeCondition)) + case _ => None + } + } + + override def maybeApplyForNewFileFormat(plan: LogicalPlan): LogicalPlan = { + plan match { + case s@ScanOperation(_, _, _, + l@LogicalRelation(fs: HadoopFsRelation, _, _, _, _)) + if fs.fileFormat.isInstanceOf[ParquetFileFormat with HoodieFormatTrait] + && !fs.fileFormat.asInstanceOf[ParquetFileFormat with HoodieFormatTrait].isProjected => + FileFormatUtilsForFileGroupReader.applyNewFileFormatChanges(s, l, fs) + case _ => plan + } + } + + override def failAnalysisForMIT(a: Attribute, cols: String): Unit = { + a.failAnalysis( + errorClass = "UNRESOLVED_COLUMN.WITH_SUGGESTION", + messageParameters = Map( + "objectName" -> a.sql, + "proposal" -> cols)) + } + + override def failTableNotFound(tableName: String): Unit = { + throw new AnalysisException( + errorClass = "TABLE_OR_VIEW_NOT_FOUND", + messageParameters = Map("relationName" -> s"`$tableName`")) + } +} diff --git a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/HoodieSpark41SchemaUtils.scala b/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/HoodieSpark4SchemaUtils.scala similarity index 93% copy from hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/HoodieSpark41SchemaUtils.scala copy to hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/HoodieSpark4SchemaUtils.scala index c68190466ca8..433394d332d1 100644 --- a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/HoodieSpark41SchemaUtils.scala +++ b/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/HoodieSpark4SchemaUtils.scala @@ -29,9 +29,9 @@ import org.apache.spark.sql.util.SchemaUtils import java.sql.{Connection, ResultSet} /** - * Utils on schema for Spark 3.4+. + * Utils on schema shared by all supported Spark 4.x versions. */ -object HoodieSpark41SchemaUtils extends HoodieSchemaUtils { +abstract class HoodieSpark4SchemaUtils extends HoodieSchemaUtils { override def checkColumnNameDuplication(columnNames: Seq[String], colType: String, caseSensitiveAnalysis: Boolean): Unit = { diff --git a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieSpark41Analysis.scala b/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieSpark4Analysis.scala similarity index 55% copy from hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieSpark41Analysis.scala copy to hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieSpark4Analysis.scala index 33dc723543e9..8d4266011183 100644 --- a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieSpark41Analysis.scala +++ b/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieSpark4Analysis.scala @@ -17,62 +17,16 @@ package org.apache.spark.sql.hudi.analysis -import org.apache.hudi.{DefaultSource, EmptyRelation, HoodieBaseRelation} -import org.apache.hudi.SparkAdapterSupport.sparkAdapter - -import org.apache.spark.sql.{AnalysisException, SparkSession} +import org.apache.spark.sql.AnalysisException import org.apache.spark.sql.catalyst.analysis.{ResolveInsertionBase, TableOutputResolver} -import org.apache.spark.sql.catalyst.catalog.{CatalogTable, HiveTableRelation} -import org.apache.spark.sql.catalyst.plans.logical._ -import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.catalyst.catalog.CatalogTable +import org.apache.spark.sql.catalyst.plans.logical.InsertIntoStatement import org.apache.spark.sql.errors.DataTypeErrors.toSQLId import org.apache.spark.sql.errors.QueryCompilationErrors -import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation, PreprocessTableInsertion} -import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation -import org.apache.spark.sql.hudi.ProvidesHoodieConfig -import org.apache.spark.sql.hudi.catalog.HoodieInternalV2Table -import org.apache.spark.sql.sources.InsertableRelation +import org.apache.spark.sql.execution.datasources.PreprocessTableInsertion import org.apache.spark.sql.types.StructType import org.apache.spark.sql.util.PartitioningUtils.normalizePartitionSpec -/** - * NOTE: PLEASE READ CAREFULLY - * - * Since Hudi relations don't currently implement DS V2 Read API, we have to fallback to V1 here. - * Such fallback will have considerable performance impact, therefore it's only performed in cases - * where V2 API have to be used. Currently only such use-case is using of Schema Evolution feature - * - * Check out HUDI-4178 for more details - */ -case class HoodieSpark41DataSourceV2ToV1Fallback(sparkSession: SparkSession) extends Rule[LogicalPlan] - with ProvidesHoodieConfig { - - override def apply(plan: LogicalPlan): LogicalPlan = plan match { - // The only place we're avoiding fallback is in [[AlterTableCommand]]s since - // current implementation relies on DSv2 features - case _: AlterTableCommand => plan - - // NOTE: Unfortunately, [[InsertIntoStatement]] is implemented in a way that doesn't expose - // target relation as a child (even though there's no good reason for that) - case iis@InsertIntoStatement(rv2@DataSourceV2Relation(v2Table: HoodieInternalV2Table, _, _, _, _, _), _, _, _, _, _, _) => - iis.copy(table = convertToV1(rv2, v2Table)) - - case _ => - plan.resolveOperatorsDown { - case rv2@DataSourceV2Relation(v2Table: HoodieInternalV2Table, _, _, _, _, _) => convertToV1(rv2, v2Table) - } - } - - private def convertToV1(rv2: DataSourceV2Relation, v2Table: HoodieInternalV2Table) = { - val output = rv2.output - val catalogTable = v2Table.catalogTable.map(_ => v2Table.v1Table) - val relation = new DefaultSource().createRelation(sparkSession.sqlContext, - buildHoodieConfig(v2Table.hoodieCatalogTable), v2Table.hoodieCatalogTable.tableSchema) - - LogicalRelation(relation, output, catalogTable, isStreaming = false, Option.empty) - } -} - /** * In Spark 3.5, the following Resolution rules are removed, * [[ResolveUserSpecifiedColumns]] and [[ResolveDefaultColumns]] @@ -85,62 +39,40 @@ case class HoodieSpark41DataSourceV2ToV1Fallback(sparkSession: SparkSession) ext * Note that [[HoodieAnalysis]] intercepts the [[InsertIntoStatement]] after Spark's built-in * Resolution rules are applies, the logic of resolving the user specified columns and default * values may no longer be applied. To make INSERT with a subset of columns specified by user - * to work, this custom resolution rule [[HoodieSpark41ResolveColumnsForInsertInto]] is added - * to achieve the same, before converting [[InsertIntoStatement]] into - * [[InsertIntoHoodieTableCommand]]. + * to work, the custom resolution rules `HoodieSpark4XResolveColumnsForInsertInto` extending + * this base class are added to achieve the same, before converting [[InsertIntoStatement]] + * into [[InsertIntoHoodieTableCommand]]. * * The implementation is copied and adapted from [[PreprocessTableInsertion]] * https://github.com/apache/spark/blob/d061aadf25fd258d2d3e7332a489c9c24a2b5530/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala#L373 * * Also note that, the project logic in [[ResolveImplementationsEarly]] for INSERT is still * needed in the case of INSERT with all columns in a different ordering. + * + * This base class carries the preprocessing logic shared by all supported Spark 4.x versions; + * the per-version subclasses keep only the plan matching that depends on version-specific + * case-class shapes of [[InsertIntoStatement]]. */ -case class HoodieSpark41ResolveColumnsForInsertInto() extends ResolveInsertionBase { - // NOTE: This is copied from [[PreprocessTableInsertion]] with additional handling of Hudi relations - override def apply(plan: LogicalPlan): LogicalPlan = { - plan match { - case i@InsertIntoStatement(table, _, _, query, _, _, _) - if table.resolved && query.resolved - && i.userSpecifiedCols.nonEmpty && i.table.isInstanceOf[LogicalRelation] - && sparkAdapter.isHoodieTable(i.table.asInstanceOf[LogicalRelation].catalogTable.get) => - table match { - case relation: HiveTableRelation => - val metadata = relation.tableMeta - preprocess(i, metadata.identifier.quotedString, metadata.partitionSchema, - Some(metadata)) - case LogicalRelation(h: HadoopFsRelation, _, catalogTable, _, _) => - preprocess(i, catalogTable, h.partitionSchema) - case LogicalRelation(_: InsertableRelation, _, catalogTable, _, _) => - preprocess(i, catalogTable, new StructType()) - // The two conditions below are adapted to Hudi relations - case LogicalRelation(_: EmptyRelation, _, catalogTable, _, _) => - preprocess(i, catalogTable) - case LogicalRelation(_: HoodieBaseRelation, _, catalogTable, _, _) => - preprocess(i, catalogTable) - case _ => i - } - case _ => plan - } - } +abstract class HoodieSpark4ResolveColumnsForInsertInto extends ResolveInsertionBase { - private def preprocess(insert: InsertIntoStatement, - catalogTable: Option[CatalogTable]): InsertIntoStatement = { + protected def preprocess(insert: InsertIntoStatement, + catalogTable: Option[CatalogTable]): InsertIntoStatement = { preprocess(insert, catalogTable, catalogTable.map(_.partitionSchema).getOrElse(new StructType())) } - private def preprocess(insert: InsertIntoStatement, - catalogTable: Option[CatalogTable], - partitionSchema: StructType): InsertIntoStatement = { + protected def preprocess(insert: InsertIntoStatement, + catalogTable: Option[CatalogTable], + partitionSchema: StructType): InsertIntoStatement = { val tblName = catalogTable.map(_.identifier.quotedString).getOrElse("unknown") preprocess(insert, tblName, partitionSchema, catalogTable) } // NOTE: this is copied from [[PreprocessTableInsertion]] with additional logic // to unset user-specified columns at the end - private def preprocess(insert: InsertIntoStatement, - tblName: String, - partColNames: StructType, - catalogTable: Option[CatalogTable]): InsertIntoStatement = { + protected def preprocess(insert: InsertIntoStatement, + tblName: String, + partColNames: StructType, + catalogTable: Option[CatalogTable]): InsertIntoStatement = { val normalizedPartSpec = normalizePartitionSpec( insert.partitionSpec, partColNames, tblName, conf.resolver) diff --git a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/HoodieSpark40CatalystExpressionUtils.scala b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/HoodieSpark40CatalystExpressionUtils.scala index a183f754483e..67c6aea40be5 100644 --- a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/HoodieSpark40CatalystExpressionUtils.scala +++ b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/HoodieSpark40CatalystExpressionUtils.scala @@ -17,101 +17,4 @@ package org.apache.spark.sql -import org.apache.spark.sql.HoodieSparkTypeUtils.isCastPreservingOrdering -import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder -import org.apache.spark.sql.catalyst.expressions.{Add, Attribute, AttributeReference, AttributeSet, BitwiseOr, Cast, DateAdd, DateDiff, DateFormatClass, DateSub, Divide, EvalMode, Exp, Expm1, Expression, FromUnixTime, FromUTCTimestamp, Log, Log10, Log1p, Log2, Lower, Multiply, ParseToDate, ParseToTimestamp, PredicateHelper, ShiftLeft, ShiftRight, ToUnixTimestamp, ToUTCTimestamp, Upper} -import org.apache.spark.sql.execution.datasources.DataSourceStrategy -import org.apache.spark.sql.types.{DataType, StructType} - -object HoodieSpark40CatalystExpressionUtils extends HoodieSpark4CatalystExpressionUtils with PredicateHelper { - - override def getEncoder(schema: StructType): ExpressionEncoder[Row] = { - ExpressionEncoder.apply(schema).resolveAndBind() - } - - override def normalizeExprs(exprs: Seq[Expression], attributes: Seq[Attribute]): Seq[Expression] = { - DataSourceStrategy.normalizeExprs(exprs, attributes) - } - - override def extractPredicatesWithinOutputSet(condition: Expression, outputSet: AttributeSet): Option[Expression] = { - super[PredicateHelper].extractPredicatesWithinOutputSet(condition, outputSet) - } - - override def matchCast(expr: Expression): Option[(Expression, DataType, Option[String])] = { - expr match { - case Cast(child, dataType, timeZoneId, _) => Some((child, dataType, timeZoneId)) - case _ => None - } - } - - override def tryMatchAttributeOrderingPreservingTransformation(expr: Expression): Option[AttributeReference] = { - expr match { - case OrderPreservingTransformation(attrRef) => Some(attrRef) - case _ => None - } - } - - def canUpCast(fromType: DataType, toType: DataType): Boolean = - Cast.canUpCast(fromType, toType) - - override def unapplyCastExpression(expr: Expression): Option[(Expression, DataType, Option[String], Boolean)] = - expr match { - case Cast(castedExpr, dataType, timeZoneId, ansiEnabled) => - Some((castedExpr, dataType, timeZoneId, if (ansiEnabled == EvalMode.ANSI) true else false)) - case _ => None - } - - private object OrderPreservingTransformation { - def unapply(expr: Expression): Option[AttributeReference] = { - expr match { - // Date/Time Expressions - case DateFormatClass(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case DateAdd(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case DateSub(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case DateDiff(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case DateDiff(_, OrderPreservingTransformation(attrRef)) => Some(attrRef) - case FromUnixTime(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case FromUTCTimestamp(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case ParseToDate(OrderPreservingTransformation(attrRef), _, _, _) => Some(attrRef) - case ParseToTimestamp(OrderPreservingTransformation(attrRef), _, _, _, _) => Some(attrRef) - case ToUnixTimestamp(OrderPreservingTransformation(attrRef), _, _, _) => Some(attrRef) - case ToUTCTimestamp(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - - // String Expressions - case Lower(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Upper(OrderPreservingTransformation(attrRef)) => Some(attrRef) - // Left API change: Improve RuntimeReplaceable - // https://issues.apache.org/jira/browse/SPARK-38240 - case org.apache.spark.sql.catalyst.expressions.Left(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - - // Math Expressions - // Binary - case Add(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case Add(_, OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case Multiply(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case Multiply(_, OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case Divide(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case BitwiseOr(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case BitwiseOr(_, OrderPreservingTransformation(attrRef)) => Some(attrRef) - // Unary - case Exp(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Expm1(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Log(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Log10(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Log1p(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Log2(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case ShiftLeft(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case ShiftRight(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - - // Other - case cast @ Cast(OrderPreservingTransformation(attrRef), _, _, _) - if isCastPreservingOrdering(cast.child.dataType, cast.dataType) => Some(attrRef) - - // Identity transformation - case attrRef: AttributeReference => Some(attrRef) - // No match - case _ => None - } - } - } -} +object HoodieSpark40CatalystExpressionUtils extends HoodieSpark4CatalystExpressionUtils diff --git a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/HoodieSpark40CatalystPlanUtils.scala b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/HoodieSpark40CatalystPlanUtils.scala index a6641def7e52..f4838b9bc8af 100644 --- a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/HoodieSpark40CatalystPlanUtils.scala +++ b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/HoodieSpark40CatalystPlanUtils.scala @@ -18,128 +18,11 @@ package org.apache.spark.sql -import org.apache.spark.sql.catalyst.TableIdentifier -import org.apache.spark.sql.catalyst.analysis.{AnalysisErrorAt, ResolvedTable} -import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeSet, Expression, ProjectionOverSchema} -import org.apache.spark.sql.catalyst.planning.ScanOperation -import org.apache.spark.sql.catalyst.plans.logical._ -import org.apache.spark.sql.connector.catalog.{Identifier, Table, TableCatalog} -import org.apache.spark.sql.execution.command.RepairTableCommand -import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation} -import org.apache.spark.sql.execution.datasources.parquet.{HoodieFormatTrait, ParquetFileFormat} +import org.apache.spark.sql.catalyst.expressions.Expression +import org.apache.spark.sql.catalyst.plans.logical.{Assignment, UpdateAction} import org.apache.spark.sql.execution.streaming.SerializedOffset -import org.apache.spark.sql.types.StructType -object HoodieSpark40CatalystPlanUtils extends BaseHoodieCatalystPlanUtils { - - def unapplyResolvedTable(plan: LogicalPlan): Option[(TableCatalog, Identifier, Table)] = - plan match { - case ResolvedTable(catalog, identifier, table, _) => Some((catalog, identifier, table)) - case _ => None - } - - override def unapplyMergeIntoTable(plan: LogicalPlan): Option[(LogicalPlan, LogicalPlan, Expression)] = { - plan match { - case MergeIntoTable(targetTable, sourceTable, mergeCondition, _, _, _, _) => - Some((targetTable, sourceTable, mergeCondition)) - case _ => None - } - } - - override def maybeApplyForNewFileFormat(plan: LogicalPlan): LogicalPlan = { - plan match { - case s@ScanOperation(_, _, _, - l@LogicalRelation(fs: HadoopFsRelation, _, _, _, _)) - if fs.fileFormat.isInstanceOf[ParquetFileFormat with HoodieFormatTrait] - && !fs.fileFormat.asInstanceOf[ParquetFileFormat with HoodieFormatTrait].isProjected => - FileFormatUtilsForFileGroupReader.applyNewFileFormatChanges(s, l, fs) - case _ => plan - } - } - - override def projectOverSchema(schema: StructType, output: AttributeSet): ProjectionOverSchema = - ProjectionOverSchema(schema, output) - - override def isRepairTable(plan: LogicalPlan): Boolean = { - plan.isInstanceOf[RepairTableCommand] - } - - override def getRepairTableChildren(plan: LogicalPlan): Option[(TableIdentifier, Boolean, Boolean, String)] = { - plan match { - case rtc: RepairTableCommand => - Some((rtc.tableName, rtc.enableAddPartitions, rtc.enableDropPartitions, rtc.cmd)) - case _ => - None - } - } - - override def failAnalysisForMIT(a: Attribute, cols: String): Unit = { - a.failAnalysis( - errorClass = "UNRESOLVED_COLUMN.WITH_SUGGESTION", - messageParameters = Map( - "objectName" -> a.sql, - "proposal" -> cols)) - } - - override def failTableNotFound(tableName: String): Unit = { - throw new AnalysisException( - errorClass = "TABLE_OR_VIEW_NOT_FOUND", - messageParameters = Map("relationName" -> s"`$tableName`")) - } - - override def unapplyCreateIndex(plan: LogicalPlan): Option[(LogicalPlan, String, String, Boolean, Seq[(Seq[String], Map[String, String])], Map[String, String])] = { - plan match { - case ci@CreateIndex(table, indexName, indexType, ignoreIfExists, columns, properties) => - Some((table, indexName, indexType, ignoreIfExists, columns.map(col => (col._1.name, col._2)), properties)) - case _ => - None - } - } - - override def unapplyDropIndex(plan: LogicalPlan): Option[(LogicalPlan, String, Boolean)] = { - plan match { - case ci@DropIndex(table, indexName, ignoreIfNotExists) => - Some((table, indexName, ignoreIfNotExists)) - case _ => - None - } - } - - override def unapplyShowIndexes(plan: LogicalPlan): Option[(LogicalPlan, Seq[Attribute])] = { - plan match { - case ci@HoodieShowIndexes(table, output) => - Some((table, output)) - case _ => - None - } - } - - override def unapplyRefreshIndex(plan: LogicalPlan): Option[(LogicalPlan, String)] = { - plan match { - case ci@RefreshIndex(table, indexName) => - Some((table, indexName)) - case _ => - None - } - } - - override def unapplyInsertIntoStatement(plan: LogicalPlan): Option[(LogicalPlan, Seq[String], Map[String, Option[String]], LogicalPlan, Boolean, Boolean)] = { - plan match { - case insert: InsertIntoStatement => - Some((insert.table, insert.userSpecifiedCols, insert.partitionSpec, insert.query, insert.overwrite, insert.ifPartitionNotExists)) - case _ => - None - } - } - - override def createProjectForByNameQuery(lr: LogicalRelation, plan: LogicalPlan): Option[LogicalPlan] = { - plan match { - case insert: InsertIntoStatement => - Some(ResolveInsertionBase.createProjectForByNameQuery(lr.catalogTable.get.qualifiedName, insert)) - case _ => - None - } - } +object HoodieSpark40CatalystPlanUtils extends HoodieSpark4CatalystPlanUtils { override def unapplyUpdateAction(mergeAction: Any): Option[(Option[Expression], Seq[Assignment])] = { mergeAction match { diff --git a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/HoodieSpark40SchemaUtils.scala b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/HoodieSpark40SchemaUtils.scala index bda84f2c1bf3..44547445db83 100644 --- a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/HoodieSpark40SchemaUtils.scala +++ b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/HoodieSpark40SchemaUtils.scala @@ -19,34 +19,7 @@ package org.apache.spark.sql -import org.apache.spark.sql.catalyst.expressions.Attribute -import org.apache.spark.sql.catalyst.types.DataTypeUtils -import org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils -import org.apache.spark.sql.jdbc.JdbcDialect -import org.apache.spark.sql.types.StructType -import org.apache.spark.sql.util.SchemaUtils - -import java.sql.{Connection, ResultSet} - /** - * Utils on schema for Spark 3.4+. + * Utils on schema for Spark 4.0. */ -object HoodieSpark40SchemaUtils extends HoodieSchemaUtils { - override def checkColumnNameDuplication(columnNames: Seq[String], - colType: String, - caseSensitiveAnalysis: Boolean): Unit = { - SchemaUtils.checkColumnNameDuplication(columnNames, caseSensitiveAnalysis) - } - - override def toAttributes(struct: StructType): Seq[Attribute] = { - DataTypeUtils.toAttributes(struct) - } - - override def getSchema(conn: Connection, - resultSet: ResultSet, - dialect: JdbcDialect, - alwaysNullable: Boolean = false, - isTimestampNTZ: Boolean = false): StructType = { - JdbcUtils.getSchema(conn, resultSet, dialect, alwaysNullable) - } -} +object HoodieSpark40SchemaUtils extends HoodieSpark4SchemaUtils diff --git a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieSpark40Analysis.scala b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieSpark40Analysis.scala index ef92ebc6504a..200fe39af471 100644 --- a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieSpark40Analysis.scala +++ b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieSpark40Analysis.scala @@ -20,21 +20,16 @@ package org.apache.spark.sql.hudi.analysis import org.apache.hudi.{DefaultSource, EmptyRelation, HoodieBaseRelation} import org.apache.hudi.SparkAdapterSupport.sparkAdapter -import org.apache.spark.sql.{AnalysisException, SparkSession, SQLContext} -import org.apache.spark.sql.catalyst.analysis.{ResolveInsertionBase, TableOutputResolver} -import org.apache.spark.sql.catalyst.catalog.{CatalogTable, HiveTableRelation} +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.catalyst.catalog.HiveTableRelation import org.apache.spark.sql.catalyst.plans.logical._ import org.apache.spark.sql.catalyst.rules.Rule -import org.apache.spark.sql.errors.DataTypeErrors.toSQLId -import org.apache.spark.sql.errors.QueryCompilationErrors -import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation, PreprocessTableInsertion} -import org.apache.spark.sql.execution.datasources.LogicalRelation +import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation} import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation import org.apache.spark.sql.hudi.ProvidesHoodieConfig import org.apache.spark.sql.hudi.catalog.HoodieInternalV2Table import org.apache.spark.sql.sources.InsertableRelation import org.apache.spark.sql.types.StructType -import org.apache.spark.sql.util.PartitioningUtils.normalizePartitionSpec /** * NOTE: PLEASE READ CAREFULLY @@ -75,28 +70,11 @@ case class HoodieSpark40DataSourceV2ToV1Fallback(sparkSession: SparkSession) ext } /** - * In Spark 3.5, the following Resolution rules are removed, - * [[ResolveUserSpecifiedColumns]] and [[ResolveDefaultColumns]] - * (see code changes in [[org.apache.spark.sql.catalyst.analysis.Analyzer]] - * from https://github.com/apache/spark/pull/41262). - * The same logic of resolving the user specified columns and default values, - * which are required for a subset of columns as user specified compared to the table - * schema to work properly, are deferred to [[PreprocessTableInsertion]] for v1 INSERT. - * - * Note that [[HoodieAnalysis]] intercepts the [[InsertIntoStatement]] after Spark's built-in - * Resolution rules are applies, the logic of resolving the user specified columns and default - * values may no longer be applied. To make INSERT with a subset of columns specified by user - * to work, this custom resolution rule [[HoodieSpark40ResolveColumnsForInsertInto]] is added - * to achieve the same, before converting [[InsertIntoStatement]] into - * [[InsertIntoHoodieTableCommand]]. - * - * The implementation is copied and adapted from [[PreprocessTableInsertion]] - * https://github.com/apache/spark/blob/d061aadf25fd258d2d3e7332a489c9c24a2b5530/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala#L373 - * - * Also note that, the project logic in [[ResolveImplementationsEarly]] for INSERT is still - * needed in the case of INSERT with all columns in a different ordering. + * Resolution rule resolving the user specified columns and default values of + * [[InsertIntoStatement]] for Spark 4.0; see [[HoodieSpark4ResolveColumnsForInsertInto]] + * for the shared preprocessing logic and the rationale. */ -case class HoodieSpark40ResolveColumnsForInsertInto() extends ResolveInsertionBase { +case class HoodieSpark40ResolveColumnsForInsertInto() extends HoodieSpark4ResolveColumnsForInsertInto { // NOTE: This is copied from [[PreprocessTableInsertion]] with additional handling of Hudi relations override def apply(plan: LogicalPlan): LogicalPlan = { plan match { @@ -123,90 +101,4 @@ case class HoodieSpark40ResolveColumnsForInsertInto() extends ResolveInsertionBa case _ => plan } } - - private def preprocess(insert: InsertIntoStatement, - catalogTable: Option[CatalogTable]): InsertIntoStatement = { - preprocess(insert, catalogTable, catalogTable.map(_.partitionSchema).getOrElse(new StructType())) - } - - private def preprocess(insert: InsertIntoStatement, - catalogTable: Option[CatalogTable], - partitionSchema: StructType): InsertIntoStatement = { - val tblName = catalogTable.map(_.identifier.quotedString).getOrElse("unknown") - preprocess(insert, tblName, partitionSchema, catalogTable) - } - - // NOTE: this is copied from [[PreprocessTableInsertion]] with additional logic - // to unset user-specified columns at the end - private def preprocess(insert: InsertIntoStatement, - tblName: String, - partColNames: StructType, - catalogTable: Option[CatalogTable]): InsertIntoStatement = { - - val normalizedPartSpec = normalizePartitionSpec( - insert.partitionSpec, partColNames, tblName, conf.resolver) - - val staticPartCols = normalizedPartSpec.filter(_._2.isDefined).keySet - val expectedColumns = insert.table.output.filterNot(a => staticPartCols.contains(a.name)) - - val partitionsTrackedByCatalog = catalogTable.isDefined && - catalogTable.get.partitionColumnNames.nonEmpty && - catalogTable.get.tracksPartitionsInCatalog - if (partitionsTrackedByCatalog && normalizedPartSpec.nonEmpty) { - // empty partition column value - if (normalizedPartSpec.values.flatten.exists(v => v != null && v.isEmpty)) { - val spec = normalizedPartSpec.map(p => p._1 + "=" + p._2).mkString("[", ", ", "]") - throw QueryCompilationErrors.invalidPartitionSpecError( - s"The spec ($spec) contains an empty partition column value") - } - } - - // Create a project if this INSERT has a user-specified column list. - val hasColumnList = insert.userSpecifiedCols.nonEmpty - val query = if (hasColumnList) { - createProjectForByNameQuery(tblName, insert) - } else { - insert.query - } - val newQuery = try { - TableOutputResolver.resolveOutputColumns( - tblName, - expectedColumns, - query, - byName = hasColumnList || insert.byName, - conf, - supportColDefaultValue = true) - } catch { - case e: AnalysisException if staticPartCols.nonEmpty && - (e.getErrorClass == "INSERT_COLUMN_ARITY_MISMATCH.NOT_ENOUGH_DATA_COLUMNS" || - e.getErrorClass == "INSERT_COLUMN_ARITY_MISMATCH.TOO_MANY_DATA_COLUMNS") => - val newException = e.copy( - errorClass = Some("INSERT_PARTITION_COLUMN_ARITY_MISMATCH"), - messageParameters = e.messageParameters ++ Map( - "tableColumns" -> insert.table.output.map(c => toSQLId(c.name)).mkString(", "), - "staticPartCols" -> staticPartCols.toSeq.sorted.map(c => toSQLId(c)).mkString(", ") - )) - newException.setStackTrace(e.getStackTrace) - throw newException - } - if (normalizedPartSpec.nonEmpty) { - if (normalizedPartSpec.size != partColNames.length) { - throw QueryCompilationErrors.requestedPartitionsMismatchTablePartitionsError( - tblName, normalizedPartSpec, partColNames) - } - - // NOTE: Hudi converts [[InsertIntoStatement]] to [[InsertIntoHoodieTableCommand]] - // and the user specified is no longer need after resolution - // (`userSpecifiedCols = Seq()`) - insert.copy(query = newQuery, partitionSpec = normalizedPartSpec, userSpecifiedCols = Seq()) - } else { - // All partition columns are dynamic because the InsertIntoTable command does - // not explicitly specify partitioning columns. - // NOTE: Hudi converts [[InsertIntoStatement]] to [[InsertIntoHoodieTableCommand]] - // and the user specified is no longer need after resolution - // (`userSpecifiedCols = Seq()`) - insert.copy(query = newQuery, partitionSpec = partColNames.map(_.name).map(_ -> None).toMap, - userSpecifiedCols = Seq()) - } - } } diff --git a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/HoodieSpark41CatalystExpressionUtils.scala b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/HoodieSpark41CatalystExpressionUtils.scala index f835d7daace0..c41389f31534 100644 --- a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/HoodieSpark41CatalystExpressionUtils.scala +++ b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/HoodieSpark41CatalystExpressionUtils.scala @@ -17,101 +17,4 @@ package org.apache.spark.sql -import org.apache.spark.sql.HoodieSparkTypeUtils.isCastPreservingOrdering -import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder -import org.apache.spark.sql.catalyst.expressions.{Add, Attribute, AttributeReference, AttributeSet, BitwiseOr, Cast, DateAdd, DateDiff, DateFormatClass, DateSub, Divide, EvalMode, Exp, Expm1, Expression, FromUnixTime, FromUTCTimestamp, Log, Log10, Log1p, Log2, Lower, Multiply, ParseToDate, ParseToTimestamp, PredicateHelper, ShiftLeft, ShiftRight, ToUnixTimestamp, ToUTCTimestamp, Upper} -import org.apache.spark.sql.execution.datasources.DataSourceStrategy -import org.apache.spark.sql.types.{DataType, StructType} - -object HoodieSpark41CatalystExpressionUtils extends HoodieSpark4CatalystExpressionUtils with PredicateHelper { - - override def getEncoder(schema: StructType): ExpressionEncoder[Row] = { - ExpressionEncoder.apply(schema).resolveAndBind() - } - - override def normalizeExprs(exprs: Seq[Expression], attributes: Seq[Attribute]): Seq[Expression] = { - DataSourceStrategy.normalizeExprs(exprs, attributes) - } - - override def extractPredicatesWithinOutputSet(condition: Expression, outputSet: AttributeSet): Option[Expression] = { - super[PredicateHelper].extractPredicatesWithinOutputSet(condition, outputSet) - } - - override def matchCast(expr: Expression): Option[(Expression, DataType, Option[String])] = { - expr match { - case Cast(child, dataType, timeZoneId, _) => Some((child, dataType, timeZoneId)) - case _ => None - } - } - - override def tryMatchAttributeOrderingPreservingTransformation(expr: Expression): Option[AttributeReference] = { - expr match { - case OrderPreservingTransformation(attrRef) => Some(attrRef) - case _ => None - } - } - - def canUpCast(fromType: DataType, toType: DataType): Boolean = - Cast.canUpCast(fromType, toType) - - override def unapplyCastExpression(expr: Expression): Option[(Expression, DataType, Option[String], Boolean)] = - expr match { - case Cast(castedExpr, dataType, timeZoneId, ansiEnabled) => - Some((castedExpr, dataType, timeZoneId, if (ansiEnabled == EvalMode.ANSI) true else false)) - case _ => None - } - - private object OrderPreservingTransformation { - def unapply(expr: Expression): Option[AttributeReference] = { - expr match { - // Date/Time Expressions - case DateFormatClass(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case DateAdd(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case DateSub(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case DateDiff(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case DateDiff(_, OrderPreservingTransformation(attrRef)) => Some(attrRef) - case FromUnixTime(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case FromUTCTimestamp(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case ParseToDate(OrderPreservingTransformation(attrRef), _, _, _) => Some(attrRef) - case ParseToTimestamp(OrderPreservingTransformation(attrRef), _, _, _, _) => Some(attrRef) - case ToUnixTimestamp(OrderPreservingTransformation(attrRef), _, _, _) => Some(attrRef) - case ToUTCTimestamp(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - - // String Expressions - case Lower(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Upper(OrderPreservingTransformation(attrRef)) => Some(attrRef) - // Left API change: Improve RuntimeReplaceable - // https://issues.apache.org/jira/browse/SPARK-38240 - case org.apache.spark.sql.catalyst.expressions.Left(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - - // Math Expressions - // Binary - case Add(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case Add(_, OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case Multiply(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case Multiply(_, OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case Divide(OrderPreservingTransformation(attrRef), _, _) => Some(attrRef) - case BitwiseOr(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case BitwiseOr(_, OrderPreservingTransformation(attrRef)) => Some(attrRef) - // Unary - case Exp(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Expm1(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Log(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Log10(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Log1p(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case Log2(OrderPreservingTransformation(attrRef)) => Some(attrRef) - case ShiftLeft(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - case ShiftRight(OrderPreservingTransformation(attrRef), _) => Some(attrRef) - - // Other - case cast@Cast(OrderPreservingTransformation(attrRef), _, _, _) - if isCastPreservingOrdering(cast.child.dataType, cast.dataType) => Some(attrRef) - - // Identity transformation - case attrRef: AttributeReference => Some(attrRef) - // No match - case _ => None - } - } - } -} +object HoodieSpark41CatalystExpressionUtils extends HoodieSpark4CatalystExpressionUtils diff --git a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/HoodieSpark41CatalystPlanUtils.scala b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/HoodieSpark41CatalystPlanUtils.scala index 6a385810aef9..24489bb4f16a 100644 --- a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/HoodieSpark41CatalystPlanUtils.scala +++ b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/HoodieSpark41CatalystPlanUtils.scala @@ -18,128 +18,11 @@ package org.apache.spark.sql -import org.apache.spark.sql.catalyst.TableIdentifier -import org.apache.spark.sql.catalyst.analysis.{AnalysisErrorAt, ResolvedTable} -import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeSet, Expression, ProjectionOverSchema} -import org.apache.spark.sql.catalyst.planning.ScanOperation -import org.apache.spark.sql.catalyst.plans.logical._ -import org.apache.spark.sql.connector.catalog.{Identifier, Table, TableCatalog} -import org.apache.spark.sql.execution.command.RepairTableCommand -import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation} -import org.apache.spark.sql.execution.datasources.parquet.{HoodieFormatTrait, ParquetFileFormat} +import org.apache.spark.sql.catalyst.expressions.Expression +import org.apache.spark.sql.catalyst.plans.logical.{Assignment, UpdateAction} import org.apache.spark.sql.execution.streaming.runtime.SerializedOffset -import org.apache.spark.sql.types.StructType -object HoodieSpark41CatalystPlanUtils extends BaseHoodieCatalystPlanUtils { - - def unapplyResolvedTable(plan: LogicalPlan): Option[(TableCatalog, Identifier, Table)] = - plan match { - case ResolvedTable(catalog, identifier, table, _) => Some((catalog, identifier, table)) - case _ => None - } - - override def unapplyMergeIntoTable(plan: LogicalPlan): Option[(LogicalPlan, LogicalPlan, Expression)] = { - plan match { - case MergeIntoTable(targetTable, sourceTable, mergeCondition, _, _, _, _) => - Some((targetTable, sourceTable, mergeCondition)) - case _ => None - } - } - - override def maybeApplyForNewFileFormat(plan: LogicalPlan): LogicalPlan = { - plan match { - case s@ScanOperation(_, _, _, - l@LogicalRelation(fs: HadoopFsRelation, _, _, _, _)) - if fs.fileFormat.isInstanceOf[ParquetFileFormat with HoodieFormatTrait] - && !fs.fileFormat.asInstanceOf[ParquetFileFormat with HoodieFormatTrait].isProjected => - FileFormatUtilsForFileGroupReader.applyNewFileFormatChanges(s, l, fs) - case _ => plan - } - } - - override def projectOverSchema(schema: StructType, output: AttributeSet): ProjectionOverSchema = - ProjectionOverSchema(schema, output) - - override def isRepairTable(plan: LogicalPlan): Boolean = { - plan.isInstanceOf[RepairTableCommand] - } - - override def getRepairTableChildren(plan: LogicalPlan): Option[(TableIdentifier, Boolean, Boolean, String)] = { - plan match { - case rtc: RepairTableCommand => - Some((rtc.tableName, rtc.enableAddPartitions, rtc.enableDropPartitions, rtc.cmd)) - case _ => - None - } - } - - override def failAnalysisForMIT(a: Attribute, cols: String): Unit = { - a.failAnalysis( - errorClass = "UNRESOLVED_COLUMN.WITH_SUGGESTION", - messageParameters = Map( - "objectName" -> a.sql, - "proposal" -> cols)) - } - - override def failTableNotFound(tableName: String): Unit = { - throw new AnalysisException( - errorClass = "TABLE_OR_VIEW_NOT_FOUND", - messageParameters = Map("relationName" -> s"`$tableName`")) - } - - override def unapplyCreateIndex(plan: LogicalPlan): Option[(LogicalPlan, String, String, Boolean, Seq[(Seq[String], Map[String, String])], Map[String, String])] = { - plan match { - case ci@CreateIndex(table, indexName, indexType, ignoreIfExists, columns, properties) => - Some((table, indexName, indexType, ignoreIfExists, columns.map(col => (col._1.name, col._2)), properties)) - case _ => - None - } - } - - override def unapplyDropIndex(plan: LogicalPlan): Option[(LogicalPlan, String, Boolean)] = { - plan match { - case ci@DropIndex(table, indexName, ignoreIfNotExists) => - Some((table, indexName, ignoreIfNotExists)) - case _ => - None - } - } - - override def unapplyShowIndexes(plan: LogicalPlan): Option[(LogicalPlan, Seq[Attribute])] = { - plan match { - case ci@HoodieShowIndexes(table, output) => - Some((table, output)) - case _ => - None - } - } - - override def unapplyRefreshIndex(plan: LogicalPlan): Option[(LogicalPlan, String)] = { - plan match { - case ci@RefreshIndex(table, indexName) => - Some((table, indexName)) - case _ => - None - } - } - - override def unapplyInsertIntoStatement(plan: LogicalPlan): Option[(LogicalPlan, Seq[String], Map[String, Option[String]], LogicalPlan, Boolean, Boolean)] = { - plan match { - case insert: InsertIntoStatement => - Some((insert.table, insert.userSpecifiedCols, insert.partitionSpec, insert.query, insert.overwrite, insert.ifPartitionNotExists)) - case _ => - None - } - } - - override def createProjectForByNameQuery(lr: LogicalRelation, plan: LogicalPlan): Option[LogicalPlan] = { - plan match { - case insert: InsertIntoStatement => - Some(ResolveInsertionBase.createProjectForByNameQuery(lr.catalogTable.get.qualifiedName, insert)) - case _ => - None - } - } +object HoodieSpark41CatalystPlanUtils extends HoodieSpark4CatalystPlanUtils { override def unapplyUpdateAction(mergeAction: Any): Option[(Option[Expression], Seq[Assignment])] = { mergeAction match { diff --git a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/HoodieSpark41SchemaUtils.scala b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/HoodieSpark41SchemaUtils.scala index c68190466ca8..149989471f95 100644 --- a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/HoodieSpark41SchemaUtils.scala +++ b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/HoodieSpark41SchemaUtils.scala @@ -19,34 +19,7 @@ package org.apache.spark.sql -import org.apache.spark.sql.catalyst.expressions.Attribute -import org.apache.spark.sql.catalyst.types.DataTypeUtils -import org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils -import org.apache.spark.sql.jdbc.JdbcDialect -import org.apache.spark.sql.types.StructType -import org.apache.spark.sql.util.SchemaUtils - -import java.sql.{Connection, ResultSet} - /** - * Utils on schema for Spark 3.4+. + * Utils on schema for Spark 4.1. */ -object HoodieSpark41SchemaUtils extends HoodieSchemaUtils { - override def checkColumnNameDuplication(columnNames: Seq[String], - colType: String, - caseSensitiveAnalysis: Boolean): Unit = { - SchemaUtils.checkColumnNameDuplication(columnNames, caseSensitiveAnalysis) - } - - override def toAttributes(struct: StructType): Seq[Attribute] = { - DataTypeUtils.toAttributes(struct) - } - - override def getSchema(conn: Connection, - resultSet: ResultSet, - dialect: JdbcDialect, - alwaysNullable: Boolean = false, - isTimestampNTZ: Boolean = false): StructType = { - JdbcUtils.getSchema(conn, resultSet, dialect, alwaysNullable) - } -} +object HoodieSpark41SchemaUtils extends HoodieSpark4SchemaUtils diff --git a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieSpark41Analysis.scala b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieSpark41Analysis.scala index 33dc723543e9..1a57c2976719 100644 --- a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieSpark41Analysis.scala +++ b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieSpark41Analysis.scala @@ -20,20 +20,16 @@ package org.apache.spark.sql.hudi.analysis import org.apache.hudi.{DefaultSource, EmptyRelation, HoodieBaseRelation} import org.apache.hudi.SparkAdapterSupport.sparkAdapter -import org.apache.spark.sql.{AnalysisException, SparkSession} -import org.apache.spark.sql.catalyst.analysis.{ResolveInsertionBase, TableOutputResolver} -import org.apache.spark.sql.catalyst.catalog.{CatalogTable, HiveTableRelation} +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.catalyst.catalog.HiveTableRelation import org.apache.spark.sql.catalyst.plans.logical._ import org.apache.spark.sql.catalyst.rules.Rule -import org.apache.spark.sql.errors.DataTypeErrors.toSQLId -import org.apache.spark.sql.errors.QueryCompilationErrors -import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation, PreprocessTableInsertion} +import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation} import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation import org.apache.spark.sql.hudi.ProvidesHoodieConfig import org.apache.spark.sql.hudi.catalog.HoodieInternalV2Table import org.apache.spark.sql.sources.InsertableRelation import org.apache.spark.sql.types.StructType -import org.apache.spark.sql.util.PartitioningUtils.normalizePartitionSpec /** * NOTE: PLEASE READ CAREFULLY @@ -74,28 +70,11 @@ case class HoodieSpark41DataSourceV2ToV1Fallback(sparkSession: SparkSession) ext } /** - * In Spark 3.5, the following Resolution rules are removed, - * [[ResolveUserSpecifiedColumns]] and [[ResolveDefaultColumns]] - * (see code changes in [[org.apache.spark.sql.catalyst.analysis.Analyzer]] - * from https://github.com/apache/spark/pull/41262). - * The same logic of resolving the user specified columns and default values, - * which are required for a subset of columns as user specified compared to the table - * schema to work properly, are deferred to [[PreprocessTableInsertion]] for v1 INSERT. - * - * Note that [[HoodieAnalysis]] intercepts the [[InsertIntoStatement]] after Spark's built-in - * Resolution rules are applies, the logic of resolving the user specified columns and default - * values may no longer be applied. To make INSERT with a subset of columns specified by user - * to work, this custom resolution rule [[HoodieSpark41ResolveColumnsForInsertInto]] is added - * to achieve the same, before converting [[InsertIntoStatement]] into - * [[InsertIntoHoodieTableCommand]]. - * - * The implementation is copied and adapted from [[PreprocessTableInsertion]] - * https://github.com/apache/spark/blob/d061aadf25fd258d2d3e7332a489c9c24a2b5530/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/rules.scala#L373 - * - * Also note that, the project logic in [[ResolveImplementationsEarly]] for INSERT is still - * needed in the case of INSERT with all columns in a different ordering. + * Resolution rule resolving the user specified columns and default values of + * [[InsertIntoStatement]] for Spark 4.1; see [[HoodieSpark4ResolveColumnsForInsertInto]] + * for the shared preprocessing logic and the rationale. */ -case class HoodieSpark41ResolveColumnsForInsertInto() extends ResolveInsertionBase { +case class HoodieSpark41ResolveColumnsForInsertInto() extends HoodieSpark4ResolveColumnsForInsertInto { // NOTE: This is copied from [[PreprocessTableInsertion]] with additional handling of Hudi relations override def apply(plan: LogicalPlan): LogicalPlan = { plan match { @@ -122,90 +101,4 @@ case class HoodieSpark41ResolveColumnsForInsertInto() extends ResolveInsertionBa case _ => plan } } - - private def preprocess(insert: InsertIntoStatement, - catalogTable: Option[CatalogTable]): InsertIntoStatement = { - preprocess(insert, catalogTable, catalogTable.map(_.partitionSchema).getOrElse(new StructType())) - } - - private def preprocess(insert: InsertIntoStatement, - catalogTable: Option[CatalogTable], - partitionSchema: StructType): InsertIntoStatement = { - val tblName = catalogTable.map(_.identifier.quotedString).getOrElse("unknown") - preprocess(insert, tblName, partitionSchema, catalogTable) - } - - // NOTE: this is copied from [[PreprocessTableInsertion]] with additional logic - // to unset user-specified columns at the end - private def preprocess(insert: InsertIntoStatement, - tblName: String, - partColNames: StructType, - catalogTable: Option[CatalogTable]): InsertIntoStatement = { - - val normalizedPartSpec = normalizePartitionSpec( - insert.partitionSpec, partColNames, tblName, conf.resolver) - - val staticPartCols = normalizedPartSpec.filter(_._2.isDefined).keySet - val expectedColumns = insert.table.output.filterNot(a => staticPartCols.contains(a.name)) - - val partitionsTrackedByCatalog = catalogTable.isDefined && - catalogTable.get.partitionColumnNames.nonEmpty && - catalogTable.get.tracksPartitionsInCatalog - if (partitionsTrackedByCatalog && normalizedPartSpec.nonEmpty) { - // empty partition column value - if (normalizedPartSpec.values.flatten.exists(v => v != null && v.isEmpty)) { - val spec = normalizedPartSpec.map(p => p._1 + "=" + p._2).mkString("[", ", ", "]") - throw QueryCompilationErrors.invalidPartitionSpecError( - s"The spec ($spec) contains an empty partition column value") - } - } - - // Create a project if this INSERT has a user-specified column list. - val hasColumnList = insert.userSpecifiedCols.nonEmpty - val query = if (hasColumnList) { - createProjectForByNameQuery(tblName, insert) - } else { - insert.query - } - val newQuery = try { - TableOutputResolver.resolveOutputColumns( - tblName, - expectedColumns, - query, - byName = hasColumnList || insert.byName, - conf, - supportColDefaultValue = true) - } catch { - case e: AnalysisException if staticPartCols.nonEmpty && - (e.getErrorClass == "INSERT_COLUMN_ARITY_MISMATCH.NOT_ENOUGH_DATA_COLUMNS" || - e.getErrorClass == "INSERT_COLUMN_ARITY_MISMATCH.TOO_MANY_DATA_COLUMNS") => - val newException = e.copy( - errorClass = Some("INSERT_PARTITION_COLUMN_ARITY_MISMATCH"), - messageParameters = e.messageParameters ++ Map( - "tableColumns" -> insert.table.output.map(c => toSQLId(c.name)).mkString(", "), - "staticPartCols" -> staticPartCols.toSeq.sorted.map(c => toSQLId(c)).mkString(", ") - )) - newException.setStackTrace(e.getStackTrace) - throw newException - } - if (normalizedPartSpec.nonEmpty) { - if (normalizedPartSpec.size != partColNames.length) { - throw QueryCompilationErrors.requestedPartitionsMismatchTablePartitionsError( - tblName, normalizedPartSpec, partColNames) - } - - // NOTE: Hudi converts [[InsertIntoStatement]] to [[InsertIntoHoodieTableCommand]] - // and the user specified is no longer need after resolution - // (`userSpecifiedCols = Seq()`) - insert.copy(query = newQuery, partitionSpec = normalizedPartSpec, userSpecifiedCols = Seq()) - } else { - // All partition columns are dynamic because the InsertIntoTable command does - // not explicitly specify partitioning columns. - // NOTE: Hudi converts [[InsertIntoStatement]] to [[InsertIntoHoodieTableCommand]] - // and the user specified is no longer need after resolution - // (`userSpecifiedCols = Seq()`) - insert.copy(query = newQuery, partitionSpec = partColNames.map(_.name).map(_ -> None).toMap, - userSpecifiedCols = Seq()) - } - } }
