You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by vh...@apache.org on 2026/08/05 17:30:34 UTC
(hudi) 04/40: refactor(spark): consolidate duplicated small utils across Spark version modules (#19150)
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 ec8e678d8766413ca1194a65fd8ede49e6786392
Author: Y Ethan Guo <et...@gmail.com>
AuthorDate: Thu Jul 2 22:50:35 2026 -0700
refactor(spark): consolidate duplicated small utils across Spark version modules (#19150)
* refactor(spark): delete unused SparkXXDataSourceUtils from all Spark version modules
The six SparkXXDataSourceUtils objects (Spark33-Spark42) were vendored
copies of Spark 3.2-era rebase-mode helpers (int96RebaseMode /
datetimeRebaseMode returning LegacyBehaviorPolicy). Nothing in the
repository references them anymore: all rebase handling in
SparkXXParquetReader, SparkXXLegacyHoodieParquetFileFormat, and the
Avro (de)serializers goes through Spark's own
org.apache.spark.sql.execution.datasources.DataSourceUtils
(datetimeRebaseSpec / int96RebaseSpec) instead.
Deleting dead code beats deduplicating it.
(cherry picked from commit ed9e368510315617f4b8e59fb32ea51c5c9ca867)
---
.../scala/org/apache/hudi/HoodieFileScanRDD.scala} | 10 +--
...uning.scala => HoodieNestedSchemaPruning.scala} | 49 ++++++++++--
.../datasources/orc/SparkOrcReaderBase.scala | 76 +++++++++++++++---
.../hudi/BaseResolveHudiAlterTableCommand.scala} | 43 +++++-----
.../spark/sql/hudi/analysis/HoodieAnalysis.scala | 34 ++------
.../spark/sql/adapter/BaseSpark3Adapter.scala | 12 ++-
.../hudi/Spark3ResolveHudiAlterTableCommand.scala | 43 ++++++++++
.../apache/spark/sql/adapter/Spark3_3Adapter.scala | 16 +---
.../datasources/Spark33NestedSchemaPruning.scala | 62 ---------------
.../datasources/orc/Spark33OrcReader.scala | 85 --------------------
.../parquet/Spark33DataSourceUtils.scala | 77 ------------------
.../hudi/Spark33ResolveHudiAlterTableCommand.scala | 67 ----------------
.../org/apache/hudi/Spark34HoodieFileScanRDD.scala | 36 ---------
.../apache/spark/sql/adapter/Spark3_4Adapter.scala | 16 +---
.../datasources/Spark34NestedSchemaPruning.scala | 62 ---------------
.../datasources/orc/Spark34OrcReader.scala | 91 ---------------------
.../parquet/Spark34DataSourceUtils.scala | 77 ------------------
.../hudi/Spark34ResolveHudiAlterTableCommand.scala | 67 ----------------
.../org/apache/hudi/Spark35HoodieFileScanRDD.scala | 36 ---------
.../apache/spark/sql/adapter/Spark3_5Adapter.scala | 16 +---
.../datasources/Spark35NestedSchemaPruning.scala | 61 --------------
.../datasources/orc/Spark35OrcReader.scala | 92 ----------------------
.../parquet/Spark35DataSourceUtils.scala | 76 ------------------
.../spark/sql/adapter/BaseSpark4Adapter.scala | 25 +++++-
.../HoodieSpark4PartitionedFileUtils.scala} | 4 +-
.../hudi/Spark4ResolveHudiAlterTableCommand.scala | 37 +++++++++
.../org/apache/hudi/Spark40HoodieFileScanRDD.scala | 36 ---------
.../apache/spark/sql/adapter/Spark4_0Adapter.scala | 32 +-------
.../HoodieSpark40PartitionedFileUtils.scala | 68 ----------------
.../datasources/Spark40NestedSchemaPruning.scala | 61 --------------
.../datasources/orc/Spark40OrcReader.scala | 92 ----------------------
.../parquet/Spark40DataSourceUtils.scala | 76 ------------------
.../hudi/Spark40ResolveHudiAlterTableCommand.scala | 68 ----------------
.../org/apache/hudi/Spark41HoodieFileScanRDD.scala | 36 ---------
.../apache/spark/sql/adapter/Spark4_1Adapter.scala | 33 +-------
.../datasources/Spark41NestedSchemaPruning.scala | 61 --------------
.../datasources/orc/Spark41OrcReader.scala | 92 ----------------------
.../parquet/Spark41DataSourceUtils.scala | 76 ------------------
.../hudi/Spark41ResolveHudiAlterTableCommand.scala | 68 ----------------
39 files changed, 272 insertions(+), 1797 deletions(-)
diff --git a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/hudi/Spark33HoodieFileScanRDD.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieFileScanRDD.scala
similarity index 77%
rename from hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/hudi/Spark33HoodieFileScanRDD.scala
rename to hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieFileScanRDD.scala
index b2ed3bce2332..92e9caf0d635 100644
--- a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/hudi/Spark33HoodieFileScanRDD.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieFileScanRDD.scala
@@ -24,11 +24,11 @@ import org.apache.spark.sql.catalyst.expressions.AttributeReference
import org.apache.spark.sql.execution.datasources.{FilePartition, FileScanRDD, PartitionedFile}
import org.apache.spark.sql.types.StructType
-class Spark33HoodieFileScanRDD(@transient private val sparkSession: SparkSession,
- read: PartitionedFile => Iterator[InternalRow],
- @transient filePartitions: Seq[FilePartition],
- readDataSchema: StructType,
- metadataColumns: Seq[AttributeReference] = Seq.empty)
+class HoodieFileScanRDD(@transient private val sparkSession: SparkSession,
+ read: PartitionedFile => Iterator[InternalRow],
+ @transient filePartitions: Seq[FilePartition],
+ readDataSchema: StructType,
+ metadataColumns: Seq[AttributeReference] = Seq.empty)
extends FileScanRDD(sparkSession, read, filePartitions, readDataSchema, metadataColumns)
with HoodieUnsafeRDD {
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/BaseHoodieNestedSchemaPruning.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/HoodieNestedSchemaPruning.scala
similarity index 77%
rename from hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/BaseHoodieNestedSchemaPruning.scala
rename to hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/HoodieNestedSchemaPruning.scala
index 9b6e26984401..42e59eea5c51 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/BaseHoodieNestedSchemaPruning.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/HoodieNestedSchemaPruning.scala
@@ -37,12 +37,9 @@ import org.apache.spark.sql.util.SchemaUtils.restoreOriginalOutputNames
* NOTE: This class is borrowed from Spark 3.2.1, with modifications adapting it to handle [[HoodieBaseRelation]],
* instead of [[HadoopFsRelation]]
*/
-abstract class BaseHoodieNestedSchemaPruning extends Rule[LogicalPlan] {
+class HoodieNestedSchemaPruning extends Rule[LogicalPlan] {
import org.apache.spark.sql.catalyst.expressions.SchemaPruning._
- // Prune the given output to make it consistent with `requiredSchema`.
- protected def getPrunedOutput(output: Seq[AttributeReference], requiredSchema: StructType): Seq[AttributeReference]
-
override def apply(plan: LogicalPlan): LogicalPlan =
if (conf.nestedSchemaPruningEnabled) {
apply0(plan)
@@ -50,13 +47,51 @@ abstract class BaseHoodieNestedSchemaPruning extends Rule[LogicalPlan] {
plan
}
- protected def apply0(plan: LogicalPlan): LogicalPlan
+ private def apply0(plan: LogicalPlan): LogicalPlan =
+ plan transformDown {
+ // NOTE: The relation is matched by type rather than by destructuring [[LogicalRelation]],
+ // since the arity of its unapply differs across the Spark versions this module
+ // compiles against. This is modified to accommodate for Hudi's custom relations,
+ // given that original [[NestedSchemaPruning]] rule is tightly coupled w/
+ // [[HadoopFsRelation]]
+ // TODO generalize to any file-based relation
+ case op @ PhysicalOperation(projects, filters, l: LogicalRelation) =>
+ l.relation match {
+ case relation: HoodieBaseRelation if relation.canPruneRelationSchema =>
+ prunePhysicalColumns(l.output, projects, filters, relation.dataSchema,
+ prunedDataSchema => {
+ val prunedRelation =
+ relation.updatePrunedDataSchema(prunedSchema = prunedDataSchema)
+ buildPrunedRelation(l, prunedRelation)
+ }).getOrElse(op)
+ case _ => op
+ }
+ }
+
+ // Prune the given output to make it consistent with `requiredSchema`.
+ private def getPrunedOutput(output: Seq[AttributeReference],
+ requiredSchema: StructType): Seq[AttributeReference] = {
+ // We need to replace the expression ids of the pruned relation output attributes
+ // with the expression ids of the original relation output attributes so that
+ // references to the original relation's output are not broken
+ val outputIdMap = output.map(att => (att.name, att.exprId)).toMap
+ // NOTE: The attributes are constructed inline (equivalent to StructType#toAttributes before
+ // Spark 3.5 and DataTypeUtils#toAttributes since, see SPARK-44353) so that this code
+ // compiles against every supported Spark version
+ requiredSchema
+ .map(f => AttributeReference(f.name, f.dataType, f.nullable, f.metadata)())
+ .map {
+ case att if outputIdMap.contains(att.name) =>
+ att.withExprId(outputIdMap(att.name))
+ case att => att
+ }
+ }
/**
* This method returns optional logical plan. `None` is returned if no nested field is required or
* all nested fields are required.
*/
- protected def prunePhysicalColumns(output: Seq[AttributeReference],
+ private def prunePhysicalColumns(output: Seq[AttributeReference],
projects: Seq[NamedExpression],
filters: Seq[Expression],
dataSchema: StructType,
@@ -148,7 +183,7 @@ abstract class BaseHoodieNestedSchemaPruning extends Rule[LogicalPlan] {
* Builds a pruned logical relation from the output of the output relation and the schema of the
* pruned base relation.
*/
- protected def buildPrunedRelation(outputRelation: LogicalRelation,
+ private def buildPrunedRelation(outputRelation: LogicalRelation,
prunedBaseRelation: BaseRelation): LogicalRelation = {
val prunedOutput = getPrunedOutput(outputRelation.output, prunedBaseRelation.schema)
outputRelation.copy(relation = prunedBaseRelation, output = prunedOutput)
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/orc/SparkOrcReaderBase.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/orc/SparkOrcReaderBase.scala
index ca16e323246d..fa056921aeac 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/orc/SparkOrcReaderBase.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/orc/SparkOrcReaderBase.scala
@@ -19,6 +19,7 @@
package org.apache.spark.sql.execution.datasources.orc
+import org.apache.hudi.SparkAdapterSupport
import org.apache.hudi.common.util
import org.apache.hudi.internal.schema.InternalSchema
import org.apache.hudi.storage.StorageConfiguration
@@ -32,19 +33,25 @@ import org.apache.orc.{OrcConf, OrcFile, TypeDescription}
import org.apache.orc.mapred.OrcStruct
import org.apache.orc.mapreduce.OrcInputFormat
import org.apache.spark.TaskContext
+import org.apache.spark.memory.MemoryMode
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.{Attribute, JoinedRow}
+import org.apache.spark.sql.catalyst.expressions.JoinedRow
import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
-import org.apache.spark.sql.execution.datasources.{PartitionedFile, RecordReaderIterator, SparkColumnarFileReader}
+import org.apache.spark.sql.execution.datasources.{FileFormat, PartitionedFile, RecordReaderIterator, SparkColumnarFileReader}
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources._
import org.apache.spark.sql.types.StructType
import org.apache.spark.util.Utils
-abstract class SparkOrcReaderBase(enableVectorizedReader: Boolean,
- dataSchema: StructType,
- orcFilterPushDown: Boolean,
- isCaseSensitive: Boolean) extends SparkColumnarFileReader {
+class SparkOrcReaderBase(enableVectorizedReader: Boolean,
+ dataSchema: StructType,
+ orcFilterPushDown: Boolean,
+ isCaseSensitive: Boolean,
+ capacity: Int,
+ memoryMode: MemoryMode,
+ batchReaderFactory: (Int, MemoryMode) => OrcColumnarBatchReader)
+ extends SparkColumnarFileReader with SparkAdapterSupport {
/**
* Read an individual ORC file
*
@@ -62,7 +69,7 @@ abstract class SparkOrcReaderBase(enableVectorizedReader: Boolean,
val resultSchema = StructType(requiredSchema.fields ++ partitionSchema.fields)
val conf = storageConf.unwrap()
- val filePath = partitionedFileToPath(file)
+ val filePath = new Path(sparkAdapter.getSparkPartitionedFileUtils.getPathFromPartitionedFile(file).toUri)
val fs = filePath.getFileSystem(conf)
val readerOptions = OrcFile.readerOptions(conf).filesystem(fs)
@@ -96,7 +103,7 @@ abstract class SparkOrcReaderBase(enableVectorizedReader: Boolean,
val taskAttemptContext = new TaskAttemptContextImpl(taskConf, attemptId)
if (enableVectorizedReader) {
- val batchReader = buildReader()
+ val batchReader = batchReaderFactory(capacity, memoryMode)
// SPARK-23399 Register a task completion listener first to call `close()` in all cases.
// There is a possibility that `initialize` and `initBatch` hit some errors (like OOM)
// after opening a file.
@@ -120,7 +127,8 @@ abstract class SparkOrcReaderBase(enableVectorizedReader: Boolean,
val iter = new RecordReaderIterator[OrcStruct](orcRecordReader)
Option(TaskContext.get()).foreach(_.addTaskCompletionListener[Unit](_ => iter.close()))
- val fullSchema = structTypeToAttributes(requiredSchema) ++ structTypeToAttributes(partitionSchema)
+ val schemaUtils = sparkAdapter.getSchemaUtils
+ val fullSchema = schemaUtils.toAttributes(requiredSchema) ++ schemaUtils.toAttributes(partitionSchema)
val unsafeProjection = GenerateUnsafeProjection.generate(fullSchema, fullSchema)
val deserializer = new OrcDeserializer(requiredSchema, requestedColIds)
@@ -134,10 +142,52 @@ abstract class SparkOrcReaderBase(enableVectorizedReader: Boolean,
}
}
}
+}
- def partitionedFileToPath(file: PartitionedFile): Path
-
- def buildReader(): OrcColumnarBatchReader
+object SparkOrcReaderBase {
+ /**
+ * Get ORC file reader
+ *
+ * @param vectorized true if vectorized reading is not prohibited due to schema, reading mode, etc
+ * @param sqlConf the [[SQLConf]] used for the read
+ * @param options passed as a param to the file format
+ * @param hadoopConf some configs will be set for the hadoopConf
+ * @param dataSchema schema of the data
+ * @param batchReaderFactory creates the [[OrcColumnarBatchReader]] from the batch size and memory
+ * mode; the reader constructor differs across Spark versions
+ * @return ORC file reader
+ */
+ def build(vectorized: Boolean,
+ sqlConf: SQLConf,
+ options: Map[String, String],
+ hadoopConf: Configuration,
+ dataSchema: StructType,
+ batchReaderFactory: (Int, MemoryMode) => OrcColumnarBatchReader): SparkOrcReaderBase = {
+ //set hadoopconf
+ hadoopConf.set(SQLConf.SESSION_LOCAL_TIMEZONE.key, sqlConf.sessionLocalTimeZone)
+ hadoopConf.setBoolean(SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key, sqlConf.nestedSchemaPruningEnabled)
+ hadoopConf.setBoolean(SQLConf.CASE_SENSITIVE.key, sqlConf.caseSensitiveAnalysis)
+
+ val memoryMode = if (sqlConf.offHeapColumnVectorEnabled) {
+ MemoryMode.OFF_HEAP
+ } else {
+ MemoryMode.ON_HEAP
+ }
- def structTypeToAttributes(schema: StructType): Seq[Attribute]
+ val enableVectorizedReader = sqlConf.orcVectorizedReaderEnabled &&
+ options.getOrElse(FileFormat.OPTION_RETURNING_BATCH,
+ throw new IllegalArgumentException(
+ "OPTION_RETURNING_BATCH should always be set for OrcFileFormat. " +
+ "To workaround this issue, set spark.sql.orc.enableVectorizedReader=false."))
+ .equals("true")
+
+ new SparkOrcReaderBase(
+ enableVectorizedReader = enableVectorizedReader && vectorized,
+ dataSchema = dataSchema,
+ orcFilterPushDown = sqlConf.orcFilterPushDown,
+ isCaseSensitive = sqlConf.caseSensitiveAnalysis,
+ capacity = sqlConf.orcVectorizedReaderBatchSize,
+ memoryMode = memoryMode,
+ batchReaderFactory = batchReaderFactory)
+ }
}
diff --git a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/hudi/Spark35ResolveHudiAlterTableCommand.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/BaseResolveHudiAlterTableCommand.scala
similarity index 52%
rename from hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/hudi/Spark35ResolveHudiAlterTableCommand.scala
rename to hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/BaseResolveHudiAlterTableCommand.scala
index 8e0f41c2b996..029ba63b4b41 100644
--- a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/hudi/Spark35ResolveHudiAlterTableCommand.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/BaseResolveHudiAlterTableCommand.scala
@@ -30,32 +30,38 @@ import org.apache.spark.sql.hudi.command.{AlterTableCommand => HudiAlterTableCom
* Rule to mostly resolve, normalize and rewrite column names based on case sensitivity.
* for alter table column commands.
*/
-class Spark35ResolveHudiAlterTableCommand(sparkSession: SparkSession) extends Rule[LogicalPlan] {
+abstract class BaseResolveHudiAlterTableCommand(sparkSession: SparkSession) extends Rule[LogicalPlan] {
def apply(plan: LogicalPlan): LogicalPlan = {
if (ProvidesHoodieConfig.isSchemaEvolutionEnabled(sparkSession)) {
- plan.resolveOperatorsUp {
- case set@SetTableProperties(ResolvedHoodieV2TablePlan(t), _) if set.resolved =>
- HudiAlterTableCommand(t.v1Table, set.changes, ColumnChangeID.PROPERTY_CHANGE)
- case unSet@UnsetTableProperties(ResolvedHoodieV2TablePlan(t), _, _) if unSet.resolved =>
- HudiAlterTableCommand(t.v1Table, unSet.changes, ColumnChangeID.PROPERTY_CHANGE)
- case drop@DropColumns(ResolvedHoodieV2TablePlan(t), _, _) if drop.resolved =>
- HudiAlterTableCommand(t.v1Table, drop.changes, ColumnChangeID.DELETE)
- case add@AddColumns(ResolvedHoodieV2TablePlan(t), _) if add.resolved =>
- HudiAlterTableCommand(t.v1Table, add.changes, ColumnChangeID.ADD)
- case renameColumn@RenameColumn(ResolvedHoodieV2TablePlan(t), _, _) if renameColumn.resolved =>
- HudiAlterTableCommand(t.v1Table, renameColumn.changes, ColumnChangeID.UPDATE)
- case alter@AlterColumn(ResolvedHoodieV2TablePlan(t), _, _, _, _, _, _) if alter.resolved =>
- HudiAlterTableCommand(t.v1Table, alter.changes, ColumnChangeID.UPDATE)
- case replace@ReplaceColumns(ResolvedHoodieV2TablePlan(t), _) if replace.resolved =>
- HudiAlterTableCommand(t.v1Table, replace.changes, ColumnChangeID.REPLACE)
- }
+ plan.resolveOperatorsUp(resolveCommonCommand.orElse(resolveAlterColumnCommand))
} else {
plan
}
}
- object ResolvedHoodieV2TablePlan {
+ private def resolveCommonCommand: PartialFunction[LogicalPlan, LogicalPlan] = {
+ case set@SetTableProperties(ResolvedHoodieV2TablePlan(t), _) if set.resolved =>
+ HudiAlterTableCommand(t.v1Table, set.changes, ColumnChangeID.PROPERTY_CHANGE)
+ case unSet@UnsetTableProperties(ResolvedHoodieV2TablePlan(t), _, _) if unSet.resolved =>
+ HudiAlterTableCommand(t.v1Table, unSet.changes, ColumnChangeID.PROPERTY_CHANGE)
+ case drop@DropColumns(ResolvedHoodieV2TablePlan(t), _, _) if drop.resolved =>
+ HudiAlterTableCommand(t.v1Table, drop.changes, ColumnChangeID.DELETE)
+ case add@AddColumns(ResolvedHoodieV2TablePlan(t), _) if add.resolved =>
+ HudiAlterTableCommand(t.v1Table, add.changes, ColumnChangeID.ADD)
+ case renameColumn@RenameColumn(ResolvedHoodieV2TablePlan(t), _, _) if renameColumn.resolved =>
+ HudiAlterTableCommand(t.v1Table, renameColumn.changes, ColumnChangeID.UPDATE)
+ case replace@ReplaceColumns(ResolvedHoodieV2TablePlan(t), _) if replace.resolved =>
+ HudiAlterTableCommand(t.v1Table, replace.changes, ColumnChangeID.REPLACE)
+ }
+
+ /**
+ * Resolves the ALTER TABLE ... ALTER COLUMN command, whose logical plan differs between
+ * the Spark 3.x (AlterColumn) and Spark 4.x (AlterColumns) branches.
+ */
+ protected def resolveAlterColumnCommand: PartialFunction[LogicalPlan, LogicalPlan]
+
+ protected object ResolvedHoodieV2TablePlan {
def unapply(plan: LogicalPlan): Option[HoodieInternalV2Table] = {
plan match {
case ResolvedTable(_, _, v2Table: HoodieInternalV2Table, _) => Some(v2Table)
@@ -64,4 +70,3 @@ class Spark35ResolveHudiAlterTableCommand(sparkSession: SparkSession) extends Ru
}
}
}
-
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala
index 41d6c2b9ae81..8872c5269f50 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala
@@ -18,8 +18,8 @@
package org.apache.spark.sql.hudi.analysis
import org.apache.hudi.{HoodieSchemaUtils, HoodieSparkUtils, SparkAdapterSupport}
-import org.apache.hudi.common.util.{ReflectionUtils, ValidationUtils}
import org.apache.hudi.common.util.ReflectionUtils.loadClass
+import org.apache.hudi.common.util.ValidationUtils
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.TableIdentifier
@@ -30,7 +30,7 @@ import org.apache.spark.sql.catalyst.optimizer.ReplaceExpressions
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution.command._
-import org.apache.spark.sql.execution.datasources.{CreateTable, LogicalRelation}
+import org.apache.spark.sql.execution.datasources.{CreateTable, HoodieNestedSchemaPruning, LogicalRelation}
import org.apache.spark.sql.hudi.HoodieSqlCommonUtils.{isMetaField, removeMetaFields}
import org.apache.spark.sql.hudi.analysis.HoodieAnalysis.{sparkAdapter, MatchCreateIndex, MatchCreateTableLike, MatchDropIndex, MatchInsertIntoStatement, MatchMergeIntoTable, MatchRefreshIndex, MatchShowIndexes, ResolvesToHudiTable}
import org.apache.spark.sql.hudi.blob.ReadBlobRule
@@ -95,16 +95,10 @@ object HoodieAnalysis extends SparkAdapterSupport {
}
val resolveAlterTableCommandsClass =
- if (HoodieSparkUtils.isSpark4_1) {
- "org.apache.spark.sql.hudi.Spark41ResolveHudiAlterTableCommand"
- } else if (HoodieSparkUtils.isSpark4_0) {
- "org.apache.spark.sql.hudi.Spark40ResolveHudiAlterTableCommand"
- } else if (HoodieSparkUtils.gteqSpark3_5) {
- "org.apache.spark.sql.hudi.Spark35ResolveHudiAlterTableCommand"
- } else if (HoodieSparkUtils.isSpark3_4) {
- "org.apache.spark.sql.hudi.Spark34ResolveHudiAlterTableCommand"
- } else if (HoodieSparkUtils.isSpark3_3) {
- "org.apache.spark.sql.hudi.Spark33ResolveHudiAlterTableCommand"
+ if (HoodieSparkUtils.isSpark4) {
+ "org.apache.spark.sql.hudi.Spark4ResolveHudiAlterTableCommand"
+ } else if (HoodieSparkUtils.isSpark3) {
+ "org.apache.spark.sql.hudi.Spark3ResolveHudiAlterTableCommand"
} else {
throw new IllegalStateException("Unsupported Spark version")
}
@@ -143,21 +137,7 @@ object HoodieAnalysis extends SparkAdapterSupport {
// Default rules
)
- val nestedSchemaPruningClass =
- if (HoodieSparkUtils.isSpark4_1) {
- "org.apache.spark.sql.execution.datasources.Spark41NestedSchemaPruning"
- } else if (HoodieSparkUtils.isSpark4_0) {
- "org.apache.spark.sql.execution.datasources.Spark40NestedSchemaPruning"
- } else if (HoodieSparkUtils.gteqSpark3_5) {
- "org.apache.spark.sql.execution.datasources.Spark35NestedSchemaPruning"
- } else if (HoodieSparkUtils.gteqSpark3_4) {
- "org.apache.spark.sql.execution.datasources.Spark34NestedSchemaPruning"
- } else {
- // spark 3.3
- "org.apache.spark.sql.execution.datasources.Spark33NestedSchemaPruning"
- }
-
- val nestedSchemaPruningRule = ReflectionUtils.loadClass(nestedSchemaPruningClass).asInstanceOf[Rule[LogicalPlan]]
+ val nestedSchemaPruningRule = new HoodieNestedSchemaPruning
rules += (_ => nestedSchemaPruningRule)
// NOTE: [[HoodiePruneFileSourcePartitions]] is a replica in kind to Spark's
diff --git a/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark3Adapter.scala b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark3Adapter.scala
index c7039c951c4e..90be1df5068d 100644
--- a/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark3Adapter.scala
+++ b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark3Adapter.scala
@@ -17,7 +17,7 @@
package org.apache.spark.sql.adapter
-import org.apache.hudi.{DefaultSource, HoodiePartitionCDCFileGroupMapping, HoodiePartitionFileSliceMapping, HoodieSchemaConversionUtils, Spark3HoodiePartitionCDCFileGroupMapping, Spark3HoodiePartitionFileSliceMapping}
+import org.apache.hudi.{DefaultSource, HoodieFileScanRDD, HoodiePartitionCDCFileGroupMapping, HoodiePartitionFileSliceMapping, HoodieSchemaConversionUtils, Spark3HoodiePartitionCDCFileGroupMapping, Spark3HoodiePartitionFileSliceMapping}
import org.apache.hudi.client.model.{HoodieInternalRow, Spark3HoodieInternalRow}
import org.apache.hudi.common.model.FileSlice
import org.apache.hudi.common.schema.HoodieSchema
@@ -36,7 +36,7 @@ import org.apache.spark.sql.FileFormatUtilsForFileGroupReader.applyFiltersToPlan
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.EliminateSubqueryAliases
import org.apache.spark.sql.catalyst.catalog.CatalogTable
-import org.apache.spark.sql.catalyst.expressions.{Expression, InterpretedPredicate, Predicate, SpecializedGetters}
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Expression, InterpretedPredicate, Predicate, SpecializedGetters}
import org.apache.spark.sql.catalyst.parser.ParseException
import org.apache.spark.sql.catalyst.planning.PhysicalOperation
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
@@ -101,6 +101,14 @@ abstract class BaseSpark3Adapter extends SparkAdapter with Logging {
Predicate.createInterpreted(e)
}
+ override def createHoodieFileScanRDD(sparkSession: SparkSession,
+ readFunction: PartitionedFile => Iterator[InternalRow],
+ filePartitions: Seq[FilePartition],
+ readDataSchema: StructType,
+ metadataColumns: Seq[AttributeReference] = Seq.empty): FileScanRDD = {
+ new HoodieFileScanRDD(sparkSession, readFunction, filePartitions, readDataSchema, metadataColumns)
+ }
+
override def createRelation(sqlContext: SQLContext,
metaClient: HoodieTableMetaClient,
schema: HoodieSchema,
diff --git a/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/hudi/Spark3ResolveHudiAlterTableCommand.scala b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/hudi/Spark3ResolveHudiAlterTableCommand.scala
new file mode 100644
index 000000000000..6246c8480ab7
--- /dev/null
+++ b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/hudi/Spark3ResolveHudiAlterTableCommand.scala
@@ -0,0 +1,43 @@
+/*
+ * 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.hudi
+
+import org.apache.hudi.internal.schema.action.TableChange.ColumnChangeID
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.plans.logical.{AlterColumn, LogicalPlan}
+import org.apache.spark.sql.hudi.command.{AlterTableCommand => HudiAlterTableCommand}
+
+/**
+ * Rule to mostly resolve, normalize and rewrite column names based on case sensitivity.
+ * for alter table column commands.
+ */
+class Spark3ResolveHudiAlterTableCommand(sparkSession: SparkSession)
+ extends BaseResolveHudiAlterTableCommand(sparkSession) {
+
+ // NOTE: The command is matched by type rather than by destructuring [[AlterColumn]], since
+ // the arity of its unapply differs between Spark 3.3 and Spark 3.4+
+ override protected def resolveAlterColumnCommand: PartialFunction[LogicalPlan, LogicalPlan] = {
+ case alter: AlterColumn if alter.resolved =>
+ alter.table match {
+ case ResolvedHoodieV2TablePlan(t) =>
+ HudiAlterTableCommand(t.v1Table, alter.changes, ColumnChangeID.UPDATE)
+ case _ => alter
+ }
+ }
+}
diff --git a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_3Adapter.scala b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_3Adapter.scala
index a4cb5b70e72e..25553f8ebdd7 100644
--- a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_3Adapter.scala
+++ b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_3Adapter.scala
@@ -17,7 +17,6 @@
package org.apache.spark.sql.adapter
-import org.apache.hudi.Spark33HoodieFileScanRDD
import org.apache.hudi.common.schema.HoodieSchema
import org.apache.hudi.storage.StorageConfiguration
@@ -31,14 +30,14 @@ import org.apache.spark.sql.avro._
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.{EliminateSubqueryAliases, ResolvedTable}
import org.apache.spark.sql.catalyst.catalog.CatalogTable
-import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Expression}
+import org.apache.spark.sql.catalyst.expressions.{Expression}
import org.apache.spark.sql.catalyst.parser.ParserInterface
import org.apache.spark.sql.catalyst.planning.PhysicalOperation
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.util.{METADATA_COL_ATTR_KEY, RebaseDateTime}
import org.apache.spark.sql.connector.catalog.{V1Table, V2TableWithV1Fallback}
import org.apache.spark.sql.execution.datasources._
-import org.apache.spark.sql.execution.datasources.orc.Spark33OrcReader
+import org.apache.spark.sql.execution.datasources.orc.{OrcColumnarBatchReader, SparkOrcReaderBase}
import org.apache.spark.sql.execution.datasources.parquet.{ParquetFileFormat, ParquetFilters, Spark33LegacyHoodieParquetFileFormat, Spark33ParquetReader}
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
import org.apache.spark.sql.hudi.analysis.TableValuedFunctions
@@ -100,14 +99,6 @@ class Spark3_3Adapter extends BaseSpark3Adapter {
Some(new Spark33LegacyHoodieParquetFileFormat(appendPartitionValues))
}
- override def createHoodieFileScanRDD(sparkSession: SparkSession,
- readFunction: PartitionedFile => Iterator[InternalRow],
- filePartitions: Seq[FilePartition],
- readDataSchema: StructType,
- metadataColumns: Seq[AttributeReference] = Seq.empty): FileScanRDD = {
- new Spark33HoodieFileScanRDD(sparkSession, readFunction, filePartitions, readDataSchema, metadataColumns)
- }
-
override def extractDeleteCondition(deleteFromTable: Command): Expression = {
deleteFromTable.asInstanceOf[DeleteFromTable].condition
}
@@ -163,7 +154,8 @@ class Spark3_3Adapter extends BaseSpark3Adapter {
}
override def createOrcFileReader(vectorized: Boolean, sqlConf: SQLConf, options: Map[String, String], hadoopConf: Configuration, dataSchema: StructType): SparkColumnarFileReader = {
- Spark33OrcReader.build(vectorized, sqlConf, options, hadoopConf, dataSchema)
+ SparkOrcReaderBase.build(vectorized, sqlConf, options, hadoopConf, dataSchema,
+ (capacity, _) => new OrcColumnarBatchReader(capacity))
}
override def createLanceFileReader(vectorized: Boolean,
diff --git a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/Spark33NestedSchemaPruning.scala b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/Spark33NestedSchemaPruning.scala
deleted file mode 100644
index c2235506eee6..000000000000
--- a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/Spark33NestedSchemaPruning.scala
+++ /dev/null
@@ -1,62 +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.execution.datasources
-
-import org.apache.hudi.HoodieBaseRelation
-
-import org.apache.spark.sql.catalyst.expressions.AttributeReference
-import org.apache.spark.sql.catalyst.planning.PhysicalOperation
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
-import org.apache.spark.sql.sources.BaseRelation
-import org.apache.spark.sql.types.StructType
-
-class Spark33NestedSchemaPruning extends BaseHoodieNestedSchemaPruning {
-
- // Prune the given output to make it consistent with `requiredSchema`.
- protected def getPrunedOutput(output: Seq[AttributeReference],
- requiredSchema: StructType): Seq[AttributeReference] = {
- // We need to replace the expression ids of the pruned relation output attributes
- // with the expression ids of the original relation output attributes so that
- // references to the original relation's output are not broken
- val outputIdMap = output.map(att => (att.name, att.exprId)).toMap
- requiredSchema
- .toAttributes
- .map {
- case att if outputIdMap.contains(att.name) =>
- att.withExprId(outputIdMap(att.name))
- case att => att
- }
- }
-
- override protected def apply0(plan: LogicalPlan): LogicalPlan =
- plan transformDown {
- case op @ PhysicalOperation(projects, filters,
- // NOTE: This is modified to accommodate for Hudi's custom relations, given that original
- // [[NestedSchemaPruning]] rule is tightly coupled w/ [[HadoopFsRelation]]
- // TODO generalize to any file-based relation
- l @ LogicalRelation(relation: HoodieBaseRelation, _, _, _))
- if relation.canPruneRelationSchema =>
-
- prunePhysicalColumns(l.output, projects, filters, relation.dataSchema,
- prunedDataSchema => {
- val prunedRelation =
- relation.updatePrunedDataSchema(prunedSchema = prunedDataSchema)
- buildPrunedRelation(l, prunedRelation)
- }).getOrElse(op)
- }
-}
diff --git a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/orc/Spark33OrcReader.scala b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/orc/Spark33OrcReader.scala
deleted file mode 100644
index 25ac5938b8d1..000000000000
--- a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/orc/Spark33OrcReader.scala
+++ /dev/null
@@ -1,85 +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.execution.datasources.orc
-
-import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.Path
-import org.apache.spark.sql.catalyst.expressions.Attribute
-import org.apache.spark.sql.execution.datasources.{FileFormat, PartitionedFile}
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.StructType
-
-import java.net.URI
-
-class Spark33OrcReader(enableVectorizedReader: Boolean,
- dataSchema: StructType,
- orcFilterPushDown: Boolean,
- isCaseSensitive: Boolean,
- capacity: Int) extends SparkOrcReaderBase(enableVectorizedReader, dataSchema, orcFilterPushDown, isCaseSensitive) {
-
- override def partitionedFileToPath(file: PartitionedFile): Path = {
- new Path(new URI(file.filePath))
- }
-
- override def buildReader(): OrcColumnarBatchReader = {
- new OrcColumnarBatchReader(capacity)
- }
-
- override def structTypeToAttributes(schema: StructType): Seq[Attribute] = {
- schema.toAttributes
- }
-}
-
-object Spark33OrcReader {
- /**
- * Get ORC file reader
- *
- * @param vectorized true if vectorized reading is not prohibited due to schema, reading mode, etc
- * @param sqlConf the [[SQLConf]] used for the read
- * @param options passed as a param to the file format
- * @param hadoopConf some configs will be set for the hadoopConf
- * @return ORC file reader
- */
- def build(vectorized: Boolean,
- sqlConf: SQLConf,
- options: Map[String, String],
- hadoopConf: Configuration,
- dataSchema: StructType): Spark33OrcReader = {
- //set hadoopconf
- hadoopConf.set(SQLConf.SESSION_LOCAL_TIMEZONE.key, sqlConf.sessionLocalTimeZone)
- hadoopConf.setBoolean(SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key, sqlConf.nestedSchemaPruningEnabled)
- hadoopConf.setBoolean(SQLConf.CASE_SENSITIVE.key, sqlConf.caseSensitiveAnalysis)
-
- val enableVectorizedReader = sqlConf.orcVectorizedReaderEnabled &&
- options.getOrElse(FileFormat.OPTION_RETURNING_BATCH,
- throw new IllegalArgumentException(
- "OPTION_RETURNING_BATCH should always be set for OrcFileFormat. " +
- "To workaround this issue, set spark.sql.orc.enableVectorizedReader=false."))
- .equals("true")
-
- new Spark33OrcReader(
- enableVectorizedReader = enableVectorizedReader && vectorized,
- isCaseSensitive = sqlConf.caseSensitiveAnalysis,
- capacity = sqlConf.orcVectorizedReaderBatchSize,
- orcFilterPushDown = sqlConf.orcFilterPushDown,
- dataSchema = dataSchema)
- }
-}
-
diff --git a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark33DataSourceUtils.scala b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark33DataSourceUtils.scala
deleted file mode 100644
index 2aa85660eb51..000000000000
--- a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark33DataSourceUtils.scala
+++ /dev/null
@@ -1,77 +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.execution.datasources.parquet
-
-import org.apache.spark.sql.SPARK_VERSION_METADATA_KEY
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy
-import org.apache.spark.util.Utils
-
-object Spark33DataSourceUtils {
-
- /**
- * NOTE: This method was copied from Spark 3.2.0, and is required to maintain runtime
- * compatibility against Spark 3.2.0
- */
- // scalastyle:off
- def int96RebaseMode(lookupFileMeta: String => String,
- modeByConfig: String): LegacyBehaviorPolicy.Value = {
- if (Utils.isTesting && SQLConf.get.getConfString("spark.test.forceNoRebase", "") == "true") {
- return LegacyBehaviorPolicy.CORRECTED
- }
- // If there is no version, we return the mode specified by the config.
- Option(lookupFileMeta(SPARK_VERSION_METADATA_KEY)).map { version =>
- // Files written by Spark 3.0 and earlier follow the legacy hybrid calendar and we need to
- // rebase the INT96 timestamp values.
- // Files written by Spark 3.1 and latter may also need the rebase if they were written with
- // the "LEGACY" rebase mode.
- if (version < "3.1.0" || lookupFileMeta("org.apache.spark.legacyINT96") != null) {
- LegacyBehaviorPolicy.LEGACY
- } else {
- LegacyBehaviorPolicy.CORRECTED
- }
- }.getOrElse(LegacyBehaviorPolicy.withName(modeByConfig))
- }
- // scalastyle:on
-
- /**
- * NOTE: This method was copied from Spark 3.2.0, and is required to maintain runtime
- * compatibility against Spark 3.2.0
- */
- // scalastyle:off
- def datetimeRebaseMode(lookupFileMeta: String => String,
- modeByConfig: String): LegacyBehaviorPolicy.Value = {
- if (Utils.isTesting && SQLConf.get.getConfString("spark.test.forceNoRebase", "") == "true") {
- return LegacyBehaviorPolicy.CORRECTED
- }
- // If there is no version, we return the mode specified by the config.
- Option(lookupFileMeta(SPARK_VERSION_METADATA_KEY)).map { version =>
- // Files written by Spark 2.4 and earlier follow the legacy hybrid calendar and we need to
- // rebase the datetime values.
- // Files written by Spark 3.0 and latter may also need the rebase if they were written with
- // the "LEGACY" rebase mode.
- if (version < "3.0.0" || lookupFileMeta("org.apache.spark.legacyDateTime") != null) {
- LegacyBehaviorPolicy.LEGACY
- } else {
- LegacyBehaviorPolicy.CORRECTED
- }
- }.getOrElse(LegacyBehaviorPolicy.withName(modeByConfig))
- }
- // scalastyle:on
-
-}
diff --git a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/hudi/Spark33ResolveHudiAlterTableCommand.scala b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/hudi/Spark33ResolveHudiAlterTableCommand.scala
deleted file mode 100644
index 55159bcc71b4..000000000000
--- a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/hudi/Spark33ResolveHudiAlterTableCommand.scala
+++ /dev/null
@@ -1,67 +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.hudi
-
-import org.apache.hudi.internal.schema.action.TableChange.ColumnChangeID
-
-import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.analysis.ResolvedTable
-import org.apache.spark.sql.catalyst.plans.logical._
-import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.hudi.catalog.HoodieInternalV2Table
-import org.apache.spark.sql.hudi.command.{AlterTableCommand => HudiAlterTableCommand}
-
-/**
- * Rule to mostly resolve, normalize and rewrite column names based on case sensitivity.
- * for alter table column commands.
- */
-class Spark33ResolveHudiAlterTableCommand(sparkSession: SparkSession) extends Rule[LogicalPlan] {
-
- def apply(plan: LogicalPlan): LogicalPlan = {
- if (ProvidesHoodieConfig.isSchemaEvolutionEnabled(sparkSession)) {
- plan.resolveOperatorsUp {
- case set@SetTableProperties(ResolvedHoodieV2TablePlan(t), _) if set.resolved =>
- HudiAlterTableCommand(t.v1Table, set.changes, ColumnChangeID.PROPERTY_CHANGE)
- case unSet@UnsetTableProperties(ResolvedHoodieV2TablePlan(t), _, _) if unSet.resolved =>
- HudiAlterTableCommand(t.v1Table, unSet.changes, ColumnChangeID.PROPERTY_CHANGE)
- case drop@DropColumns(ResolvedHoodieV2TablePlan(t), _, _) if drop.resolved =>
- HudiAlterTableCommand(t.v1Table, drop.changes, ColumnChangeID.DELETE)
- case add@AddColumns(ResolvedHoodieV2TablePlan(t), _) if add.resolved =>
- HudiAlterTableCommand(t.v1Table, add.changes, ColumnChangeID.ADD)
- case renameColumn@RenameColumn(ResolvedHoodieV2TablePlan(t), _, _) if renameColumn.resolved =>
- HudiAlterTableCommand(t.v1Table, renameColumn.changes, ColumnChangeID.UPDATE)
- case alter@AlterColumn(ResolvedHoodieV2TablePlan(t), _, _, _, _, _) if alter.resolved =>
- HudiAlterTableCommand(t.v1Table, alter.changes, ColumnChangeID.UPDATE)
- case replace@ReplaceColumns(ResolvedHoodieV2TablePlan(t), _) if replace.resolved =>
- HudiAlterTableCommand(t.v1Table, replace.changes, ColumnChangeID.REPLACE)
- }
- } else {
- plan
- }
- }
-
- object ResolvedHoodieV2TablePlan {
- def unapply(plan: LogicalPlan): Option[HoodieInternalV2Table] = {
- plan match {
- case ResolvedTable(_, _, v2Table: HoodieInternalV2Table, _) => Some(v2Table)
- case _ => None
- }
- }
- }
-}
-
diff --git a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/hudi/Spark34HoodieFileScanRDD.scala b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/hudi/Spark34HoodieFileScanRDD.scala
deleted file mode 100644
index df86e5b169c0..000000000000
--- a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/hudi/Spark34HoodieFileScanRDD.scala
+++ /dev/null
@@ -1,36 +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.hudi
-
-import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.AttributeReference
-import org.apache.spark.sql.execution.datasources.{FilePartition, FileScanRDD, PartitionedFile}
-import org.apache.spark.sql.types.StructType
-
-class Spark34HoodieFileScanRDD(@transient private val sparkSession: SparkSession,
- read: PartitionedFile => Iterator[InternalRow],
- @transient filePartitions: Seq[FilePartition],
- readDataSchema: StructType,
- metadataColumns: Seq[AttributeReference] = Seq.empty)
- extends FileScanRDD(sparkSession, read, filePartitions, readDataSchema, metadataColumns)
- with HoodieUnsafeRDD {
-
- override final def collect(): Array[InternalRow] = super[HoodieUnsafeRDD].collect()
-}
diff --git a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_4Adapter.scala b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_4Adapter.scala
index 923c8ac91959..3d62ba1edc0c 100644
--- a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_4Adapter.scala
+++ b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_4Adapter.scala
@@ -17,7 +17,6 @@
package org.apache.spark.sql.adapter
-import org.apache.hudi.Spark34HoodieFileScanRDD
import org.apache.hudi.common.schema.HoodieSchema
import org.apache.hudi.storage.StorageConfiguration
@@ -31,7 +30,7 @@ import org.apache.spark.sql.avro._
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.{EliminateSubqueryAliases, ResolvedTable}
import org.apache.spark.sql.catalyst.catalog.CatalogTable
-import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Expression}
+import org.apache.spark.sql.catalyst.expressions.{Expression}
import org.apache.spark.sql.catalyst.parser.ParserInterface
import org.apache.spark.sql.catalyst.planning.PhysicalOperation
import org.apache.spark.sql.catalyst.plans.logical._
@@ -39,7 +38,7 @@ import org.apache.spark.sql.catalyst.util.{METADATA_COL_ATTR_KEY, RebaseDateTime
import org.apache.spark.sql.connector.catalog.{V1Table, V2TableWithV1Fallback}
import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.execution.datasources.lance.SparkLanceReaderBase
-import org.apache.spark.sql.execution.datasources.orc.Spark34OrcReader
+import org.apache.spark.sql.execution.datasources.orc.{OrcColumnarBatchReader, SparkOrcReaderBase}
import org.apache.spark.sql.execution.datasources.parquet.{ParquetFileFormat, ParquetFilters, Spark34LegacyHoodieParquetFileFormat, Spark34ParquetReader}
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
import org.apache.spark.sql.hudi.analysis.TableValuedFunctions
@@ -101,14 +100,6 @@ class Spark3_4Adapter extends BaseSpark3Adapter {
Some(new Spark34LegacyHoodieParquetFileFormat(appendPartitionValues))
}
- override def createHoodieFileScanRDD(sparkSession: SparkSession,
- readFunction: PartitionedFile => Iterator[InternalRow],
- filePartitions: Seq[FilePartition],
- readDataSchema: StructType,
- metadataColumns: Seq[AttributeReference] = Seq.empty): FileScanRDD = {
- new Spark34HoodieFileScanRDD(sparkSession, readFunction, filePartitions, readDataSchema, metadataColumns)
- }
-
override def extractDeleteCondition(deleteFromTable: Command): Expression = {
deleteFromTable.asInstanceOf[DeleteFromTable].condition
}
@@ -164,7 +155,8 @@ class Spark3_4Adapter extends BaseSpark3Adapter {
}
override def createOrcFileReader(vectorized: Boolean, sqlConf: SQLConf, options: Map[String, String], hadoopConf: Configuration, dataSchema: StructType): SparkColumnarFileReader = {
- Spark34OrcReader.build(vectorized, sqlConf, options, hadoopConf, dataSchema)
+ SparkOrcReaderBase.build(vectorized, sqlConf, options, hadoopConf, dataSchema,
+ (capacity, memoryMode) => new OrcColumnarBatchReader(capacity, memoryMode))
}
override def createLanceFileReader(vectorized: Boolean,
diff --git a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/Spark34NestedSchemaPruning.scala b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/Spark34NestedSchemaPruning.scala
deleted file mode 100644
index 6f3e6d12e23e..000000000000
--- a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/Spark34NestedSchemaPruning.scala
+++ /dev/null
@@ -1,62 +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.execution.datasources
-
-import org.apache.hudi.HoodieBaseRelation
-
-import org.apache.spark.sql.catalyst.expressions.AttributeReference
-import org.apache.spark.sql.catalyst.planning.PhysicalOperation
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
-import org.apache.spark.sql.sources.BaseRelation
-import org.apache.spark.sql.types.StructType
-
-class Spark34NestedSchemaPruning extends BaseHoodieNestedSchemaPruning {
-
- // Prune the given output to make it consistent with `requiredSchema`.
- protected def getPrunedOutput(output: Seq[AttributeReference],
- requiredSchema: StructType): Seq[AttributeReference] = {
- // We need to replace the expression ids of the pruned relation output attributes
- // with the expression ids of the original relation output attributes so that
- // references to the original relation's output are not broken
- val outputIdMap = output.map(att => (att.name, att.exprId)).toMap
- requiredSchema
- .toAttributes
- .map {
- case att if outputIdMap.contains(att.name) =>
- att.withExprId(outputIdMap(att.name))
- case att => att
- }
- }
-
- override protected def apply0(plan: LogicalPlan): LogicalPlan =
- plan transformDown {
- case op @ PhysicalOperation(projects, filters,
- // NOTE: This is modified to accommodate for Hudi's custom relations, given that original
- // [[NestedSchemaPruning]] rule is tightly coupled w/ [[HadoopFsRelation]]
- // TODO generalize to any file-based relation
- l @ LogicalRelation(relation: HoodieBaseRelation, _, _, _))
- if relation.canPruneRelationSchema =>
-
- prunePhysicalColumns(l.output, projects, filters, relation.dataSchema,
- prunedDataSchema => {
- val prunedRelation =
- relation.updatePrunedDataSchema(prunedSchema = prunedDataSchema)
- buildPrunedRelation(l, prunedRelation)
- }).getOrElse(op)
- }
-}
diff --git a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/orc/Spark34OrcReader.scala b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/orc/Spark34OrcReader.scala
deleted file mode 100644
index a1463aa01aaf..000000000000
--- a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/orc/Spark34OrcReader.scala
+++ /dev/null
@@ -1,91 +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.execution.datasources.orc
-
-import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.Path
-import org.apache.spark.memory.MemoryMode
-import org.apache.spark.sql.catalyst.expressions.Attribute
-import org.apache.spark.sql.execution.datasources.{FileFormat, PartitionedFile}
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.StructType
-
-class Spark34OrcReader(enableVectorizedReader: Boolean,
- memoryMode: MemoryMode,
- dataSchema: StructType,
- orcFilterPushDown: Boolean,
- isCaseSensitive: Boolean,
- capacity: Int) extends SparkOrcReaderBase(enableVectorizedReader, dataSchema, orcFilterPushDown, isCaseSensitive) {
-
- override def partitionedFileToPath(file: PartitionedFile): Path = {
- file.toPath
- }
-
- override def buildReader(): OrcColumnarBatchReader = {
- new OrcColumnarBatchReader(capacity, memoryMode)
- }
-
- override def structTypeToAttributes(schema: StructType): Seq[Attribute] = {
- schema.toAttributes
- }
-}
-
-object Spark34OrcReader {
- /**
- * Get ORC file reader
- *
- * @param vectorized true if vectorized reading is not prohibited due to schema, reading mode, etc
- * @param sqlConf the [[SQLConf]] used for the read
- * @param options passed as a param to the file format
- * @param hadoopConf some configs will be set for the hadoopConf
- * @return ORC file reader
- */
- def build(vectorized: Boolean,
- sqlConf: SQLConf,
- options: Map[String, String],
- hadoopConf: Configuration,
- dataSchema: StructType): Spark34OrcReader = {
- //set hadoopconf
- hadoopConf.set(SQLConf.SESSION_LOCAL_TIMEZONE.key, sqlConf.sessionLocalTimeZone)
- hadoopConf.setBoolean(SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key, sqlConf.nestedSchemaPruningEnabled)
- hadoopConf.setBoolean(SQLConf.CASE_SENSITIVE.key, sqlConf.caseSensitiveAnalysis)
-
- val memoryMode = if (sqlConf.offHeapColumnVectorEnabled) {
- MemoryMode.OFF_HEAP
- } else {
- MemoryMode.ON_HEAP
- }
-
- val enableVectorizedReader = sqlConf.orcVectorizedReaderEnabled &&
- options.getOrElse(FileFormat.OPTION_RETURNING_BATCH,
- throw new IllegalArgumentException(
- "OPTION_RETURNING_BATCH should always be set for OrcFileFormat. " +
- "To workaround this issue, set spark.sql.orc.enableVectorizedReader=false."))
- .equals("true")
-
- new Spark34OrcReader(
- enableVectorizedReader = enableVectorizedReader && vectorized,
- memoryMode = memoryMode,
- isCaseSensitive = sqlConf.caseSensitiveAnalysis,
- capacity = sqlConf.orcVectorizedReaderBatchSize,
- orcFilterPushDown = sqlConf.orcFilterPushDown,
- dataSchema = dataSchema)
- }
-}
diff --git a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark34DataSourceUtils.scala b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark34DataSourceUtils.scala
deleted file mode 100644
index d404bc8c24b5..000000000000
--- a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark34DataSourceUtils.scala
+++ /dev/null
@@ -1,77 +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.execution.datasources.parquet
-
-import org.apache.spark.sql.SPARK_VERSION_METADATA_KEY
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy
-import org.apache.spark.util.Utils
-
-object Spark34DataSourceUtils {
-
- /**
- * NOTE: This method was copied from Spark 3.2.0, and is required to maintain runtime
- * compatibility against Spark 3.2.0
- */
- // scalastyle:off
- def int96RebaseMode(lookupFileMeta: String => String,
- modeByConfig: String): LegacyBehaviorPolicy.Value = {
- if (Utils.isTesting && SQLConf.get.getConfString("spark.test.forceNoRebase", "") == "true") {
- return LegacyBehaviorPolicy.CORRECTED
- }
- // If there is no version, we return the mode specified by the config.
- Option(lookupFileMeta(SPARK_VERSION_METADATA_KEY)).map { version =>
- // Files written by Spark 3.0 and earlier follow the legacy hybrid calendar and we need to
- // rebase the INT96 timestamp values.
- // Files written by Spark 3.1 and latter may also need the rebase if they were written with
- // the "LEGACY" rebase mode.
- if (version < "3.1.0" || lookupFileMeta("org.apache.spark.legacyINT96") != null) {
- LegacyBehaviorPolicy.LEGACY
- } else {
- LegacyBehaviorPolicy.CORRECTED
- }
- }.getOrElse(LegacyBehaviorPolicy.withName(modeByConfig))
- }
- // scalastyle:on
-
- /**
- * NOTE: This method was copied from Spark 3.2.0, and is required to maintain runtime
- * compatibility against Spark 3.2.0
- */
- // scalastyle:off
- def datetimeRebaseMode(lookupFileMeta: String => String,
- modeByConfig: String): LegacyBehaviorPolicy.Value = {
- if (Utils.isTesting && SQLConf.get.getConfString("spark.test.forceNoRebase", "") == "true") {
- return LegacyBehaviorPolicy.CORRECTED
- }
- // If there is no version, we return the mode specified by the config.
- Option(lookupFileMeta(SPARK_VERSION_METADATA_KEY)).map { version =>
- // Files written by Spark 2.4 and earlier follow the legacy hybrid calendar and we need to
- // rebase the datetime values.
- // Files written by Spark 3.0 and latter may also need the rebase if they were written with
- // the "LEGACY" rebase mode.
- if (version < "3.0.0" || lookupFileMeta("org.apache.spark.legacyDateTime") != null) {
- LegacyBehaviorPolicy.LEGACY
- } else {
- LegacyBehaviorPolicy.CORRECTED
- }
- }.getOrElse(LegacyBehaviorPolicy.withName(modeByConfig))
- }
- // scalastyle:on
-
-}
diff --git a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/hudi/Spark34ResolveHudiAlterTableCommand.scala b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/hudi/Spark34ResolveHudiAlterTableCommand.scala
deleted file mode 100644
index 31d2f93efbde..000000000000
--- a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/hudi/Spark34ResolveHudiAlterTableCommand.scala
+++ /dev/null
@@ -1,67 +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.hudi
-
-import org.apache.hudi.internal.schema.action.TableChange.ColumnChangeID
-
-import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.analysis.ResolvedTable
-import org.apache.spark.sql.catalyst.plans.logical._
-import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.hudi.catalog.HoodieInternalV2Table
-import org.apache.spark.sql.hudi.command.{AlterTableCommand => HudiAlterTableCommand}
-
-/**
- * Rule to mostly resolve, normalize and rewrite column names based on case sensitivity.
- * for alter table column commands.
- */
-class Spark34ResolveHudiAlterTableCommand(sparkSession: SparkSession) extends Rule[LogicalPlan] {
-
- def apply(plan: LogicalPlan): LogicalPlan = {
- if (ProvidesHoodieConfig.isSchemaEvolutionEnabled(sparkSession)) {
- plan.resolveOperatorsUp {
- case set@SetTableProperties(ResolvedHoodieV2TablePlan(t), _) if set.resolved =>
- HudiAlterTableCommand(t.v1Table, set.changes, ColumnChangeID.PROPERTY_CHANGE)
- case unSet@UnsetTableProperties(ResolvedHoodieV2TablePlan(t), _, _) if unSet.resolved =>
- HudiAlterTableCommand(t.v1Table, unSet.changes, ColumnChangeID.PROPERTY_CHANGE)
- case drop@DropColumns(ResolvedHoodieV2TablePlan(t), _, _) if drop.resolved =>
- HudiAlterTableCommand(t.v1Table, drop.changes, ColumnChangeID.DELETE)
- case add@AddColumns(ResolvedHoodieV2TablePlan(t), _) if add.resolved =>
- HudiAlterTableCommand(t.v1Table, add.changes, ColumnChangeID.ADD)
- case renameColumn@RenameColumn(ResolvedHoodieV2TablePlan(t), _, _) if renameColumn.resolved =>
- HudiAlterTableCommand(t.v1Table, renameColumn.changes, ColumnChangeID.UPDATE)
- case alter@AlterColumn(ResolvedHoodieV2TablePlan(t), _, _, _, _, _, _) if alter.resolved =>
- HudiAlterTableCommand(t.v1Table, alter.changes, ColumnChangeID.UPDATE)
- case replace@ReplaceColumns(ResolvedHoodieV2TablePlan(t), _) if replace.resolved =>
- HudiAlterTableCommand(t.v1Table, replace.changes, ColumnChangeID.REPLACE)
- }
- } else {
- plan
- }
- }
-
- object ResolvedHoodieV2TablePlan {
- def unapply(plan: LogicalPlan): Option[HoodieInternalV2Table] = {
- plan match {
- case ResolvedTable(_, _, v2Table: HoodieInternalV2Table, _) => Some(v2Table)
- case _ => None
- }
- }
- }
-}
-
diff --git a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/hudi/Spark35HoodieFileScanRDD.scala b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/hudi/Spark35HoodieFileScanRDD.scala
deleted file mode 100644
index 9ab3c04605d5..000000000000
--- a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/hudi/Spark35HoodieFileScanRDD.scala
+++ /dev/null
@@ -1,36 +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.hudi
-
-import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.AttributeReference
-import org.apache.spark.sql.execution.datasources.{FilePartition, FileScanRDD, PartitionedFile}
-import org.apache.spark.sql.types.StructType
-
-class Spark35HoodieFileScanRDD(@transient private val sparkSession: SparkSession,
- read: PartitionedFile => Iterator[InternalRow],
- @transient filePartitions: Seq[FilePartition],
- readDataSchema: StructType,
- metadataColumns: Seq[AttributeReference] = Seq.empty)
- extends FileScanRDD(sparkSession, read, filePartitions, readDataSchema, metadataColumns)
- with HoodieUnsafeRDD {
-
- override final def collect(): Array[InternalRow] = super[HoodieUnsafeRDD].collect()
-}
diff --git a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_5Adapter.scala b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_5Adapter.scala
index a2dfe3e7be60..3fd01ae7085b 100644
--- a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_5Adapter.scala
+++ b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/adapter/Spark3_5Adapter.scala
@@ -17,7 +17,6 @@
package org.apache.spark.sql.adapter
-import org.apache.hudi.Spark35HoodieFileScanRDD
import org.apache.hudi.common.schema.HoodieSchema
import org.apache.hudi.storage.StorageConfiguration
@@ -30,7 +29,7 @@ import org.apache.spark.sql.avro._
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.{EliminateSubqueryAliases, ResolvedTable}
import org.apache.spark.sql.catalyst.catalog.CatalogTable
-import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Expression}
+import org.apache.spark.sql.catalyst.expressions.{Expression}
import org.apache.spark.sql.catalyst.parser.ParserInterface
import org.apache.spark.sql.catalyst.planning.PhysicalOperation
import org.apache.spark.sql.catalyst.plans.logical._
@@ -38,7 +37,7 @@ import org.apache.spark.sql.catalyst.util.{METADATA_COL_ATTR_KEY, RebaseDateTime
import org.apache.spark.sql.connector.catalog.{V1Table, V2TableWithV1Fallback}
import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.execution.datasources.lance.SparkLanceReaderBase
-import org.apache.spark.sql.execution.datasources.orc.Spark35OrcReader
+import org.apache.spark.sql.execution.datasources.orc.{OrcColumnarBatchReader, SparkOrcReaderBase}
import org.apache.spark.sql.execution.datasources.parquet.{ParquetFileFormat, ParquetFilters, Spark35LegacyHoodieParquetFileFormat, Spark35ParquetReader}
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
import org.apache.spark.sql.hudi.analysis.TableValuedFunctions
@@ -103,14 +102,6 @@ class Spark3_5Adapter extends BaseSpark3Adapter {
Some(new Spark35LegacyHoodieParquetFileFormat(appendPartitionValues))
}
- override def createHoodieFileScanRDD(sparkSession: SparkSession,
- readFunction: PartitionedFile => Iterator[InternalRow],
- filePartitions: Seq[FilePartition],
- readDataSchema: StructType,
- metadataColumns: Seq[AttributeReference] = Seq.empty): FileScanRDD = {
- new Spark35HoodieFileScanRDD(sparkSession, readFunction, filePartitions, readDataSchema, metadataColumns)
- }
-
override def extractDeleteCondition(deleteFromTable: Command): Expression = {
deleteFromTable.asInstanceOf[DeleteFromTable].condition
}
@@ -179,7 +170,8 @@ class Spark3_5Adapter extends BaseSpark3Adapter {
options: Map[String, String],
hadoopConf: Configuration,
dataSchema: StructType): SparkColumnarFileReader = {
- Spark35OrcReader.build(vectorized, sqlConf, options, hadoopConf, dataSchema)
+ SparkOrcReaderBase.build(vectorized, sqlConf, options, hadoopConf, dataSchema,
+ (capacity, memoryMode) => new OrcColumnarBatchReader(capacity, memoryMode))
}
override def createLanceFileReader(vectorized: Boolean,
diff --git a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/Spark35NestedSchemaPruning.scala b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/Spark35NestedSchemaPruning.scala
deleted file mode 100644
index 24d07b085af9..000000000000
--- a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/Spark35NestedSchemaPruning.scala
+++ /dev/null
@@ -1,61 +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.execution.datasources
-
-import org.apache.hudi.HoodieBaseRelation
-
-import org.apache.spark.sql.catalyst.expressions.AttributeReference
-import org.apache.spark.sql.catalyst.planning.PhysicalOperation
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
-import org.apache.spark.sql.catalyst.types.DataTypeUtils
-import org.apache.spark.sql.types.StructType
-
-class Spark35NestedSchemaPruning extends BaseHoodieNestedSchemaPruning {
-
- // Prune the given output to make it consistent with `requiredSchema`.
- protected def getPrunedOutput(output: Seq[AttributeReference],
- requiredSchema: StructType): Seq[AttributeReference] = {
- // We need to replace the expression ids of the pruned relation output attributes
- // with the expression ids of the original relation output attributes so that
- // references to the original relation's output are not broken
- val outputIdMap = output.map(att => (att.name, att.exprId)).toMap
- DataTypeUtils.toAttributes(requiredSchema)
- .map {
- case att if outputIdMap.contains(att.name) =>
- att.withExprId(outputIdMap(att.name))
- case att => att
- }
- }
-
- override protected def apply0(plan: LogicalPlan): LogicalPlan =
- plan transformDown {
- case op @ PhysicalOperation(projects, filters,
- // NOTE: This is modified to accommodate for Hudi's custom relations, given that original
- // [[NestedSchemaPruning]] rule is tightly coupled w/ [[HadoopFsRelation]]
- // TODO generalize to any file-based relation
- l @ LogicalRelation(relation: HoodieBaseRelation, _, _, _))
- if relation.canPruneRelationSchema =>
-
- prunePhysicalColumns(l.output, projects, filters, relation.dataSchema,
- prunedDataSchema => {
- val prunedRelation =
- relation.updatePrunedDataSchema(prunedSchema = prunedDataSchema)
- buildPrunedRelation(l, prunedRelation)
- }).getOrElse(op)
- }
-}
diff --git a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/orc/Spark35OrcReader.scala b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/orc/Spark35OrcReader.scala
deleted file mode 100644
index badd76c6a885..000000000000
--- a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/orc/Spark35OrcReader.scala
+++ /dev/null
@@ -1,92 +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.execution.datasources.orc
-
-import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.Path
-import org.apache.spark.memory.MemoryMode
-import org.apache.spark.sql.catalyst.expressions.Attribute
-import org.apache.spark.sql.catalyst.types.DataTypeUtils.toAttributes
-import org.apache.spark.sql.execution.datasources.{FileFormat, PartitionedFile}
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.StructType
-
-class Spark35OrcReader(enableVectorizedReader: Boolean,
- memoryMode: MemoryMode,
- dataSchema: StructType,
- orcFilterPushDown: Boolean,
- isCaseSensitive: Boolean,
- capacity: Int) extends SparkOrcReaderBase(enableVectorizedReader, dataSchema, orcFilterPushDown, isCaseSensitive) {
-
- override def partitionedFileToPath(file: PartitionedFile): Path = {
- file.toPath
- }
-
- override def buildReader(): OrcColumnarBatchReader = {
- new OrcColumnarBatchReader(capacity, memoryMode)
- }
-
- override def structTypeToAttributes(schema: StructType): Seq[Attribute] = {
- toAttributes(schema)
- }
-}
-
-object Spark35OrcReader {
- /**
- * Get ORC file reader
- *
- * @param vectorized true if vectorized reading is not prohibited due to schema, reading mode, etc
- * @param sqlConf the [[SQLConf]] used for the read
- * @param options passed as a param to the file format
- * @param hadoopConf some configs will be set for the hadoopConf
- * @return ORC file reader
- */
- def build(vectorized: Boolean,
- sqlConf: SQLConf,
- options: Map[String, String],
- hadoopConf: Configuration,
- dataSchema: StructType): Spark35OrcReader = {
- //set hadoopconf
- hadoopConf.set(SQLConf.SESSION_LOCAL_TIMEZONE.key, sqlConf.sessionLocalTimeZone)
- hadoopConf.setBoolean(SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key, sqlConf.nestedSchemaPruningEnabled)
- hadoopConf.setBoolean(SQLConf.CASE_SENSITIVE.key, sqlConf.caseSensitiveAnalysis)
-
- val memoryMode = if (sqlConf.offHeapColumnVectorEnabled) {
- MemoryMode.OFF_HEAP
- } else {
- MemoryMode.ON_HEAP
- }
-
- val enableVectorizedReader = sqlConf.orcVectorizedReaderEnabled &&
- options.getOrElse(FileFormat.OPTION_RETURNING_BATCH,
- throw new IllegalArgumentException(
- "OPTION_RETURNING_BATCH should always be set for OrcFileFormat. " +
- "To workaround this issue, set spark.sql.orc.enableVectorizedReader=false."))
- .equals("true")
-
- new Spark35OrcReader(
- enableVectorizedReader = enableVectorizedReader && vectorized,
- memoryMode = memoryMode,
- isCaseSensitive = sqlConf.caseSensitiveAnalysis,
- capacity = sqlConf.orcVectorizedReaderBatchSize,
- orcFilterPushDown = sqlConf.orcFilterPushDown,
- dataSchema = dataSchema)
- }
-}
diff --git a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark35DataSourceUtils.scala b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark35DataSourceUtils.scala
deleted file mode 100644
index 9e3c63529b48..000000000000
--- a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark35DataSourceUtils.scala
+++ /dev/null
@@ -1,76 +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.execution.datasources.parquet
-
-import org.apache.spark.sql.SPARK_VERSION_METADATA_KEY
-import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
-import org.apache.spark.util.Utils
-
-object Spark35DataSourceUtils {
-
- /**
- * NOTE: This method was copied from [[Spark32PlusDataSourceUtils]], and is required to maintain runtime
- * compatibility against Spark 3.5.0
- */
- // scalastyle:off
- def int96RebaseMode(lookupFileMeta: String => String,
- modeByConfig: String): LegacyBehaviorPolicy.Value = {
- if (Utils.isTesting && SQLConf.get.getConfString("spark.test.forceNoRebase", "") == "true") {
- return LegacyBehaviorPolicy.CORRECTED
- }
- // If there is no version, we return the mode specified by the config.
- Option(lookupFileMeta(SPARK_VERSION_METADATA_KEY)).map { version =>
- // Files written by Spark 3.0 and earlier follow the legacy hybrid calendar and we need to
- // rebase the INT96 timestamp values.
- // Files written by Spark 3.1 and latter may also need the rebase if they were written with
- // the "LEGACY" rebase mode.
- if (version < "3.1.0" || lookupFileMeta("org.apache.spark.legacyINT96") != null) {
- LegacyBehaviorPolicy.LEGACY
- } else {
- LegacyBehaviorPolicy.CORRECTED
- }
- }.getOrElse(LegacyBehaviorPolicy.withName(modeByConfig))
- }
- // scalastyle:on
-
- /**
- * NOTE: This method was copied from Spark 3.2.0, and is required to maintain runtime
- * compatibility against Spark 3.2.0
- */
- // scalastyle:off
- def datetimeRebaseMode(lookupFileMeta: String => String,
- modeByConfig: String): LegacyBehaviorPolicy.Value = {
- if (Utils.isTesting && SQLConf.get.getConfString("spark.test.forceNoRebase", "") == "true") {
- return LegacyBehaviorPolicy.CORRECTED
- }
- // If there is no version, we return the mode specified by the config.
- Option(lookupFileMeta(SPARK_VERSION_METADATA_KEY)).map { version =>
- // Files written by Spark 2.4 and earlier follow the legacy hybrid calendar and we need to
- // rebase the datetime values.
- // Files written by Spark 3.0 and latter may also need the rebase if they were written with
- // the "LEGACY" rebase mode.
- if (version < "3.0.0" || lookupFileMeta("org.apache.spark.legacyDateTime") != null) {
- LegacyBehaviorPolicy.LEGACY
- } else {
- LegacyBehaviorPolicy.CORRECTED
- }
- }.getOrElse(LegacyBehaviorPolicy.withName(modeByConfig))
- }
- // scalastyle:on
-
-}
diff --git a/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark4Adapter.scala b/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark4Adapter.scala
index ea6e96943a69..27a835b821f2 100644
--- a/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark4Adapter.scala
+++ b/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark4Adapter.scala
@@ -17,13 +17,14 @@
package org.apache.spark.sql.adapter
-import org.apache.hudi.{AvroConversionUtils, DefaultSource, HoodieSchemaConversionUtils}
+import org.apache.hudi.{AvroConversionUtils, DefaultSource, HoodieFileScanRDD, HoodieSchemaConversionUtils}
import org.apache.hudi.common.schema.HoodieSchema
import org.apache.hudi.common.table.HoodieTableMetaClient
import org.apache.hudi.common.util.JsonUtils
import org.apache.hudi.spark.internal.ReflectUtil
import org.apache.hudi.storage.StorageConfiguration
+import org.apache.hadoop.conf.Configuration
import org.apache.parquet.schema.{GroupType, MessageType, PrimitiveType, Type, Types}
import org.apache.parquet.schema.Type.Repetition
import org.apache.spark.api.java.JavaSparkContext
@@ -34,7 +35,7 @@ import org.apache.spark.sql.FileFormatUtilsForFileGroupReader.applyFiltersToPlan
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.EliminateSubqueryAliases
import org.apache.spark.sql.catalyst.catalog.CatalogTable
-import org.apache.spark.sql.catalyst.expressions.{Expression, InterpretedPredicate, Predicate, SpecializedGetters}
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Expression, InterpretedPredicate, Predicate, SpecializedGetters}
import org.apache.spark.sql.catalyst.parser.ParseException
import org.apache.spark.sql.catalyst.planning.PhysicalOperation
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
@@ -42,6 +43,7 @@ import org.apache.spark.sql.catalyst.util.DateFormatter
import org.apache.spark.sql.classic.ColumnConversions
import org.apache.spark.sql.execution.{PartitionedFileUtil, QueryExecution, SQLExecution}
import org.apache.spark.sql.execution.datasources._
+import org.apache.spark.sql.execution.datasources.orc.{OrcColumnarBatchReader, SparkOrcReaderBase}
import org.apache.spark.sql.execution.datasources.parquet.{HoodieFormatTrait, ParquetFilters, SparkShreddingUtils}
import org.apache.spark.sql.hudi.SparkAdapter
import org.apache.spark.sql.internal.SQLConf
@@ -101,6 +103,23 @@ abstract class BaseSpark4Adapter extends SparkAdapter with Logging {
Predicate.createInterpreted(e)
}
+ override def createHoodieFileScanRDD(sparkSession: SparkSession,
+ readFunction: PartitionedFile => Iterator[InternalRow],
+ filePartitions: Seq[FilePartition],
+ readDataSchema: StructType,
+ metadataColumns: Seq[AttributeReference] = Seq.empty): FileScanRDD = {
+ new HoodieFileScanRDD(sparkSession, readFunction, filePartitions, readDataSchema, metadataColumns)
+ }
+
+ override def createOrcFileReader(vectorized: Boolean,
+ sqlConf: SQLConf,
+ options: Map[String, String],
+ hadoopConf: Configuration,
+ dataSchema: StructType): SparkColumnarFileReader = {
+ SparkOrcReaderBase.build(vectorized, sqlConf, options, hadoopConf, dataSchema,
+ (capacity, memoryMode) => new OrcColumnarBatchReader(capacity, memoryMode))
+ }
+
override def createRelation(sqlContext: SQLContext,
metaClient: HoodieTableMetaClient,
schema: HoodieSchema,
@@ -130,6 +149,8 @@ abstract class BaseSpark4Adapter extends SparkAdapter with Logging {
override def getUTF8StringFactory: HoodieUTF8StringFactory = Spark4HoodieUTF8StringFactory
+ override def getSparkPartitionedFileUtils: HoodieSparkPartitionedFileUtils = HoodieSpark4PartitionedFileUtils
+
override def splitFiles(sparkSession: SparkSession,
partitionDirectory: PartitionDirectory,
isSplitable: Boolean,
diff --git a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/HoodieSpark41PartitionedFileUtils.scala b/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/execution/datasources/HoodieSpark4PartitionedFileUtils.scala
similarity index 96%
rename from hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/HoodieSpark41PartitionedFileUtils.scala
rename to hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/execution/datasources/HoodieSpark4PartitionedFileUtils.scala
index d11ec6baee4f..824ff9f93080 100644
--- a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/HoodieSpark41PartitionedFileUtils.scala
+++ b/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/execution/datasources/HoodieSpark4PartitionedFileUtils.scala
@@ -27,9 +27,9 @@ import org.apache.spark.paths.SparkPath
import org.apache.spark.sql.catalyst.InternalRow
/**
- * Utils on Spark [[PartitionedFile]] and [[PartitionDirectory]] for Spark 4.0.
+ * Utils on Spark [[PartitionedFile]] and [[PartitionDirectory]] for Spark 4.x.
*/
-object HoodieSpark41PartitionedFileUtils extends HoodieSparkPartitionedFileUtils {
+object HoodieSpark4PartitionedFileUtils extends HoodieSparkPartitionedFileUtils {
override def getPathFromPartitionedFile(partitionedFile: PartitionedFile): StoragePath = {
new StoragePath(partitionedFile.filePath.toUri)
}
diff --git a/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/hudi/Spark4ResolveHudiAlterTableCommand.scala b/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/hudi/Spark4ResolveHudiAlterTableCommand.scala
new file mode 100644
index 000000000000..73ada28f7ccd
--- /dev/null
+++ b/hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/hudi/Spark4ResolveHudiAlterTableCommand.scala
@@ -0,0 +1,37 @@
+/*
+ * 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.hudi
+
+import org.apache.hudi.internal.schema.action.TableChange.ColumnChangeID
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.plans.logical.{AlterColumns, LogicalPlan}
+import org.apache.spark.sql.hudi.command.{AlterTableCommand => HudiAlterTableCommand}
+
+/**
+ * Rule to mostly resolve, normalize and rewrite column names based on case sensitivity.
+ * for alter table column commands.
+ */
+class Spark4ResolveHudiAlterTableCommand(sparkSession: SparkSession)
+ extends BaseResolveHudiAlterTableCommand(sparkSession) {
+
+ override protected def resolveAlterColumnCommand: PartialFunction[LogicalPlan, LogicalPlan] = {
+ case alter@AlterColumns(ResolvedHoodieV2TablePlan(t), _) if alter.resolved =>
+ HudiAlterTableCommand(t.v1Table, alter.changes, ColumnChangeID.UPDATE)
+ }
+}
diff --git a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/hudi/Spark40HoodieFileScanRDD.scala b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/hudi/Spark40HoodieFileScanRDD.scala
deleted file mode 100644
index 5e9792a0677d..000000000000
--- a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/hudi/Spark40HoodieFileScanRDD.scala
+++ /dev/null
@@ -1,36 +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.hudi
-
-import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.AttributeReference
-import org.apache.spark.sql.execution.datasources.{FilePartition, FileScanRDD, PartitionedFile}
-import org.apache.spark.sql.types.StructType
-
-class Spark40HoodieFileScanRDD(@transient private val sparkSession: SparkSession,
- read: PartitionedFile => Iterator[InternalRow],
- @transient filePartitions: Seq[FilePartition],
- readDataSchema: StructType,
- metadataColumns: Seq[AttributeReference] = Seq.empty)
- extends FileScanRDD(sparkSession, read, filePartitions, readDataSchema, metadataColumns)
- with HoodieUnsafeRDD {
-
- override final def collect(): Array[InternalRow] = super[HoodieUnsafeRDD].collect()
-}
diff --git a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_0Adapter.scala b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_0Adapter.scala
index 7a3d8bec2403..519ec97cd771 100644
--- a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_0Adapter.scala
+++ b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_0Adapter.scala
@@ -17,7 +17,7 @@
package org.apache.spark.sql.adapter
-import org.apache.hudi.{HoodiePartitionCDCFileGroupMapping, HoodiePartitionFileSliceMapping, Spark40HoodieFileScanRDD, Spark40HoodiePartitionCDCFileGroupMapping, Spark40HoodiePartitionFileSliceMapping}
+import org.apache.hudi.{HoodiePartitionCDCFileGroupMapping, HoodiePartitionFileSliceMapping, Spark40HoodiePartitionCDCFileGroupMapping, Spark40HoodiePartitionFileSliceMapping}
import org.apache.hudi.client.model.{HoodieInternalRow, Spark40HoodieInternalRow}
import org.apache.hudi.common.model.FileSlice
import org.apache.hudi.common.schema.HoodieSchema
@@ -33,7 +33,7 @@ import org.apache.spark.sql.avro._
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.{EliminateSubqueryAliases, ResolvedTable}
import org.apache.spark.sql.catalyst.catalog.CatalogTable
-import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Expression}
+import org.apache.spark.sql.catalyst.expressions.{Expression}
import org.apache.spark.sql.catalyst.parser.{ParseException, ParserInterface}
import org.apache.spark.sql.catalyst.planning.PhysicalOperation
import org.apache.spark.sql.catalyst.plans.logical._
@@ -43,7 +43,6 @@ import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec
import org.apache.spark.sql.connector.catalog.{V1Table, V2TableWithV1Fallback}
import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.execution.datasources.lance.SparkLanceReaderBase
-import org.apache.spark.sql.execution.datasources.orc.Spark40OrcReader
import org.apache.spark.sql.execution.datasources.parquet.{HoodieParquetReadSupport, ParquetFileFormat, Spark40HoodieParquetReadSupport, Spark40LegacyHoodieParquetFileFormat, Spark40ParquetReader}
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
import org.apache.spark.sql.execution.streaming.MemoryStream
@@ -98,8 +97,6 @@ class Spark4_0Adapter extends BaseSpark4Adapter {
override def getSchemaUtils: HoodieSchemaUtils = HoodieSpark40SchemaUtils
- override def getSparkPartitionedFileUtils: HoodieSparkPartitionedFileUtils = HoodieSpark40PartitionedFileUtils
-
override def newParseException(command: Option[String],
exception: AnalysisException,
start: Origin,
@@ -136,14 +133,6 @@ class Spark4_0Adapter extends BaseSpark4Adapter {
new Spark40HoodiePartitionFileSliceMapping(values, slices)
}
- override def createHoodieFileScanRDD(sparkSession: SparkSession,
- readFunction: PartitionedFile => Iterator[InternalRow],
- filePartitions: Seq[FilePartition],
- readDataSchema: StructType,
- metadataColumns: Seq[AttributeReference] = Seq.empty): FileScanRDD = {
- new Spark40HoodieFileScanRDD(sparkSession, readFunction, filePartitions, readDataSchema, metadataColumns)
- }
-
override def extractDeleteCondition(deleteFromTable: Command): Expression = {
deleteFromTable.asInstanceOf[DeleteFromTable].condition
}
@@ -209,23 +198,6 @@ class Spark4_0Adapter extends BaseSpark4Adapter {
datetimeRebaseSpec, getRebaseSpec("LEGACY"), tableSchemaOpt)
}
- /**
- * TODO
- *
- * @param vectorized
- * @param sqlConf
- * @param options
- * @param hadoopConf
- * @return
- */
- override def createOrcFileReader(vectorized: Boolean,
- sqlConf: SQLConf,
- options: Map[String, String],
- hadoopConf: Configuration,
- dataSchema: StructType): SparkColumnarFileReader = {
- Spark40OrcReader.build(vectorized, sqlConf, options, hadoopConf, dataSchema)
- }
-
override def createLanceFileReader(vectorized: Boolean,
sqlConf: SQLConf,
options: Map[String, String],
diff --git a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/HoodieSpark40PartitionedFileUtils.scala b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/HoodieSpark40PartitionedFileUtils.scala
deleted file mode 100644
index bc83633383a9..000000000000
--- a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/HoodieSpark40PartitionedFileUtils.scala
+++ /dev/null
@@ -1,68 +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.execution.datasources
-
-import org.apache.hudi.common.util.ReflectionUtils
-import org.apache.hudi.storage.StoragePath
-
-import org.apache.hadoop.fs.FileStatus
-import org.apache.spark.paths.SparkPath
-import org.apache.spark.sql.catalyst.InternalRow
-
-/**
- * Utils on Spark [[PartitionedFile]] and [[PartitionDirectory]] for Spark 4.0.
- */
-object HoodieSpark40PartitionedFileUtils extends HoodieSparkPartitionedFileUtils {
- override def getPathFromPartitionedFile(partitionedFile: PartitionedFile): StoragePath = {
- new StoragePath(partitionedFile.filePath.toUri)
- }
-
- override def getStringPathFromPartitionedFile(partitionedFile: PartitionedFile): String = {
- partitionedFile.filePath.toPath.toString
- }
-
- override def createPartitionedFile(partitionValues: InternalRow,
- filePath: StoragePath,
- start: Long,
- length: Long): PartitionedFile = {
- PartitionedFile(partitionValues, SparkPath.fromUri(filePath.toUri), start, length, Array.empty)
- }
-
- override def toFileStatuses(partitionDirs: Seq[PartitionDirectory]): Seq[FileStatus] = {
- val files: Seq[FileStatusWithMetadata] = partitionDirs.flatMap(_.files)
- try {
- files.map(_.fileStatus)
- } catch {
- case _: NoSuchMethodException | _: NoSuchMethodError | _: IllegalArgumentException =>
- val methodOpt = ReflectionUtils.getMethod(classOf[FileStatusWithMetadata], "toFileStatus")
- if (methodOpt.isPresent) {
- val method = methodOpt.get()
- files.map(f => method.invoke(f).asInstanceOf[FileStatus])
- } else {
- throw new RuntimeException(
- "Cannot find toFileStatus method on FileStatusWithMetadata in custom Spark Runtime")
- }
- }
- }
-
- override def newPartitionDirectory(internalRow: InternalRow, statuses: Seq[FileStatus]): PartitionDirectory = {
- PartitionDirectory(internalRow, statuses.toArray)
- }
-}
diff --git a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/Spark40NestedSchemaPruning.scala b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/Spark40NestedSchemaPruning.scala
deleted file mode 100644
index c10989d89d9f..000000000000
--- a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/Spark40NestedSchemaPruning.scala
+++ /dev/null
@@ -1,61 +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.execution.datasources
-
-import org.apache.hudi.HoodieBaseRelation
-
-import org.apache.spark.sql.catalyst.expressions.AttributeReference
-import org.apache.spark.sql.catalyst.planning.PhysicalOperation
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
-import org.apache.spark.sql.catalyst.types.DataTypeUtils
-import org.apache.spark.sql.types.StructType
-
-class Spark40NestedSchemaPruning extends BaseHoodieNestedSchemaPruning {
-
- // Prune the given output to make it consistent with `requiredSchema`.
- protected def getPrunedOutput(output: Seq[AttributeReference],
- requiredSchema: StructType): Seq[AttributeReference] = {
- // We need to replace the expression ids of the pruned relation output attributes
- // with the expression ids of the original relation output attributes so that
- // references to the original relation's output are not broken
- val outputIdMap = output.map(att => (att.name, att.exprId)).toMap
- DataTypeUtils.toAttributes(requiredSchema)
- .map {
- case att if outputIdMap.contains(att.name) =>
- att.withExprId(outputIdMap(att.name))
- case att => att
- }
- }
-
- override protected def apply0(plan: LogicalPlan): LogicalPlan =
- plan transformDown {
- case op @ PhysicalOperation(projects, filters,
- // NOTE: This is modified to accommodate for Hudi's custom relations, given that original
- // [[NestedSchemaPruning]] rule is tightly coupled w/ [[HadoopFsRelation]]
- // TODO generalize to any file-based relation
- l @ LogicalRelation(relation: HoodieBaseRelation, _, _, _, _))
- if relation.canPruneRelationSchema =>
-
- prunePhysicalColumns(l.output, projects, filters, relation.dataSchema,
- prunedDataSchema => {
- val prunedRelation =
- relation.updatePrunedDataSchema(prunedSchema = prunedDataSchema)
- buildPrunedRelation(l, prunedRelation)
- }).getOrElse(op)
- }
-}
diff --git a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/orc/Spark40OrcReader.scala b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/orc/Spark40OrcReader.scala
deleted file mode 100644
index d1da60f5bf08..000000000000
--- a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/orc/Spark40OrcReader.scala
+++ /dev/null
@@ -1,92 +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.execution.datasources.orc
-
-import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.Path
-import org.apache.spark.memory.MemoryMode
-import org.apache.spark.sql.catalyst.expressions.Attribute
-import org.apache.spark.sql.catalyst.types.DataTypeUtils.toAttributes
-import org.apache.spark.sql.execution.datasources.{FileFormat, PartitionedFile}
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.StructType
-
-class Spark40OrcReader(enableVectorizedReader: Boolean,
- memoryMode: MemoryMode,
- dataSchema: StructType,
- orcFilterPushDown: Boolean,
- isCaseSensitive: Boolean,
- capacity: Int) extends SparkOrcReaderBase(enableVectorizedReader, dataSchema, orcFilterPushDown, isCaseSensitive) {
-
- override def partitionedFileToPath(file: PartitionedFile): Path = {
- file.toPath
- }
-
- override def buildReader(): OrcColumnarBatchReader = {
- new OrcColumnarBatchReader(capacity, memoryMode)
- }
-
- override def structTypeToAttributes(schema: StructType): Seq[Attribute] = {
- toAttributes(schema)
- }
-}
-
-object Spark40OrcReader {
- /**
- * Get ORC file reader
- *
- * @param vectorized true if vectorized reading is not prohibited due to schema, reading mode, etc
- * @param sqlConf the [[SQLConf]] used for the read
- * @param options passed as a param to the file format
- * @param hadoopConf some configs will be set for the hadoopConf
- * @return ORC file reader
- */
- def build(vectorized: Boolean,
- sqlConf: SQLConf,
- options: Map[String, String],
- hadoopConf: Configuration,
- dataSchema: StructType): Spark40OrcReader = {
- //set hadoopconf
- hadoopConf.set(SQLConf.SESSION_LOCAL_TIMEZONE.key, sqlConf.sessionLocalTimeZone)
- hadoopConf.setBoolean(SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key, sqlConf.nestedSchemaPruningEnabled)
- hadoopConf.setBoolean(SQLConf.CASE_SENSITIVE.key, sqlConf.caseSensitiveAnalysis)
-
- val memoryMode = if (sqlConf.offHeapColumnVectorEnabled) {
- MemoryMode.OFF_HEAP
- } else {
- MemoryMode.ON_HEAP
- }
-
- val enableVectorizedReader = sqlConf.orcVectorizedReaderEnabled &&
- options.getOrElse(FileFormat.OPTION_RETURNING_BATCH,
- throw new IllegalArgumentException(
- "OPTION_RETURNING_BATCH should always be set for OrcFileFormat. " +
- "To workaround this issue, set spark.sql.orc.enableVectorizedReader=false."))
- .equals("true")
-
- new Spark40OrcReader(
- enableVectorizedReader = enableVectorizedReader && vectorized,
- memoryMode = memoryMode,
- isCaseSensitive = sqlConf.caseSensitiveAnalysis,
- capacity = sqlConf.orcVectorizedReaderBatchSize,
- orcFilterPushDown = sqlConf.orcFilterPushDown,
- dataSchema = dataSchema)
- }
-}
diff --git a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark40DataSourceUtils.scala b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark40DataSourceUtils.scala
deleted file mode 100644
index 3b84a3e164be..000000000000
--- a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark40DataSourceUtils.scala
+++ /dev/null
@@ -1,76 +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.execution.datasources.parquet
-
-import org.apache.spark.sql.SPARK_VERSION_METADATA_KEY
-import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
-import org.apache.spark.util.Utils
-
-object Spark40DataSourceUtils {
-
- /**
- * NOTE: This method was copied from [[Spark32PlusDataSourceUtils]], and is required to maintain runtime
- * compatibility against Spark 3.5.0
- */
- // scalastyle:off
- def int96RebaseMode(lookupFileMeta: String => String,
- modeByConfig: String): LegacyBehaviorPolicy.Value = {
- if (Utils.isTesting && SQLConf.get.getConfString("spark.test.forceNoRebase", "") == "true") {
- return LegacyBehaviorPolicy.CORRECTED
- }
- // If there is no version, we return the mode specified by the config.
- Option(lookupFileMeta(SPARK_VERSION_METADATA_KEY)).map { version =>
- // Files written by Spark 3.0 and earlier follow the legacy hybrid calendar and we need to
- // rebase the INT96 timestamp values.
- // Files written by Spark 3.1 and latter may also need the rebase if they were written with
- // the "LEGACY" rebase mode.
- if (version < "3.1.0" || lookupFileMeta("org.apache.spark.legacyINT96") != null) {
- LegacyBehaviorPolicy.LEGACY
- } else {
- LegacyBehaviorPolicy.CORRECTED
- }
- }.getOrElse(LegacyBehaviorPolicy.withName(modeByConfig))
- }
- // scalastyle:on
-
- /**
- * NOTE: This method was copied from Spark 3.2.0, and is required to maintain runtime
- * compatibility against Spark 3.2.0
- */
- // scalastyle:off
- def datetimeRebaseMode(lookupFileMeta: String => String,
- modeByConfig: String): LegacyBehaviorPolicy.Value = {
- if (Utils.isTesting && SQLConf.get.getConfString("spark.test.forceNoRebase", "") == "true") {
- return LegacyBehaviorPolicy.CORRECTED
- }
- // If there is no version, we return the mode specified by the config.
- Option(lookupFileMeta(SPARK_VERSION_METADATA_KEY)).map { version =>
- // Files written by Spark 2.4 and earlier follow the legacy hybrid calendar and we need to
- // rebase the datetime values.
- // Files written by Spark 3.0 and latter may also need the rebase if they were written with
- // the "LEGACY" rebase mode.
- if (version < "3.0.0" || lookupFileMeta("org.apache.spark.legacyDateTime") != null) {
- LegacyBehaviorPolicy.LEGACY
- } else {
- LegacyBehaviorPolicy.CORRECTED
- }
- }.getOrElse(LegacyBehaviorPolicy.withName(modeByConfig))
- }
- // scalastyle:on
-
-}
diff --git a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/hudi/Spark40ResolveHudiAlterTableCommand.scala b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/hudi/Spark40ResolveHudiAlterTableCommand.scala
deleted file mode 100644
index e64dae370d4e..000000000000
--- a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/hudi/Spark40ResolveHudiAlterTableCommand.scala
+++ /dev/null
@@ -1,68 +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.hudi
-
-import org.apache.hudi.common.config.HoodieCommonConfig
-import org.apache.hudi.internal.schema.action.TableChange.ColumnChangeID
-
-import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.analysis.ResolvedTable
-import org.apache.spark.sql.catalyst.plans.logical._
-import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.hudi.catalog.HoodieInternalV2Table
-import org.apache.spark.sql.hudi.command.{AlterTableCommand => HudiAlterTableCommand}
-
-/**
- * Rule to mostly resolve, normalize and rewrite column names based on case sensitivity.
- * for alter table column commands.
- */
-class Spark40ResolveHudiAlterTableCommand(sparkSession: SparkSession) extends Rule[LogicalPlan] {
-
- def apply(plan: LogicalPlan): LogicalPlan = {
- if (ProvidesHoodieConfig.isSchemaEvolutionEnabled(sparkSession)) {
- plan.resolveOperatorsUp {
- case set@SetTableProperties(ResolvedHoodieV2TablePlan(t), _) if set.resolved =>
- HudiAlterTableCommand(t.v1Table, set.changes, ColumnChangeID.PROPERTY_CHANGE)
- case unSet@UnsetTableProperties(ResolvedHoodieV2TablePlan(t), _, _) if unSet.resolved =>
- HudiAlterTableCommand(t.v1Table, unSet.changes, ColumnChangeID.PROPERTY_CHANGE)
- case drop@DropColumns(ResolvedHoodieV2TablePlan(t), _, _) if drop.resolved =>
- HudiAlterTableCommand(t.v1Table, drop.changes, ColumnChangeID.DELETE)
- case add@AddColumns(ResolvedHoodieV2TablePlan(t), _) if add.resolved =>
- HudiAlterTableCommand(t.v1Table, add.changes, ColumnChangeID.ADD)
- case renameColumn@RenameColumn(ResolvedHoodieV2TablePlan(t), _, _) if renameColumn.resolved =>
- HudiAlterTableCommand(t.v1Table, renameColumn.changes, ColumnChangeID.UPDATE)
- case alter@AlterColumns(ResolvedHoodieV2TablePlan(t), _) if alter.resolved =>
- HudiAlterTableCommand(t.v1Table, alter.changes, ColumnChangeID.UPDATE)
- case replace@ReplaceColumns(ResolvedHoodieV2TablePlan(t), _) if replace.resolved =>
- HudiAlterTableCommand(t.v1Table, replace.changes, ColumnChangeID.REPLACE)
- }
- } else {
- plan
- }
- }
-
- object ResolvedHoodieV2TablePlan {
- def unapply(plan: LogicalPlan): Option[HoodieInternalV2Table] = {
- plan match {
- case ResolvedTable(_, _, v2Table: HoodieInternalV2Table, _) => Some(v2Table)
- case _ => None
- }
- }
- }
-}
-
diff --git a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/hudi/Spark41HoodieFileScanRDD.scala b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/hudi/Spark41HoodieFileScanRDD.scala
deleted file mode 100644
index 4b27188b51c1..000000000000
--- a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/hudi/Spark41HoodieFileScanRDD.scala
+++ /dev/null
@@ -1,36 +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.hudi
-
-import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.AttributeReference
-import org.apache.spark.sql.execution.datasources.{FilePartition, FileScanRDD, PartitionedFile}
-import org.apache.spark.sql.types.StructType
-
-class Spark41HoodieFileScanRDD(@transient private val sparkSession: SparkSession,
- read: PartitionedFile => Iterator[InternalRow],
- @transient filePartitions: Seq[FilePartition],
- readDataSchema: StructType,
- metadataColumns: Seq[AttributeReference] = Seq.empty)
- extends FileScanRDD(sparkSession, read, filePartitions, readDataSchema, metadataColumns)
- with HoodieUnsafeRDD {
-
- override final def collect(): Array[InternalRow] = super[HoodieUnsafeRDD].collect()
-}
diff --git a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_1Adapter.scala b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_1Adapter.scala
index 195979548bc4..f6a8e19b0e6c 100644
--- a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_1Adapter.scala
+++ b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_1Adapter.scala
@@ -17,7 +17,7 @@
package org.apache.spark.sql.adapter
-import org.apache.hudi.{HoodiePartitionCDCFileGroupMapping, HoodiePartitionFileSliceMapping, Spark41HoodieFileScanRDD, Spark41HoodiePartitionCDCFileGroupMapping, Spark41HoodiePartitionFileSliceMapping}
+import org.apache.hudi.{HoodiePartitionCDCFileGroupMapping, HoodiePartitionFileSliceMapping, Spark41HoodiePartitionCDCFileGroupMapping, Spark41HoodiePartitionFileSliceMapping}
import org.apache.hudi.client.model.{HoodieInternalRow, Spark41HoodieInternalRow}
import org.apache.hudi.common.model.FileSlice
import org.apache.hudi.common.schema.HoodieSchema
@@ -32,7 +32,7 @@ import org.apache.spark.sql.avro._
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.{EliminateSubqueryAliases, ResolvedTable}
import org.apache.spark.sql.catalyst.catalog.CatalogTable
-import org.apache.spark.sql.catalyst.expressions.{AttributeReference, BoundReference, CreateNamedStruct, Expression, Literal, UnsafeProjection}
+import org.apache.spark.sql.catalyst.expressions.{BoundReference, CreateNamedStruct, Expression, Literal, UnsafeProjection}
import org.apache.spark.sql.catalyst.expressions.variant.VariantGet
import org.apache.spark.sql.catalyst.parser.{ParseException, ParserInterface}
import org.apache.spark.sql.catalyst.planning.PhysicalOperation
@@ -43,7 +43,6 @@ import org.apache.spark.sql.catalyst.util.{METADATA_COL_ATTR_KEY, RebaseDateTime
import org.apache.spark.sql.connector.catalog.{V1Table, V2TableWithV1Fallback}
import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.execution.datasources.lance.SparkLanceReaderBase
-import org.apache.spark.sql.execution.datasources.orc.Spark41OrcReader
import org.apache.spark.sql.execution.datasources.parquet.{ParquetFileFormat, Spark41LegacyHoodieParquetFileFormat, Spark41ParquetReader}
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
@@ -98,8 +97,6 @@ class Spark4_1Adapter extends BaseSpark4Adapter {
override def getSchemaUtils: HoodieSchemaUtils = HoodieSpark41SchemaUtils
- override def getSparkPartitionedFileUtils: HoodieSparkPartitionedFileUtils = HoodieSpark41PartitionedFileUtils
-
override def newParseException(command: Option[String],
exception: AnalysisException,
start: Origin,
@@ -138,14 +135,6 @@ class Spark4_1Adapter extends BaseSpark4Adapter {
new Spark41HoodiePartitionFileSliceMapping(values, slices)
}
- override def createHoodieFileScanRDD(sparkSession: SparkSession,
- readFunction: PartitionedFile => Iterator[InternalRow],
- filePartitions: Seq[FilePartition],
- readDataSchema: StructType,
- metadataColumns: Seq[AttributeReference] = Seq.empty): FileScanRDD = {
- new Spark41HoodieFileScanRDD(sparkSession, readFunction, filePartitions, readDataSchema, metadataColumns)
- }
-
override def extractDeleteCondition(deleteFromTable: Command): Expression = {
deleteFromTable.asInstanceOf[DeleteFromTable].condition
}
@@ -200,24 +189,6 @@ class Spark4_1Adapter extends BaseSpark4Adapter {
Spark41ParquetReader.build(vectorized, sqlConf, options, hadoopConf)
}
- /**
- * Get ORC file reader
- *
- * @param vectorized true if vectorized reading is not prohibited due to schema, reading mode, etc
- * @param sqlConf the [[SQLConf]] used for the read
- * @param options passed as a param to the file format
- * @param hadoopConf some configs will be set for the hadoopConf
- * @param dataSchema the data schema of the ORC file
- * @return ORC file reader
- */
- override def createOrcFileReader(vectorized: Boolean,
- sqlConf: SQLConf,
- options: Map[String, String],
- hadoopConf: Configuration,
- dataSchema: StructType): SparkColumnarFileReader = {
- Spark41OrcReader.build(vectorized, sqlConf, options, hadoopConf, dataSchema)
- }
-
override def createLanceFileReader(vectorized: Boolean,
sqlConf: SQLConf,
options: Map[String, String],
diff --git a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/Spark41NestedSchemaPruning.scala b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/Spark41NestedSchemaPruning.scala
deleted file mode 100644
index e9cac2a66ac2..000000000000
--- a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/Spark41NestedSchemaPruning.scala
+++ /dev/null
@@ -1,61 +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.execution.datasources
-
-import org.apache.hudi.HoodieBaseRelation
-
-import org.apache.spark.sql.catalyst.expressions.AttributeReference
-import org.apache.spark.sql.catalyst.planning.PhysicalOperation
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
-import org.apache.spark.sql.catalyst.types.DataTypeUtils
-import org.apache.spark.sql.types.StructType
-
-class Spark41NestedSchemaPruning extends BaseHoodieNestedSchemaPruning {
-
- // Prune the given output to make it consistent with `requiredSchema`.
- protected def getPrunedOutput(output: Seq[AttributeReference],
- requiredSchema: StructType): Seq[AttributeReference] = {
- // We need to replace the expression ids of the pruned relation output attributes
- // with the expression ids of the original relation output attributes so that
- // references to the original relation's output are not broken
- val outputIdMap = output.map(att => (att.name, att.exprId)).toMap
- DataTypeUtils.toAttributes(requiredSchema)
- .map {
- case att if outputIdMap.contains(att.name) =>
- att.withExprId(outputIdMap(att.name))
- case att => att
- }
- }
-
- override protected def apply0(plan: LogicalPlan): LogicalPlan =
- plan transformDown {
- case op@PhysicalOperation(projects, filters,
- // NOTE: This is modified to accommodate for Hudi's custom relations, given that original
- // [[NestedSchemaPruning]] rule is tightly coupled w/ [[HadoopFsRelation]]
- // TODO generalize to any file-based relation
- l@LogicalRelation(relation: HoodieBaseRelation, _, _, _, _))
- if relation.canPruneRelationSchema =>
-
- prunePhysicalColumns(l.output, projects, filters, relation.dataSchema,
- prunedDataSchema => {
- val prunedRelation =
- relation.updatePrunedDataSchema(prunedSchema = prunedDataSchema)
- buildPrunedRelation(l, prunedRelation)
- }).getOrElse(op)
- }
-}
diff --git a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/orc/Spark41OrcReader.scala b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/orc/Spark41OrcReader.scala
deleted file mode 100644
index 413196a236a4..000000000000
--- a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/orc/Spark41OrcReader.scala
+++ /dev/null
@@ -1,92 +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.execution.datasources.orc
-
-import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.Path
-import org.apache.spark.memory.MemoryMode
-import org.apache.spark.sql.catalyst.expressions.Attribute
-import org.apache.spark.sql.catalyst.types.DataTypeUtils.toAttributes
-import org.apache.spark.sql.execution.datasources.{FileFormat, PartitionedFile}
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.StructType
-
-class Spark41OrcReader(enableVectorizedReader: Boolean,
- memoryMode: MemoryMode,
- dataSchema: StructType,
- orcFilterPushDown: Boolean,
- isCaseSensitive: Boolean,
- capacity: Int) extends SparkOrcReaderBase(enableVectorizedReader, dataSchema, orcFilterPushDown, isCaseSensitive) {
-
- override def partitionedFileToPath(file: PartitionedFile): Path = {
- file.toPath
- }
-
- override def buildReader(): OrcColumnarBatchReader = {
- new OrcColumnarBatchReader(capacity, memoryMode)
- }
-
- override def structTypeToAttributes(schema: StructType): Seq[Attribute] = {
- toAttributes(schema)
- }
-}
-
-object Spark41OrcReader {
- /**
- * Get ORC file reader
- *
- * @param vectorized true if vectorized reading is not prohibited due to schema, reading mode, etc
- * @param sqlConf the [[SQLConf]] used for the read
- * @param options passed as a param to the file format
- * @param hadoopConf some configs will be set for the hadoopConf
- * @return ORC file reader
- */
- def build(vectorized: Boolean,
- sqlConf: SQLConf,
- options: Map[String, String],
- hadoopConf: Configuration,
- dataSchema: StructType): Spark41OrcReader = {
- //set hadoopconf
- hadoopConf.set(SQLConf.SESSION_LOCAL_TIMEZONE.key, sqlConf.sessionLocalTimeZone)
- hadoopConf.setBoolean(SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key, sqlConf.nestedSchemaPruningEnabled)
- hadoopConf.setBoolean(SQLConf.CASE_SENSITIVE.key, sqlConf.caseSensitiveAnalysis)
-
- val memoryMode = if (sqlConf.offHeapColumnVectorEnabled) {
- MemoryMode.OFF_HEAP
- } else {
- MemoryMode.ON_HEAP
- }
-
- val enableVectorizedReader = sqlConf.orcVectorizedReaderEnabled &&
- options.getOrElse(FileFormat.OPTION_RETURNING_BATCH,
- throw new IllegalArgumentException(
- "OPTION_RETURNING_BATCH should always be set for OrcFileFormat. " +
- "To workaround this issue, set spark.sql.orc.enableVectorizedReader=false."))
- .equals("true")
-
- new Spark41OrcReader(
- enableVectorizedReader = enableVectorizedReader && vectorized,
- memoryMode = memoryMode,
- isCaseSensitive = sqlConf.caseSensitiveAnalysis,
- capacity = sqlConf.orcVectorizedReaderBatchSize,
- orcFilterPushDown = sqlConf.orcFilterPushDown,
- dataSchema = dataSchema)
- }
-}
diff --git a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark41DataSourceUtils.scala b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark41DataSourceUtils.scala
deleted file mode 100644
index 9c4b28d2423a..000000000000
--- a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark41DataSourceUtils.scala
+++ /dev/null
@@ -1,76 +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.execution.datasources.parquet
-
-import org.apache.spark.sql.SPARK_VERSION_METADATA_KEY
-import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
-import org.apache.spark.util.Utils
-
-object Spark41DataSourceUtils {
-
- /**
- * NOTE: This method was copied from [[Spark32PlusDataSourceUtils]], and is required to maintain runtime
- * compatibility against Spark 3.5.0
- */
- // scalastyle:off
- def int96RebaseMode(lookupFileMeta: String => String,
- modeByConfig: String): LegacyBehaviorPolicy.Value = {
- if (Utils.isTesting && SQLConf.get.getConfString("spark.test.forceNoRebase", "") == "true") {
- return LegacyBehaviorPolicy.CORRECTED
- }
- // If there is no version, we return the mode specified by the config.
- Option(lookupFileMeta(SPARK_VERSION_METADATA_KEY)).map { version =>
- // Files written by Spark 3.0 and earlier follow the legacy hybrid calendar and we need to
- // rebase the INT96 timestamp values.
- // Files written by Spark 3.1 and latter may also need the rebase if they were written with
- // the "LEGACY" rebase mode.
- if (version < "3.1.0" || lookupFileMeta("org.apache.spark.legacyINT96") != null) {
- LegacyBehaviorPolicy.LEGACY
- } else {
- LegacyBehaviorPolicy.CORRECTED
- }
- }.getOrElse(LegacyBehaviorPolicy.withName(modeByConfig))
- }
- // scalastyle:on
-
- /**
- * NOTE: This method was copied from Spark 3.2.0, and is required to maintain runtime
- * compatibility against Spark 3.2.0
- */
- // scalastyle:off
- def datetimeRebaseMode(lookupFileMeta: String => String,
- modeByConfig: String): LegacyBehaviorPolicy.Value = {
- if (Utils.isTesting && SQLConf.get.getConfString("spark.test.forceNoRebase", "") == "true") {
- return LegacyBehaviorPolicy.CORRECTED
- }
- // If there is no version, we return the mode specified by the config.
- Option(lookupFileMeta(SPARK_VERSION_METADATA_KEY)).map { version =>
- // Files written by Spark 2.4 and earlier follow the legacy hybrid calendar and we need to
- // rebase the datetime values.
- // Files written by Spark 3.0 and latter may also need the rebase if they were written with
- // the "LEGACY" rebase mode.
- if (version < "3.0.0" || lookupFileMeta("org.apache.spark.legacyDateTime") != null) {
- LegacyBehaviorPolicy.LEGACY
- } else {
- LegacyBehaviorPolicy.CORRECTED
- }
- }.getOrElse(LegacyBehaviorPolicy.withName(modeByConfig))
- }
- // scalastyle:on
-
-}
diff --git a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/hudi/Spark41ResolveHudiAlterTableCommand.scala b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/hudi/Spark41ResolveHudiAlterTableCommand.scala
deleted file mode 100644
index 92cbc2803bf6..000000000000
--- a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/hudi/Spark41ResolveHudiAlterTableCommand.scala
+++ /dev/null
@@ -1,68 +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.hudi
-
-import org.apache.hudi.common.config.HoodieCommonConfig
-import org.apache.hudi.internal.schema.action.TableChange.ColumnChangeID
-
-import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.analysis.ResolvedTable
-import org.apache.spark.sql.catalyst.plans.logical._
-import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.hudi.catalog.HoodieInternalV2Table
-import org.apache.spark.sql.hudi.command.{AlterTableCommand => HudiAlterTableCommand}
-
-/**
- * Rule to mostly resolve, normalize and rewrite column names based on case sensitivity.
- * for alter table column commands.
- */
-class Spark41ResolveHudiAlterTableCommand(sparkSession: SparkSession) extends Rule[LogicalPlan] {
-
- def apply(plan: LogicalPlan): LogicalPlan = {
- if (ProvidesHoodieConfig.isSchemaEvolutionEnabled(sparkSession)) {
- plan.resolveOperatorsUp {
- case set@SetTableProperties(ResolvedHoodieV2TablePlan(t), _) if set.resolved =>
- HudiAlterTableCommand(t.v1Table, set.changes, ColumnChangeID.PROPERTY_CHANGE)
- case unSet@UnsetTableProperties(ResolvedHoodieV2TablePlan(t), _, _) if unSet.resolved =>
- HudiAlterTableCommand(t.v1Table, unSet.changes, ColumnChangeID.PROPERTY_CHANGE)
- case drop@DropColumns(ResolvedHoodieV2TablePlan(t), _, _) if drop.resolved =>
- HudiAlterTableCommand(t.v1Table, drop.changes, ColumnChangeID.DELETE)
- case add@AddColumns(ResolvedHoodieV2TablePlan(t), _) if add.resolved =>
- HudiAlterTableCommand(t.v1Table, add.changes, ColumnChangeID.ADD)
- case renameColumn@RenameColumn(ResolvedHoodieV2TablePlan(t), _, _) if renameColumn.resolved =>
- HudiAlterTableCommand(t.v1Table, renameColumn.changes, ColumnChangeID.UPDATE)
- case alter@AlterColumns(ResolvedHoodieV2TablePlan(t), _) if alter.resolved =>
- HudiAlterTableCommand(t.v1Table, alter.changes, ColumnChangeID.UPDATE)
- case replace@ReplaceColumns(ResolvedHoodieV2TablePlan(t), _) if replace.resolved =>
- HudiAlterTableCommand(t.v1Table, replace.changes, ColumnChangeID.REPLACE)
- }
- } else {
- plan
- }
- }
-
- object ResolvedHoodieV2TablePlan {
- def unapply(plan: LogicalPlan): Option[HoodieInternalV2Table] = {
- plan match {
- case ResolvedTable(_, _, v2Table: HoodieInternalV2Table, _) => Some(v2Table)
- case _ => None
- }
- }
- }
-}
-