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 265768229f3c3e0e49b66cef142dc4338e90e4f6 Author: Y Ethan Guo <[email protected]> AuthorDate: Fri Jul 10 06:32:00 2026 -0700 refactor(spark): consolidate the vendored 3.x Avro serde forks into hudi-spark3-common (#19168) * refactor(spark): consolidate the vendored 3.x Avro serde forks into hudi-spark3-common The vendored spark-avro AvroSerializer and AvroDeserializer are duplicated across hudi-spark3.3.x, hudi-spark3.4.x and hudi-spark3.5.x. The 3.3 and 3.4 copies are byte-identical; the 3.5 copies differ only in how they import LegacyBehaviorPolicy. Move a single copy of each into hudi-spark3-common, which every 3.x version module already depends on, and delete the six duplicates. The one cross-version snag is the LegacyBehaviorPolicy enum: it is nested in SQLConf (org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy) on Spark 3.3/3.4 but a top-level object (org.apache.spark.sql.internal.LegacyBehaviorPolicy) on Spark 3.5, so no single explicit import resolves on all three (verified by compiling against the 3.3.4/3.4.3/3.5.5 catalyst jars). The shared source imports both containers via wildcards; exactly one of them contributes the enum on any given version, so there is no ambiguity and the bodies stay byte-identical to the originals. The per-version HoodieSpark3_xAvro{Serializer, Deserializer} wrappers are unchanged, since the shared classes keep the same constructors they called. Consolidating also gives each file a unique repo path, so it re-enters the coverage denominator: the identically-pathed copies were dropped by report-path resolution and were invisible to Codecov. * refactor(spark): consolidate the vendored 4.x Avro serde forks into hudi-spark4-common The vendored spark-avro AvroSerializer and AvroDeserializer are duplicated across hudi-spark4.0.x, hudi-spark4.1.x and hudi-spark4.2.x. AvroSerializer: the 4.0/4.1/4.2 bodies are identical except for the private convenience constructor that reads AVRO_REBASE_MODE_IN_WRITE from SQLConf. On Spark 4.0 that ConfigEntry is typed as String (so the read wraps it in LegacyBehaviorPolicy.withName), while on 4.1+ it is already a LegacyBehaviorPolicy.Value; a single shared source cannot express both. Since the only callers are the per-version HoodieSpark4_xAvroSerializer wrappers, move that read into each wrapper (verbatim per version) and drop the convenience constructor, leaving one shared AvroSerializer in hudi-spark4-common that all three versions use. AvroDeserializer: 4.1 and 4.2 are identical apart from one comment word, so they collapse to a single shared copy in hudi-spark4-common. Spark 4.0 is kept separate: it pulls in Avro 1.12.0 (fast reader off) and lacks the ~71-line read-side java.time normalization that 4.1+ need for Avro 1.12.1, and that block does not parameterize cleanly. Because hudi-spark4.0.x depends on hudi-spark4-common, a same-named copy would collide on the classpath, so the 4.0-only copy is renamed to Spark40AvroDeserializer (body unchanged). Consolidating also gives each file a unique repo path, so it re-enters the coverage denominator: the identically-pathed copies were dropped by report-path resolution and were invisible to Codecov. * Revert the 4.x Avro serde consolidation (unsound: vendored fork must shadow spark-sql per-module) Reverts commit 5a8169f05bbe. On Spark 4.x the avro connector is merged into the spark-sql artifact, so org.apache.spark.sql.avro.{AvroDeserializer, AvroSerializer,AvroUtils} ship inside spark-sql_2.13 itself (verified in the 4.0.2/4.1.1/4.2.0-preview4 jars). spark-sql is a provided dependency of every 4.x version module, so Spark's own AvroDeserializer/AvroSerializer are always on the 4.x compile classpath. The vendored fork shares Spark's FQN. While it lived as source in each version module, same-module source shadowed the spark-sql class during that module's compile. Moving it into the hudi-spark4-common jar turned it into a peer dependency-jar class that competes with spark-sql's copy, and spark-sql wins: the HoodieSpark4_xAvro{Deserializer,Serializer} wrappers then resolve the wrong class (compile error in 4.2.x: the 3-arg (Schema, DataType, LegacyBehaviorPolicy.Value) call cannot bind to Spark's constructors). Even where it compiled, the wrapper would silently link Spark's serde on a data path. The 3.x consolidation is kept: spark-sql/spark-catalyst 3.x do not contain the avro serde classes (they live only in the separate spark-avro artifact, which is not a dependency of the 3.x modules), so the vendored fork in hudi-spark3-common is the only such class on the 3.x classpath and resolves correctly. This revert restores the six 4.x serde files to be byte-identical to apache/master. (cherry picked from commit 7b843ae2c35fa00993a2a1b102c9ad7e1e116267) --- .../apache/spark/sql/avro/AvroDeserializer.scala | 7 +- .../org/apache/spark/sql/avro/AvroSerializer.scala | 9 +- .../apache/spark/sql/avro/AvroDeserializer.scala | 531 --------------------- .../org/apache/spark/sql/avro/AvroSerializer.scala | 490 ------------------- .../apache/spark/sql/avro/AvroDeserializer.scala | 531 --------------------- .../org/apache/spark/sql/avro/AvroSerializer.scala | 489 ------------------- 6 files changed, 13 insertions(+), 2044 deletions(-) diff --git a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala similarity index 97% rename from hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala rename to hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala index b5eba6be24cd..2a9244508182 100644 --- a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala +++ b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala @@ -33,7 +33,12 @@ import org.apache.spark.sql.catalyst.expressions.{SpecificInternalRow, UnsafeArr import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, ArrayData, DateTimeUtils, GenericArrayData, RebaseDateTime} import org.apache.spark.sql.catalyst.util.DateTimeConstants.MILLIS_PER_DAY import org.apache.spark.sql.execution.datasources.DataSourceUtils -import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy +// LegacyBehaviorPolicy is nested in SQLConf on Spark 3.3/3.4 (org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy) +// but a top-level object on Spark 3.5 (org.apache.spark.sql.internal.LegacyBehaviorPolicy). Importing both +// containers via wildcards lets this single shared source resolve the enum on every 3.x version: exactly one of +// the two wildcards contributes LegacyBehaviorPolicy on any given version, so there is no ambiguity. +import org.apache.spark.sql.internal._ +import org.apache.spark.sql.internal.SQLConf._ import org.apache.spark.sql.types._ import org.apache.spark.unsafe.types.UTF8String diff --git a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala similarity index 97% rename from hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala rename to hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala index a1241b72e58b..a432eec0ffed 100644 --- a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala +++ b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala @@ -34,8 +34,13 @@ import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.expressions.{SpecializedGetters, SpecificInternalRow} import org.apache.spark.sql.catalyst.util.{DateTimeUtils, RebaseDateTime} import org.apache.spark.sql.execution.datasources.DataSourceUtils -import org.apache.spark.sql.internal.SQLConf -import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy +// LegacyBehaviorPolicy is nested in SQLConf on Spark 3.3/3.4 (org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy) +// but a top-level object on Spark 3.5 (org.apache.spark.sql.internal.LegacyBehaviorPolicy). Importing both +// containers via wildcards lets this single shared source resolve the enum on every 3.x version: exactly one of +// the two wildcards contributes LegacyBehaviorPolicy on any given version, so there is no ambiguity. The same two +// wildcards also keep the SQLConf object in scope for the SQLConf.get / SQLConf.AVRO_REBASE_MODE_IN_WRITE usages. +import org.apache.spark.sql.internal._ +import org.apache.spark.sql.internal.SQLConf._ import org.apache.spark.sql.types._ import java.nio.ByteBuffer diff --git a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala deleted file mode 100644 index b5eba6be24cd..000000000000 --- a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala +++ /dev/null @@ -1,531 +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.avro - -import org.apache.hudi.common.schema.HoodieSchema -import org.apache.hudi.common.schema.HoodieSchema.VectorLogicalType - -import org.apache.avro.{LogicalTypes, Schema, SchemaBuilder} -import org.apache.avro.Conversions.DecimalConversion -import org.apache.avro.LogicalTypes.{LocalTimestampMicros, LocalTimestampMillis, TimestampMicros, TimestampMillis} -import org.apache.avro.Schema.Type._ -import org.apache.avro.generic._ -import org.apache.avro.util.Utf8 -import org.apache.spark.sql.avro.AvroDeserializer.{createDateRebaseFuncInRead, createTimestampRebaseFuncInRead, RebaseSpec} -import org.apache.spark.sql.avro.AvroUtils.{toFieldStr, AvroMatchedField} -import org.apache.spark.sql.catalyst.{InternalRow, NoopFilters, StructFilters} -import org.apache.spark.sql.catalyst.expressions.{SpecificInternalRow, UnsafeArrayData} -import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, ArrayData, DateTimeUtils, GenericArrayData, RebaseDateTime} -import org.apache.spark.sql.catalyst.util.DateTimeConstants.MILLIS_PER_DAY -import org.apache.spark.sql.execution.datasources.DataSourceUtils -import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy -import org.apache.spark.sql.types._ -import org.apache.spark.unsafe.types.UTF8String - -import java.math.BigDecimal -import java.nio.ByteBuffer -import java.nio.ByteOrder -import java.util.TimeZone - -import scala.collection.JavaConverters._ - -/** - * A deserializer to deserialize data in avro format to data in catalyst format. - * - * NOTE: This code is borrowed from Spark 3.3.0 - * This code is borrowed, so that we can better control compatibility w/in Spark minor - * branches (3.2.x, 3.1.x, etc) - * - * PLEASE REFRAIN MAKING ANY CHANGES TO THIS CODE UNLESS ABSOLUTELY NECESSARY - */ -private[sql] class AvroDeserializer(rootAvroType: Schema, - rootCatalystType: DataType, - positionalFieldMatch: Boolean, - datetimeRebaseSpec: RebaseSpec, - filters: StructFilters) { - - def this(rootAvroType: Schema, - rootCatalystType: DataType, - datetimeRebaseMode: String) = { - this( - rootAvroType, - rootCatalystType, - positionalFieldMatch = false, - RebaseSpec(LegacyBehaviorPolicy.withName(datetimeRebaseMode)), - new NoopFilters) - } - - private lazy val decimalConversions = new DecimalConversion() - - private val dateRebaseFunc = createDateRebaseFuncInRead(datetimeRebaseSpec.mode, "Avro") - - private val timestampRebaseFunc = createTimestampRebaseFuncInRead(datetimeRebaseSpec, "Avro") - - private val converter: Any => Option[Any] = try { - rootCatalystType match { - // A shortcut for empty schema. - case st: StructType if st.isEmpty => - (_: Any) => Some(InternalRow.empty) - - case st: StructType => - val resultRow = new SpecificInternalRow(st.map(_.dataType)) - val fieldUpdater = new RowUpdater(resultRow) - val applyFilters = filters.skipRow(resultRow, _) - val writer = getRecordWriter(rootAvroType, st, Nil, Nil, applyFilters) - (data: Any) => { - val record = data.asInstanceOf[GenericRecord] - val skipRow = writer(fieldUpdater, record) - if (skipRow) None else Some(resultRow) - } - - case _ => - val tmpRow = new SpecificInternalRow(Seq(rootCatalystType)) - val fieldUpdater = new RowUpdater(tmpRow) - val writer = newWriter(rootAvroType, rootCatalystType, Nil, Nil) - (data: Any) => { - writer(fieldUpdater, 0, data) - Some(tmpRow.get(0, rootCatalystType)) - } - } - } catch { - case ise: IncompatibleSchemaException => throw new IncompatibleSchemaException( - s"Cannot convert Avro type $rootAvroType to SQL type ${rootCatalystType.sql}.", ise) - } - - def deserialize(data: Any): Option[Any] = converter(data) - - /** - * Creates a writer to write avro values to Catalyst values at the given ordinal with the given - * updater. - */ - private def newWriter(avroType: Schema, - catalystType: DataType, - avroPath: Seq[String], - catalystPath: Seq[String]): (CatalystDataUpdater, Int, Any) => Unit = { - val errorPrefix = s"Cannot convert Avro ${toFieldStr(avroPath)} to " + - s"SQL ${toFieldStr(catalystPath)} because " - val incompatibleMsg = errorPrefix + - s"schema is incompatible (avroType = $avroType, sqlType = ${catalystType.sql})" - - (avroType.getType, catalystType) match { - case (NULL, NullType) => (updater, ordinal, _) => - updater.setNullAt(ordinal) - - // TODO: we can avoid boxing if future version of avro provide primitive accessors. - case (BOOLEAN, BooleanType) => (updater, ordinal, value) => - updater.setBoolean(ordinal, value.asInstanceOf[Boolean]) - - case (INT, IntegerType) => (updater, ordinal, value) => - updater.setInt(ordinal, value.asInstanceOf[Int]) - - case (INT, DateType) => (updater, ordinal, value) => - updater.setInt(ordinal, dateRebaseFunc(value.asInstanceOf[Int])) - - case (LONG, LongType) => (updater, ordinal, value) => - updater.setLong(ordinal, value.asInstanceOf[Long]) - - case (LONG, TimestampType) => avroType.getLogicalType match { - // For backward compatibility, if the Avro type is Long and it is not logical type - // (the `null` case), the value is processed as timestamp type with millisecond precision. - case null | _: TimestampMillis => (updater, ordinal, value) => - val millis = value.asInstanceOf[Long] - val micros = DateTimeUtils.millisToMicros(millis) - updater.setLong(ordinal, timestampRebaseFunc(micros)) - case _: TimestampMicros => (updater, ordinal, value) => - val micros = value.asInstanceOf[Long] - updater.setLong(ordinal, timestampRebaseFunc(micros)) - case other => throw new IncompatibleSchemaException(errorPrefix + - s"Avro logical type $other cannot be converted to SQL type ${TimestampType.sql}.") - } - - case (LONG, TimestampNTZType) => avroType.getLogicalType match { - // To keep consistent with TimestampType, if the Avro type is Long and it is not - // logical type (the `null` case), the value is processed as TimestampNTZ - // with millisecond precision. - case null | _: LocalTimestampMillis => (updater, ordinal, value) => - val millis = value.asInstanceOf[Long] - val micros = DateTimeUtils.millisToMicros(millis) - updater.setLong(ordinal, micros) - case _: LocalTimestampMicros => (updater, ordinal, value) => - val micros = value.asInstanceOf[Long] - updater.setLong(ordinal, micros) - case other => throw new IncompatibleSchemaException(errorPrefix + - s"Avro logical type $other cannot be converted to SQL type ${TimestampNTZType.sql}.") - } - - // Handle VECTOR logical type (FLOAT, DOUBLE, INT8) - case (FIXED, ArrayType(elementType, false)) => avroType.getLogicalType match { - case vectorLogicalType: VectorLogicalType => - val dimension = vectorLogicalType.getDimension - val vecElementType = HoodieSchema.Vector.VectorElementType.fromString(vectorLogicalType.getElementType) - val elementSize = vecElementType.getElementSize - (updater, ordinal, value) => { - val bytes = value.asInstanceOf[GenericData.Fixed].bytes() - val expectedSize = Math.multiplyExact(dimension, elementSize) - if (bytes.length != expectedSize) { - throw new IncompatibleSchemaException( - s"VECTOR byte size mismatch: expected=$expectedSize, actual=${bytes.length}") - } - elementType match { - case FloatType => - val buffer = ByteBuffer.wrap(bytes).order(VectorLogicalType.VECTOR_BYTE_ORDER) - val floats = new Array[Float](dimension) - var i = 0; while (i < dimension) { floats(i) = buffer.getFloat(); i += 1 } - updater.set(ordinal, ArrayData.toArrayData(floats)) - case DoubleType => - val buffer = ByteBuffer.wrap(bytes).order(VectorLogicalType.VECTOR_BYTE_ORDER) - val doubles = new Array[Double](dimension) - var i = 0; while (i < dimension) { doubles(i) = buffer.getDouble(); i += 1 } - updater.set(ordinal, ArrayData.toArrayData(doubles)) - case ByteType => - updater.set(ordinal, ArrayData.toArrayData(bytes.clone())) - } - } - case _ => throw new IncompatibleSchemaException(incompatibleMsg) - } - - // Before we upgrade Avro to 1.8 for logical type support, spark-avro converts Long to Date. - // For backward compatibility, we still keep this conversion. - case (LONG, DateType) => (updater, ordinal, value) => - updater.setInt(ordinal, (value.asInstanceOf[Long] / MILLIS_PER_DAY).toInt) - - case (FLOAT, FloatType) => (updater, ordinal, value) => - updater.setFloat(ordinal, value.asInstanceOf[Float]) - - case (DOUBLE, DoubleType) => (updater, ordinal, value) => - updater.setDouble(ordinal, value.asInstanceOf[Double]) - - case (STRING, StringType) => (updater, ordinal, value) => - val str = value match { - case s: String => UTF8String.fromString(s) - case s: Utf8 => - val bytes = new Array[Byte](s.getByteLength) - System.arraycopy(s.getBytes, 0, bytes, 0, s.getByteLength) - UTF8String.fromBytes(bytes) - case s: GenericData.EnumSymbol => UTF8String.fromString(s.toString) - } - updater.set(ordinal, str) - - case (ENUM, StringType) => (updater, ordinal, value) => - updater.set(ordinal, UTF8String.fromString(value.toString)) - - case (FIXED, BinaryType) => (updater, ordinal, value) => - updater.set(ordinal, value.asInstanceOf[GenericFixed].bytes().clone()) - - case (BYTES, BinaryType) => (updater, ordinal, value) => - val bytes = value match { - case b: ByteBuffer => - val bytes = new Array[Byte](b.remaining) - b.get(bytes) - // Do not forget to reset the position - b.rewind() - bytes - case b: Array[Byte] => b - case other => - throw new RuntimeException(errorPrefix + s"$other is not a valid avro binary.") - } - updater.set(ordinal, bytes) - - case (FIXED, _: DecimalType) => (updater, ordinal, value) => - val d = avroType.getLogicalType.asInstanceOf[LogicalTypes.Decimal] - val bigDecimal = decimalConversions.fromFixed(value.asInstanceOf[GenericFixed], avroType, d) - val decimal = createDecimal(bigDecimal, d.getPrecision, d.getScale) - updater.setDecimal(ordinal, decimal) - - case (BYTES, _: DecimalType) => (updater, ordinal, value) => - val d = avroType.getLogicalType.asInstanceOf[LogicalTypes.Decimal] - val bigDecimal = decimalConversions.fromBytes(value.asInstanceOf[ByteBuffer], avroType, d) - val decimal = createDecimal(bigDecimal, d.getPrecision, d.getScale) - updater.setDecimal(ordinal, decimal) - - case (RECORD, st: StructType) => - // Avro datasource doesn't accept filters with nested attributes. See SPARK-32328. - // We can always return `false` from `applyFilters` for nested records. - val writeRecord = - getRecordWriter(avroType, st, avroPath, catalystPath, applyFilters = _ => false) - (updater, ordinal, value) => - val row = new SpecificInternalRow(st) - writeRecord(new RowUpdater(row), value.asInstanceOf[GenericRecord]) - updater.set(ordinal, row) - - case (ARRAY, ArrayType(elementType, containsNull)) => - val avroElementPath = avroPath :+ "element" - val elementWriter = newWriter(avroType.getElementType, elementType, - avroElementPath, catalystPath :+ "element") - (updater, ordinal, value) => - val collection = value.asInstanceOf[java.util.Collection[Any]] - val result = createArrayData(elementType, collection.size()) - val elementUpdater = new ArrayDataUpdater(result) - - var i = 0 - val iter = collection.iterator() - while (iter.hasNext) { - val element = iter.next() - if (element == null) { - if (!containsNull) { - throw new RuntimeException( - s"Array value at path ${toFieldStr(avroElementPath)} is not allowed to be null") - } else { - elementUpdater.setNullAt(i) - } - } else { - elementWriter(elementUpdater, i, element) - } - i += 1 - } - - updater.set(ordinal, result) - - case (MAP, MapType(keyType, valueType, valueContainsNull)) if keyType == StringType => - val keyWriter = newWriter(SchemaBuilder.builder().stringType(), StringType, - avroPath :+ "key", catalystPath :+ "key") - val valueWriter = newWriter(avroType.getValueType, valueType, - avroPath :+ "value", catalystPath :+ "value") - (updater, ordinal, value) => - val map = value.asInstanceOf[java.util.Map[AnyRef, AnyRef]] - val keyArray = createArrayData(keyType, map.size()) - val keyUpdater = new ArrayDataUpdater(keyArray) - val valueArray = createArrayData(valueType, map.size()) - val valueUpdater = new ArrayDataUpdater(valueArray) - val iter = map.entrySet().iterator() - var i = 0 - while (iter.hasNext) { - val entry = iter.next() - assert(entry.getKey != null) - keyWriter(keyUpdater, i, entry.getKey) - if (entry.getValue == null) { - if (!valueContainsNull) { - throw new RuntimeException( - s"Map value at path ${toFieldStr(avroPath :+ "value")} is not allowed to be null") - } else { - valueUpdater.setNullAt(i) - } - } else { - valueWriter(valueUpdater, i, entry.getValue) - } - i += 1 - } - - // The Avro map will never have null or duplicated map keys, it's safe to create a - // ArrayBasedMapData directly here. - updater.set(ordinal, new ArrayBasedMapData(keyArray, valueArray)) - - case (UNION, _) => - val allTypes = avroType.getTypes.asScala - val nonNullTypes = allTypes.filter(_.getType != NULL) - val nonNullAvroType = Schema.createUnion(nonNullTypes.asJava) - if (nonNullTypes.nonEmpty) { - if (nonNullTypes.length == 1) { - newWriter(nonNullTypes.head, catalystType, avroPath, catalystPath) - } else { - nonNullTypes.map(_.getType).toSeq match { - case Seq(a, b) if Set(a, b) == Set(INT, LONG) && catalystType == LongType => - (updater, ordinal, value) => value match { - case null => updater.setNullAt(ordinal) - case l: java.lang.Long => updater.setLong(ordinal, l) - case i: java.lang.Integer => updater.setLong(ordinal, i.longValue()) - } - - case Seq(a, b) if Set(a, b) == Set(FLOAT, DOUBLE) && catalystType == DoubleType => - (updater, ordinal, value) => value match { - case null => updater.setNullAt(ordinal) - case d: java.lang.Double => updater.setDouble(ordinal, d) - case f: java.lang.Float => updater.setDouble(ordinal, f.doubleValue()) - } - - case _ => - catalystType match { - case st: StructType if st.length == nonNullTypes.size => - val fieldWriters = nonNullTypes.zip(st.fields).map { - case (schema, field) => - newWriter(schema, field.dataType, avroPath, catalystPath :+ field.name) - }.toArray - (updater, ordinal, value) => { - val row = new SpecificInternalRow(st) - val fieldUpdater = new RowUpdater(row) - val i = GenericData.get().resolveUnion(nonNullAvroType, value) - fieldWriters(i)(fieldUpdater, i, value) - updater.set(ordinal, row) - } - - case _ => throw new IncompatibleSchemaException(incompatibleMsg) - } - } - } - } else { - (updater, ordinal, _) => updater.setNullAt(ordinal) - } - - case (INT, _: YearMonthIntervalType) => (updater, ordinal, value) => - updater.setInt(ordinal, value.asInstanceOf[Int]) - - case (LONG, _: DayTimeIntervalType) => (updater, ordinal, value) => - updater.setLong(ordinal, value.asInstanceOf[Long]) - - case _ => throw new IncompatibleSchemaException(incompatibleMsg) - } - } - - // TODO: move the following method in Decimal object on creating Decimal from BigDecimal? - private def createDecimal(decimal: BigDecimal, precision: Int, scale: Int): Decimal = { - if (precision <= Decimal.MAX_LONG_DIGITS) { - // Constructs a `Decimal` with an unscaled `Long` value if possible. - Decimal(decimal.unscaledValue().longValue(), precision, scale) - } else { - // Otherwise, resorts to an unscaled `BigInteger` instead. - Decimal(decimal, precision, scale) - } - } - - private def getRecordWriter( - avroType: Schema, - catalystType: StructType, - avroPath: Seq[String], - catalystPath: Seq[String], - applyFilters: Int => Boolean): (CatalystDataUpdater, GenericRecord) => Boolean = { - - val avroSchemaHelper = new AvroUtils.AvroSchemaHelper( - avroType, catalystType, avroPath, catalystPath, positionalFieldMatch) - - avroSchemaHelper.validateNoExtraCatalystFields(ignoreNullable = true) - // no need to validateNoExtraAvroFields since extra Avro fields are ignored - - val (validFieldIndexes, fieldWriters) = avroSchemaHelper.matchedFields.map { - case AvroMatchedField(catalystField, ordinal, avroField) => - val baseWriter = newWriter(avroField.schema(), catalystField.dataType, - avroPath :+ avroField.name, catalystPath :+ catalystField.name) - val fieldWriter = (fieldUpdater: CatalystDataUpdater, value: Any) => { - if (value == null) { - fieldUpdater.setNullAt(ordinal) - } else { - baseWriter(fieldUpdater, ordinal, value) - } - } - (avroField.pos(), fieldWriter) - }.toArray.unzip - - (fieldUpdater, record) => { - var i = 0 - var skipRow = false - while (i < validFieldIndexes.length && !skipRow) { - fieldWriters(i)(fieldUpdater, record.get(validFieldIndexes(i))) - skipRow = applyFilters(i) - i += 1 - } - skipRow - } - } - - private def createArrayData(elementType: DataType, length: Int): ArrayData = elementType match { - case BooleanType => UnsafeArrayData.fromPrimitiveArray(new Array[Boolean](length)) - case ByteType => UnsafeArrayData.fromPrimitiveArray(new Array[Byte](length)) - case ShortType => UnsafeArrayData.fromPrimitiveArray(new Array[Short](length)) - case IntegerType => UnsafeArrayData.fromPrimitiveArray(new Array[Int](length)) - case LongType => UnsafeArrayData.fromPrimitiveArray(new Array[Long](length)) - case FloatType => UnsafeArrayData.fromPrimitiveArray(new Array[Float](length)) - case DoubleType => UnsafeArrayData.fromPrimitiveArray(new Array[Double](length)) - case _ => new GenericArrayData(new Array[Any](length)) - } - - /** - * A base interface for updating values inside catalyst data structure like `InternalRow` and - * `ArrayData`. - */ - sealed trait CatalystDataUpdater { - def set(ordinal: Int, value: Any): Unit - - def setNullAt(ordinal: Int): Unit = set(ordinal, null) - def setBoolean(ordinal: Int, value: Boolean): Unit = set(ordinal, value) - def setByte(ordinal: Int, value: Byte): Unit = set(ordinal, value) - def setShort(ordinal: Int, value: Short): Unit = set(ordinal, value) - def setInt(ordinal: Int, value: Int): Unit = set(ordinal, value) - def setLong(ordinal: Int, value: Long): Unit = set(ordinal, value) - def setDouble(ordinal: Int, value: Double): Unit = set(ordinal, value) - def setFloat(ordinal: Int, value: Float): Unit = set(ordinal, value) - def setDecimal(ordinal: Int, value: Decimal): Unit = set(ordinal, value) - } - - final class RowUpdater(row: InternalRow) extends CatalystDataUpdater { - override def set(ordinal: Int, value: Any): Unit = row.update(ordinal, value) - - override def setNullAt(ordinal: Int): Unit = row.setNullAt(ordinal) - override def setBoolean(ordinal: Int, value: Boolean): Unit = row.setBoolean(ordinal, value) - override def setByte(ordinal: Int, value: Byte): Unit = row.setByte(ordinal, value) - override def setShort(ordinal: Int, value: Short): Unit = row.setShort(ordinal, value) - override def setInt(ordinal: Int, value: Int): Unit = row.setInt(ordinal, value) - override def setLong(ordinal: Int, value: Long): Unit = row.setLong(ordinal, value) - override def setDouble(ordinal: Int, value: Double): Unit = row.setDouble(ordinal, value) - override def setFloat(ordinal: Int, value: Float): Unit = row.setFloat(ordinal, value) - override def setDecimal(ordinal: Int, value: Decimal): Unit = - row.setDecimal(ordinal, value, value.precision) - } - - final class ArrayDataUpdater(array: ArrayData) extends CatalystDataUpdater { - override def set(ordinal: Int, value: Any): Unit = array.update(ordinal, value) - - override def setNullAt(ordinal: Int): Unit = array.setNullAt(ordinal) - override def setBoolean(ordinal: Int, value: Boolean): Unit = array.setBoolean(ordinal, value) - override def setByte(ordinal: Int, value: Byte): Unit = array.setByte(ordinal, value) - override def setShort(ordinal: Int, value: Short): Unit = array.setShort(ordinal, value) - override def setInt(ordinal: Int, value: Int): Unit = array.setInt(ordinal, value) - override def setLong(ordinal: Int, value: Long): Unit = array.setLong(ordinal, value) - override def setDouble(ordinal: Int, value: Double): Unit = array.setDouble(ordinal, value) - override def setFloat(ordinal: Int, value: Float): Unit = array.setFloat(ordinal, value) - override def setDecimal(ordinal: Int, value: Decimal): Unit = array.update(ordinal, value) - } -} - -object AvroDeserializer { - - // NOTE: Following methods have been renamed in Spark 3.2.1 [1] making [[AvroDeserializer]] implementation - // (which relies on it) be only compatible with the exact same version of [[DataSourceUtils]]. - // To make sure this implementation is compatible w/ all Spark versions w/in Spark 3.2.x branch, - // we're preemptively cloned those methods to make sure Hudi is compatible w/ Spark 3.2.0 as well as - // w/ Spark >= 3.2.1 - // - // [1] https://github.com/apache/spark/pull/34978 - - // Specification of rebase operation including `mode` and the time zone in which it is performed - case class RebaseSpec(mode: LegacyBehaviorPolicy.Value, originTimeZone: Option[String] = None) { - // Use the default JVM time zone for backward compatibility - def timeZone: String = originTimeZone.getOrElse(TimeZone.getDefault.getID) - } - - def createDateRebaseFuncInRead(rebaseMode: LegacyBehaviorPolicy.Value, - format: String): Int => Int = rebaseMode match { - case LegacyBehaviorPolicy.EXCEPTION => days: Int => - if (days < RebaseDateTime.lastSwitchJulianDay) { - throw DataSourceUtils.newRebaseExceptionInRead(format) - } - days - case LegacyBehaviorPolicy.LEGACY => RebaseDateTime.rebaseJulianToGregorianDays - case LegacyBehaviorPolicy.CORRECTED => identity[Int] - } - - def createTimestampRebaseFuncInRead(rebaseSpec: RebaseSpec, - format: String): Long => Long = rebaseSpec.mode match { - case LegacyBehaviorPolicy.EXCEPTION => micros: Long => - if (micros < RebaseDateTime.lastSwitchJulianTs) { - throw DataSourceUtils.newRebaseExceptionInRead(format) - } - micros - case LegacyBehaviorPolicy.LEGACY => micros: Long => - RebaseDateTime.rebaseJulianToGregorianMicros(TimeZone.getTimeZone(rebaseSpec.timeZone), micros) - case LegacyBehaviorPolicy.CORRECTED => identity[Long] - } -} diff --git a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala deleted file mode 100644 index a1241b72e58b..000000000000 --- a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala +++ /dev/null @@ -1,490 +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.avro - -import org.apache.hudi.common.schema.HoodieSchema -import org.apache.hudi.common.schema.HoodieSchema.VectorLogicalType - -import org.apache.avro.{LogicalTypes, Schema} -import org.apache.avro.Conversions.DecimalConversion -import org.apache.avro.LogicalTypes.{LocalTimestampMicros, LocalTimestampMillis, TimestampMicros, TimestampMillis} -import org.apache.avro.Schema.Type -import org.apache.avro.Schema.Type._ -import org.apache.avro.generic.GenericData.{EnumSymbol, Fixed, Record} -import org.apache.avro.util.Utf8 -import org.apache.spark.internal.Logging -import org.apache.spark.sql.avro.AvroSerializer.{createDateRebaseFuncInWrite, createTimestampRebaseFuncInWrite} -import org.apache.spark.sql.avro.AvroUtils.{toFieldStr, AvroMatchedField} -import org.apache.spark.sql.catalyst.InternalRow -import org.apache.spark.sql.catalyst.expressions.{SpecializedGetters, SpecificInternalRow} -import org.apache.spark.sql.catalyst.util.{DateTimeUtils, RebaseDateTime} -import org.apache.spark.sql.execution.datasources.DataSourceUtils -import org.apache.spark.sql.internal.SQLConf -import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy -import org.apache.spark.sql.types._ - -import java.nio.ByteBuffer -import java.nio.ByteOrder -import java.util.TimeZone - -import scala.collection.JavaConverters._ - -/** - * A serializer to serialize data in catalyst format to data in avro format. - * - * NOTE: This code is borrowed from Spark 3.3.0 - * This code is borrowed, so that we can better control compatibility w/in Spark minor - * branches (3.2.x, 3.1.x, etc) - * - * NOTE: THIS IMPLEMENTATION HAS BEEN MODIFIED FROM ITS ORIGINAL VERSION WITH THE MODIFICATION - * BEING EXPLICITLY ANNOTATED INLINE. PLEASE MAKE SURE TO UNDERSTAND PROPERLY ALL THE - * MODIFICATIONS. - * - * PLEASE REFRAIN MAKING ANY CHANGES TO THIS CODE UNLESS ABSOLUTELY NECESSARY - */ -private[sql] class AvroSerializer(rootCatalystType: DataType, - rootAvroType: Schema, - nullable: Boolean, - positionalFieldMatch: Boolean, - datetimeRebaseMode: LegacyBehaviorPolicy.Value) extends Logging { - - def this(rootCatalystType: DataType, rootAvroType: Schema, nullable: Boolean) = { - this(rootCatalystType, rootAvroType, nullable, positionalFieldMatch = false, - LegacyBehaviorPolicy.withName(SQLConf.get.getConf(SQLConf.AVRO_REBASE_MODE_IN_WRITE, - LegacyBehaviorPolicy.CORRECTED.toString))) - } - - def serialize(catalystData: Any): Any = { - converter.apply(catalystData) - } - - private val dateRebaseFunc = createDateRebaseFuncInWrite( - datetimeRebaseMode, "Avro") - - private val timestampRebaseFunc = createTimestampRebaseFuncInWrite( - datetimeRebaseMode, "Avro") - - private val converter: Any => Any = { - val actualAvroType = resolveNullableType(rootAvroType, nullable) - val baseConverter = try { - rootCatalystType match { - case st: StructType => - newStructConverter(st, actualAvroType, Nil, Nil).asInstanceOf[Any => Any] - case _ => - val tmpRow = new SpecificInternalRow(Seq(rootCatalystType)) - val converter = newConverter(rootCatalystType, actualAvroType, Nil, Nil) - (data: Any) => - tmpRow.update(0, data) - converter.apply(tmpRow, 0) - } - } catch { - case ise: IncompatibleSchemaException => throw new IncompatibleSchemaException( - s"Cannot convert SQL type ${rootCatalystType.sql} to Avro type $rootAvroType.", ise) - } - if (nullable) { - (data: Any) => - if (data == null) { - null - } else { - baseConverter.apply(data) - } - } else { - baseConverter - } - } - - private type Converter = (SpecializedGetters, Int) => Any - - private lazy val decimalConversions = new DecimalConversion() - - private def newConverter(catalystType: DataType, - avroType: Schema, - catalystPath: Seq[String], - avroPath: Seq[String]): Converter = { - val errorPrefix = s"Cannot convert SQL ${toFieldStr(catalystPath)} " + - s"to Avro ${toFieldStr(avroPath)} because " - (catalystType, avroType.getType) match { - case (NullType, NULL) => - (getter, ordinal) => null - case (BooleanType, BOOLEAN) => - (getter, ordinal) => getter.getBoolean(ordinal) - case (ByteType, INT) => - (getter, ordinal) => getter.getByte(ordinal).toInt - case (ShortType, INT) => - (getter, ordinal) => getter.getShort(ordinal).toInt - case (IntegerType, INT) => - (getter, ordinal) => getter.getInt(ordinal) - case (LongType, LONG) => - (getter, ordinal) => getter.getLong(ordinal) - case (FloatType, FLOAT) => - (getter, ordinal) => getter.getFloat(ordinal) - case (DoubleType, DOUBLE) => - (getter, ordinal) => getter.getDouble(ordinal) - case (d: DecimalType, FIXED) - if avroType.getLogicalType == LogicalTypes.decimal(d.precision, d.scale) => - (getter, ordinal) => - val decimal = getter.getDecimal(ordinal, d.precision, d.scale) - decimalConversions.toFixed(decimal.toJavaBigDecimal, avroType, - LogicalTypes.decimal(d.precision, d.scale)) - - case (d: DecimalType, BYTES) - if avroType.getLogicalType == LogicalTypes.decimal(d.precision, d.scale) => - (getter, ordinal) => - val decimal = getter.getDecimal(ordinal, d.precision, d.scale) - decimalConversions.toBytes(decimal.toJavaBigDecimal, avroType, - LogicalTypes.decimal(d.precision, d.scale)) - - // Handle VECTOR logical type (FLOAT, DOUBLE, INT8) - case (ArrayType(elementType, false), FIXED) => avroType.getLogicalType match { - case vectorLogicalType: VectorLogicalType => - val dimension = vectorLogicalType.getDimension - val vecElementType = HoodieSchema.Vector.VectorElementType.fromString(vectorLogicalType.getElementType) - val bufferSize = Math.multiplyExact(dimension, vecElementType.getElementSize) - (getter, ordinal) => { - val arrayData = getter.getArray(ordinal) - if (arrayData.numElements() != dimension) { - throw new IncompatibleSchemaException( - s"VECTOR dimension mismatch at ${toFieldStr(catalystPath)}: " + - s"expected=$dimension, actual=${arrayData.numElements()}") - } - elementType match { - case FloatType => - val buffer = ByteBuffer.allocate(bufferSize).order(VectorLogicalType.VECTOR_BYTE_ORDER) - var i = 0; while (i < dimension) { buffer.putFloat(arrayData.getFloat(i)); i += 1 } - new Fixed(avroType, buffer.array()) - case DoubleType => - val buffer = ByteBuffer.allocate(bufferSize).order(VectorLogicalType.VECTOR_BYTE_ORDER) - var i = 0; while (i < dimension) { buffer.putDouble(arrayData.getDouble(i)); i += 1 } - new Fixed(avroType, buffer.array()) - case ByteType => - val bytes = new Array[Byte](dimension) - var i = 0; while (i < dimension) { bytes(i) = arrayData.getByte(i); i += 1 } - new Fixed(avroType, bytes) - case _ => throw new IncompatibleSchemaException(errorPrefix + - s"schema is incompatible (sqlType = ${catalystType.sql}, avroType = $avroType)") - } - } - case _ => throw new IncompatibleSchemaException(errorPrefix + - s"schema is incompatible (sqlType = ${catalystType.sql}, avroType = $avroType)") - } - - case (StringType, ENUM) => - val enumSymbols: Set[String] = avroType.getEnumSymbols.asScala.toSet - (getter, ordinal) => - val data = getter.getUTF8String(ordinal).toString - if (!enumSymbols.contains(data)) { - throw new IncompatibleSchemaException(errorPrefix + - s""""$data" cannot be written since it's not defined in enum """ + - enumSymbols.mkString("\"", "\", \"", "\"")) - } - new EnumSymbol(avroType, data) - - case (StringType, STRING) => - (getter, ordinal) => new Utf8(getter.getUTF8String(ordinal).getBytes) - - case (BinaryType, FIXED) => - val size = avroType.getFixedSize - (getter, ordinal) => - val data: Array[Byte] = getter.getBinary(ordinal) - if (data.length != size) { - def len2str(len: Int): String = s"$len ${if (len > 1) "bytes" else "byte"}" - - throw new IncompatibleSchemaException(errorPrefix + len2str(data.length) + - " of binary data cannot be written into FIXED type with size of " + len2str(size)) - } - new Fixed(avroType, data) - - case (BinaryType, BYTES) => - (getter, ordinal) => ByteBuffer.wrap(getter.getBinary(ordinal)) - - case (DateType, INT) => - (getter, ordinal) => dateRebaseFunc(getter.getInt(ordinal)) - - case (TimestampType, LONG) => avroType.getLogicalType match { - // For backward compatibility, if the Avro type is Long and it is not logical type - // (the `null` case), output the timestamp value as with millisecond precision. - case null | _: TimestampMillis => (getter, ordinal) => - DateTimeUtils.microsToMillis(timestampRebaseFunc(getter.getLong(ordinal))) - case _: TimestampMicros => (getter, ordinal) => - timestampRebaseFunc(getter.getLong(ordinal)) - case other => throw new IncompatibleSchemaException(errorPrefix + - s"SQL type ${TimestampType.sql} cannot be converted to Avro logical type $other") - } - - case (TimestampNTZType, LONG) => avroType.getLogicalType match { - // To keep consistent with TimestampType, if the Avro type is Long and it is not - // logical type (the `null` case), output the TimestampNTZ as long value - // in millisecond precision. - case null | _: LocalTimestampMillis => (getter, ordinal) => - DateTimeUtils.microsToMillis(getter.getLong(ordinal)) - case _: LocalTimestampMicros => (getter, ordinal) => - getter.getLong(ordinal) - case other => throw new IncompatibleSchemaException(errorPrefix + - s"SQL type ${TimestampNTZType.sql} cannot be converted to Avro logical type $other") - } - - case (ArrayType(et, containsNull), ARRAY) => - val elementConverter = newConverter( - et, resolveNullableType(avroType.getElementType, containsNull), - catalystPath :+ "element", avroPath :+ "element") - (getter, ordinal) => { - val arrayData = getter.getArray(ordinal) - val len = arrayData.numElements() - val result = new Array[Any](len) - var i = 0 - while (i < len) { - if (containsNull && arrayData.isNullAt(i)) { - result(i) = null - } else { - result(i) = elementConverter(arrayData, i) - } - i += 1 - } - // avro writer is expecting a Java Collection, so we convert it into - // `ArrayList` backed by the specified array without data copying. - java.util.Arrays.asList(result: _*) - } - - case (st: StructType, RECORD) => - val structConverter = newStructConverter(st, avroType, catalystPath, avroPath) - val numFields = st.length - (getter, ordinal) => structConverter(getter.getStruct(ordinal, numFields)) - - //////////////////////////////////////////////////////////////////////////////////////////// - // Following section is amended to the original (Spark's) implementation - // >>> BEGINS - //////////////////////////////////////////////////////////////////////////////////////////// - - case (st: StructType, UNION) => - val unionConverter = newUnionConverter(st, avroType, catalystPath, avroPath) - val numFields = st.length - (getter, ordinal) => unionConverter(getter.getStruct(ordinal, numFields)) - - //////////////////////////////////////////////////////////////////////////////////////////// - // <<< ENDS - //////////////////////////////////////////////////////////////////////////////////////////// - - case (MapType(kt, vt, valueContainsNull), MAP) if kt == StringType => - val valueConverter = newConverter( - vt, resolveNullableType(avroType.getValueType, valueContainsNull), - catalystPath :+ "value", avroPath :+ "value") - (getter, ordinal) => - val mapData = getter.getMap(ordinal) - val len = mapData.numElements() - val result = new java.util.HashMap[String, Any](len) - val keyArray = mapData.keyArray() - val valueArray = mapData.valueArray() - var i = 0 - while (i < len) { - val key = keyArray.getUTF8String(i).toString - if (valueContainsNull && valueArray.isNullAt(i)) { - result.put(key, null) - } else { - result.put(key, valueConverter(valueArray, i)) - } - i += 1 - } - result - - case (_: YearMonthIntervalType, INT) => - (getter, ordinal) => getter.getInt(ordinal) - - case (_: DayTimeIntervalType, LONG) => - (getter, ordinal) => getter.getLong(ordinal) - - case _ => - throw new IncompatibleSchemaException(errorPrefix + - s"schema is incompatible (sqlType = ${catalystType.sql}, avroType = $avroType)") - } - } - - private def newStructConverter(catalystStruct: StructType, - avroStruct: Schema, - catalystPath: Seq[String], - avroPath: Seq[String]): InternalRow => Record = { - - val avroSchemaHelper = new AvroUtils.AvroSchemaHelper( - avroStruct, catalystStruct, avroPath, catalystPath, positionalFieldMatch) - - avroSchemaHelper.validateNoExtraCatalystFields(ignoreNullable = false) - avroSchemaHelper.validateNoExtraRequiredAvroFields() - - val (avroIndices, fieldConverters) = avroSchemaHelper.matchedFields.map { - case AvroMatchedField(catalystField, _, avroField) => - val converter = newConverter(catalystField.dataType, - resolveNullableType(avroField.schema(), catalystField.nullable), - catalystPath :+ catalystField.name, avroPath :+ avroField.name) - (avroField.pos(), converter) - }.toArray.unzip - - val numFields = catalystStruct.length - row: InternalRow => - val result = new Record(avroStruct) - var i = 0 - while (i < numFields) { - if (row.isNullAt(i)) { - result.put(avroIndices(i), null) - } else { - result.put(avroIndices(i), fieldConverters(i).apply(row, i)) - } - i += 1 - } - result - } - - //////////////////////////////////////////////////////////////////////////////////////////// - // Following section is amended to the original (Spark's) implementation - // >>> BEGINS - //////////////////////////////////////////////////////////////////////////////////////////// - - private def newUnionConverter(catalystStruct: StructType, - avroUnion: Schema, - catalystPath: Seq[String], - avroPath: Seq[String]): InternalRow => Any = { - if (avroUnion.getType != UNION || !canMapUnion(catalystStruct, avroUnion)) { - throw new IncompatibleSchemaException(s"Cannot convert Catalyst type $catalystStruct to " + - s"Avro type $avroUnion.") - } - val nullable = avroUnion.getTypes.size() > 0 && avroUnion.getTypes.get(0).getType == Type.NULL - val avroInnerTypes = if (nullable) { - avroUnion.getTypes.asScala.tail - } else { - avroUnion.getTypes.asScala - } - val fieldConverters = catalystStruct.zip(avroInnerTypes).map { - case (f1, f2) => newConverter(f1.dataType, f2, catalystPath, avroPath) - } - val numFields = catalystStruct.length - (row: InternalRow) => - var i = 0 - var result: Any = null - while (i < numFields) { - if (!row.isNullAt(i)) { - if (result != null) { - throw new IncompatibleSchemaException(s"Cannot convert Catalyst record $catalystStruct to " + - s"Avro union $avroUnion. Record has more than one optional values set") - } - result = fieldConverters(i).apply(row, i) - } - i += 1 - } - if (!nullable && result == null) { - throw new IncompatibleSchemaException(s"Cannot convert Catalyst record $catalystStruct to " + - s"Avro union $avroUnion. Record has no values set, while should have exactly one") - } - result - } - - private def canMapUnion(catalystStruct: StructType, avroStruct: Schema): Boolean = { - (avroStruct.getTypes.size() > 0 && - avroStruct.getTypes.get(0).getType == Type.NULL && - avroStruct.getTypes.size() - 1 == catalystStruct.length) || avroStruct.getTypes.size() == catalystStruct.length - } - - //////////////////////////////////////////////////////////////////////////////////////////// - // <<< ENDS - //////////////////////////////////////////////////////////////////////////////////////////// - - - /** - * Resolve a possibly nullable Avro Type. - * - * An Avro type is nullable when it is a [[UNION]] of two types: one null type and another - * non-null type. This method will check the nullability of the input Avro type and return the - * non-null type within when it is nullable. Otherwise it will return the input Avro type - * unchanged. It will throw an [[UnsupportedAvroTypeException]] when the input Avro type is an - * unsupported nullable type. - * - * It will also log a warning message if the nullability for Avro and catalyst types are - * different. - */ - private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = { - val (avroNullable, resolvedAvroType) = resolveAvroType(avroType) - warnNullabilityDifference(avroNullable, nullable) - resolvedAvroType - } - - /** - * Check the nullability of the input Avro type and resolve it when it is nullable. The first - * return value is a [[Boolean]] indicating if the input Avro type is nullable. The second - * return value is the possibly resolved type. - */ - private def resolveAvroType(avroType: Schema): (Boolean, Schema) = { - if (avroType.getType == Type.UNION) { - val fields = avroType.getTypes.asScala - val actualType = fields.filter(_.getType != Type.NULL) - if (fields.length == 2 && actualType.length == 1) { - (true, actualType.head) - } else { - // This is just a normal union, not used to designate nullability - (false, avroType) - } - } else { - (false, avroType) - } - } - - /** - * log a warning message if the nullability for Avro and catalyst types are different. - */ - private def warnNullabilityDifference(avroNullable: Boolean, catalystNullable: Boolean): Unit = { - if (avroNullable && !catalystNullable) { - logWarning("Writing Avro files with nullable Avro schema and non-nullable catalyst schema.") - } - if (!avroNullable && catalystNullable) { - logWarning("Writing Avro files with non-nullable Avro schema and nullable catalyst " + - "schema will throw runtime exception if there is a record with null value.") - } - } -} - -object AvroSerializer { - - // NOTE: Following methods have been renamed in Spark 3.2.1 [1] making [[AvroSerializer]] implementation - // (which relies on it) be only compatible with the exact same version of [[DataSourceUtils]]. - // To make sure this implementation is compatible w/ all Spark versions w/in Spark 3.2.x branch, - // we're preemptively cloned those methods to make sure Hudi is compatible w/ Spark 3.2.0 as well as - // w/ Spark >= 3.2.1 - // - // [1] https://github.com/apache/spark/pull/34978 - - def createDateRebaseFuncInWrite(rebaseMode: LegacyBehaviorPolicy.Value, - format: String): Int => Int = rebaseMode match { - case LegacyBehaviorPolicy.EXCEPTION => days: Int => - if (days < RebaseDateTime.lastSwitchGregorianDay) { - throw DataSourceUtils.newRebaseExceptionInWrite(format) - } - days - case LegacyBehaviorPolicy.LEGACY => RebaseDateTime.rebaseGregorianToJulianDays - case LegacyBehaviorPolicy.CORRECTED => identity[Int] - } - - def createTimestampRebaseFuncInWrite(rebaseMode: LegacyBehaviorPolicy.Value, - format: String): Long => Long = rebaseMode match { - case LegacyBehaviorPolicy.EXCEPTION => micros: Long => - if (micros < RebaseDateTime.lastSwitchGregorianTs) { - throw DataSourceUtils.newRebaseExceptionInWrite(format) - } - micros - case LegacyBehaviorPolicy.LEGACY => - val timeZone = SQLConf.get.sessionLocalTimeZone - RebaseDateTime.rebaseGregorianToJulianMicros(TimeZone.getTimeZone(timeZone), _) - case LegacyBehaviorPolicy.CORRECTED => identity[Long] - } - -} diff --git a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala deleted file mode 100644 index d10ff1edfe87..000000000000 --- a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala +++ /dev/null @@ -1,531 +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.avro - -import org.apache.hudi.common.schema.HoodieSchema -import org.apache.hudi.common.schema.HoodieSchema.VectorLogicalType - -import org.apache.avro.{LogicalTypes, Schema, SchemaBuilder} -import org.apache.avro.Conversions.DecimalConversion -import org.apache.avro.LogicalTypes.{LocalTimestampMicros, LocalTimestampMillis, TimestampMicros, TimestampMillis} -import org.apache.avro.Schema.Type._ -import org.apache.avro.generic._ -import org.apache.avro.util.Utf8 -import org.apache.spark.sql.avro.AvroDeserializer.{createDateRebaseFuncInRead, createTimestampRebaseFuncInRead, RebaseSpec} -import org.apache.spark.sql.avro.AvroUtils.{toFieldStr, AvroMatchedField} -import org.apache.spark.sql.catalyst.{InternalRow, NoopFilters, StructFilters} -import org.apache.spark.sql.catalyst.expressions.{SpecificInternalRow, UnsafeArrayData} -import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, ArrayData, DateTimeUtils, GenericArrayData, RebaseDateTime} -import org.apache.spark.sql.catalyst.util.DateTimeConstants.MILLIS_PER_DAY -import org.apache.spark.sql.execution.datasources.DataSourceUtils -import org.apache.spark.sql.internal.LegacyBehaviorPolicy -import org.apache.spark.sql.types._ -import org.apache.spark.unsafe.types.UTF8String - -import java.math.BigDecimal -import java.nio.ByteBuffer -import java.nio.ByteOrder -import java.util.TimeZone - -import scala.collection.JavaConverters._ - -/** - * A deserializer to deserialize data in avro format to data in catalyst format. - * - * NOTE: This code is borrowed from Spark 3.3.0 - * This code is borrowed, so that we can better control compatibility w/in Spark minor - * branches (3.2.x, 3.1.x, etc) - * - * PLEASE REFRAIN MAKING ANY CHANGES TO THIS CODE UNLESS ABSOLUTELY NECESSARY - */ -private[sql] class AvroDeserializer(rootAvroType: Schema, - rootCatalystType: DataType, - positionalFieldMatch: Boolean, - datetimeRebaseSpec: RebaseSpec, - filters: StructFilters) { - - def this(rootAvroType: Schema, - rootCatalystType: DataType, - datetimeRebaseMode: String) = { - this( - rootAvroType, - rootCatalystType, - positionalFieldMatch = false, - RebaseSpec(LegacyBehaviorPolicy.withName(datetimeRebaseMode)), - new NoopFilters) - } - - private lazy val decimalConversions = new DecimalConversion() - - private val dateRebaseFunc = createDateRebaseFuncInRead(datetimeRebaseSpec.mode, "Avro") - - private val timestampRebaseFunc = createTimestampRebaseFuncInRead(datetimeRebaseSpec, "Avro") - - private val converter: Any => Option[Any] = try { - rootCatalystType match { - // A shortcut for empty schema. - case st: StructType if st.isEmpty => - (_: Any) => Some(InternalRow.empty) - - case st: StructType => - val resultRow = new SpecificInternalRow(st.map(_.dataType)) - val fieldUpdater = new RowUpdater(resultRow) - val applyFilters = filters.skipRow(resultRow, _) - val writer = getRecordWriter(rootAvroType, st, Nil, Nil, applyFilters) - (data: Any) => { - val record = data.asInstanceOf[GenericRecord] - val skipRow = writer(fieldUpdater, record) - if (skipRow) None else Some(resultRow) - } - - case _ => - val tmpRow = new SpecificInternalRow(Seq(rootCatalystType)) - val fieldUpdater = new RowUpdater(tmpRow) - val writer = newWriter(rootAvroType, rootCatalystType, Nil, Nil) - (data: Any) => { - writer(fieldUpdater, 0, data) - Some(tmpRow.get(0, rootCatalystType)) - } - } - } catch { - case ise: IncompatibleSchemaException => throw new IncompatibleSchemaException( - s"Cannot convert Avro type $rootAvroType to SQL type ${rootCatalystType.sql}.", ise) - } - - def deserialize(data: Any): Option[Any] = converter(data) - - /** - * Creates a writer to write avro values to Catalyst values at the given ordinal with the given - * updater. - */ - private def newWriter(avroType: Schema, - catalystType: DataType, - avroPath: Seq[String], - catalystPath: Seq[String]): (CatalystDataUpdater, Int, Any) => Unit = { - val errorPrefix = s"Cannot convert Avro ${toFieldStr(avroPath)} to " + - s"SQL ${toFieldStr(catalystPath)} because " - val incompatibleMsg = errorPrefix + - s"schema is incompatible (avroType = $avroType, sqlType = ${catalystType.sql})" - - (avroType.getType, catalystType) match { - case (NULL, NullType) => (updater, ordinal, _) => - updater.setNullAt(ordinal) - - // TODO: we can avoid boxing if future version of avro provide primitive accessors. - case (BOOLEAN, BooleanType) => (updater, ordinal, value) => - updater.setBoolean(ordinal, value.asInstanceOf[Boolean]) - - case (INT, IntegerType) => (updater, ordinal, value) => - updater.setInt(ordinal, value.asInstanceOf[Int]) - - case (INT, DateType) => (updater, ordinal, value) => - updater.setInt(ordinal, dateRebaseFunc(value.asInstanceOf[Int])) - - case (LONG, LongType) => (updater, ordinal, value) => - updater.setLong(ordinal, value.asInstanceOf[Long]) - - case (LONG, TimestampType) => avroType.getLogicalType match { - // For backward compatibility, if the Avro type is Long and it is not logical type - // (the `null` case), the value is processed as timestamp type with millisecond precision. - case null | _: TimestampMillis => (updater, ordinal, value) => - val millis = value.asInstanceOf[Long] - val micros = DateTimeUtils.millisToMicros(millis) - updater.setLong(ordinal, timestampRebaseFunc(micros)) - case _: TimestampMicros => (updater, ordinal, value) => - val micros = value.asInstanceOf[Long] - updater.setLong(ordinal, timestampRebaseFunc(micros)) - case other => throw new IncompatibleSchemaException(errorPrefix + - s"Avro logical type $other cannot be converted to SQL type ${TimestampType.sql}.") - } - - case (LONG, TimestampNTZType) => avroType.getLogicalType match { - // To keep consistent with TimestampType, if the Avro type is Long and it is not - // logical type (the `null` case), the value is processed as TimestampNTZ - // with millisecond precision. - case null | _: LocalTimestampMillis => (updater, ordinal, value) => - val millis = value.asInstanceOf[Long] - val micros = DateTimeUtils.millisToMicros(millis) - updater.setLong(ordinal, micros) - case _: LocalTimestampMicros => (updater, ordinal, value) => - val micros = value.asInstanceOf[Long] - updater.setLong(ordinal, micros) - case other => throw new IncompatibleSchemaException(errorPrefix + - s"Avro logical type $other cannot be converted to SQL type ${TimestampNTZType.sql}.") - } - - // Handle VECTOR logical type (FLOAT, DOUBLE, INT8) - case (FIXED, ArrayType(elementType, false)) => avroType.getLogicalType match { - case vectorLogicalType: VectorLogicalType => - val dimension = vectorLogicalType.getDimension - val vecElementType = HoodieSchema.Vector.VectorElementType.fromString(vectorLogicalType.getElementType) - val elementSize = vecElementType.getElementSize - (updater, ordinal, value) => { - val bytes = value.asInstanceOf[GenericData.Fixed].bytes() - val expectedSize = Math.multiplyExact(dimension, elementSize) - if (bytes.length != expectedSize) { - throw new IncompatibleSchemaException( - s"VECTOR byte size mismatch: expected=$expectedSize, actual=${bytes.length}") - } - elementType match { - case FloatType => - val buffer = ByteBuffer.wrap(bytes).order(VectorLogicalType.VECTOR_BYTE_ORDER) - val floats = new Array[Float](dimension) - var i = 0; while (i < dimension) { floats(i) = buffer.getFloat(); i += 1 } - updater.set(ordinal, ArrayData.toArrayData(floats)) - case DoubleType => - val buffer = ByteBuffer.wrap(bytes).order(VectorLogicalType.VECTOR_BYTE_ORDER) - val doubles = new Array[Double](dimension) - var i = 0; while (i < dimension) { doubles(i) = buffer.getDouble(); i += 1 } - updater.set(ordinal, ArrayData.toArrayData(doubles)) - case ByteType => - updater.set(ordinal, ArrayData.toArrayData(bytes.clone())) - } - } - case _ => throw new IncompatibleSchemaException(incompatibleMsg) - } - - // Before we upgrade Avro to 1.8 for logical type support, spark-avro converts Long to Date. - // For backward compatibility, we still keep this conversion. - case (LONG, DateType) => (updater, ordinal, value) => - updater.setInt(ordinal, (value.asInstanceOf[Long] / MILLIS_PER_DAY).toInt) - - case (FLOAT, FloatType) => (updater, ordinal, value) => - updater.setFloat(ordinal, value.asInstanceOf[Float]) - - case (DOUBLE, DoubleType) => (updater, ordinal, value) => - updater.setDouble(ordinal, value.asInstanceOf[Double]) - - case (STRING, StringType) => (updater, ordinal, value) => - val str = value match { - case s: String => UTF8String.fromString(s) - case s: Utf8 => - val bytes = new Array[Byte](s.getByteLength) - System.arraycopy(s.getBytes, 0, bytes, 0, s.getByteLength) - UTF8String.fromBytes(bytes) - case s: GenericData.EnumSymbol => UTF8String.fromString(s.toString) - } - updater.set(ordinal, str) - - case (ENUM, StringType) => (updater, ordinal, value) => - updater.set(ordinal, UTF8String.fromString(value.toString)) - - case (FIXED, BinaryType) => (updater, ordinal, value) => - updater.set(ordinal, value.asInstanceOf[GenericFixed].bytes().clone()) - - case (BYTES, BinaryType) => (updater, ordinal, value) => - val bytes = value match { - case b: ByteBuffer => - val bytes = new Array[Byte](b.remaining) - b.get(bytes) - // Do not forget to reset the position - b.rewind() - bytes - case b: Array[Byte] => b - case other => - throw new RuntimeException(errorPrefix + s"$other is not a valid avro binary.") - } - updater.set(ordinal, bytes) - - case (FIXED, _: DecimalType) => (updater, ordinal, value) => - val d = avroType.getLogicalType.asInstanceOf[LogicalTypes.Decimal] - val bigDecimal = decimalConversions.fromFixed(value.asInstanceOf[GenericFixed], avroType, d) - val decimal = createDecimal(bigDecimal, d.getPrecision, d.getScale) - updater.setDecimal(ordinal, decimal) - - case (BYTES, _: DecimalType) => (updater, ordinal, value) => - val d = avroType.getLogicalType.asInstanceOf[LogicalTypes.Decimal] - val bigDecimal = decimalConversions.fromBytes(value.asInstanceOf[ByteBuffer], avroType, d) - val decimal = createDecimal(bigDecimal, d.getPrecision, d.getScale) - updater.setDecimal(ordinal, decimal) - - case (RECORD, st: StructType) => - // Avro datasource doesn't accept filters with nested attributes. See SPARK-32328. - // We can always return `false` from `applyFilters` for nested records. - val writeRecord = - getRecordWriter(avroType, st, avroPath, catalystPath, applyFilters = _ => false) - (updater, ordinal, value) => - val row = new SpecificInternalRow(st) - writeRecord(new RowUpdater(row), value.asInstanceOf[GenericRecord]) - updater.set(ordinal, row) - - case (ARRAY, ArrayType(elementType, containsNull)) => - val avroElementPath = avroPath :+ "element" - val elementWriter = newWriter(avroType.getElementType, elementType, - avroElementPath, catalystPath :+ "element") - (updater, ordinal, value) => - val collection = value.asInstanceOf[java.util.Collection[Any]] - val result = createArrayData(elementType, collection.size()) - val elementUpdater = new ArrayDataUpdater(result) - - var i = 0 - val iter = collection.iterator() - while (iter.hasNext) { - val element = iter.next() - if (element == null) { - if (!containsNull) { - throw new RuntimeException( - s"Array value at path ${toFieldStr(avroElementPath)} is not allowed to be null") - } else { - elementUpdater.setNullAt(i) - } - } else { - elementWriter(elementUpdater, i, element) - } - i += 1 - } - - updater.set(ordinal, result) - - case (MAP, MapType(keyType, valueType, valueContainsNull)) if keyType == StringType => - val keyWriter = newWriter(SchemaBuilder.builder().stringType(), StringType, - avroPath :+ "key", catalystPath :+ "key") - val valueWriter = newWriter(avroType.getValueType, valueType, - avroPath :+ "value", catalystPath :+ "value") - (updater, ordinal, value) => - val map = value.asInstanceOf[java.util.Map[AnyRef, AnyRef]] - val keyArray = createArrayData(keyType, map.size()) - val keyUpdater = new ArrayDataUpdater(keyArray) - val valueArray = createArrayData(valueType, map.size()) - val valueUpdater = new ArrayDataUpdater(valueArray) - val iter = map.entrySet().iterator() - var i = 0 - while (iter.hasNext) { - val entry = iter.next() - assert(entry.getKey != null) - keyWriter(keyUpdater, i, entry.getKey) - if (entry.getValue == null) { - if (!valueContainsNull) { - throw new RuntimeException( - s"Map value at path ${toFieldStr(avroPath :+ "value")} is not allowed to be null") - } else { - valueUpdater.setNullAt(i) - } - } else { - valueWriter(valueUpdater, i, entry.getValue) - } - i += 1 - } - - // The Avro map will never have null or duplicated map keys, it's safe to create a - // ArrayBasedMapData directly here. - updater.set(ordinal, new ArrayBasedMapData(keyArray, valueArray)) - - case (UNION, _) => - val allTypes = avroType.getTypes.asScala - val nonNullTypes = allTypes.filter(_.getType != NULL) - val nonNullAvroType = Schema.createUnion(nonNullTypes.asJava) - if (nonNullTypes.nonEmpty) { - if (nonNullTypes.length == 1) { - newWriter(nonNullTypes.head, catalystType, avroPath, catalystPath) - } else { - nonNullTypes.map(_.getType).toSeq match { - case Seq(a, b) if Set(a, b) == Set(INT, LONG) && catalystType == LongType => - (updater, ordinal, value) => value match { - case null => updater.setNullAt(ordinal) - case l: java.lang.Long => updater.setLong(ordinal, l) - case i: java.lang.Integer => updater.setLong(ordinal, i.longValue()) - } - - case Seq(a, b) if Set(a, b) == Set(FLOAT, DOUBLE) && catalystType == DoubleType => - (updater, ordinal, value) => value match { - case null => updater.setNullAt(ordinal) - case d: java.lang.Double => updater.setDouble(ordinal, d) - case f: java.lang.Float => updater.setDouble(ordinal, f.doubleValue()) - } - - case _ => - catalystType match { - case st: StructType if st.length == nonNullTypes.size => - val fieldWriters = nonNullTypes.zip(st.fields).map { - case (schema, field) => - newWriter(schema, field.dataType, avroPath, catalystPath :+ field.name) - }.toArray - (updater, ordinal, value) => { - val row = new SpecificInternalRow(st) - val fieldUpdater = new RowUpdater(row) - val i = GenericData.get().resolveUnion(nonNullAvroType, value) - fieldWriters(i)(fieldUpdater, i, value) - updater.set(ordinal, row) - } - - case _ => throw new IncompatibleSchemaException(incompatibleMsg) - } - } - } - } else { - (updater, ordinal, _) => updater.setNullAt(ordinal) - } - - case (INT, _: YearMonthIntervalType) => (updater, ordinal, value) => - updater.setInt(ordinal, value.asInstanceOf[Int]) - - case (LONG, _: DayTimeIntervalType) => (updater, ordinal, value) => - updater.setLong(ordinal, value.asInstanceOf[Long]) - - case _ => throw new IncompatibleSchemaException(incompatibleMsg) - } - } - - // TODO: move the following method in Decimal object on creating Decimal from BigDecimal? - private def createDecimal(decimal: BigDecimal, precision: Int, scale: Int): Decimal = { - if (precision <= Decimal.MAX_LONG_DIGITS) { - // Constructs a `Decimal` with an unscaled `Long` value if possible. - Decimal(decimal.unscaledValue().longValue(), precision, scale) - } else { - // Otherwise, resorts to an unscaled `BigInteger` instead. - Decimal(decimal, precision, scale) - } - } - - private def getRecordWriter( - avroType: Schema, - catalystType: StructType, - avroPath: Seq[String], - catalystPath: Seq[String], - applyFilters: Int => Boolean): (CatalystDataUpdater, GenericRecord) => Boolean = { - - val avroSchemaHelper = new AvroUtils.AvroSchemaHelper( - avroType, catalystType, avroPath, catalystPath, positionalFieldMatch) - - avroSchemaHelper.validateNoExtraCatalystFields(ignoreNullable = true) - // no need to validateNoExtraAvroFields since extra Avro fields are ignored - - val (validFieldIndexes, fieldWriters) = avroSchemaHelper.matchedFields.map { - case AvroMatchedField(catalystField, ordinal, avroField) => - val baseWriter = newWriter(avroField.schema(), catalystField.dataType, - avroPath :+ avroField.name, catalystPath :+ catalystField.name) - val fieldWriter = (fieldUpdater: CatalystDataUpdater, value: Any) => { - if (value == null) { - fieldUpdater.setNullAt(ordinal) - } else { - baseWriter(fieldUpdater, ordinal, value) - } - } - (avroField.pos(), fieldWriter) - }.toArray.unzip - - (fieldUpdater, record) => { - var i = 0 - var skipRow = false - while (i < validFieldIndexes.length && !skipRow) { - fieldWriters(i)(fieldUpdater, record.get(validFieldIndexes(i))) - skipRow = applyFilters(i) - i += 1 - } - skipRow - } - } - - private def createArrayData(elementType: DataType, length: Int): ArrayData = elementType match { - case BooleanType => UnsafeArrayData.fromPrimitiveArray(new Array[Boolean](length)) - case ByteType => UnsafeArrayData.fromPrimitiveArray(new Array[Byte](length)) - case ShortType => UnsafeArrayData.fromPrimitiveArray(new Array[Short](length)) - case IntegerType => UnsafeArrayData.fromPrimitiveArray(new Array[Int](length)) - case LongType => UnsafeArrayData.fromPrimitiveArray(new Array[Long](length)) - case FloatType => UnsafeArrayData.fromPrimitiveArray(new Array[Float](length)) - case DoubleType => UnsafeArrayData.fromPrimitiveArray(new Array[Double](length)) - case _ => new GenericArrayData(new Array[Any](length)) - } - - /** - * A base interface for updating values inside catalyst data structure like `InternalRow` and - * `ArrayData`. - */ - sealed trait CatalystDataUpdater { - def set(ordinal: Int, value: Any): Unit - - def setNullAt(ordinal: Int): Unit = set(ordinal, null) - def setBoolean(ordinal: Int, value: Boolean): Unit = set(ordinal, value) - def setByte(ordinal: Int, value: Byte): Unit = set(ordinal, value) - def setShort(ordinal: Int, value: Short): Unit = set(ordinal, value) - def setInt(ordinal: Int, value: Int): Unit = set(ordinal, value) - def setLong(ordinal: Int, value: Long): Unit = set(ordinal, value) - def setDouble(ordinal: Int, value: Double): Unit = set(ordinal, value) - def setFloat(ordinal: Int, value: Float): Unit = set(ordinal, value) - def setDecimal(ordinal: Int, value: Decimal): Unit = set(ordinal, value) - } - - final class RowUpdater(row: InternalRow) extends CatalystDataUpdater { - override def set(ordinal: Int, value: Any): Unit = row.update(ordinal, value) - - override def setNullAt(ordinal: Int): Unit = row.setNullAt(ordinal) - override def setBoolean(ordinal: Int, value: Boolean): Unit = row.setBoolean(ordinal, value) - override def setByte(ordinal: Int, value: Byte): Unit = row.setByte(ordinal, value) - override def setShort(ordinal: Int, value: Short): Unit = row.setShort(ordinal, value) - override def setInt(ordinal: Int, value: Int): Unit = row.setInt(ordinal, value) - override def setLong(ordinal: Int, value: Long): Unit = row.setLong(ordinal, value) - override def setDouble(ordinal: Int, value: Double): Unit = row.setDouble(ordinal, value) - override def setFloat(ordinal: Int, value: Float): Unit = row.setFloat(ordinal, value) - override def setDecimal(ordinal: Int, value: Decimal): Unit = - row.setDecimal(ordinal, value, value.precision) - } - - final class ArrayDataUpdater(array: ArrayData) extends CatalystDataUpdater { - override def set(ordinal: Int, value: Any): Unit = array.update(ordinal, value) - - override def setNullAt(ordinal: Int): Unit = array.setNullAt(ordinal) - override def setBoolean(ordinal: Int, value: Boolean): Unit = array.setBoolean(ordinal, value) - override def setByte(ordinal: Int, value: Byte): Unit = array.setByte(ordinal, value) - override def setShort(ordinal: Int, value: Short): Unit = array.setShort(ordinal, value) - override def setInt(ordinal: Int, value: Int): Unit = array.setInt(ordinal, value) - override def setLong(ordinal: Int, value: Long): Unit = array.setLong(ordinal, value) - override def setDouble(ordinal: Int, value: Double): Unit = array.setDouble(ordinal, value) - override def setFloat(ordinal: Int, value: Float): Unit = array.setFloat(ordinal, value) - override def setDecimal(ordinal: Int, value: Decimal): Unit = array.update(ordinal, value) - } -} - -object AvroDeserializer { - - // NOTE: Following methods have been renamed in Spark 3.2.1 [1] making [[AvroDeserializer]] implementation - // (which relies on it) be only compatible with the exact same version of [[DataSourceUtils]]. - // To make sure this implementation is compatible w/ all Spark versions w/in Spark 3.2.x branch, - // we're preemptively cloned those methods to make sure Hudi is compatible w/ Spark 3.2.0 as well as - // w/ Spark >= 3.2.1 - // - // [1] https://github.com/apache/spark/pull/34978 - - // Specification of rebase operation including `mode` and the time zone in which it is performed - case class RebaseSpec(mode: LegacyBehaviorPolicy.Value, originTimeZone: Option[String] = None) { - // Use the default JVM time zone for backward compatibility - def timeZone: String = originTimeZone.getOrElse(TimeZone.getDefault.getID) - } - - def createDateRebaseFuncInRead(rebaseMode: LegacyBehaviorPolicy.Value, - format: String): Int => Int = rebaseMode match { - case LegacyBehaviorPolicy.EXCEPTION => days: Int => - if (days < RebaseDateTime.lastSwitchJulianDay) { - throw DataSourceUtils.newRebaseExceptionInRead(format) - } - days - case LegacyBehaviorPolicy.LEGACY => RebaseDateTime.rebaseJulianToGregorianDays - case LegacyBehaviorPolicy.CORRECTED => identity[Int] - } - - def createTimestampRebaseFuncInRead(rebaseSpec: RebaseSpec, - format: String): Long => Long = rebaseSpec.mode match { - case LegacyBehaviorPolicy.EXCEPTION => micros: Long => - if (micros < RebaseDateTime.lastSwitchJulianTs) { - throw DataSourceUtils.newRebaseExceptionInRead(format) - } - micros - case LegacyBehaviorPolicy.LEGACY => micros: Long => - RebaseDateTime.rebaseJulianToGregorianMicros(TimeZone.getTimeZone(rebaseSpec.timeZone), micros) - case LegacyBehaviorPolicy.CORRECTED => identity[Long] - } -} diff --git a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala deleted file mode 100644 index 756ef82a55d2..000000000000 --- a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/avro/AvroSerializer.scala +++ /dev/null @@ -1,489 +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.avro - -import org.apache.hudi.common.schema.HoodieSchema -import org.apache.hudi.common.schema.HoodieSchema.VectorLogicalType - -import org.apache.avro.{LogicalTypes, Schema} -import org.apache.avro.Conversions.DecimalConversion -import org.apache.avro.LogicalTypes.{LocalTimestampMicros, LocalTimestampMillis, TimestampMicros, TimestampMillis} -import org.apache.avro.Schema.Type -import org.apache.avro.Schema.Type._ -import org.apache.avro.generic.GenericData.{EnumSymbol, Fixed, Record} -import org.apache.avro.util.Utf8 -import org.apache.spark.internal.Logging -import org.apache.spark.sql.avro.AvroSerializer.{createDateRebaseFuncInWrite, createTimestampRebaseFuncInWrite} -import org.apache.spark.sql.avro.AvroUtils.{toFieldStr, AvroMatchedField} -import org.apache.spark.sql.catalyst.InternalRow -import org.apache.spark.sql.catalyst.expressions.{SpecializedGetters, SpecificInternalRow} -import org.apache.spark.sql.catalyst.util.{DateTimeUtils, RebaseDateTime} -import org.apache.spark.sql.execution.datasources.DataSourceUtils -import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf} -import org.apache.spark.sql.types._ - -import java.nio.ByteBuffer -import java.nio.ByteOrder -import java.util.TimeZone - -import scala.collection.JavaConverters._ - -/** - * A serializer to serialize data in catalyst format to data in avro format. - * - * NOTE: This code is borrowed from Spark 3.3.0 - * This code is borrowed, so that we can better control compatibility w/in Spark minor - * branches (3.2.x, 3.1.x, etc) - * - * NOTE: THIS IMPLEMENTATION HAS BEEN MODIFIED FROM ITS ORIGINAL VERSION WITH THE MODIFICATION - * BEING EXPLICITLY ANNOTATED INLINE. PLEASE MAKE SURE TO UNDERSTAND PROPERLY ALL THE - * MODIFICATIONS. - * - * PLEASE REFRAIN MAKING ANY CHANGES TO THIS CODE UNLESS ABSOLUTELY NECESSARY - */ -private[sql] class AvroSerializer(rootCatalystType: DataType, - rootAvroType: Schema, - nullable: Boolean, - positionalFieldMatch: Boolean, - datetimeRebaseMode: LegacyBehaviorPolicy.Value) extends Logging { - - def this(rootCatalystType: DataType, rootAvroType: Schema, nullable: Boolean) = { - this(rootCatalystType, rootAvroType, nullable, positionalFieldMatch = false, - LegacyBehaviorPolicy.withName(SQLConf.get.getConf(SQLConf.AVRO_REBASE_MODE_IN_WRITE, - LegacyBehaviorPolicy.CORRECTED.toString))) - } - - def serialize(catalystData: Any): Any = { - converter.apply(catalystData) - } - - private val dateRebaseFunc = createDateRebaseFuncInWrite( - datetimeRebaseMode, "Avro") - - private val timestampRebaseFunc = createTimestampRebaseFuncInWrite( - datetimeRebaseMode, "Avro") - - private val converter: Any => Any = { - val actualAvroType = resolveNullableType(rootAvroType, nullable) - val baseConverter = try { - rootCatalystType match { - case st: StructType => - newStructConverter(st, actualAvroType, Nil, Nil).asInstanceOf[Any => Any] - case _ => - val tmpRow = new SpecificInternalRow(Seq(rootCatalystType)) - val converter = newConverter(rootCatalystType, actualAvroType, Nil, Nil) - (data: Any) => - tmpRow.update(0, data) - converter.apply(tmpRow, 0) - } - } catch { - case ise: IncompatibleSchemaException => throw new IncompatibleSchemaException( - s"Cannot convert SQL type ${rootCatalystType.sql} to Avro type $rootAvroType.", ise) - } - if (nullable) { - (data: Any) => - if (data == null) { - null - } else { - baseConverter.apply(data) - } - } else { - baseConverter - } - } - - private type Converter = (SpecializedGetters, Int) => Any - - private lazy val decimalConversions = new DecimalConversion() - - private def newConverter(catalystType: DataType, - avroType: Schema, - catalystPath: Seq[String], - avroPath: Seq[String]): Converter = { - val errorPrefix = s"Cannot convert SQL ${toFieldStr(catalystPath)} " + - s"to Avro ${toFieldStr(avroPath)} because " - (catalystType, avroType.getType) match { - case (NullType, NULL) => - (getter, ordinal) => null - case (BooleanType, BOOLEAN) => - (getter, ordinal) => getter.getBoolean(ordinal) - case (ByteType, INT) => - (getter, ordinal) => getter.getByte(ordinal).toInt - case (ShortType, INT) => - (getter, ordinal) => getter.getShort(ordinal).toInt - case (IntegerType, INT) => - (getter, ordinal) => getter.getInt(ordinal) - case (LongType, LONG) => - (getter, ordinal) => getter.getLong(ordinal) - case (FloatType, FLOAT) => - (getter, ordinal) => getter.getFloat(ordinal) - case (DoubleType, DOUBLE) => - (getter, ordinal) => getter.getDouble(ordinal) - case (d: DecimalType, FIXED) - if avroType.getLogicalType == LogicalTypes.decimal(d.precision, d.scale) => - (getter, ordinal) => - val decimal = getter.getDecimal(ordinal, d.precision, d.scale) - decimalConversions.toFixed(decimal.toJavaBigDecimal, avroType, - LogicalTypes.decimal(d.precision, d.scale)) - - case (d: DecimalType, BYTES) - if avroType.getLogicalType == LogicalTypes.decimal(d.precision, d.scale) => - (getter, ordinal) => - val decimal = getter.getDecimal(ordinal, d.precision, d.scale) - decimalConversions.toBytes(decimal.toJavaBigDecimal, avroType, - LogicalTypes.decimal(d.precision, d.scale)) - - // Handle VECTOR logical type (FLOAT, DOUBLE, INT8) - case (ArrayType(elementType, false), FIXED) => avroType.getLogicalType match { - case vectorLogicalType: VectorLogicalType => - val dimension = vectorLogicalType.getDimension - val vecElementType = HoodieSchema.Vector.VectorElementType.fromString(vectorLogicalType.getElementType) - val bufferSize = Math.multiplyExact(dimension, vecElementType.getElementSize) - (getter, ordinal) => { - val arrayData = getter.getArray(ordinal) - if (arrayData.numElements() != dimension) { - throw new IncompatibleSchemaException( - s"VECTOR dimension mismatch at ${toFieldStr(catalystPath)}: " + - s"expected=$dimension, actual=${arrayData.numElements()}") - } - elementType match { - case FloatType => - val buffer = ByteBuffer.allocate(bufferSize).order(VectorLogicalType.VECTOR_BYTE_ORDER) - var i = 0; while (i < dimension) { buffer.putFloat(arrayData.getFloat(i)); i += 1 } - new Fixed(avroType, buffer.array()) - case DoubleType => - val buffer = ByteBuffer.allocate(bufferSize).order(VectorLogicalType.VECTOR_BYTE_ORDER) - var i = 0; while (i < dimension) { buffer.putDouble(arrayData.getDouble(i)); i += 1 } - new Fixed(avroType, buffer.array()) - case ByteType => - val bytes = new Array[Byte](dimension) - var i = 0; while (i < dimension) { bytes(i) = arrayData.getByte(i); i += 1 } - new Fixed(avroType, bytes) - case _ => throw new IncompatibleSchemaException(errorPrefix + - s"schema is incompatible (sqlType = ${catalystType.sql}, avroType = $avroType)") - } - } - case _ => throw new IncompatibleSchemaException(errorPrefix + - s"schema is incompatible (sqlType = ${catalystType.sql}, avroType = $avroType)") - } - - case (StringType, ENUM) => - val enumSymbols: Set[String] = avroType.getEnumSymbols.asScala.toSet - (getter, ordinal) => - val data = getter.getUTF8String(ordinal).toString - if (!enumSymbols.contains(data)) { - throw new IncompatibleSchemaException(errorPrefix + - s""""$data" cannot be written since it's not defined in enum """ + - enumSymbols.mkString("\"", "\", \"", "\"")) - } - new EnumSymbol(avroType, data) - - case (StringType, STRING) => - (getter, ordinal) => new Utf8(getter.getUTF8String(ordinal).getBytes) - - case (BinaryType, FIXED) => - val size = avroType.getFixedSize - (getter, ordinal) => - val data: Array[Byte] = getter.getBinary(ordinal) - if (data.length != size) { - def len2str(len: Int): String = s"$len ${if (len > 1) "bytes" else "byte"}" - - throw new IncompatibleSchemaException(errorPrefix + len2str(data.length) + - " of binary data cannot be written into FIXED type with size of " + len2str(size)) - } - new Fixed(avroType, data) - - case (BinaryType, BYTES) => - (getter, ordinal) => ByteBuffer.wrap(getter.getBinary(ordinal)) - - case (DateType, INT) => - (getter, ordinal) => dateRebaseFunc(getter.getInt(ordinal)) - - case (TimestampType, LONG) => avroType.getLogicalType match { - // For backward compatibility, if the Avro type is Long and it is not logical type - // (the `null` case), output the timestamp value as with millisecond precision. - case null | _: TimestampMillis => (getter, ordinal) => - DateTimeUtils.microsToMillis(timestampRebaseFunc(getter.getLong(ordinal))) - case _: TimestampMicros => (getter, ordinal) => - timestampRebaseFunc(getter.getLong(ordinal)) - case other => throw new IncompatibleSchemaException(errorPrefix + - s"SQL type ${TimestampType.sql} cannot be converted to Avro logical type $other") - } - - case (TimestampNTZType, LONG) => avroType.getLogicalType match { - // To keep consistent with TimestampType, if the Avro type is Long and it is not - // logical type (the `null` case), output the TimestampNTZ as long value - // in millisecond precision. - case null | _: LocalTimestampMillis => (getter, ordinal) => - DateTimeUtils.microsToMillis(getter.getLong(ordinal)) - case _: LocalTimestampMicros => (getter, ordinal) => - getter.getLong(ordinal) - case other => throw new IncompatibleSchemaException(errorPrefix + - s"SQL type ${TimestampNTZType.sql} cannot be converted to Avro logical type $other") - } - - case (ArrayType(et, containsNull), ARRAY) => - val elementConverter = newConverter( - et, resolveNullableType(avroType.getElementType, containsNull), - catalystPath :+ "element", avroPath :+ "element") - (getter, ordinal) => { - val arrayData = getter.getArray(ordinal) - val len = arrayData.numElements() - val result = new Array[Any](len) - var i = 0 - while (i < len) { - if (containsNull && arrayData.isNullAt(i)) { - result(i) = null - } else { - result(i) = elementConverter(arrayData, i) - } - i += 1 - } - // avro writer is expecting a Java Collection, so we convert it into - // `ArrayList` backed by the specified array without data copying. - java.util.Arrays.asList(result: _*) - } - - case (st: StructType, RECORD) => - val structConverter = newStructConverter(st, avroType, catalystPath, avroPath) - val numFields = st.length - (getter, ordinal) => structConverter(getter.getStruct(ordinal, numFields)) - - //////////////////////////////////////////////////////////////////////////////////////////// - // Following section is amended to the original (Spark's) implementation - // >>> BEGINS - //////////////////////////////////////////////////////////////////////////////////////////// - - case (st: StructType, UNION) => - val unionConverter = newUnionConverter(st, avroType, catalystPath, avroPath) - val numFields = st.length - (getter, ordinal) => unionConverter(getter.getStruct(ordinal, numFields)) - - //////////////////////////////////////////////////////////////////////////////////////////// - // <<< ENDS - //////////////////////////////////////////////////////////////////////////////////////////// - - case (MapType(kt, vt, valueContainsNull), MAP) if kt == StringType => - val valueConverter = newConverter( - vt, resolveNullableType(avroType.getValueType, valueContainsNull), - catalystPath :+ "value", avroPath :+ "value") - (getter, ordinal) => - val mapData = getter.getMap(ordinal) - val len = mapData.numElements() - val result = new java.util.HashMap[String, Any](len) - val keyArray = mapData.keyArray() - val valueArray = mapData.valueArray() - var i = 0 - while (i < len) { - val key = keyArray.getUTF8String(i).toString - if (valueContainsNull && valueArray.isNullAt(i)) { - result.put(key, null) - } else { - result.put(key, valueConverter(valueArray, i)) - } - i += 1 - } - result - - case (_: YearMonthIntervalType, INT) => - (getter, ordinal) => getter.getInt(ordinal) - - case (_: DayTimeIntervalType, LONG) => - (getter, ordinal) => getter.getLong(ordinal) - - case _ => - throw new IncompatibleSchemaException(errorPrefix + - s"schema is incompatible (sqlType = ${catalystType.sql}, avroType = $avroType)") - } - } - - private def newStructConverter(catalystStruct: StructType, - avroStruct: Schema, - catalystPath: Seq[String], - avroPath: Seq[String]): InternalRow => Record = { - - val avroSchemaHelper = new AvroUtils.AvroSchemaHelper( - avroStruct, catalystStruct, avroPath, catalystPath, positionalFieldMatch) - - avroSchemaHelper.validateNoExtraCatalystFields(ignoreNullable = false) - avroSchemaHelper.validateNoExtraRequiredAvroFields() - - val (avroIndices, fieldConverters) = avroSchemaHelper.matchedFields.map { - case AvroMatchedField(catalystField, _, avroField) => - val converter = newConverter(catalystField.dataType, - resolveNullableType(avroField.schema(), catalystField.nullable), - catalystPath :+ catalystField.name, avroPath :+ avroField.name) - (avroField.pos(), converter) - }.toArray.unzip - - val numFields = catalystStruct.length - row: InternalRow => - val result = new Record(avroStruct) - var i = 0 - while (i < numFields) { - if (row.isNullAt(i)) { - result.put(avroIndices(i), null) - } else { - result.put(avroIndices(i), fieldConverters(i).apply(row, i)) - } - i += 1 - } - result - } - - //////////////////////////////////////////////////////////////////////////////////////////// - // Following section is amended to the original (Spark's) implementation - // >>> BEGINS - //////////////////////////////////////////////////////////////////////////////////////////// - - private def newUnionConverter(catalystStruct: StructType, - avroUnion: Schema, - catalystPath: Seq[String], - avroPath: Seq[String]): InternalRow => Any = { - if (avroUnion.getType != UNION || !canMapUnion(catalystStruct, avroUnion)) { - throw new IncompatibleSchemaException(s"Cannot convert Catalyst type $catalystStruct to " + - s"Avro type $avroUnion.") - } - val nullable = avroUnion.getTypes.size() > 0 && avroUnion.getTypes.get(0).getType == Type.NULL - val avroInnerTypes = if (nullable) { - avroUnion.getTypes.asScala.tail - } else { - avroUnion.getTypes.asScala - } - val fieldConverters = catalystStruct.zip(avroInnerTypes).map { - case (f1, f2) => newConverter(f1.dataType, f2, catalystPath, avroPath) - } - val numFields = catalystStruct.length - (row: InternalRow) => - var i = 0 - var result: Any = null - while (i < numFields) { - if (!row.isNullAt(i)) { - if (result != null) { - throw new IncompatibleSchemaException(s"Cannot convert Catalyst record $catalystStruct to " + - s"Avro union $avroUnion. Record has more than one optional values set") - } - result = fieldConverters(i).apply(row, i) - } - i += 1 - } - if (!nullable && result == null) { - throw new IncompatibleSchemaException(s"Cannot convert Catalyst record $catalystStruct to " + - s"Avro union $avroUnion. Record has no values set, while should have exactly one") - } - result - } - - private def canMapUnion(catalystStruct: StructType, avroStruct: Schema): Boolean = { - (avroStruct.getTypes.size() > 0 && - avroStruct.getTypes.get(0).getType == Type.NULL && - avroStruct.getTypes.size() - 1 == catalystStruct.length) || avroStruct.getTypes.size() == catalystStruct.length - } - - //////////////////////////////////////////////////////////////////////////////////////////// - // <<< ENDS - //////////////////////////////////////////////////////////////////////////////////////////// - - - /** - * Resolve a possibly nullable Avro Type. - * - * An Avro type is nullable when it is a [[UNION]] of two types: one null type and another - * non-null type. This method will check the nullability of the input Avro type and return the - * non-null type within when it is nullable. Otherwise it will return the input Avro type - * unchanged. It will throw an [[UnsupportedAvroTypeException]] when the input Avro type is an - * unsupported nullable type. - * - * It will also log a warning message if the nullability for Avro and catalyst types are - * different. - */ - private def resolveNullableType(avroType: Schema, nullable: Boolean): Schema = { - val (avroNullable, resolvedAvroType) = resolveAvroType(avroType) - warnNullabilityDifference(avroNullable, nullable) - resolvedAvroType - } - - /** - * Check the nullability of the input Avro type and resolve it when it is nullable. The first - * return value is a [[Boolean]] indicating if the input Avro type is nullable. The second - * return value is the possibly resolved type. - */ - private def resolveAvroType(avroType: Schema): (Boolean, Schema) = { - if (avroType.getType == Type.UNION) { - val fields = avroType.getTypes.asScala - val actualType = fields.filter(_.getType != Type.NULL) - if (fields.length == 2 && actualType.length == 1) { - (true, actualType.head) - } else { - // This is just a normal union, not used to designate nullability - (false, avroType) - } - } else { - (false, avroType) - } - } - - /** - * log a warning message if the nullability for Avro and catalyst types are different. - */ - private def warnNullabilityDifference(avroNullable: Boolean, catalystNullable: Boolean): Unit = { - if (avroNullable && !catalystNullable) { - logWarning("Writing Avro files with nullable Avro schema and non-nullable catalyst schema.") - } - if (!avroNullable && catalystNullable) { - logWarning("Writing Avro files with non-nullable Avro schema and nullable catalyst " + - "schema will throw runtime exception if there is a record with null value.") - } - } -} - -object AvroSerializer { - - // NOTE: Following methods have been renamed in Spark 3.2.1 [1] making [[AvroSerializer]] implementation - // (which relies on it) be only compatible with the exact same version of [[DataSourceUtils]]. - // To make sure this implementation is compatible w/ all Spark versions w/in Spark 3.2.x branch, - // we're preemptively cloned those methods to make sure Hudi is compatible w/ Spark 3.2.0 as well as - // w/ Spark >= 3.2.1 - // - // [1] https://github.com/apache/spark/pull/34978 - - def createDateRebaseFuncInWrite(rebaseMode: LegacyBehaviorPolicy.Value, - format: String): Int => Int = rebaseMode match { - case LegacyBehaviorPolicy.EXCEPTION => days: Int => - if (days < RebaseDateTime.lastSwitchGregorianDay) { - throw DataSourceUtils.newRebaseExceptionInWrite(format) - } - days - case LegacyBehaviorPolicy.LEGACY => RebaseDateTime.rebaseGregorianToJulianDays - case LegacyBehaviorPolicy.CORRECTED => identity[Int] - } - - def createTimestampRebaseFuncInWrite(rebaseMode: LegacyBehaviorPolicy.Value, - format: String): Long => Long = rebaseMode match { - case LegacyBehaviorPolicy.EXCEPTION => micros: Long => - if (micros < RebaseDateTime.lastSwitchGregorianTs) { - throw DataSourceUtils.newRebaseExceptionInWrite(format) - } - micros - case LegacyBehaviorPolicy.LEGACY => - val timeZone = SQLConf.get.sessionLocalTimeZone - RebaseDateTime.rebaseGregorianToJulianMicros(TimeZone.getTimeZone(timeZone), _) - case LegacyBehaviorPolicy.CORRECTED => identity[Long] - } - -}
