This is an automated email from the ASF dual-hosted git repository.
wombatu-kun 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 9953687a2c93 refactor(spark): consolidate the vendored AvroUtils fork
into hudi-spark-common (#19147)
9953687a2c93 is described below
commit 9953687a2c9351c42da313548f86e471d1dd810d
Author: Y Ethan Guo <[email protected]>
AuthorDate: Thu Jul 2 22:47:56 2026 -0700
refactor(spark): consolidate the vendored AvroUtils fork into
hudi-spark-common (#19147)
The vendored spark-avro AvroUtils.scala is duplicated across all six
Spark version modules (hudi-spark3.3.x through hudi-spark4.2.x). The six
copies are identical except for a stray space inside one import line of
the 3.3.x copy. Move a single copy to hudi-spark-common, which every
version module already depends on, and delete the six duplicates.
The consolidated copy omits the upstream RowReader trait and the three
imports only it used: the trait references the version-specific vendored
AvroDeserializer, which stays in the version modules and is not visible
to hudi-spark-common, and it has no usages anywhere in the repo. The
members the version modules actually use (toFieldStr, AvroMatchedField,
AvroSchemaHelper) are unchanged, and IncompatibleSchemaException still
resolves from hudi-spark-client, same as before.
Consolidating also gives the file a unique repo path, so it re-enters
the coverage denominator: the six identically-pathed copies were dropped
by report-path resolution and invisible to Codecov.
---
.../org/apache/spark/sql/avro/AvroUtils.scala | 45 +---
.../org/apache/spark/sql/avro/AvroUtils.scala | 227 ---------------------
.../org/apache/spark/sql/avro/AvroUtils.scala | 227 ---------------------
.../org/apache/spark/sql/avro/AvroUtils.scala | 227 ---------------------
.../org/apache/spark/sql/avro/AvroUtils.scala | 227 ---------------------
.../org/apache/spark/sql/avro/AvroUtils.scala | 227 ---------------------
6 files changed, 3 insertions(+), 1177 deletions(-)
diff --git
a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
similarity index 85%
rename from
hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
rename to
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
index 028ebebf0bf5..b11987a40c67 100644
---
a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
@@ -18,10 +18,7 @@
package org.apache.spark.sql.avro
import org.apache.avro.Schema
-import org.apache.avro.file.FileReader
-import org.apache.avro.generic.GenericRecord
import org.apache.spark.internal.Logging
-import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
@@ -34,6 +31,9 @@ import scala.collection.JavaConverters._
* This code is borrowed, so that we can better control compatibility
w/in Spark minor
* branches (3.2.x, 3.1.x, etc)
*
+ * This copy omits the upstream `RowReader` trait: it is unused in Hudi
and references
+ * the version-specific vendored `AvroDeserializer`, which is not
visible to this module.
+ *
* PLEASE REFRAIN MAKING ANY CHANGES TO THIS CODE UNLESS ABSOLUTELY
NECESSARY
*/
private[sql] object AvroUtils extends Logging {
@@ -55,45 +55,6 @@ private[sql] object AvroUtils extends Logging {
case _ => false
}
- // The trait provides iterator-like interface for reading records from an
Avro file,
- // deserializing and returning them as internal rows.
- trait RowReader {
- protected val fileReader: FileReader[GenericRecord]
- protected val deserializer: AvroDeserializer
- protected val stopPosition: Long
-
- private[this] var completed = false
- private[this] var currentRow: Option[InternalRow] = None
-
- def hasNextRow: Boolean = {
- while (!completed && currentRow.isEmpty) {
- val r = fileReader.hasNext && !fileReader.pastSync(stopPosition)
- if (!r) {
- fileReader.close()
- completed = true
- currentRow = None
- } else {
- val record = fileReader.next()
- // the row must be deserialized in hasNextRow, because
AvroDeserializer#deserialize
- // potentially filters rows
- currentRow =
deserializer.deserialize(record).asInstanceOf[Option[InternalRow]]
- }
- }
- currentRow.isDefined
- }
-
- def nextRow: InternalRow = {
- if (currentRow.isEmpty) {
- hasNextRow
- }
- val returnRow = currentRow
- currentRow = None // free up hasNextRow to consume more Avro records, if
not exhausted
- returnRow.getOrElse {
- throw new NoSuchElementException("next on empty iterator")
- }
- }
- }
-
/** Wrapper for a pair of matched fields, one Catalyst and one corresponding
Avro field. */
private[sql] case class AvroMatchedField(
catalystField: StructField,
diff --git
a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
deleted file mode 100644
index 8aae6b442f8a..000000000000
---
a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
+++ /dev/null
@@ -1,227 +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.avro.Schema
-import org.apache.avro.file. FileReader
-import org.apache.avro.generic.GenericRecord
-import org.apache.spark.internal.Logging
-import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types._
-
-import java.util.Locale
-
-import scala.collection.JavaConverters._
-
-/**
- * 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] object AvroUtils extends Logging {
-
- def supportsDataType(dataType: DataType): Boolean = dataType match {
- case _: AtomicType => true
-
- case st: StructType => st.forall { f => supportsDataType(f.dataType) }
-
- case ArrayType(elementType, _) => supportsDataType(elementType)
-
- case MapType(keyType, valueType, _) =>
- supportsDataType(keyType) && supportsDataType(valueType)
-
- case udt: UserDefinedType[_] => supportsDataType(udt.sqlType)
-
- case _: NullType => true
-
- case _ => false
- }
-
- // The trait provides iterator-like interface for reading records from an
Avro file,
- // deserializing and returning them as internal rows.
- trait RowReader {
- protected val fileReader: FileReader[GenericRecord]
- protected val deserializer: AvroDeserializer
- protected val stopPosition: Long
-
- private[this] var completed = false
- private[this] var currentRow: Option[InternalRow] = None
-
- def hasNextRow: Boolean = {
- while (!completed && currentRow.isEmpty) {
- val r = fileReader.hasNext && !fileReader.pastSync(stopPosition)
- if (!r) {
- fileReader.close()
- completed = true
- currentRow = None
- } else {
- val record = fileReader.next()
- // the row must be deserialized in hasNextRow, because
AvroDeserializer#deserialize
- // potentially filters rows
- currentRow =
deserializer.deserialize(record).asInstanceOf[Option[InternalRow]]
- }
- }
- currentRow.isDefined
- }
-
- def nextRow: InternalRow = {
- if (currentRow.isEmpty) {
- hasNextRow
- }
- val returnRow = currentRow
- currentRow = None // free up hasNextRow to consume more Avro records, if
not exhausted
- returnRow.getOrElse {
- throw new NoSuchElementException("next on empty iterator")
- }
- }
- }
-
- /** Wrapper for a pair of matched fields, one Catalyst and one corresponding
Avro field. */
- private[sql] case class AvroMatchedField(
- catalystField: StructField,
- catalystPosition: Int,
- avroField: Schema.Field)
-
- /**
- * Helper class to perform field lookup/matching on Avro schemas.
- *
- * This will match `avroSchema` against `catalystSchema`, attempting to find
a matching field in
- * the Avro schema for each field in the Catalyst schema and vice-versa,
respecting settings for
- * case sensitivity. The match results can be accessed using the getter
methods.
- *
- * @param avroSchema The schema in which to search for fields. Must be of
type RECORD.
- * @param catalystSchema The Catalyst schema to use for matching.
- * @param avroPath The seq of parent field names leading to `avroSchema`.
- * @param catalystPath The seq of parent field names leading to
`catalystSchema`.
- * @param positionalFieldMatch If true, perform field matching in a
positional fashion
- * (structural comparison between schemas,
ignoring names);
- * otherwise, perform field matching using field
names.
- */
- class AvroSchemaHelper(
- avroSchema: Schema,
- catalystSchema: StructType,
- avroPath: Seq[String],
- catalystPath: Seq[String],
- positionalFieldMatch: Boolean) {
- if (avroSchema.getType != Schema.Type.RECORD) {
- throw new IncompatibleSchemaException(
- s"Attempting to treat ${avroSchema.getName} as a RECORD, but it was:
${avroSchema.getType}")
- }
-
- private[this] val avroFieldArray = avroSchema.getFields.asScala.toArray
- private[this] val fieldMap = avroSchema.getFields.asScala
- .groupBy(_.name.toLowerCase(Locale.ROOT))
- .mapValues(_.toSeq) // toSeq needed for scala 2.13
-
- /** The fields which have matching equivalents in both Avro and Catalyst
schemas. */
- val matchedFields: Seq[AvroMatchedField] =
catalystSchema.zipWithIndex.flatMap {
- case (sqlField, sqlPos) =>
- getAvroField(sqlField.name, sqlPos).map(AvroMatchedField(sqlField,
sqlPos, _))
- }
-
- /**
- * Validate that there are no Catalyst fields which don't have a matching
Avro field, throwing
- * [[IncompatibleSchemaException]] if such extra fields are found. If
`ignoreNullable` is false,
- * consider nullable Catalyst fields to be eligible to be an extra field;
otherwise,
- * ignore nullable Catalyst fields when checking for extras.
- */
- def validateNoExtraCatalystFields(ignoreNullable: Boolean): Unit =
- catalystSchema.zipWithIndex.foreach { case (sqlField, sqlPos) =>
- if (getAvroField(sqlField.name, sqlPos).isEmpty &&
- (!ignoreNullable || !sqlField.nullable)) {
- if (positionalFieldMatch) {
- throw new IncompatibleSchemaException("Cannot find field at
position " +
- s"$sqlPos of ${toFieldStr(avroPath)} from Avro schema (using
positional matching)")
- } else {
- throw new IncompatibleSchemaException(
- s"Cannot find ${toFieldStr(catalystPath :+ sqlField.name)} in
Avro schema")
- }
- }
- }
-
- /**
- * Validate that there are no Avro fields which don't have a matching
Catalyst field, throwing
- * [[IncompatibleSchemaException]] if such extra fields are found. Only
required (non-nullable)
- * fields are checked; nullable fields are ignored.
- */
- def validateNoExtraRequiredAvroFields(): Unit = {
- val extraFields = avroFieldArray.toSet -- matchedFields.map(_.avroField)
- extraFields.filterNot(isNullable).foreach { extraField =>
- if (positionalFieldMatch) {
- throw new IncompatibleSchemaException(s"Found field
'${extraField.name()}' at position " +
- s"${extraField.pos()} of ${toFieldStr(avroPath)} from Avro schema
but there is no " +
- s"match in the SQL schema at ${toFieldStr(catalystPath)} (using
positional matching)")
- } else {
- throw new IncompatibleSchemaException(
- s"Found ${toFieldStr(avroPath :+ extraField.name())} in Avro
schema but there is no " +
- "match in the SQL schema")
- }
- }
- }
-
- /**
- * Extract a single field from the contained avro schema which has the
desired field name,
- * performing the matching with proper case sensitivity according to
SQLConf.resolver.
- *
- * @param name The name of the field to search for.
- * @return `Some(match)` if a matching Avro field is found, otherwise
`None`.
- */
- private[avro] def getFieldByName(name: String): Option[Schema.Field] = {
-
- // get candidates, ignoring case of field name
- val candidates = fieldMap.getOrElse(name.toLowerCase(Locale.ROOT),
Seq.empty)
-
- // search candidates, taking into account case sensitivity settings
- candidates.filter(f => SQLConf.get.resolver(f.name(), name)) match {
- case Seq(avroField) => Some(avroField)
- case Seq() => None
- case matches => throw new IncompatibleSchemaException(s"Searching for
'$name' in Avro " +
- s"schema at ${toFieldStr(avroPath)} gave ${matches.size} matches.
Candidates: " +
- matches.map(_.name()).mkString("[", ", ", "]")
- )
- }
- }
-
- /** Get the Avro field corresponding to the provided Catalyst field
name/position, if any. */
- def getAvroField(fieldName: String, catalystPos: Int):
Option[Schema.Field] = {
- if (positionalFieldMatch) {
- avroFieldArray.lift(catalystPos)
- } else {
- getFieldByName(fieldName)
- }
- }
- }
-
- /**
- * Convert a sequence of hierarchical field names (like `Seq(foo, bar)`)
into a human-readable
- * string representing the field, like "field 'foo.bar'". If `names` is
empty, the string
- * "top-level record" is returned.
- */
- private[avro] def toFieldStr(names: Seq[String]): String = names match {
- case Seq() => "top-level record"
- case n => s"field '${n.mkString(".")}'"
- }
-
- /** Return true iff `avroField` is nullable, i.e. `UNION` type and has
`NULL` as an option. */
- private[avro] def isNullable(avroField: Schema.Field): Boolean =
- avroField.schema().getType == Schema.Type.UNION &&
- avroField.schema().getTypes.asScala.exists(_.getType == Schema.Type.NULL)
-}
diff --git
a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
deleted file mode 100644
index 8aae6b442f8a..000000000000
---
a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
+++ /dev/null
@@ -1,227 +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.avro.Schema
-import org.apache.avro.file. FileReader
-import org.apache.avro.generic.GenericRecord
-import org.apache.spark.internal.Logging
-import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types._
-
-import java.util.Locale
-
-import scala.collection.JavaConverters._
-
-/**
- * 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] object AvroUtils extends Logging {
-
- def supportsDataType(dataType: DataType): Boolean = dataType match {
- case _: AtomicType => true
-
- case st: StructType => st.forall { f => supportsDataType(f.dataType) }
-
- case ArrayType(elementType, _) => supportsDataType(elementType)
-
- case MapType(keyType, valueType, _) =>
- supportsDataType(keyType) && supportsDataType(valueType)
-
- case udt: UserDefinedType[_] => supportsDataType(udt.sqlType)
-
- case _: NullType => true
-
- case _ => false
- }
-
- // The trait provides iterator-like interface for reading records from an
Avro file,
- // deserializing and returning them as internal rows.
- trait RowReader {
- protected val fileReader: FileReader[GenericRecord]
- protected val deserializer: AvroDeserializer
- protected val stopPosition: Long
-
- private[this] var completed = false
- private[this] var currentRow: Option[InternalRow] = None
-
- def hasNextRow: Boolean = {
- while (!completed && currentRow.isEmpty) {
- val r = fileReader.hasNext && !fileReader.pastSync(stopPosition)
- if (!r) {
- fileReader.close()
- completed = true
- currentRow = None
- } else {
- val record = fileReader.next()
- // the row must be deserialized in hasNextRow, because
AvroDeserializer#deserialize
- // potentially filters rows
- currentRow =
deserializer.deserialize(record).asInstanceOf[Option[InternalRow]]
- }
- }
- currentRow.isDefined
- }
-
- def nextRow: InternalRow = {
- if (currentRow.isEmpty) {
- hasNextRow
- }
- val returnRow = currentRow
- currentRow = None // free up hasNextRow to consume more Avro records, if
not exhausted
- returnRow.getOrElse {
- throw new NoSuchElementException("next on empty iterator")
- }
- }
- }
-
- /** Wrapper for a pair of matched fields, one Catalyst and one corresponding
Avro field. */
- private[sql] case class AvroMatchedField(
- catalystField: StructField,
- catalystPosition: Int,
- avroField: Schema.Field)
-
- /**
- * Helper class to perform field lookup/matching on Avro schemas.
- *
- * This will match `avroSchema` against `catalystSchema`, attempting to find
a matching field in
- * the Avro schema for each field in the Catalyst schema and vice-versa,
respecting settings for
- * case sensitivity. The match results can be accessed using the getter
methods.
- *
- * @param avroSchema The schema in which to search for fields. Must be of
type RECORD.
- * @param catalystSchema The Catalyst schema to use for matching.
- * @param avroPath The seq of parent field names leading to `avroSchema`.
- * @param catalystPath The seq of parent field names leading to
`catalystSchema`.
- * @param positionalFieldMatch If true, perform field matching in a
positional fashion
- * (structural comparison between schemas,
ignoring names);
- * otherwise, perform field matching using field
names.
- */
- class AvroSchemaHelper(
- avroSchema: Schema,
- catalystSchema: StructType,
- avroPath: Seq[String],
- catalystPath: Seq[String],
- positionalFieldMatch: Boolean) {
- if (avroSchema.getType != Schema.Type.RECORD) {
- throw new IncompatibleSchemaException(
- s"Attempting to treat ${avroSchema.getName} as a RECORD, but it was:
${avroSchema.getType}")
- }
-
- private[this] val avroFieldArray = avroSchema.getFields.asScala.toArray
- private[this] val fieldMap = avroSchema.getFields.asScala
- .groupBy(_.name.toLowerCase(Locale.ROOT))
- .mapValues(_.toSeq) // toSeq needed for scala 2.13
-
- /** The fields which have matching equivalents in both Avro and Catalyst
schemas. */
- val matchedFields: Seq[AvroMatchedField] =
catalystSchema.zipWithIndex.flatMap {
- case (sqlField, sqlPos) =>
- getAvroField(sqlField.name, sqlPos).map(AvroMatchedField(sqlField,
sqlPos, _))
- }
-
- /**
- * Validate that there are no Catalyst fields which don't have a matching
Avro field, throwing
- * [[IncompatibleSchemaException]] if such extra fields are found. If
`ignoreNullable` is false,
- * consider nullable Catalyst fields to be eligible to be an extra field;
otherwise,
- * ignore nullable Catalyst fields when checking for extras.
- */
- def validateNoExtraCatalystFields(ignoreNullable: Boolean): Unit =
- catalystSchema.zipWithIndex.foreach { case (sqlField, sqlPos) =>
- if (getAvroField(sqlField.name, sqlPos).isEmpty &&
- (!ignoreNullable || !sqlField.nullable)) {
- if (positionalFieldMatch) {
- throw new IncompatibleSchemaException("Cannot find field at
position " +
- s"$sqlPos of ${toFieldStr(avroPath)} from Avro schema (using
positional matching)")
- } else {
- throw new IncompatibleSchemaException(
- s"Cannot find ${toFieldStr(catalystPath :+ sqlField.name)} in
Avro schema")
- }
- }
- }
-
- /**
- * Validate that there are no Avro fields which don't have a matching
Catalyst field, throwing
- * [[IncompatibleSchemaException]] if such extra fields are found. Only
required (non-nullable)
- * fields are checked; nullable fields are ignored.
- */
- def validateNoExtraRequiredAvroFields(): Unit = {
- val extraFields = avroFieldArray.toSet -- matchedFields.map(_.avroField)
- extraFields.filterNot(isNullable).foreach { extraField =>
- if (positionalFieldMatch) {
- throw new IncompatibleSchemaException(s"Found field
'${extraField.name()}' at position " +
- s"${extraField.pos()} of ${toFieldStr(avroPath)} from Avro schema
but there is no " +
- s"match in the SQL schema at ${toFieldStr(catalystPath)} (using
positional matching)")
- } else {
- throw new IncompatibleSchemaException(
- s"Found ${toFieldStr(avroPath :+ extraField.name())} in Avro
schema but there is no " +
- "match in the SQL schema")
- }
- }
- }
-
- /**
- * Extract a single field from the contained avro schema which has the
desired field name,
- * performing the matching with proper case sensitivity according to
SQLConf.resolver.
- *
- * @param name The name of the field to search for.
- * @return `Some(match)` if a matching Avro field is found, otherwise
`None`.
- */
- private[avro] def getFieldByName(name: String): Option[Schema.Field] = {
-
- // get candidates, ignoring case of field name
- val candidates = fieldMap.getOrElse(name.toLowerCase(Locale.ROOT),
Seq.empty)
-
- // search candidates, taking into account case sensitivity settings
- candidates.filter(f => SQLConf.get.resolver(f.name(), name)) match {
- case Seq(avroField) => Some(avroField)
- case Seq() => None
- case matches => throw new IncompatibleSchemaException(s"Searching for
'$name' in Avro " +
- s"schema at ${toFieldStr(avroPath)} gave ${matches.size} matches.
Candidates: " +
- matches.map(_.name()).mkString("[", ", ", "]")
- )
- }
- }
-
- /** Get the Avro field corresponding to the provided Catalyst field
name/position, if any. */
- def getAvroField(fieldName: String, catalystPos: Int):
Option[Schema.Field] = {
- if (positionalFieldMatch) {
- avroFieldArray.lift(catalystPos)
- } else {
- getFieldByName(fieldName)
- }
- }
- }
-
- /**
- * Convert a sequence of hierarchical field names (like `Seq(foo, bar)`)
into a human-readable
- * string representing the field, like "field 'foo.bar'". If `names` is
empty, the string
- * "top-level record" is returned.
- */
- private[avro] def toFieldStr(names: Seq[String]): String = names match {
- case Seq() => "top-level record"
- case n => s"field '${n.mkString(".")}'"
- }
-
- /** Return true iff `avroField` is nullable, i.e. `UNION` type and has
`NULL` as an option. */
- private[avro] def isNullable(avroField: Schema.Field): Boolean =
- avroField.schema().getType == Schema.Type.UNION &&
- avroField.schema().getTypes.asScala.exists(_.getType == Schema.Type.NULL)
-}
diff --git
a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
deleted file mode 100644
index 8aae6b442f8a..000000000000
---
a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
+++ /dev/null
@@ -1,227 +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.avro.Schema
-import org.apache.avro.file. FileReader
-import org.apache.avro.generic.GenericRecord
-import org.apache.spark.internal.Logging
-import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types._
-
-import java.util.Locale
-
-import scala.collection.JavaConverters._
-
-/**
- * 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] object AvroUtils extends Logging {
-
- def supportsDataType(dataType: DataType): Boolean = dataType match {
- case _: AtomicType => true
-
- case st: StructType => st.forall { f => supportsDataType(f.dataType) }
-
- case ArrayType(elementType, _) => supportsDataType(elementType)
-
- case MapType(keyType, valueType, _) =>
- supportsDataType(keyType) && supportsDataType(valueType)
-
- case udt: UserDefinedType[_] => supportsDataType(udt.sqlType)
-
- case _: NullType => true
-
- case _ => false
- }
-
- // The trait provides iterator-like interface for reading records from an
Avro file,
- // deserializing and returning them as internal rows.
- trait RowReader {
- protected val fileReader: FileReader[GenericRecord]
- protected val deserializer: AvroDeserializer
- protected val stopPosition: Long
-
- private[this] var completed = false
- private[this] var currentRow: Option[InternalRow] = None
-
- def hasNextRow: Boolean = {
- while (!completed && currentRow.isEmpty) {
- val r = fileReader.hasNext && !fileReader.pastSync(stopPosition)
- if (!r) {
- fileReader.close()
- completed = true
- currentRow = None
- } else {
- val record = fileReader.next()
- // the row must be deserialized in hasNextRow, because
AvroDeserializer#deserialize
- // potentially filters rows
- currentRow =
deserializer.deserialize(record).asInstanceOf[Option[InternalRow]]
- }
- }
- currentRow.isDefined
- }
-
- def nextRow: InternalRow = {
- if (currentRow.isEmpty) {
- hasNextRow
- }
- val returnRow = currentRow
- currentRow = None // free up hasNextRow to consume more Avro records, if
not exhausted
- returnRow.getOrElse {
- throw new NoSuchElementException("next on empty iterator")
- }
- }
- }
-
- /** Wrapper for a pair of matched fields, one Catalyst and one corresponding
Avro field. */
- private[sql] case class AvroMatchedField(
- catalystField: StructField,
- catalystPosition: Int,
- avroField: Schema.Field)
-
- /**
- * Helper class to perform field lookup/matching on Avro schemas.
- *
- * This will match `avroSchema` against `catalystSchema`, attempting to find
a matching field in
- * the Avro schema for each field in the Catalyst schema and vice-versa,
respecting settings for
- * case sensitivity. The match results can be accessed using the getter
methods.
- *
- * @param avroSchema The schema in which to search for fields. Must be of
type RECORD.
- * @param catalystSchema The Catalyst schema to use for matching.
- * @param avroPath The seq of parent field names leading to `avroSchema`.
- * @param catalystPath The seq of parent field names leading to
`catalystSchema`.
- * @param positionalFieldMatch If true, perform field matching in a
positional fashion
- * (structural comparison between schemas,
ignoring names);
- * otherwise, perform field matching using field
names.
- */
- class AvroSchemaHelper(
- avroSchema: Schema,
- catalystSchema: StructType,
- avroPath: Seq[String],
- catalystPath: Seq[String],
- positionalFieldMatch: Boolean) {
- if (avroSchema.getType != Schema.Type.RECORD) {
- throw new IncompatibleSchemaException(
- s"Attempting to treat ${avroSchema.getName} as a RECORD, but it was:
${avroSchema.getType}")
- }
-
- private[this] val avroFieldArray = avroSchema.getFields.asScala.toArray
- private[this] val fieldMap = avroSchema.getFields.asScala
- .groupBy(_.name.toLowerCase(Locale.ROOT))
- .mapValues(_.toSeq) // toSeq needed for scala 2.13
-
- /** The fields which have matching equivalents in both Avro and Catalyst
schemas. */
- val matchedFields: Seq[AvroMatchedField] =
catalystSchema.zipWithIndex.flatMap {
- case (sqlField, sqlPos) =>
- getAvroField(sqlField.name, sqlPos).map(AvroMatchedField(sqlField,
sqlPos, _))
- }
-
- /**
- * Validate that there are no Catalyst fields which don't have a matching
Avro field, throwing
- * [[IncompatibleSchemaException]] if such extra fields are found. If
`ignoreNullable` is false,
- * consider nullable Catalyst fields to be eligible to be an extra field;
otherwise,
- * ignore nullable Catalyst fields when checking for extras.
- */
- def validateNoExtraCatalystFields(ignoreNullable: Boolean): Unit =
- catalystSchema.zipWithIndex.foreach { case (sqlField, sqlPos) =>
- if (getAvroField(sqlField.name, sqlPos).isEmpty &&
- (!ignoreNullable || !sqlField.nullable)) {
- if (positionalFieldMatch) {
- throw new IncompatibleSchemaException("Cannot find field at
position " +
- s"$sqlPos of ${toFieldStr(avroPath)} from Avro schema (using
positional matching)")
- } else {
- throw new IncompatibleSchemaException(
- s"Cannot find ${toFieldStr(catalystPath :+ sqlField.name)} in
Avro schema")
- }
- }
- }
-
- /**
- * Validate that there are no Avro fields which don't have a matching
Catalyst field, throwing
- * [[IncompatibleSchemaException]] if such extra fields are found. Only
required (non-nullable)
- * fields are checked; nullable fields are ignored.
- */
- def validateNoExtraRequiredAvroFields(): Unit = {
- val extraFields = avroFieldArray.toSet -- matchedFields.map(_.avroField)
- extraFields.filterNot(isNullable).foreach { extraField =>
- if (positionalFieldMatch) {
- throw new IncompatibleSchemaException(s"Found field
'${extraField.name()}' at position " +
- s"${extraField.pos()} of ${toFieldStr(avroPath)} from Avro schema
but there is no " +
- s"match in the SQL schema at ${toFieldStr(catalystPath)} (using
positional matching)")
- } else {
- throw new IncompatibleSchemaException(
- s"Found ${toFieldStr(avroPath :+ extraField.name())} in Avro
schema but there is no " +
- "match in the SQL schema")
- }
- }
- }
-
- /**
- * Extract a single field from the contained avro schema which has the
desired field name,
- * performing the matching with proper case sensitivity according to
SQLConf.resolver.
- *
- * @param name The name of the field to search for.
- * @return `Some(match)` if a matching Avro field is found, otherwise
`None`.
- */
- private[avro] def getFieldByName(name: String): Option[Schema.Field] = {
-
- // get candidates, ignoring case of field name
- val candidates = fieldMap.getOrElse(name.toLowerCase(Locale.ROOT),
Seq.empty)
-
- // search candidates, taking into account case sensitivity settings
- candidates.filter(f => SQLConf.get.resolver(f.name(), name)) match {
- case Seq(avroField) => Some(avroField)
- case Seq() => None
- case matches => throw new IncompatibleSchemaException(s"Searching for
'$name' in Avro " +
- s"schema at ${toFieldStr(avroPath)} gave ${matches.size} matches.
Candidates: " +
- matches.map(_.name()).mkString("[", ", ", "]")
- )
- }
- }
-
- /** Get the Avro field corresponding to the provided Catalyst field
name/position, if any. */
- def getAvroField(fieldName: String, catalystPos: Int):
Option[Schema.Field] = {
- if (positionalFieldMatch) {
- avroFieldArray.lift(catalystPos)
- } else {
- getFieldByName(fieldName)
- }
- }
- }
-
- /**
- * Convert a sequence of hierarchical field names (like `Seq(foo, bar)`)
into a human-readable
- * string representing the field, like "field 'foo.bar'". If `names` is
empty, the string
- * "top-level record" is returned.
- */
- private[avro] def toFieldStr(names: Seq[String]): String = names match {
- case Seq() => "top-level record"
- case n => s"field '${n.mkString(".")}'"
- }
-
- /** Return true iff `avroField` is nullable, i.e. `UNION` type and has
`NULL` as an option. */
- private[avro] def isNullable(avroField: Schema.Field): Boolean =
- avroField.schema().getType == Schema.Type.UNION &&
- avroField.schema().getTypes.asScala.exists(_.getType == Schema.Type.NULL)
-}
diff --git
a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
deleted file mode 100644
index 8aae6b442f8a..000000000000
---
a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
+++ /dev/null
@@ -1,227 +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.avro.Schema
-import org.apache.avro.file. FileReader
-import org.apache.avro.generic.GenericRecord
-import org.apache.spark.internal.Logging
-import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types._
-
-import java.util.Locale
-
-import scala.collection.JavaConverters._
-
-/**
- * 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] object AvroUtils extends Logging {
-
- def supportsDataType(dataType: DataType): Boolean = dataType match {
- case _: AtomicType => true
-
- case st: StructType => st.forall { f => supportsDataType(f.dataType) }
-
- case ArrayType(elementType, _) => supportsDataType(elementType)
-
- case MapType(keyType, valueType, _) =>
- supportsDataType(keyType) && supportsDataType(valueType)
-
- case udt: UserDefinedType[_] => supportsDataType(udt.sqlType)
-
- case _: NullType => true
-
- case _ => false
- }
-
- // The trait provides iterator-like interface for reading records from an
Avro file,
- // deserializing and returning them as internal rows.
- trait RowReader {
- protected val fileReader: FileReader[GenericRecord]
- protected val deserializer: AvroDeserializer
- protected val stopPosition: Long
-
- private[this] var completed = false
- private[this] var currentRow: Option[InternalRow] = None
-
- def hasNextRow: Boolean = {
- while (!completed && currentRow.isEmpty) {
- val r = fileReader.hasNext && !fileReader.pastSync(stopPosition)
- if (!r) {
- fileReader.close()
- completed = true
- currentRow = None
- } else {
- val record = fileReader.next()
- // the row must be deserialized in hasNextRow, because
AvroDeserializer#deserialize
- // potentially filters rows
- currentRow =
deserializer.deserialize(record).asInstanceOf[Option[InternalRow]]
- }
- }
- currentRow.isDefined
- }
-
- def nextRow: InternalRow = {
- if (currentRow.isEmpty) {
- hasNextRow
- }
- val returnRow = currentRow
- currentRow = None // free up hasNextRow to consume more Avro records, if
not exhausted
- returnRow.getOrElse {
- throw new NoSuchElementException("next on empty iterator")
- }
- }
- }
-
- /** Wrapper for a pair of matched fields, one Catalyst and one corresponding
Avro field. */
- private[sql] case class AvroMatchedField(
- catalystField: StructField,
- catalystPosition: Int,
- avroField: Schema.Field)
-
- /**
- * Helper class to perform field lookup/matching on Avro schemas.
- *
- * This will match `avroSchema` against `catalystSchema`, attempting to find
a matching field in
- * the Avro schema for each field in the Catalyst schema and vice-versa,
respecting settings for
- * case sensitivity. The match results can be accessed using the getter
methods.
- *
- * @param avroSchema The schema in which to search for fields. Must be of
type RECORD.
- * @param catalystSchema The Catalyst schema to use for matching.
- * @param avroPath The seq of parent field names leading to `avroSchema`.
- * @param catalystPath The seq of parent field names leading to
`catalystSchema`.
- * @param positionalFieldMatch If true, perform field matching in a
positional fashion
- * (structural comparison between schemas,
ignoring names);
- * otherwise, perform field matching using field
names.
- */
- class AvroSchemaHelper(
- avroSchema: Schema,
- catalystSchema: StructType,
- avroPath: Seq[String],
- catalystPath: Seq[String],
- positionalFieldMatch: Boolean) {
- if (avroSchema.getType != Schema.Type.RECORD) {
- throw new IncompatibleSchemaException(
- s"Attempting to treat ${avroSchema.getName} as a RECORD, but it was:
${avroSchema.getType}")
- }
-
- private[this] val avroFieldArray = avroSchema.getFields.asScala.toArray
- private[this] val fieldMap = avroSchema.getFields.asScala
- .groupBy(_.name.toLowerCase(Locale.ROOT))
- .mapValues(_.toSeq) // toSeq needed for scala 2.13
-
- /** The fields which have matching equivalents in both Avro and Catalyst
schemas. */
- val matchedFields: Seq[AvroMatchedField] =
catalystSchema.zipWithIndex.flatMap {
- case (sqlField, sqlPos) =>
- getAvroField(sqlField.name, sqlPos).map(AvroMatchedField(sqlField,
sqlPos, _))
- }
-
- /**
- * Validate that there are no Catalyst fields which don't have a matching
Avro field, throwing
- * [[IncompatibleSchemaException]] if such extra fields are found. If
`ignoreNullable` is false,
- * consider nullable Catalyst fields to be eligible to be an extra field;
otherwise,
- * ignore nullable Catalyst fields when checking for extras.
- */
- def validateNoExtraCatalystFields(ignoreNullable: Boolean): Unit =
- catalystSchema.zipWithIndex.foreach { case (sqlField, sqlPos) =>
- if (getAvroField(sqlField.name, sqlPos).isEmpty &&
- (!ignoreNullable || !sqlField.nullable)) {
- if (positionalFieldMatch) {
- throw new IncompatibleSchemaException("Cannot find field at
position " +
- s"$sqlPos of ${toFieldStr(avroPath)} from Avro schema (using
positional matching)")
- } else {
- throw new IncompatibleSchemaException(
- s"Cannot find ${toFieldStr(catalystPath :+ sqlField.name)} in
Avro schema")
- }
- }
- }
-
- /**
- * Validate that there are no Avro fields which don't have a matching
Catalyst field, throwing
- * [[IncompatibleSchemaException]] if such extra fields are found. Only
required (non-nullable)
- * fields are checked; nullable fields are ignored.
- */
- def validateNoExtraRequiredAvroFields(): Unit = {
- val extraFields = avroFieldArray.toSet -- matchedFields.map(_.avroField)
- extraFields.filterNot(isNullable).foreach { extraField =>
- if (positionalFieldMatch) {
- throw new IncompatibleSchemaException(s"Found field
'${extraField.name()}' at position " +
- s"${extraField.pos()} of ${toFieldStr(avroPath)} from Avro schema
but there is no " +
- s"match in the SQL schema at ${toFieldStr(catalystPath)} (using
positional matching)")
- } else {
- throw new IncompatibleSchemaException(
- s"Found ${toFieldStr(avroPath :+ extraField.name())} in Avro
schema but there is no " +
- "match in the SQL schema")
- }
- }
- }
-
- /**
- * Extract a single field from the contained avro schema which has the
desired field name,
- * performing the matching with proper case sensitivity according to
SQLConf.resolver.
- *
- * @param name The name of the field to search for.
- * @return `Some(match)` if a matching Avro field is found, otherwise
`None`.
- */
- private[avro] def getFieldByName(name: String): Option[Schema.Field] = {
-
- // get candidates, ignoring case of field name
- val candidates = fieldMap.getOrElse(name.toLowerCase(Locale.ROOT),
Seq.empty)
-
- // search candidates, taking into account case sensitivity settings
- candidates.filter(f => SQLConf.get.resolver(f.name(), name)) match {
- case Seq(avroField) => Some(avroField)
- case Seq() => None
- case matches => throw new IncompatibleSchemaException(s"Searching for
'$name' in Avro " +
- s"schema at ${toFieldStr(avroPath)} gave ${matches.size} matches.
Candidates: " +
- matches.map(_.name()).mkString("[", ", ", "]")
- )
- }
- }
-
- /** Get the Avro field corresponding to the provided Catalyst field
name/position, if any. */
- def getAvroField(fieldName: String, catalystPos: Int):
Option[Schema.Field] = {
- if (positionalFieldMatch) {
- avroFieldArray.lift(catalystPos)
- } else {
- getFieldByName(fieldName)
- }
- }
- }
-
- /**
- * Convert a sequence of hierarchical field names (like `Seq(foo, bar)`)
into a human-readable
- * string representing the field, like "field 'foo.bar'". If `names` is
empty, the string
- * "top-level record" is returned.
- */
- private[avro] def toFieldStr(names: Seq[String]): String = names match {
- case Seq() => "top-level record"
- case n => s"field '${n.mkString(".")}'"
- }
-
- /** Return true iff `avroField` is nullable, i.e. `UNION` type and has
`NULL` as an option. */
- private[avro] def isNullable(avroField: Schema.Field): Boolean =
- avroField.schema().getType == Schema.Type.UNION &&
- avroField.schema().getTypes.asScala.exists(_.getType == Schema.Type.NULL)
-}
diff --git
a/hudi-spark-datasource/hudi-spark4.2.x/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
b/hudi-spark-datasource/hudi-spark4.2.x/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
deleted file mode 100644
index 8aae6b442f8a..000000000000
---
a/hudi-spark-datasource/hudi-spark4.2.x/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
+++ /dev/null
@@ -1,227 +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.avro.Schema
-import org.apache.avro.file. FileReader
-import org.apache.avro.generic.GenericRecord
-import org.apache.spark.internal.Logging
-import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types._
-
-import java.util.Locale
-
-import scala.collection.JavaConverters._
-
-/**
- * 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] object AvroUtils extends Logging {
-
- def supportsDataType(dataType: DataType): Boolean = dataType match {
- case _: AtomicType => true
-
- case st: StructType => st.forall { f => supportsDataType(f.dataType) }
-
- case ArrayType(elementType, _) => supportsDataType(elementType)
-
- case MapType(keyType, valueType, _) =>
- supportsDataType(keyType) && supportsDataType(valueType)
-
- case udt: UserDefinedType[_] => supportsDataType(udt.sqlType)
-
- case _: NullType => true
-
- case _ => false
- }
-
- // The trait provides iterator-like interface for reading records from an
Avro file,
- // deserializing and returning them as internal rows.
- trait RowReader {
- protected val fileReader: FileReader[GenericRecord]
- protected val deserializer: AvroDeserializer
- protected val stopPosition: Long
-
- private[this] var completed = false
- private[this] var currentRow: Option[InternalRow] = None
-
- def hasNextRow: Boolean = {
- while (!completed && currentRow.isEmpty) {
- val r = fileReader.hasNext && !fileReader.pastSync(stopPosition)
- if (!r) {
- fileReader.close()
- completed = true
- currentRow = None
- } else {
- val record = fileReader.next()
- // the row must be deserialized in hasNextRow, because
AvroDeserializer#deserialize
- // potentially filters rows
- currentRow =
deserializer.deserialize(record).asInstanceOf[Option[InternalRow]]
- }
- }
- currentRow.isDefined
- }
-
- def nextRow: InternalRow = {
- if (currentRow.isEmpty) {
- hasNextRow
- }
- val returnRow = currentRow
- currentRow = None // free up hasNextRow to consume more Avro records, if
not exhausted
- returnRow.getOrElse {
- throw new NoSuchElementException("next on empty iterator")
- }
- }
- }
-
- /** Wrapper for a pair of matched fields, one Catalyst and one corresponding
Avro field. */
- private[sql] case class AvroMatchedField(
- catalystField: StructField,
- catalystPosition: Int,
- avroField: Schema.Field)
-
- /**
- * Helper class to perform field lookup/matching on Avro schemas.
- *
- * This will match `avroSchema` against `catalystSchema`, attempting to find
a matching field in
- * the Avro schema for each field in the Catalyst schema and vice-versa,
respecting settings for
- * case sensitivity. The match results can be accessed using the getter
methods.
- *
- * @param avroSchema The schema in which to search for fields. Must be of
type RECORD.
- * @param catalystSchema The Catalyst schema to use for matching.
- * @param avroPath The seq of parent field names leading to `avroSchema`.
- * @param catalystPath The seq of parent field names leading to
`catalystSchema`.
- * @param positionalFieldMatch If true, perform field matching in a
positional fashion
- * (structural comparison between schemas,
ignoring names);
- * otherwise, perform field matching using field
names.
- */
- class AvroSchemaHelper(
- avroSchema: Schema,
- catalystSchema: StructType,
- avroPath: Seq[String],
- catalystPath: Seq[String],
- positionalFieldMatch: Boolean) {
- if (avroSchema.getType != Schema.Type.RECORD) {
- throw new IncompatibleSchemaException(
- s"Attempting to treat ${avroSchema.getName} as a RECORD, but it was:
${avroSchema.getType}")
- }
-
- private[this] val avroFieldArray = avroSchema.getFields.asScala.toArray
- private[this] val fieldMap = avroSchema.getFields.asScala
- .groupBy(_.name.toLowerCase(Locale.ROOT))
- .mapValues(_.toSeq) // toSeq needed for scala 2.13
-
- /** The fields which have matching equivalents in both Avro and Catalyst
schemas. */
- val matchedFields: Seq[AvroMatchedField] =
catalystSchema.zipWithIndex.flatMap {
- case (sqlField, sqlPos) =>
- getAvroField(sqlField.name, sqlPos).map(AvroMatchedField(sqlField,
sqlPos, _))
- }
-
- /**
- * Validate that there are no Catalyst fields which don't have a matching
Avro field, throwing
- * [[IncompatibleSchemaException]] if such extra fields are found. If
`ignoreNullable` is false,
- * consider nullable Catalyst fields to be eligible to be an extra field;
otherwise,
- * ignore nullable Catalyst fields when checking for extras.
- */
- def validateNoExtraCatalystFields(ignoreNullable: Boolean): Unit =
- catalystSchema.zipWithIndex.foreach { case (sqlField, sqlPos) =>
- if (getAvroField(sqlField.name, sqlPos).isEmpty &&
- (!ignoreNullable || !sqlField.nullable)) {
- if (positionalFieldMatch) {
- throw new IncompatibleSchemaException("Cannot find field at
position " +
- s"$sqlPos of ${toFieldStr(avroPath)} from Avro schema (using
positional matching)")
- } else {
- throw new IncompatibleSchemaException(
- s"Cannot find ${toFieldStr(catalystPath :+ sqlField.name)} in
Avro schema")
- }
- }
- }
-
- /**
- * Validate that there are no Avro fields which don't have a matching
Catalyst field, throwing
- * [[IncompatibleSchemaException]] if such extra fields are found. Only
required (non-nullable)
- * fields are checked; nullable fields are ignored.
- */
- def validateNoExtraRequiredAvroFields(): Unit = {
- val extraFields = avroFieldArray.toSet -- matchedFields.map(_.avroField)
- extraFields.filterNot(isNullable).foreach { extraField =>
- if (positionalFieldMatch) {
- throw new IncompatibleSchemaException(s"Found field
'${extraField.name()}' at position " +
- s"${extraField.pos()} of ${toFieldStr(avroPath)} from Avro schema
but there is no " +
- s"match in the SQL schema at ${toFieldStr(catalystPath)} (using
positional matching)")
- } else {
- throw new IncompatibleSchemaException(
- s"Found ${toFieldStr(avroPath :+ extraField.name())} in Avro
schema but there is no " +
- "match in the SQL schema")
- }
- }
- }
-
- /**
- * Extract a single field from the contained avro schema which has the
desired field name,
- * performing the matching with proper case sensitivity according to
SQLConf.resolver.
- *
- * @param name The name of the field to search for.
- * @return `Some(match)` if a matching Avro field is found, otherwise
`None`.
- */
- private[avro] def getFieldByName(name: String): Option[Schema.Field] = {
-
- // get candidates, ignoring case of field name
- val candidates = fieldMap.getOrElse(name.toLowerCase(Locale.ROOT),
Seq.empty)
-
- // search candidates, taking into account case sensitivity settings
- candidates.filter(f => SQLConf.get.resolver(f.name(), name)) match {
- case Seq(avroField) => Some(avroField)
- case Seq() => None
- case matches => throw new IncompatibleSchemaException(s"Searching for
'$name' in Avro " +
- s"schema at ${toFieldStr(avroPath)} gave ${matches.size} matches.
Candidates: " +
- matches.map(_.name()).mkString("[", ", ", "]")
- )
- }
- }
-
- /** Get the Avro field corresponding to the provided Catalyst field
name/position, if any. */
- def getAvroField(fieldName: String, catalystPos: Int):
Option[Schema.Field] = {
- if (positionalFieldMatch) {
- avroFieldArray.lift(catalystPos)
- } else {
- getFieldByName(fieldName)
- }
- }
- }
-
- /**
- * Convert a sequence of hierarchical field names (like `Seq(foo, bar)`)
into a human-readable
- * string representing the field, like "field 'foo.bar'". If `names` is
empty, the string
- * "top-level record" is returned.
- */
- private[avro] def toFieldStr(names: Seq[String]): String = names match {
- case Seq() => "top-level record"
- case n => s"field '${n.mkString(".")}'"
- }
-
- /** Return true iff `avroField` is nullable, i.e. `UNION` type and has
`NULL` as an option. */
- private[avro] def isNullable(avroField: Schema.Field): Boolean =
- avroField.schema().getType == Schema.Type.UNION &&
- avroField.schema().getTypes.asScala.exists(_.getType == Schema.Type.NULL)
-}
