This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 7b843ae2c35f refactor(spark): consolidate the vendored 3.x Avro serde
forks into hudi-spark3-common (#19168)
7b843ae2c35f is described below
commit 7b843ae2c35fa00993a2a1b102c9ad7e1e116267
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.
---
.../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]
- }
-
-}
