You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "eason-yuchen-liu (via GitHub)" <gi...@apache.org> on 2024/07/19 17:14:20 UTC
[PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
eason-yuchen-liu opened a new pull request, #47425:
URL: https://github.com/apache/spark/pull/47425
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
The builtin ProtoBuf connector first supports recursive schema reference. It is approached by letting users specify an option “recursive.fields.max.depth”, and at the start of the execution, unroll the recursive field by this level. It converts a problem of dynamic schema for each row to a fixed schema which is supported by Spark. Avro can just adopt a similar method. This PR defines an option "recursiveFieldMaxDepth" to both Avro data source and `from_avro` function. With this option, Spark can support Avro recursive schema up to certain depth.
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
Recursive reference denotes the case that the type of a field can be defined before in the parent nodes. A simple example is:
```
{
"type": "record",
"name": "LongList",
"fields" : [
{"name": "value", "type": "long"},
{"name": "next", "type": ["null", "LongList"]}
]
}
```
This is written in Avro Schema DSL and represents a linked list data structure. Spark currently will throw an error on this schema. Many users used schema like this, so we should support it.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
Yes. Previously, it will throw error on recursive schemas like above. With this change, it will still throw the same error by default but when users specify the option to a number greater than 0, the schema will be unrolled to that depth.
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
Added new unit tests and integration tests to AvroSuite and AvroFunctionSuite.
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
No.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #47425:
URL: https://github.com/apache/spark/pull/47425#issuecomment-2277939564
Friendly reminder, @gengliangwang
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "bogao007 (via GitHub)" <gi...@apache.org>.
bogao007 commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1690240057
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroOptions.scala:
##########
@@ -136,6 +136,13 @@ private[sql] class AvroOptions(
val stableIdPrefixForUnionType: String = parameters
.getOrElse(STABLE_ID_PREFIX_FOR_UNION_TYPE, "member_")
+
+ val recursiveFieldMaxDepth: Int =
+ parameters.get(RECURSIVE_FIELD_MAX_DEPTH).map(_.toInt).getOrElse(-1)
+
+ if (recursiveFieldMaxDepth > 15) {
+ throw new IllegalArgumentException(s"Valid range of $RECURSIVE_FIELD_MAX_DEPTH is 0 - 15.")
Review Comment:
Can we follow the error class strategy to classify this error? You can refer to something like [this](https://github.com/apache/spark/blob/34e65a8e72513d0445b5ff9b251e3388625ad1ec/connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroDataToCatalyst.scala#L81-L87), but consider creating your own type.
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroOptions.scala:
##########
@@ -136,6 +136,13 @@ private[sql] class AvroOptions(
val stableIdPrefixForUnionType: String = parameters
.getOrElse(STABLE_ID_PREFIX_FOR_UNION_TYPE, "member_")
+
+ val recursiveFieldMaxDepth: Int =
+ parameters.get(RECURSIVE_FIELD_MAX_DEPTH).map(_.toInt).getOrElse(-1)
+
+ if (recursiveFieldMaxDepth > 15) {
Review Comment:
Is 15 the max depth we allow? Can we using a constant value like `RECURSIVE_FIELD_DEPTH_LIMIT` to represent it?
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala:
##########
@@ -128,62 +137,113 @@ object SchemaConverters {
case NULL => SchemaType(NullType, nullable = true)
case RECORD =>
- if (existingRecordNames.contains(avroSchema.getFullName)) {
+ val recursiveDepth: Int = existingRecordNames.getOrElse(avroSchema.getFullName, 0)
+ if (recursiveDepth > 0 && recursiveFieldMaxDepth <= 0) {
throw new IncompatibleSchemaException(s"""
- |Found recursive reference in Avro schema, which can not be processed by Spark:
- |${avroSchema.toString(true)}
+ |Found recursive reference in Avro schema, which can not be processed by Spark by
+ | default: ${avroSchema.toString(true)}. Try setting the option `recursiveFieldMaxDepth`
+ | to 1 - 15. Going beyond 15 levels of recursion is not allowed.
""".stripMargin)
- }
- val newRecordNames = existingRecordNames + avroSchema.getFullName
- val fields = avroSchema.getFields.asScala.map { f =>
- val schemaType = toSqlTypeHelper(
- f.schema(),
- newRecordNames,
- useStableIdForUnionType,
- stableIdPrefixForUnionType)
- StructField(f.name, schemaType.dataType, schemaType.nullable)
- }
+ } else if (recursiveDepth > 0 && recursiveDepth >= recursiveFieldMaxDepth) {
+ logInfo(
+ log"The field ${MDC(FIELD_NAME, avroSchema.getFullName)} of type " +
+ log"${MDC(FIELD_TYPE, avroSchema.getType.getName)} is dropped at recursive depth " +
+ log"${MDC(RECURSIVE_DEPTH, recursiveDepth)}."
+ )
+ null
+ } else {
+ val newRecordNames =
+ existingRecordNames + (avroSchema.getFullName -> (recursiveDepth + 1))
+ val fields = avroSchema.getFields.asScala.map { f =>
+ val schemaType = toSqlTypeHelper(
+ f.schema(),
+ newRecordNames,
+ useStableIdForUnionType,
+ stableIdPrefixForUnionType,
+ recursiveFieldMaxDepth)
+ if (schemaType == null) {
+ null
+ }
+ else {
+ StructField(f.name, schemaType.dataType, schemaType.nullable)
+ }
+ }.filter(_ != null).toSeq
- SchemaType(StructType(fields.toArray), nullable = false)
+ SchemaType(StructType(fields), nullable = false)
+ }
case ARRAY =>
val schemaType = toSqlTypeHelper(
avroSchema.getElementType,
existingRecordNames,
useStableIdForUnionType,
- stableIdPrefixForUnionType)
- SchemaType(
- ArrayType(schemaType.dataType, containsNull = schemaType.nullable),
- nullable = false)
+ stableIdPrefixForUnionType,
+ recursiveFieldMaxDepth)
+ if (schemaType == null) {
+ logInfo(
+ log"Dropping ${MDC(FIELD_NAME, avroSchema.getFullName)} of type " +
+ log"${MDC(FIELD_TYPE, avroSchema.getType.getName)} as it does not have any " +
+ log"fields left likely due to recursive depth limit."
+ )
+ null
+ } else {
+ SchemaType(
+ ArrayType(schemaType.dataType, containsNull = schemaType.nullable),
+ nullable = false)
+ }
case MAP =>
val schemaType = toSqlTypeHelper(avroSchema.getValueType,
- existingRecordNames, useStableIdForUnionType, stableIdPrefixForUnionType)
- SchemaType(
- MapType(StringType, schemaType.dataType, valueContainsNull = schemaType.nullable),
- nullable = false)
+ existingRecordNames, useStableIdForUnionType, stableIdPrefixForUnionType,
+ recursiveFieldMaxDepth)
+ if (schemaType == null) {
+ logInfo(
+ log"Dropping ${MDC(FIELD_NAME, avroSchema.getFullName)} of type " +
+ log"${MDC(FIELD_TYPE, avroSchema.getType.getName)} as it does not have any " +
+ log"fields left likely due to recursive depth limit."
+ )
+ null
+ } else {
+ SchemaType(
+ MapType(StringType, schemaType.dataType, valueContainsNull = schemaType.nullable),
+ nullable = false)
+ }
case UNION =>
if (avroSchema.getTypes.asScala.exists(_.getType == NULL)) {
// In case of a union with null, eliminate it and make a recursive call
val remainingUnionTypes = AvroUtils.nonNullUnionBranches(avroSchema)
- if (remainingUnionTypes.size == 1) {
- toSqlTypeHelper(
- remainingUnionTypes.head,
- existingRecordNames,
- useStableIdForUnionType,
- stableIdPrefixForUnionType).copy(nullable = true)
+ val schemaType =
+ if (remainingUnionTypes.size == 1) {
+ toSqlTypeHelper(
+ remainingUnionTypes.head,
+ existingRecordNames,
+ useStableIdForUnionType,
+ stableIdPrefixForUnionType,
+ recursiveFieldMaxDepth)
+ } else {
+ toSqlTypeHelper(
+ Schema.createUnion(remainingUnionTypes.asJava),
Review Comment:
nit: Maybe we could just use if-else clause to define the union types since all other parameters in `toSqlTypeHelper ` are the same?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "hkulyc (via GitHub)" <gi...@apache.org>.
hkulyc commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1720363091
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala:
##########
@@ -128,62 +138,110 @@ object SchemaConverters {
case NULL => SchemaType(NullType, nullable = true)
case RECORD =>
- if (existingRecordNames.contains(avroSchema.getFullName)) {
+ val recursiveDepth: Int = existingRecordNames.getOrElse(avroSchema.getFullName, 0)
+ if (recursiveDepth > 0 && recursiveFieldMaxDepth <= 0) {
throw new IncompatibleSchemaException(s"""
- |Found recursive reference in Avro schema, which can not be processed by Spark:
- |${avroSchema.toString(true)}
+ |Found recursive reference in Avro schema, which can not be processed by Spark by
+ | default: ${avroSchema.toString(true)}. Try setting the option `recursiveFieldMaxDepth`
+ | to 1 - $RECURSIVE_FIELD_MAX_DEPTH_LIMIT.
""".stripMargin)
- }
- val newRecordNames = existingRecordNames + avroSchema.getFullName
- val fields = avroSchema.getFields.asScala.map { f =>
- val schemaType = toSqlTypeHelper(
- f.schema(),
- newRecordNames,
- useStableIdForUnionType,
- stableIdPrefixForUnionType)
- StructField(f.name, schemaType.dataType, schemaType.nullable)
- }
+ } else if (recursiveDepth > 0 && recursiveDepth >= recursiveFieldMaxDepth) {
+ logInfo(
+ log"The field ${MDC(FIELD_NAME, avroSchema.getFullName)} of type " +
+ log"${MDC(FIELD_TYPE, avroSchema.getType.getName)} is dropped at recursive depth " +
+ log"${MDC(RECURSIVE_DEPTH, recursiveDepth)}."
+ )
+ null
+ } else {
+ val newRecordNames =
+ existingRecordNames + (avroSchema.getFullName -> (recursiveDepth + 1))
+ val fields = avroSchema.getFields.asScala.map { f =>
+ val schemaType = toSqlTypeHelper(
+ f.schema(),
+ newRecordNames,
+ useStableIdForUnionType,
+ stableIdPrefixForUnionType,
+ recursiveFieldMaxDepth)
+ if (schemaType == null) {
+ null
+ }
+ else {
+ StructField(f.name, schemaType.dataType, schemaType.nullable)
+ }
+ }.filter(_ != null).toSeq
Review Comment:
We need to drop them because StructType does not take an array that has null values. One alternative is to wrap the null value with StructField(..., nullable = true), but we are not doing it for Protobuf. In Protobuf, we just directly drop them.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "WweiL (via GitHub)" <gi...@apache.org>.
WweiL commented on PR #47425:
URL: https://github.com/apache/spark/pull/47425#issuecomment-2372555938
With this one merged https://github.com/apache/spark/pull/48043
This PR can be closed. Thanks for everyone who worked / reviewed this PR!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "eason-yuchen-liu (via GitHub)" <gi...@apache.org>.
eason-yuchen-liu commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1690380093
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroOptions.scala:
##########
@@ -136,6 +136,13 @@ private[sql] class AvroOptions(
val stableIdPrefixForUnionType: String = parameters
.getOrElse(STABLE_ID_PREFIX_FOR_UNION_TYPE, "member_")
+
+ val recursiveFieldMaxDepth: Int =
+ parameters.get(RECURSIVE_FIELD_MAX_DEPTH).map(_.toInt).getOrElse(-1)
+
+ if (recursiveFieldMaxDepth > 15) {
+ throw new IllegalArgumentException(s"Valid range of $RECURSIVE_FIELD_MAX_DEPTH is 0 - 15.")
Review Comment:
Sure.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #47425:
URL: https://github.com/apache/spark/pull/47425#issuecomment-2270296789
@gengliangwang Would you mind helping reviewing the change as you've been one of the main reviewers for Avro? I can give a try, but I don't feel like I'm qualified to review and sign-off.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1724060621
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroOptions.scala:
##########
@@ -170,4 +180,46 @@ private[sql] object AvroOptions extends DataSourceOptions {
// When STABLE_ID_FOR_UNION_TYPE is enabled, the option allows to configure the prefix for fields
// of Avro Union type.
val STABLE_ID_PREFIX_FOR_UNION_TYPE = newOption("stableIdentifierPrefixForUnionType")
+
+ /**
+ * Adds support for recursive fields. If this option is not specified or is set to 0, recursive
+ * fields are not permitted. Setting it to 1 drops all recursive fields, 2 allows recursive
+ * fields to be recursed once, and 3 allows it to be recursed twice and so on, up to 15.
+ * Values larger than 15 are not allowed in order to avoid inadvertently creating very large
+ * schemas. If an avro message has depth beyond this limit, the Spark struct returned is
+ * truncated after the recursion limit.
+ *
+ * Examples: Consider an Avro schema with a recursive field:
+ * {"type" : "record", "name" : "Node", "fields" : [{"name": "Id", "type": "int"},
+ * {"name": "Next", "type": ["null", "Node"]}]}
+ * The following lists the parsed schema with different values for this setting.
+ * 1: `struct<Id: int>`
+ * 2: `struct<Id: int, Next: struct<Id: int>>`
+ * 3: `struct<Id: int, Next: struct<Id: int, Next: struct<Id: int>>>`
+ * and so on.
+ */
+ val RECURSIVE_FIELD_MAX_DEPTH = newOption("recursiveFieldMaxDepth")
+
+ val RECURSIVE_FIELD_MAX_DEPTH_LIMIT: Int = 15
+}
+
+abstract class AvroOptionsException(
+ errorClass: String,
+ messageParameters: Map[String, String],
+ cause: Throwable)
+ extends SparkRuntimeException(
+ errorClass,
+ messageParameters,
+ cause)
+
+object AvroOptionsError {
Review Comment:
cc @hkulyc
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "hkulyc (via GitHub)" <gi...@apache.org>.
hkulyc commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1731810938
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala:
##########
@@ -201,29 +259,33 @@ object SchemaConverters {
s,
existingRecordNames,
useStableIdForUnionType,
- stableIdPrefixForUnionType)
-
- val fieldName = if (useStableIdForUnionType) {
- // Avro's field name may be case sensitive, so field names for two named type
- // could be "a" and "A" and we need to distinguish them. In this case, we throw
- // an exception.
- // Stable id prefix can be empty so the name of the field can be just the type.
- val tempFieldName = s"${stableIdPrefixForUnionType}${s.getName}"
- if (!fieldNameSet.add(tempFieldName.toLowerCase(Locale.ROOT))) {
- throw new IncompatibleSchemaException(
- "Cannot generate stable identifier for Avro union type due to name " +
- s"conflict of type name ${s.getName}")
- }
- tempFieldName
+ stableIdPrefixForUnionType,
+ recursiveFieldMaxDepth)
+ if (schemaType == null) {
+ null
} else {
- s"member$i"
- }
+ val fieldName = if (useStableIdForUnionType) {
Review Comment:
Thanks for pointing this out, we currently do not have a test case for this. Will add it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "hkulyc (via GitHub)" <gi...@apache.org>.
hkulyc commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1720369350
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala:
##########
@@ -201,29 +259,33 @@ object SchemaConverters {
s,
existingRecordNames,
useStableIdForUnionType,
- stableIdPrefixForUnionType)
-
- val fieldName = if (useStableIdForUnionType) {
- // Avro's field name may be case sensitive, so field names for two named type
- // could be "a" and "A" and we need to distinguish them. In this case, we throw
- // an exception.
- // Stable id prefix can be empty so the name of the field can be just the type.
- val tempFieldName = s"${stableIdPrefixForUnionType}${s.getName}"
- if (!fieldNameSet.add(tempFieldName.toLowerCase(Locale.ROOT))) {
- throw new IncompatibleSchemaException(
- "Cannot generate stable identifier for Avro union type due to name " +
- s"conflict of type name ${s.getName}")
- }
- tempFieldName
+ stableIdPrefixForUnionType,
+ recursiveFieldMaxDepth)
+ if (schemaType == null) {
+ null
} else {
- s"member$i"
- }
+ val fieldName = if (useStableIdForUnionType) {
Review Comment:
If you are talking about `useStableIdForUnionType`, the code was originally added by another PR: https://github.com/apache/spark/pull/44964. And I believe the test is already included in that PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "eason-yuchen-liu (via GitHub)" <gi...@apache.org>.
eason-yuchen-liu commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1690106801
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroOptions.scala:
##########
@@ -170,4 +173,23 @@ private[sql] object AvroOptions extends DataSourceOptions {
// When STABLE_ID_FOR_UNION_TYPE is enabled, the option allows to configure the prefix for fields
// of Avro Union type.
val STABLE_ID_PREFIX_FOR_UNION_TYPE = newOption("stableIdentifierPrefixForUnionType")
+
+ /**
+ * Adds support for recursive fields. If this option is not specified or is set to 0, recursive
+ * fields are not permitted. Setting it to 1 drops all recursive fields, 2 allows recursive
+ * fields to be recursed once, and 3 allows it to be recursed twice and so on, up to 15.
+ * Values larger than 15 are not allowed in order avoid inadvertently creating very large schemas.
Review Comment:
Protobuf has max depth of 10 and it is hardcoded. 15 is used because some users have demand for up to 12 and 3 more is given as buffer. I agree that it will be better if users can increase the max depth at will. Since Protobuf also does not support it. This config can be added in a future PR for both Protobuf and Avro.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1762244492
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala:
##########
@@ -201,29 +259,33 @@ object SchemaConverters {
s,
existingRecordNames,
useStableIdForUnionType,
- stableIdPrefixForUnionType)
-
- val fieldName = if (useStableIdForUnionType) {
- // Avro's field name may be case sensitive, so field names for two named type
- // could be "a" and "A" and we need to distinguish them. In this case, we throw
- // an exception.
- // Stable id prefix can be empty so the name of the field can be just the type.
- val tempFieldName = s"${stableIdPrefixForUnionType}${s.getName}"
- if (!fieldNameSet.add(tempFieldName.toLowerCase(Locale.ROOT))) {
- throw new IncompatibleSchemaException(
- "Cannot generate stable identifier for Avro union type due to name " +
- s"conflict of type name ${s.getName}")
- }
- tempFieldName
+ stableIdPrefixForUnionType,
+ recursiveFieldMaxDepth)
+ if (schemaType == null) {
+ null
} else {
- s"member$i"
- }
+ val fieldName = if (useStableIdForUnionType) {
Review Comment:
> Do you think it is sufficient?
No, it doesn't covert the case when the length of non-null fields > 1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "hkulyc (via GitHub)" <gi...@apache.org>.
hkulyc commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1731810938
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala:
##########
@@ -201,29 +259,33 @@ object SchemaConverters {
s,
existingRecordNames,
useStableIdForUnionType,
- stableIdPrefixForUnionType)
-
- val fieldName = if (useStableIdForUnionType) {
- // Avro's field name may be case sensitive, so field names for two named type
- // could be "a" and "A" and we need to distinguish them. In this case, we throw
- // an exception.
- // Stable id prefix can be empty so the name of the field can be just the type.
- val tempFieldName = s"${stableIdPrefixForUnionType}${s.getName}"
- if (!fieldNameSet.add(tempFieldName.toLowerCase(Locale.ROOT))) {
- throw new IncompatibleSchemaException(
- "Cannot generate stable identifier for Avro union type due to name " +
- s"conflict of type name ${s.getName}")
- }
- tempFieldName
+ stableIdPrefixForUnionType,
+ recursiveFieldMaxDepth)
+ if (schemaType == null) {
+ null
} else {
- s"member$i"
- }
+ val fieldName = if (useStableIdForUnionType) {
Review Comment:
We have a test case for a simplified version of this: https://github.com/apache/spark/blob/ac5a940b46b65c826041a2cd8c5a8e08a4c8ec96/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala#L2271
Do you think it is sufficient? Thanks for pointing this out. @gengliangwang
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on PR #47425:
URL: https://github.com/apache/spark/pull/47425#issuecomment-2278021564
@HeartSaVioR Thanks for the ping. I will find time to review this one recently.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "eason-yuchen-liu (via GitHub)" <gi...@apache.org>.
eason-yuchen-liu commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1690125376
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala:
##########
@@ -128,62 +135,109 @@ object SchemaConverters {
case NULL => SchemaType(NullType, nullable = true)
case RECORD =>
- if (existingRecordNames.contains(avroSchema.getFullName)) {
+ val recursiveDepth: Int = existingRecordNames.getOrElse(avroSchema.getFullName, 0)
+ if (recursiveDepth > 0 && (recursiveFieldMaxDepth <= 0 || recursiveFieldMaxDepth > 15)) {
throw new IncompatibleSchemaException(s"""
- |Found recursive reference in Avro schema, which can not be processed by Spark:
- |${avroSchema.toString(true)}
+ |Found recursive reference in Avro schema, which can not be processed by Spark by
Review Comment:
Good idea.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "WweiL (via GitHub)" <gi...@apache.org>.
WweiL commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1690245997
##########
connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala:
##########
@@ -2252,6 +2252,151 @@ abstract class AvroSuite
""".stripMargin)
}
+ private def checkSparkSchemaEquals(
Review Comment:
Can we also have some boundary checks (< 0, > 15)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "WweiL (via GitHub)" <gi...@apache.org>.
WweiL commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1690501108
##########
connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala:
##########
@@ -2252,6 +2252,151 @@ abstract class AvroSuite
""".stripMargin)
}
+ private def checkSparkSchemaEquals(
+ avroSchema: String, expectedSchema: StructType, recursiveFieldMaxDepth: Int): Unit = {
+ val sparkSchema =
+ SchemaConverters.toSqlType(
+ new Schema.Parser().parse(avroSchema), false, "", recursiveFieldMaxDepth).dataType
+
+ assert(sparkSchema === expectedSchema)
+ }
+
+ test("Translate recursive schema - union") {
Review Comment:
nvm it is included already, let's resolve this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1717670601
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroOptions.scala:
##########
@@ -170,4 +180,46 @@ private[sql] object AvroOptions extends DataSourceOptions {
// When STABLE_ID_FOR_UNION_TYPE is enabled, the option allows to configure the prefix for fields
// of Avro Union type.
val STABLE_ID_PREFIX_FOR_UNION_TYPE = newOption("stableIdentifierPrefixForUnionType")
+
+ /**
+ * Adds support for recursive fields. If this option is not specified or is set to 0, recursive
+ * fields are not permitted. Setting it to 1 drops all recursive fields, 2 allows recursive
+ * fields to be recursed once, and 3 allows it to be recursed twice and so on, up to 15.
+ * Values larger than 15 are not allowed in order to avoid inadvertently creating very large
+ * schemas. If an avro message has depth beyond this limit, the Spark struct returned is
+ * truncated after the recursion limit.
+ *
+ * Examples: Consider an Avro schema with a recursive field:
+ * {"type" : "record", "name" : "Node", "fields" : [{"name": "Id", "type": "int"},
+ * {"name": "Next", "type": ["null", "Node"]}]}
+ * The following lists the parsed schema with different values for this setting.
+ * 1: `struct<Id: int>`
+ * 2: `struct<Id: int, Next: struct<Id: int>>`
+ * 3: `struct<Id: int, Next: struct<Id: int, Next: struct<Id: int>>>`
+ * and so on.
+ */
+ val RECURSIVE_FIELD_MAX_DEPTH = newOption("recursiveFieldMaxDepth")
+
+ val RECURSIVE_FIELD_MAX_DEPTH_LIMIT: Int = 15
+}
+
+abstract class AvroOptionsException(
+ errorClass: String,
+ messageParameters: Map[String, String],
+ cause: Throwable)
+ extends SparkRuntimeException(
+ errorClass,
+ messageParameters,
+ cause)
+
+object AvroOptionsError {
Review Comment:
Let's move this to QueryCompilationErrors. There are some Avro errors in it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1717676903
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala:
##########
@@ -128,62 +138,110 @@ object SchemaConverters {
case NULL => SchemaType(NullType, nullable = true)
case RECORD =>
- if (existingRecordNames.contains(avroSchema.getFullName)) {
+ val recursiveDepth: Int = existingRecordNames.getOrElse(avroSchema.getFullName, 0)
+ if (recursiveDepth > 0 && recursiveFieldMaxDepth <= 0) {
throw new IncompatibleSchemaException(s"""
- |Found recursive reference in Avro schema, which can not be processed by Spark:
- |${avroSchema.toString(true)}
+ |Found recursive reference in Avro schema, which can not be processed by Spark by
+ | default: ${avroSchema.toString(true)}. Try setting the option `recursiveFieldMaxDepth`
+ | to 1 - $RECURSIVE_FIELD_MAX_DEPTH_LIMIT.
""".stripMargin)
- }
- val newRecordNames = existingRecordNames + avroSchema.getFullName
- val fields = avroSchema.getFields.asScala.map { f =>
- val schemaType = toSqlTypeHelper(
- f.schema(),
- newRecordNames,
- useStableIdForUnionType,
- stableIdPrefixForUnionType)
- StructField(f.name, schemaType.dataType, schemaType.nullable)
- }
+ } else if (recursiveDepth > 0 && recursiveDepth >= recursiveFieldMaxDepth) {
+ logInfo(
+ log"The field ${MDC(FIELD_NAME, avroSchema.getFullName)} of type " +
+ log"${MDC(FIELD_TYPE, avroSchema.getType.getName)} is dropped at recursive depth " +
+ log"${MDC(RECURSIVE_DEPTH, recursiveDepth)}."
+ )
+ null
+ } else {
+ val newRecordNames =
+ existingRecordNames + (avroSchema.getFullName -> (recursiveDepth + 1))
+ val fields = avroSchema.getFields.asScala.map { f =>
+ val schemaType = toSqlTypeHelper(
+ f.schema(),
+ newRecordNames,
+ useStableIdForUnionType,
+ stableIdPrefixForUnionType,
+ recursiveFieldMaxDepth)
+ if (schemaType == null) {
+ null
+ }
+ else {
+ StructField(f.name, schemaType.dataType, schemaType.nullable)
+ }
+ }.filter(_ != null).toSeq
Review Comment:
Don't we need to keep the null fields?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "hkulyc (via GitHub)" <gi...@apache.org>.
hkulyc commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1720370503
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala:
##########
@@ -201,29 +259,33 @@ object SchemaConverters {
s,
existingRecordNames,
useStableIdForUnionType,
- stableIdPrefixForUnionType)
-
- val fieldName = if (useStableIdForUnionType) {
- // Avro's field name may be case sensitive, so field names for two named type
- // could be "a" and "A" and we need to distinguish them. In this case, we throw
- // an exception.
- // Stable id prefix can be empty so the name of the field can be just the type.
- val tempFieldName = s"${stableIdPrefixForUnionType}${s.getName}"
- if (!fieldNameSet.add(tempFieldName.toLowerCase(Locale.ROOT))) {
- throw new IncompatibleSchemaException(
- "Cannot generate stable identifier for Avro union type due to name " +
- s"conflict of type name ${s.getName}")
- }
- tempFieldName
+ stableIdPrefixForUnionType,
+ recursiveFieldMaxDepth)
+ if (schemaType == null) {
+ null
} else {
- s"member$i"
- }
+ val fieldName = if (useStableIdForUnionType) {
Review Comment:
This PR only wraps this block with a condition checking.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "hkulyc (via GitHub)" <gi...@apache.org>.
hkulyc commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1724151371
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala:
##########
@@ -201,29 +259,33 @@ object SchemaConverters {
s,
existingRecordNames,
useStableIdForUnionType,
- stableIdPrefixForUnionType)
-
- val fieldName = if (useStableIdForUnionType) {
- // Avro's field name may be case sensitive, so field names for two named type
- // could be "a" and "A" and we need to distinguish them. In this case, we throw
- // an exception.
- // Stable id prefix can be empty so the name of the field can be just the type.
- val tempFieldName = s"${stableIdPrefixForUnionType}${s.getName}"
- if (!fieldNameSet.add(tempFieldName.toLowerCase(Locale.ROOT))) {
- throw new IncompatibleSchemaException(
- "Cannot generate stable identifier for Avro union type due to name " +
- s"conflict of type name ${s.getName}")
- }
- tempFieldName
+ stableIdPrefixForUnionType,
+ recursiveFieldMaxDepth)
+ if (schemaType == null) {
+ null
} else {
- s"member$i"
- }
+ val fieldName = if (useStableIdForUnionType) {
Review Comment:
Recursive and non-recursive union schema shares this code branch. For non-recursive schema we have this test that covers this branch.https://github.com/apache/spark/blob/33059396bd2d529d499498106f51a42328b457ca/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroFunctionsSuite.scala#L290
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "eason-yuchen-liu (via GitHub)" <gi...@apache.org>.
eason-yuchen-liu commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1690470130
##########
connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala:
##########
@@ -2252,6 +2252,151 @@ abstract class AvroSuite
""".stripMargin)
}
+ private def checkSparkSchemaEquals(
Review Comment:
I put it in the integration test to minimize code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1730635996
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala:
##########
@@ -201,29 +259,33 @@ object SchemaConverters {
s,
existingRecordNames,
useStableIdForUnionType,
- stableIdPrefixForUnionType)
-
- val fieldName = if (useStableIdForUnionType) {
- // Avro's field name may be case sensitive, so field names for two named type
- // could be "a" and "A" and we need to distinguish them. In this case, we throw
- // an exception.
- // Stable id prefix can be empty so the name of the field can be just the type.
- val tempFieldName = s"${stableIdPrefixForUnionType}${s.getName}"
- if (!fieldNameSet.add(tempFieldName.toLowerCase(Locale.ROOT))) {
- throw new IncompatibleSchemaException(
- "Cannot generate stable identifier for Avro union type due to name " +
- s"conflict of type name ${s.getName}")
- }
- tempFieldName
+ stableIdPrefixForUnionType,
+ recursiveFieldMaxDepth)
+ if (schemaType == null) {
+ null
} else {
- s"member$i"
- }
+ val fieldName = if (useStableIdForUnionType) {
Review Comment:
@hkulyc yes, what about the recursive union schema?
For example
```
{
"type": "record",
"name": "TreeNode",
"fields": [
{
"name": "name",
"type": "string"
},
{
"name": "value",
"type": [
"string",
"int"
]
},
{
"name": "children",
"type": [
"null",
{
"type": "array",
"items": "TreeNode"
}
],
"default": null
}
]
}
```
Do we have a test case for that?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "hkulyc (via GitHub)" <gi...@apache.org>.
hkulyc commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1731810938
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala:
##########
@@ -201,29 +259,33 @@ object SchemaConverters {
s,
existingRecordNames,
useStableIdForUnionType,
- stableIdPrefixForUnionType)
-
- val fieldName = if (useStableIdForUnionType) {
- // Avro's field name may be case sensitive, so field names for two named type
- // could be "a" and "A" and we need to distinguish them. In this case, we throw
- // an exception.
- // Stable id prefix can be empty so the name of the field can be just the type.
- val tempFieldName = s"${stableIdPrefixForUnionType}${s.getName}"
- if (!fieldNameSet.add(tempFieldName.toLowerCase(Locale.ROOT))) {
- throw new IncompatibleSchemaException(
- "Cannot generate stable identifier for Avro union type due to name " +
- s"conflict of type name ${s.getName}")
- }
- tempFieldName
+ stableIdPrefixForUnionType,
+ recursiveFieldMaxDepth)
+ if (schemaType == null) {
+ null
} else {
- s"member$i"
- }
+ val fieldName = if (useStableIdForUnionType) {
Review Comment:
We have a test case for a simplified version of this: https://github.com/apache/spark/blob/ac5a940b46b65c826041a2cd8c5a8e08a4c8ec96/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala#L2271
Do you think it is sufficient? Thanks for pointing this out.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1717672874
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala:
##########
@@ -54,36 +58,42 @@ object SchemaConverters {
def toSqlType(
avroSchema: Schema,
useStableIdForUnionType: Boolean,
- stableIdPrefixForUnionType: String): SchemaType = {
- toSqlTypeHelper(avroSchema, Set.empty, useStableIdForUnionType, stableIdPrefixForUnionType)
+ stableIdPrefixForUnionType: String,
+ recursiveFieldMaxDepth: Int = -1): SchemaType = {
Review Comment:
let's add code comment for each parameter and explain -1 means not supported.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "eason-yuchen-liu (via GitHub)" <gi...@apache.org>.
eason-yuchen-liu commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1690379854
##########
connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala:
##########
@@ -2252,6 +2252,151 @@ abstract class AvroSuite
""".stripMargin)
}
+ private def checkSparkSchemaEquals(
Review Comment:
The value can be negative. I will add a test for > 15 case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "bogao007 (via GitHub)" <gi...@apache.org>.
bogao007 commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1690573517
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala:
##########
@@ -128,62 +138,110 @@ object SchemaConverters {
case NULL => SchemaType(NullType, nullable = true)
case RECORD =>
- if (existingRecordNames.contains(avroSchema.getFullName)) {
+ val recursiveDepth: Int = existingRecordNames.getOrElse(avroSchema.getFullName, 0)
+ if (recursiveDepth > 0 && recursiveFieldMaxDepth <= 0) {
throw new IncompatibleSchemaException(s"""
- |Found recursive reference in Avro schema, which can not be processed by Spark:
- |${avroSchema.toString(true)}
+ |Found recursive reference in Avro schema, which can not be processed by Spark by
+ | default: ${avroSchema.toString(true)}. Try setting the option `recursiveFieldMaxDepth`
+ | to 1 - $RECURSIVE_FIELD_MAX_DEPTH_LIMIT.
""".stripMargin)
- }
- val newRecordNames = existingRecordNames + avroSchema.getFullName
- val fields = avroSchema.getFields.asScala.map { f =>
- val schemaType = toSqlTypeHelper(
- f.schema(),
- newRecordNames,
- useStableIdForUnionType,
- stableIdPrefixForUnionType)
- StructField(f.name, schemaType.dataType, schemaType.nullable)
- }
+ } else if (recursiveDepth > 0 && recursiveDepth >= recursiveFieldMaxDepth) {
+ logInfo(
+ log"The field ${MDC(FIELD_NAME, avroSchema.getFullName)} of type " +
Review Comment:
What's the behavior for Protobuf? Do we drop the fields or do we throw errors?
##########
docs/sql-data-sources-avro.md:
##########
@@ -353,6 +353,13 @@ Data source options of Avro can be set via:
<td>read</td>
<td>4.0.0</td>
</tr>
+ <tr>
Review Comment:
Thanks for adding the doc for the new option! Apart from this, should we add a block on recursion support similar to what we do for [protobuf](https://github.com/apache/spark/blob/34e65a8e72513d0445b5ff9b251e3388625ad1ec/docs/sql-data-sources-protobuf.md?plain=1#L402-L434)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "eason-yuchen-liu (via GitHub)" <gi...@apache.org>.
eason-yuchen-liu commented on PR #47425:
URL: https://github.com/apache/spark/pull/47425#issuecomment-2245926099
@WweiL @bogao007 PTAL. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "eason-yuchen-liu (via GitHub)" <gi...@apache.org>.
eason-yuchen-liu commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1690609318
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala:
##########
@@ -128,62 +138,110 @@ object SchemaConverters {
case NULL => SchemaType(NullType, nullable = true)
case RECORD =>
- if (existingRecordNames.contains(avroSchema.getFullName)) {
+ val recursiveDepth: Int = existingRecordNames.getOrElse(avroSchema.getFullName, 0)
+ if (recursiveDepth > 0 && recursiveFieldMaxDepth <= 0) {
throw new IncompatibleSchemaException(s"""
- |Found recursive reference in Avro schema, which can not be processed by Spark:
- |${avroSchema.toString(true)}
+ |Found recursive reference in Avro schema, which can not be processed by Spark by
+ | default: ${avroSchema.toString(true)}. Try setting the option `recursiveFieldMaxDepth`
+ | to 1 - $RECURSIVE_FIELD_MAX_DEPTH_LIMIT.
""".stripMargin)
- }
- val newRecordNames = existingRecordNames + avroSchema.getFullName
- val fields = avroSchema.getFields.asScala.map { f =>
- val schemaType = toSqlTypeHelper(
- f.schema(),
- newRecordNames,
- useStableIdForUnionType,
- stableIdPrefixForUnionType)
- StructField(f.name, schemaType.dataType, schemaType.nullable)
- }
+ } else if (recursiveDepth > 0 && recursiveDepth >= recursiveFieldMaxDepth) {
+ logInfo(
+ log"The field ${MDC(FIELD_NAME, avroSchema.getFullName)} of type " +
Review Comment:
Protobuf also drops the field and logs the action.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #47425:
URL: https://github.com/apache/spark/pull/47425#issuecomment-2264324280
cc @HeartSaVioR and @rangadi
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "eason-yuchen-liu (via GitHub)" <gi...@apache.org>.
eason-yuchen-liu commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1690624578
##########
docs/sql-data-sources-avro.md:
##########
@@ -353,6 +353,13 @@ Data source options of Avro can be set via:
<td>read</td>
<td>4.0.0</td>
</tr>
+ <tr>
Review Comment:
Good idea.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "eason-yuchen-liu (via GitHub)" <gi...@apache.org>.
eason-yuchen-liu commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1690120247
##########
connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSerdeSuite.scala:
##########
@@ -228,7 +228,8 @@ object AvroSerdeSuite {
RebaseSpec(CORRECTED),
new NoopFilters,
false,
- "")
+ "",
+ -1)
Review Comment:
Yeah I have thought about it. I did not add default value for two reasons. First is some newly added options (`stableIdPrefixForUnionType`) did not specify a default value either. Second is there are two constructors for the class, one with 5 arguements, the other with 7 arguements, if we were adding default values for both of them, there will be a clash of definition which is confusing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "WweiL (via GitHub)" <gi...@apache.org>.
WweiL commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1690257811
##########
connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala:
##########
@@ -2252,6 +2252,151 @@ abstract class AvroSuite
""".stripMargin)
}
+ private def checkSparkSchemaEquals(
+ avroSchema: String, expectedSchema: StructType, recursiveFieldMaxDepth: Int): Unit = {
+ val sparkSchema =
+ SchemaConverters.toSqlType(
+ new Schema.Parser().parse(avroSchema), false, "", recursiveFieldMaxDepth).dataType
+
+ assert(sparkSchema === expectedSchema)
+ }
+
+ test("Translate recursive schema - union") {
Review Comment:
Can we also have a test for more complex recursive behavior, partially used as documentation (what the behavior would look like), and in case in the future something is changed?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "hkulyc (via GitHub)" <gi...@apache.org>.
hkulyc commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1720370503
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala:
##########
@@ -201,29 +259,33 @@ object SchemaConverters {
s,
existingRecordNames,
useStableIdForUnionType,
- stableIdPrefixForUnionType)
-
- val fieldName = if (useStableIdForUnionType) {
- // Avro's field name may be case sensitive, so field names for two named type
- // could be "a" and "A" and we need to distinguish them. In this case, we throw
- // an exception.
- // Stable id prefix can be empty so the name of the field can be just the type.
- val tempFieldName = s"${stableIdPrefixForUnionType}${s.getName}"
- if (!fieldNameSet.add(tempFieldName.toLowerCase(Locale.ROOT))) {
- throw new IncompatibleSchemaException(
- "Cannot generate stable identifier for Avro union type due to name " +
- s"conflict of type name ${s.getName}")
- }
- tempFieldName
+ stableIdPrefixForUnionType,
+ recursiveFieldMaxDepth)
+ if (schemaType == null) {
+ null
} else {
- s"member$i"
- }
+ val fieldName = if (useStableIdForUnionType) {
Review Comment:
This PR only wraps this block with a condition checking whether the child type is null or not.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #47425: [SPARK-48939][AVRO] Support reading Avro with recursive schema reference
URL: https://github.com/apache/spark/pull/47425
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #47425:
URL: https://github.com/apache/spark/pull/47425#issuecomment-2568552769
We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "eason-yuchen-liu (via GitHub)" <gi...@apache.org>.
eason-yuchen-liu commented on PR #47425:
URL: https://github.com/apache/spark/pull/47425#issuecomment-2256862538
@cloud-fan Could you please review it? Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "WweiL (via GitHub)" <gi...@apache.org>.
WweiL commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1689265340
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroOptions.scala:
##########
@@ -170,4 +173,23 @@ private[sql] object AvroOptions extends DataSourceOptions {
// When STABLE_ID_FOR_UNION_TYPE is enabled, the option allows to configure the prefix for fields
// of Avro Union type.
val STABLE_ID_PREFIX_FOR_UNION_TYPE = newOption("stableIdentifierPrefixForUnionType")
+
+ /**
+ * Adds support for recursive fields. If this option is not specified or is set to 0, recursive
+ * fields are not permitted. Setting it to 1 drops all recursive fields, 2 allows recursive
+ * fields to be recursed once, and 3 allows it to be recursed twice and so on, up to 15.
+ * Values larger than 15 are not allowed in order avoid inadvertently creating very large schemas.
Review Comment:
In order to?
##########
connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala:
##########
@@ -2252,6 +2252,151 @@ abstract class AvroSuite
""".stripMargin)
}
+ private def checkSparkSchemaEquals(
+ avroSchema: String, expectedSchema: StructType, recursiveFieldMaxDepth: Int): Unit = {
+ val sparkSchema =
+ SchemaConverters.toSqlType(
+ new Schema.Parser().parse(avroSchema), false, "", recursiveFieldMaxDepth).dataType
+
+ assert(sparkSchema === expectedSchema)
+ }
+
+ test("Translate recursive schema - 1") {
+ val avroSchema = """
+ |{
+ | "type": "record",
+ | "name": "LongList",
+ | "fields" : [
+ | {"name": "value", "type": "long"}, // each element has a long
+ | {"name": "next", "type": ["null", "LongList"]} // optional next element
+ | ]
+ |}
+ """.stripMargin
+ val nonRecursiveFields = new StructType().add("value", LongType, nullable = false)
+ var expectedSchema = nonRecursiveFields
+ for (i <- 1 to 5) {
+ checkSparkSchemaEquals(avroSchema, expectedSchema, i)
+ expectedSchema = nonRecursiveFields.add("next", expectedSchema)
+ }
+ }
+
+ test("Translate recursive schema - 2") {
+ val avroSchema = """
+ |{
+ | "type": "record",
+ | "name": "LongList",
+ | "fields": [
+ | {
+ | "name": "value",
+ | "type": {
+ | "type": "record",
+ | "name": "foo",
+ | "fields": [
+ | {
+ | "name": "parent",
+ | "type": "LongList"
+ | }
+ | ]
+ | }
+ | }
+ | ]
+ |}
+ """.stripMargin
+ val nonRecursiveFields = new StructType().add("value", StructType(Seq()), nullable = false)
+ var expectedSchema = nonRecursiveFields
+ for (i <- 1 to 5) {
+ checkSparkSchemaEquals(avroSchema, expectedSchema, i)
+ expectedSchema = new StructType().add("value",
+ new StructType().add("parent", expectedSchema, nullable = false), nullable = false)
+ }
+ }
+
+ test("Translate recursive schema - 3") {
+ val avroSchema = """
+ |{
+ | "type": "record",
+ | "name": "LongList",
+ | "fields" : [
+ | {"name": "value", "type": "long"},
+ | {"name": "array", "type": {"type": "array", "items": "LongList"}}
+ | ]
+ |}
+ """.stripMargin
+ val nonRecursiveFields = new StructType().add("value", LongType, nullable = false)
+ var expectedSchema = nonRecursiveFields
+ for (i <- 1 to 5) {
+ checkSparkSchemaEquals(avroSchema, expectedSchema, i)
+ expectedSchema =
+ nonRecursiveFields.add("array", new ArrayType(expectedSchema, false), nullable = false)
+ }
+ }
+
+ test("Translate recursive schema - 4") {
Review Comment:
Can we have better names for the test?
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala:
##########
@@ -128,62 +135,109 @@ object SchemaConverters {
case NULL => SchemaType(NullType, nullable = true)
case RECORD =>
- if (existingRecordNames.contains(avroSchema.getFullName)) {
+ val recursiveDepth: Int = existingRecordNames.getOrElse(avroSchema.getFullName, 0)
+ if (recursiveDepth > 0 && (recursiveFieldMaxDepth <= 0 || recursiveFieldMaxDepth > 15)) {
throw new IncompatibleSchemaException(s"""
- |Found recursive reference in Avro schema, which can not be processed by Spark:
- |${avroSchema.toString(true)}
+ |Found recursive reference in Avro schema, which can not be processed by Spark by
Review Comment:
IIRC protobuf does similar here. But this logic looks a bit weird. If we do want to limit the max recursive depth, I feel that it should be checked in the option and throw `IllegalArgumentException`
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala:
##########
@@ -128,62 +135,109 @@ object SchemaConverters {
case NULL => SchemaType(NullType, nullable = true)
case RECORD =>
- if (existingRecordNames.contains(avroSchema.getFullName)) {
+ val recursiveDepth: Int = existingRecordNames.getOrElse(avroSchema.getFullName, 0)
+ if (recursiveDepth > 0 && (recursiveFieldMaxDepth <= 0 || recursiveFieldMaxDepth > 15)) {
throw new IncompatibleSchemaException(s"""
- |Found recursive reference in Avro schema, which can not be processed by Spark:
- |${avroSchema.toString(true)}
+ |Found recursive reference in Avro schema, which can not be processed by Spark by
+ | default: ${avroSchema.toString(true)}. Try setting the option `recursiveFieldMaxDepth`
+ | to 1 - 15. Going beyond 15 levels of recursion is not allowed.
""".stripMargin)
- }
- val newRecordNames = existingRecordNames + avroSchema.getFullName
- val fields = avroSchema.getFields.asScala.map { f =>
- val schemaType = toSqlTypeHelper(
- f.schema(),
- newRecordNames,
- useStableIdForUnionType,
- stableIdPrefixForUnionType)
- StructField(f.name, schemaType.dataType, schemaType.nullable)
- }
+ } else if (recursiveDepth > 0 && recursiveDepth >= recursiveFieldMaxDepth) {
+ log.info(
Review Comment:
Please try to use the new MDC logging, example: https://github.com/apache/spark/pull/46192
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroOptions.scala:
##########
@@ -170,4 +173,23 @@ private[sql] object AvroOptions extends DataSourceOptions {
// When STABLE_ID_FOR_UNION_TYPE is enabled, the option allows to configure the prefix for fields
// of Avro Union type.
val STABLE_ID_PREFIX_FOR_UNION_TYPE = newOption("stableIdentifierPrefixForUnionType")
+
+ /**
+ * Adds support for recursive fields. If this option is not specified or is set to 0, recursive
+ * fields are not permitted. Setting it to 1 drops all recursive fields, 2 allows recursive
+ * fields to be recursed once, and 3 allows it to be recursed twice and so on, up to 15.
+ * Values larger than 15 are not allowed in order avoid inadvertently creating very large schemas.
Review Comment:
I feel this should be a spark conf
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroOptions.scala:
##########
@@ -170,4 +173,23 @@ private[sql] object AvroOptions extends DataSourceOptions {
// When STABLE_ID_FOR_UNION_TYPE is enabled, the option allows to configure the prefix for fields
// of Avro Union type.
val STABLE_ID_PREFIX_FOR_UNION_TYPE = newOption("stableIdentifierPrefixForUnionType")
+
+ /**
+ * Adds support for recursive fields. If this option is not specified or is set to 0, recursive
+ * fields are not permitted. Setting it to 1 drops all recursive fields, 2 allows recursive
+ * fields to be recursed once, and 3 allows it to be recursed twice and so on, up to 15.
+ * Values larger than 15 are not allowed in order avoid inadvertently creating very large schemas.
Review Comment:
Does protobuf also have max depth of 15?
##########
connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSerdeSuite.scala:
##########
@@ -228,7 +228,8 @@ object AvroSerdeSuite {
RebaseSpec(CORRECTED),
new NoopFilters,
false,
- "")
+ "",
+ -1)
Review Comment:
Should we just add a default value in the definition to prevent multiple API change?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1724060133
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala:
##########
@@ -201,29 +259,33 @@ object SchemaConverters {
s,
existingRecordNames,
useStableIdForUnionType,
- stableIdPrefixForUnionType)
-
- val fieldName = if (useStableIdForUnionType) {
- // Avro's field name may be case sensitive, so field names for two named type
- // could be "a" and "A" and we need to distinguish them. In this case, we throw
- // an exception.
- // Stable id prefix can be empty so the name of the field can be just the type.
- val tempFieldName = s"${stableIdPrefixForUnionType}${s.getName}"
- if (!fieldNameSet.add(tempFieldName.toLowerCase(Locale.ROOT))) {
- throw new IncompatibleSchemaException(
- "Cannot generate stable identifier for Avro union type due to name " +
- s"conflict of type name ${s.getName}")
- }
- tempFieldName
+ stableIdPrefixForUnionType,
+ recursiveFieldMaxDepth)
+ if (schemaType == null) {
+ null
} else {
- s"member$i"
- }
+ val fieldName = if (useStableIdForUnionType) {
Review Comment:
I meant the recursive union schema with more than 2 non-null fields.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "gengliangwang (via GitHub)" <gi...@apache.org>.
gengliangwang commented on code in PR #47425:
URL: https://github.com/apache/spark/pull/47425#discussion_r1717896570
##########
connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala:
##########
@@ -201,29 +259,33 @@ object SchemaConverters {
s,
existingRecordNames,
useStableIdForUnionType,
- stableIdPrefixForUnionType)
-
- val fieldName = if (useStableIdForUnionType) {
- // Avro's field name may be case sensitive, so field names for two named type
- // could be "a" and "A" and we need to distinguish them. In this case, we throw
- // an exception.
- // Stable id prefix can be empty so the name of the field can be just the type.
- val tempFieldName = s"${stableIdPrefixForUnionType}${s.getName}"
- if (!fieldNameSet.add(tempFieldName.toLowerCase(Locale.ROOT))) {
- throw new IncompatibleSchemaException(
- "Cannot generate stable identifier for Avro union type due to name " +
- s"conflict of type name ${s.getName}")
- }
- tempFieldName
+ stableIdPrefixForUnionType,
+ recursiveFieldMaxDepth)
+ if (schemaType == null) {
+ null
} else {
- s"member$i"
- }
+ val fieldName = if (useStableIdForUnionType) {
Review Comment:
Do we have a test case for this code branch?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48939][AVRO] Support reading Avro with recursive schema reference [spark]
Posted by "WweiL (via GitHub)" <gi...@apache.org>.
WweiL commented on PR #47425:
URL: https://github.com/apache/spark/pull/47425#issuecomment-2290108890
Thanks for the review routing! When it's convenient, can people assign this PR to @WweiL and @hkulyc so that we can keep track of reviews? Thanks again!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org