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