This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new cc9de4d1174e fix(schema): stop union walks looping forever (#19834)
cc9de4d1174e is described below
commit cc9de4d1174e75ae7db2dde5f2ea9e15e22f428a
Author: voonhous <[email protected]>
AuthorDate: Tue Sep 8 18:28:25 2026 +0800
fix(schema): stop union walks looping forever (#19834)
HoodieSchema.getNonNullType() returns this for a union with no null
branch, and a union of the non-null branches when there are two or
more of them. Six recursive schema walkers called themselves on that
result inside their UNION arm, so any schema carrying a field typed
["null","string","int"] or ["string","int"] recursed until
StackOverflowError:
- HoodieSchemaUtils.hasDecimalField (Streamer JSON sources)
- HoodieSchemaRepair.hasTimestampMillisField
- InternalSchemaConverter.collectColNamesFromSchema
- ValueType.fromSchema (column stats)
- HoodieTableMetadataUtil.coerceToComparable (column stats)
- HoodieSchemaUtils.findNestedField
The predicates now iterate every branch. The two walkers that need a
single answer throw an explicit unsupported-union error, and
findNestedField returns empty, since such a union has no one type to
descend into. The name collector walks the branch
visitSchemaToBuildType keeps, the only one the internal schema
carries ids for. HoodieSchema#isComplexUnion() names the shape once,
and getNonNullType()'s javadoc spells out that its result can itself
be a union.
Ending the loop only moved the failure, so the read and stats paths
follow.
Column stats reached ValueType#fromSchema and #coerceToComparable
with such a union because isColumnTypeSupported's deny lists never
named UNION. Both throw, and readColumnRangeMetadataFrom catches
Exception, so one union column silently dropped the stats for every
column in the file. Skip the column up front the way RECORD, MAP and
ARRAY are.
On the Spark read path, pruneDataSchema rejected a union on either
side. A union is a leaf for pruning -- Avro picks a branch by its
type, so dropping one changes the column's type instead of narrowing
it -- so the data schema passes through whole. Spark reads a union as
a struct of nullable member0..memberN fields and prunes it like any
other struct, so the scan's output projection is now bound to the
shape the reader emits and drops the fields Spark did not ask for by
name at every depth. That also covers the BLOB and VARIANT columns
pruneDataSchema keeps whole.
The member-struct heuristic ran on the root of the requested schema
too, where an all-memberN projection is the row and not a union. It
handed the reader the entire table schema, so SELECT member0 came
back with _hoodie_commit_time, and two such columns of one type
failed the conversion outright; it is guarded on depth now.
Closes #19825
---
.../sql/avro/HoodieSparkSchemaConverters.scala | 8 +-
.../datasources/SparkSchemaTransformUtils.scala | 90 +++++++++++++-
.../TestSparkSchemaTransformUtils.scala | 86 ++++++++++++++
.../apache/hudi/common/schema/HoodieSchema.java | 23 +++-
.../hudi/common/schema/HoodieSchemaRepair.java | 2 +-
.../hudi/common/schema/HoodieSchemaUtils.java | 33 ++++--
.../internal/convert/InternalSchemaConverter.java | 12 +-
.../hudi/metadata/HoodieTableMetadataUtil.java | 16 ++-
.../org/apache/hudi/metadata/stats/ValueType.java | 4 +
.../hudi/common/schema/TestHoodieSchema.java | 16 +++
.../hudi/common/schema/TestHoodieSchemaRepair.java | 20 ++++
.../hudi/common/schema/TestHoodieSchemaUtils.java | 115 ++++++++++++++++++
.../convert/TestInternalSchemaConverter.java | 28 +++++
.../hudi/metadata/TestHoodieTableMetadataUtil.java | 30 +++++
.../apache/hudi/metadata/stats/TestValueType.java | 6 +
.../HoodieFileGroupReaderBasedFileFormat.scala | 38 +++++-
.../spark/sql/avro/TestSchemaConverters.scala | 60 +++++++++-
.../TestNestedSchemaPruningOptimization.scala | 131 ++++++++++++++++++++-
18 files changed, 694 insertions(+), 24 deletions(-)
diff --git
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/avro/HoodieSparkSchemaConverters.scala
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/avro/HoodieSparkSchemaConverters.scala
index dc925e60a624..37fdafddf01b 100644
---
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/avro/HoodieSparkSchemaConverters.scala
+++
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/avro/HoodieSparkSchemaConverters.scala
@@ -199,8 +199,12 @@ object HoodieSparkSchemaConverters extends
SparkAdapterSupport {
case st: StructType =>
val childNameSpace = if (nameSpace != "") s"$nameSpace.$recordName"
else recordName
- // Check if this might be a union (using heuristic like Avro converter)
- if (canBeUnion(st)) {
+ // Check if this might be a union (using heuristic like Avro
converter). The root struct is
+ // never one: it is the row, so a projection whose columns are all
nullable and named
+ // member0..memberN would otherwise convert to a union, which
pruneDataSchema keeps whole --
+ // handing the reader the entire table schema -- or which Avro rejects
outright once two of
+ // those columns share a type ("Duplicate in union").
+ if (depth > 0 && canBeUnion(st)) {
val nonNullUnionFieldTypes = st.map { f =>
toHoodieTypeNested(f.dataType, nullable = false, f.name,
childNameSpace, f.metadata, depth + 1)
}
diff --git
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/execution/datasources/SparkSchemaTransformUtils.scala
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/execution/datasources/SparkSchemaTransformUtils.scala
index 3150216c5e1e..2d2ba6f1f8fb 100644
---
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/execution/datasources/SparkSchemaTransformUtils.scala
+++
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/execution/datasources/SparkSchemaTransformUtils.scala
@@ -22,7 +22,7 @@ package org.apache.spark.sql.execution.datasources
import org.apache.hudi.HoodieSparkUtils
import org.apache.spark.sql.HoodieSchemaUtils
import
org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
-import org.apache.spark.sql.catalyst.expressions.{ArrayTransform, Attribute,
AttributeReference, Cast, CreateNamedStruct, CreateStruct, Expression,
GetStructField, LambdaFunction, Literal, MapEntries, MapFromEntries,
NamedLambdaVariable, UnsafeProjection}
+import org.apache.spark.sql.catalyst.expressions.{ArrayTransform, Attribute,
AttributeReference, Cast, CreateNamedStruct, CreateStruct, Expression,
GetStructField, If, IsNull, LambdaFunction, Literal, MapEntries,
MapFromEntries, NamedLambdaVariable, UnsafeProjection}
import org.apache.spark.sql.types.{ArrayType, DataType, DateType, DecimalType,
DoubleType, FloatType, IntegerType, LongType, MapType, StringType, StructField,
StructType, TimestampNTZType}
import scala.util.Try
@@ -33,6 +33,8 @@ import scala.util.Try
*
* These utilities are used by file format readers that need to:
* - Pad missing columns with NULL literals (required for Lance)
+ * - Drop nested fields the reader returns but the scan did not ask for
(unions, BLOB and VARIANT are
+ * read whole)
* - Handle nested struct/array/map type conversions
* - Work around Spark unsafe cast issues (float->double, numeric->decimal)
*
@@ -165,6 +167,92 @@ object SparkSchemaTransformUtils {
expr
}
+ /**
+ * Generate UnsafeProjection that narrows nested structs down to the fields
the target names, matched
+ * by name at every depth. The counterpart of
[[generateNullPaddingProjection]] for input that is wider
+ * than the target rather than narrower: the file group reader hands back a
union (a member0..memberN
+ * struct on the Spark side), a BLOB and a VARIANT whole because
pruneDataSchema cannot prune their
+ * inner fields, while Spark's nested schema pruning may have asked for only
some of them.
+ *
+ * @param inputSchema Schema of the rows the reader emits
+ * @param targetSchema Schema the scan has to produce (a subset of
inputSchema at every depth)
+ * @return UnsafeProjection that drops the nested fields the target does not
name
+ */
+ def generateNestedPruningProjection(inputSchema: StructType, targetSchema:
StructType): UnsafeProjection = {
+ val inputAttributes = inputSchema.fields.map(f =>
AttributeReference(f.name, f.dataType, f.nullable)())
+ val inputFieldMap = inputAttributes.map(a => a.name -> a).toMap
+ val expressions = targetSchema.fields.map { field =>
+ val attr = inputFieldMap(field.name)
+ recursivelyPruneExpression(attr, attr.dataType, field.dataType)
+ }
+ GenerateUnsafeProjection.generate(expressions, inputAttributes)
+ }
+
+ /**
+ * Used to determine if [[generateNestedPruningProjection]] has anything to
drop.
+ *
+ * @param readType Type of the value the reader emits
+ * @param requestedType Type the scan has to produce
+ * @return true if readType names a struct field, at any depth, that
requestedType does not
+ */
+ def needsNestedPruning(readType: DataType, requestedType: DataType): Boolean
= (readType, requestedType) match {
+ case (readStruct: StructType, requestedStruct: StructType) =>
+ readStruct.fields.exists(f =>
requestedStruct.getFieldIndex(f.name).isEmpty) ||
+ requestedStruct.fields.exists { requestedField =>
+ readStruct.getFieldIndex(requestedField.name)
+ .exists(i => needsNestedPruning(readStruct.fields(i).dataType,
requestedField.dataType))
+ }
+ case (ArrayType(readElem, _), ArrayType(requestedElem, _)) =>
+ needsNestedPruning(readElem, requestedElem)
+ case (MapType(readKey, readVal, _), MapType(requestedKey, requestedVal,
_)) =>
+ needsNestedPruning(readKey, requestedKey) || needsNestedPruning(readVal,
requestedVal)
+ case _ => false
+ }
+
+ /**
+ * Recursively rebuild nested struct/array/map values with only the fields
the destination names.
+ *
+ * @param expr Source expression
+ * @param srcType Source data type (may have additional nested fields)
+ * @param dstType Destination data type
+ * @return Expression carrying only the nested fields the destination names
+ */
+ private def recursivelyPruneExpression(
+ expr: Expression,
+ srcType: DataType,
+ dstType: DataType
+ ): Expression = (srcType, dstType) match {
+ case (s: StructType, d: StructType) if needsNestedPruning(s, d) =>
+ val children = d.fields.toSeq.flatMap { dstField =>
+ val srcIndex = s.fieldIndex(dstField.name)
+ val child = GetStructField(expr, srcIndex, Some(dstField.name))
+ Seq(Literal(dstField.name), recursivelyPruneExpression(child,
s.fields(srcIndex).dataType, dstField.dataType))
+ }
+ val pruned = CreateNamedStruct(children)
+ // CreateNamedStruct is never null, so without this guard a null struct
comes back as a struct of nulls
+ If(IsNull(expr), Literal(null, pruned.dataType), pruned)
+
+ case (ArrayType(sElementType, containsNull), ArrayType(dElementType, _))
+ if needsNestedPruning(sElementType, dElementType) =>
+ val lambdaVar = NamedLambdaVariable("element", sElementType,
containsNull)
+ val body = recursivelyPruneExpression(lambdaVar, sElementType,
dElementType)
+ ArrayTransform(expr, LambdaFunction(body, Seq(lambdaVar)))
+
+ case (MapType(sKeyType, sValType, valueContainsNull), MapType(dKeyType,
dValType, _))
+ if needsNestedPruning(sKeyType, dKeyType) ||
needsNestedPruning(sValType, dValType) =>
+ val kv = NamedLambdaVariable("kv", new StructType()
+ .add("key", sKeyType, nullable = false)
+ .add("value", sValType, nullable = valueContainsNull), nullable =
false)
+ val newKey = recursivelyPruneExpression(GetStructField(kv, 0), sKeyType,
dKeyType)
+ val newVal = recursivelyPruneExpression(GetStructField(kv, 1), sValType,
dValType)
+ val entry = CreateStruct(Seq(newKey, newVal))
+ MapFromEntries(ArrayTransform(MapEntries(expr), LambdaFunction(entry,
Seq(kv))))
+
+ case _ =>
+ // Nothing below this point is wider than requested
+ expr
+ }
+
/**
* Recursively cast expressions with special handling for unsupported
conversions.
*
diff --git
a/hudi-client/hudi-spark-client/src/test/scala/org/apache/spark/sql/execution/datasources/TestSparkSchemaTransformUtils.scala
b/hudi-client/hudi-spark-client/src/test/scala/org/apache/spark/sql/execution/datasources/TestSparkSchemaTransformUtils.scala
index bab3a6f9c09f..a12e0c78de61 100644
---
a/hudi-client/hudi-spark-client/src/test/scala/org/apache/spark/sql/execution/datasources/TestSparkSchemaTransformUtils.scala
+++
b/hudi-client/hudi-spark-client/src/test/scala/org/apache/spark/sql/execution/datasources/TestSparkSchemaTransformUtils.scala
@@ -265,6 +265,92 @@ class TestSparkSchemaTransformUtils {
assertTrue(outputStruct.isNullAt(2), "city field should be NULL")
}
+ @Test
+ def testGenerateNestedPruningProjection_nestedStructNarrowedByName(): Unit =
{
+ // Input: (id: int, choice: struct<member0: string, member1: int, member2:
long>), the member struct a
+ // union comes back as from the reader
+ val inputSchema = StructType(Seq(
+ StructField("id", IntegerType, nullable = false),
+ StructField("choice", StructType(Seq(
+ StructField("member0", StringType, nullable = true),
+ StructField("member1", IntegerType, nullable = true),
+ StructField("member2", LongType, nullable = true)
+ )), nullable = true)
+ ))
+
+ // Target keeps the last two members only, so a positional read of the
struct would be off by one
+ val targetSchema = StructType(Seq(
+ StructField("id", IntegerType, nullable = false),
+ StructField("choice", StructType(Seq(
+ StructField("member1", IntegerType, nullable = true),
+ StructField("member2", LongType, nullable = true)
+ )), nullable = true)
+ ))
+
+ assertTrue(SparkSchemaTransformUtils.needsNestedPruning(inputSchema,
targetSchema))
+ // The other direction is padding, not pruning
+ assertFalse(SparkSchemaTransformUtils.needsNestedPruning(targetSchema,
inputSchema))
+
+ val projection =
SparkSchemaTransformUtils.generateNestedPruningProjection(inputSchema,
targetSchema)
+
+ val outputRow = projection.apply(new GenericInternalRow(Array[Any](
+ 1,
+ new GenericInternalRow(Array[Any](UTF8String.fromString("a1"), 7, 100L))
+ )))
+ assertEquals(2, outputRow.numFields)
+ assertEquals(1, outputRow.getInt(0))
+ val choice = outputRow.getStruct(1, 2)
+ assertEquals(7, choice.getInt(0))
+ assertEquals(100L, choice.getLong(1))
+
+ // A null struct stays null instead of coming back as a struct of nulls
+ val nullRow = projection.apply(new GenericInternalRow(Array[Any](2, null)))
+ assertTrue(nullRow.isNullAt(1), "choice should stay NULL")
+ }
+
+ @Test
+ def testGenerateNestedPruningProjection_arrayAndMapOfStructs(): Unit = {
+ val wideElement = StructType(Seq(
+ StructField("k", StringType, nullable = true),
+ StructField("v", StringType, nullable = true)
+ ))
+ val narrowElement = StructType(Seq(StructField("v", StringType, nullable =
true)))
+ val inputSchema = StructType(Seq(
+ StructField("tags", ArrayType(wideElement, containsNull = true),
nullable = true),
+ StructField("props", MapType(StringType, wideElement, valueContainsNull
= true), nullable = true)
+ ))
+ val targetSchema = StructType(Seq(
+ StructField("tags", ArrayType(narrowElement, containsNull = true),
nullable = true),
+ StructField("props", MapType(StringType, narrowElement,
valueContainsNull = true), nullable = true)
+ ))
+
+ assertTrue(SparkSchemaTransformUtils.needsNestedPruning(inputSchema,
targetSchema))
+ val projection =
SparkSchemaTransformUtils.generateNestedPruningProjection(inputSchema,
targetSchema)
+
+ val element = new
GenericInternalRow(Array[Any](UTF8String.fromString("k0"),
UTF8String.fromString("v0")))
+ val outputRow = projection.apply(new GenericInternalRow(Array[Any](
+ ArrayData.toArrayData(Array[Any](element)),
+ new
ArrayBasedMapData(ArrayData.toArrayData(Array[Any](UTF8String.fromString("m0"))),
ArrayData.toArrayData(Array[Any](element)))
+ )))
+
+ val tags = outputRow.getArray(0)
+ assertEquals(1, tags.numElements())
+ assertEquals(UTF8String.fromString("v0"), tags.getStruct(0,
1).getUTF8String(0))
+ val props = outputRow.getMap(1)
+ assertEquals(UTF8String.fromString("m0"),
props.keyArray().getUTF8String(0))
+ assertEquals(UTF8String.fromString("v0"), props.valueArray().getStruct(0,
1).getUTF8String(0))
+ }
+
+ @Test
+ def testNeedsNestedPruning_ignoresNullabilityAndLeafTypes(): Unit = {
+ val readType = StructType(Seq(StructField("x", IntegerType, nullable =
true)))
+ val requestedType = StructType(Seq(StructField("x", IntegerType, nullable
= false)))
+ assertFalse(SparkSchemaTransformUtils.needsNestedPruning(readType,
requestedType))
+
assertFalse(SparkSchemaTransformUtils.needsNestedPruning(ArrayType(readType),
ArrayType(requestedType)))
+ assertFalse(SparkSchemaTransformUtils.needsNestedPruning(StringType,
StringType))
+ assertFalse(SparkSchemaTransformUtils.needsNestedPruning(readType,
StringType))
+ }
+
@Test
def testFilterSchemaByFileSchema_allFieldsPresent(): Unit = {
// Both schemas have (id, name, age)
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchema.java
b/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchema.java
index 53fe10e362e9..e236aebf6027 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchema.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchema.java
@@ -1378,9 +1378,28 @@ public class HoodieSchema implements Serializable {
}
/**
- * If this is a union schema, returns the non-null type. Otherwise, returns
this schema.
+ * Whether this is a union other than the nullable wrapper of a single type:
two or more non-null
+ * branches ({@code [A, B]}, {@code ["null", A, B]}), a lone branch with no
null ({@code [T]}), or null
+ * alone ({@code ["null"]}). {@link #getNonNullType()} cannot reduce such a
union to one non-null type,
+ * so walkers that need one check this before recursing on it.
*
- * @return the non-null schema from a union or the current schema
+ * @return true if this is a union with anything other than exactly one null
and one non-null branch
+ */
+ public boolean isComplexUnion() {
+ return type == HoodieSchemaType.UNION && !(avroSchema.getTypes().size() ==
2 && isNullable());
+ }
+
+ /**
+ * Strips the null branch from a union. {@code ["null", T]} (in either
order) yields {@code T}. A union
+ * with two or more non-null branches yields a union of just those branches
(this schema itself when
+ * there was no null branch to strip), so the result can still be a UNION. A
lone-branch union
+ * {@code [T]} is returned as-is, still a UNION. {@code ["null"]} has
nothing left once the null is
+ * stripped and throws. Callers that need one non-null type out of a union
check
+ * {@link #isComplexUnion()} first: recursing on this result without it
never terminates. Non-union
+ * schemas are returned as-is.
+ *
+ * @return the non-null schema from a nullable union, a union of the
non-null branches, or this schema
+ * @throws IllegalArgumentException if this is the union {@code ["null"]}
*/
public HoodieSchema getNonNullType() {
if (type != HoodieSchemaType.UNION) {
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaRepair.java
b/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaRepair.java
index c00470238d1c..905b8d37dc13 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaRepair.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaRepair.java
@@ -246,7 +246,7 @@ public class HoodieSchemaRepair {
return hasTimestampMillisField(tableSchema.getValueType());
case UNION:
- return hasTimestampMillisField(tableSchema.getNonNullType());
+ return
tableSchema.getTypes().stream().anyMatch(HoodieSchemaRepair::hasTimestampMillisField);
case TIMESTAMP:
HoodieSchema.Timestamp timestampType = (HoodieSchema.Timestamp)
tableSchema;
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaUtils.java
b/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaUtils.java
index 1c5e74fcd7ed..db440d941abb 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaUtils.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchemaUtils.java
@@ -545,6 +545,18 @@ public final class HoodieSchemaUtils {
}
private static HoodieSchema pruneDataSchemaInternal(HoodieSchema dataSchema,
HoodieSchema requiredSchema, Set<String> mandatoryFields) {
+ // A union is a leaf as far as pruning goes: Avro resolves a branch by its
type, so dropping a branch
+ // changes the column's type instead of narrowing it. Hand the data schema
back unpruned whichever
+ // side still holds a union once the null branch is stripped, and let the
caller's projection drop
+ // what it did not ask for. Spark reads a union as a struct of nullable
member0..memberN fields and
+ // its nested schema pruning can project a subset of those members, so the
required schema comes back
+ // as a union when two or more members survive and as the surviving
member's own type when one does:
+ // a record, array or map there belongs to a branch and must not be
matched against the data union.
+ // The reverse pairing is a plain record whose fields happen to be named
member0..memberN, which
+ // HoodieSparkSchemaConverters also reads back as a union.
+ if (dataSchema.getType() == HoodieSchemaType.UNION ||
requiredSchema.getType() == HoodieSchemaType.UNION) {
+ return dataSchema;
+ }
switch (requiredSchema.getType()) {
case RECORD:
// BLOB and VARIANT are represented as Avro RECORDs but carry a
logical type
@@ -593,9 +605,6 @@ public final class HoodieSchemaUtils {
}
return
HoodieSchema.createMap(pruneDataSchema(dataSchema.getValueType(),
requiredSchema.getValueType(), Collections.emptySet()));
- case UNION:
- throw new IllegalArgumentException("Data schema is a union");
-
default:
return dataSchema;
}
@@ -682,6 +691,10 @@ public final class HoodieSchemaUtils {
private static Option<HoodieSchemaField> findNestedField(HoodieSchema
schema, String[] fieldParts, int index) {
if (schema.getType() == HoodieSchemaType.UNION) {
+ if (schema.isComplexUnion()) {
+ // No single record to descend into
+ return Option.empty();
+ }
Option<HoodieSchemaField> notUnion =
findNestedField(schema.getNonNullType(), fieldParts, index);
if (!notUnion.isPresent()) {
return Option.empty();
@@ -820,6 +833,13 @@ public final class HoodieSchemaUtils {
return "hoodie." + sanitizedTableName + "." + sanitizedTableName +
"_record";
}
+ /**
+ * Checks whether the schema is, or nests, a DECIMAL at any depth,
descending through records, arrays,
+ * maps and every branch of a union.
+ *
+ * @param schema the schema to search
+ * @return true if a decimal type is found anywhere in the schema
+ */
public static boolean hasDecimalField(HoodieSchema schema) {
switch (schema.getType()) {
case RECORD:
@@ -834,7 +854,7 @@ public final class HoodieSchemaUtils {
case MAP:
return hasDecimalField(schema.getValueType());
case UNION:
- return hasDecimalField(schema.getNonNullType());
+ return
schema.getTypes().stream().anyMatch(HoodieSchemaUtils::hasDecimalField);
case DECIMAL:
return true;
default:
@@ -863,13 +883,12 @@ public final class HoodieSchemaUtils {
return schema;
}
- List<HoodieSchema> innerTypes = schema.getTypes();
- if (innerTypes.size() == 2 && schema.isNullable()) {
+ if (!schema.isComplexUnion()) {
// this is a basic nullable field so handle it more efficiently
return schema.getNonNullType();
}
- HoodieSchema nonNullType = innerTypes.stream()
+ HoodieSchema nonNullType = schema.getTypes().stream()
.filter(it -> it.getType() != HoodieSchemaType.NULL &&
Objects.equals(it.getFullName(), fieldSchemaFullName))
.findFirst()
.orElse(null);
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/convert/InternalSchemaConverter.java
b/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/convert/InternalSchemaConverter.java
index 7b39c934e28e..45163934705d 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/convert/InternalSchemaConverter.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/schema/internal/convert/InternalSchemaConverter.java
@@ -130,7 +130,17 @@ public class InternalSchemaConverter {
return;
case UNION:
- collectColNamesFromSchema(schema.getNonNullType(), visited, resultSet);
+ // visitSchemaToBuildType keeps a single branch of a union -- the
first one unless that is the
+ // null branch -- and drops the rest, so the internal schema only
carries ids for that branch's
+ // leaves. Walk the same branch: naming a leaf of any other one makes
pruneInternalSchema fail
+ // with "cannot prune col: x.y which does not exist in hudi table". A
union branch is never
+ // itself a union, so this terminates where recursing on
getNonNullType() did not (#19825).
+ for (HoodieSchema branch : schema.getTypes()) {
+ if (branch.getType() != HoodieSchemaType.NULL) {
+ collectColNamesFromSchema(branch, visited, resultSet);
+ break;
+ }
+ }
return;
case ARRAY:
diff --git
a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
index 655fa6e58a9a..1ae0b1664a2c 100644
---
a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
+++
b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
@@ -1385,6 +1385,10 @@ public class HoodieTableMetadataUtil {
switch (schemaType) {
case UNION:
// TODO we need to handle unions in general case as well
+ if (schema.isComplexUnion()) {
+ throw new HoodieNotSupportedException(String.format(
+ "Unsupported UNION type %s: Only UNION of a null type and a
non-null type is supported", schema));
+ }
return coerceToComparable(schema.getNonNullType(), val);
case FIXED:
@@ -1507,6 +1511,8 @@ public class HoodieTableMetadataUtil {
}
public static boolean isColumnTypeSupported(HoodieSchema schema,
Option<HoodieRecordType> recordType, HoodieIndexVersion indexVersion) {
+ // getNonNullType() strips the null branch of a nullable column, so a
UNION still standing after this
+ // is a complex one (HoodieSchema#isComplexUnion), which has no single
value type to collect stats for.
HoodieSchema schemaToCheck = schema.getNonNullType();
if (indexVersion.lowerThan(HoodieIndexVersion.V2)) {
return isColumnTypeSupportedV1(schemaToCheck, recordType);
@@ -1526,11 +1532,12 @@ public class HoodieTableMetadataUtil {
}
HoodieSchemaType type = schema.getType();
- // if record type is set and if its AVRO, MAP, ARRAY, RECORD and ENUM
types are unsupported.
+ // if record type is set and if its AVRO, MAP, ARRAY, RECORD, ENUM and
multi-branch UNION types are unsupported.
if (recordType.isPresent() && recordType.get() == HoodieRecordType.AVRO) {
return (type != HoodieSchemaType.RECORD && type !=
HoodieSchemaType.ARRAY && type != HoodieSchemaType.MAP
&& type != HoodieSchemaType.ENUM && type != HoodieSchemaType.VARIANT
- && type != HoodieSchemaType.BLOB && type != HoodieSchemaType.VECTOR);
+ && type != HoodieSchemaType.BLOB && type != HoodieSchemaType.VECTOR
+ && type != HoodieSchemaType.UNION);
}
// if record Type is not set or if recordType is SPARK then we cannot
support AVRO, MAP, ARRAY, RECORD, ENUM and FIXED and BYTES type as well.
// HUDI-8585 will add support for BYTES and FIXED
@@ -1539,7 +1546,8 @@ public class HoodieTableMetadataUtil {
&& type != HoodieSchemaType.DECIMAL // DECIMAL's underlying type is
BYTES
&& type != HoodieSchemaType.BLOB
&& type != HoodieSchemaType.VECTOR
- && type != HoodieSchemaType.VARIANT;
+ && type != HoodieSchemaType.VARIANT
+ && type != HoodieSchemaType.UNION;
}
private static boolean isColumnTypeSupportedV2(HoodieSchema schema) {
@@ -1551,7 +1559,7 @@ public class HoodieTableMetadataUtil {
return type != HoodieSchemaType.RECORD && type != HoodieSchemaType.MAP
&& type != HoodieSchemaType.ARRAY && type != HoodieSchemaType.ENUM
&& type != HoodieSchemaType.BLOB && type != HoodieSchemaType.VECTOR
- && type != HoodieSchemaType.VARIANT;
+ && type != HoodieSchemaType.VARIANT && type != HoodieSchemaType.UNION;
}
public static Set<String> getInflightMetadataPartitions(HoodieTableConfig
tableConfig) {
diff --git
a/hudi-common/src/main/java/org/apache/hudi/metadata/stats/ValueType.java
b/hudi-common/src/main/java/org/apache/hudi/metadata/stats/ValueType.java
index 02dc1e59c709..6a3d8526d09b 100644
--- a/hudi-common/src/main/java/org/apache/hudi/metadata/stats/ValueType.java
+++ b/hudi-common/src/main/java/org/apache/hudi/metadata/stats/ValueType.java
@@ -294,6 +294,10 @@ public enum ValueType {
case UUID:
return ValueType.UUID;
case UNION:
+ if (schema.isComplexUnion()) {
+ throw new IllegalArgumentException(String.format(
+ "Unsupported UNION type %s: Only UNION of a null type and a
non-null type is supported", schema));
+ }
return fromSchema(schema.getNonNullType());
default:
throw new IllegalArgumentException("Unsupported type: " + type);
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchema.java
b/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchema.java
index c5e9b8b090b2..491897fda0a4 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchema.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchema.java
@@ -391,6 +391,22 @@ public class TestHoodieSchema {
), complexNonNullType.getTypes());
assertSame(complexNonNullType, complexNonNullType.getNonNullType());
+
+ // isComplexUnion() is what a walker checks before recursing on
getNonNullType()
+ assertTrue(union.isComplexUnion());
+ assertTrue(unionWithoutNull.isComplexUnion());
+ assertTrue(complexUnion.isComplexUnion());
+
assertFalse(HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.STRING)).isComplexUnion());
+
assertFalse(HoodieSchema.createUnion(HoodieSchema.create(HoodieSchemaType.STRING),
HoodieSchema.create(HoodieSchemaType.NULL)).isComplexUnion());
+ assertFalse(HoodieSchema.create(HoodieSchemaType.STRING).isComplexUnion());
+
+ // A lone branch has no null to strip and comes back as-is, still a union;
null alone has nothing left
+ HoodieSchema loneBranch =
HoodieSchema.createUnion(HoodieSchema.create(HoodieSchemaType.STRING));
+ assertSame(loneBranch, loneBranch.getNonNullType());
+ assertTrue(loneBranch.isComplexUnion());
+ HoodieSchema nullOnly =
HoodieSchema.createUnion(HoodieSchema.create(HoodieSchemaType.NULL));
+ assertThrows(IllegalArgumentException.class, nullOnly::getNonNullType);
+ assertTrue(nullOnly.isComplexUnion());
}
@Test
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaRepair.java
b/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaRepair.java
index d5de60aa3316..073509d4d031 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaRepair.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaRepair.java
@@ -780,6 +780,26 @@ public class TestHoodieSchemaRepair {
"Should return false for nullable union without timestamp-millis");
}
+ @Test
+ public void testHasTimestampMillisFieldMultiBranchUnion() {
+ // Unions with two or more non-null branches used to recurse forever
(#19825)
+ HoodieSchema withoutMillis = HoodieSchema.createUnion(
+ HoodieSchema.create(HoodieSchemaType.NULL),
+ HoodieSchema.create(HoodieSchemaType.STRING),
+ HoodieSchema.create(HoodieSchemaType.LONG)
+ );
+ assertFalse(HoodieSchemaRepair.hasTimestampMillisField(withoutMillis),
+ "Should return false for multi-branch union without timestamp-millis");
+
+ HoodieSchema withMillis = HoodieSchema.createUnion(
+ HoodieSchema.create(HoodieSchemaType.NULL),
+ HoodieSchema.create(HoodieSchemaType.STRING),
+ HoodieSchema.createTimestampMillis()
+ );
+ assertTrue(HoodieSchemaRepair.hasTimestampMillisField(withMillis),
+ "Should return true for multi-branch union containing
timestamp-millis");
+ }
+
@Test
public void
testHasTimestampMillisFieldUnionWithRecordContainingTimestampMillis() {
HoodieSchema recordSchema = HoodieSchema.createRecord("Record", null,
null, Collections.singletonList(
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaUtils.java
b/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaUtils.java
index e87913c85b8f..d1ec2d09529e 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaUtils.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/schema/TestHoodieSchemaUtils.java
@@ -879,6 +879,87 @@ public class TestHoodieSchemaUtils {
assertInstanceOf(HoodieSchema.Blob.class, prunedPayload.get().schema());
}
+ @Test
+ void testPruningPreservesMultiBranchUnion() {
+ // A field typed ["null", "string", "int"] reaches the pruner as a union
even after the null branch
+ // is stripped. A union branch is picked by type, so there is nothing to
narrow: the pruner must pass
+ // the data schema through rather than reject it (#19825).
+ HoodieSchema unionSchema = HoodieSchema.createUnion(
+ HoodieSchema.create(HoodieSchemaType.NULL),
+ HoodieSchema.create(HoodieSchemaType.STRING),
+ HoodieSchema.create(HoodieSchemaType.INT));
+ HoodieSchema dataSchema = HoodieSchema.createRecord("test_record", null,
null, Arrays.asList(
+ HoodieSchemaField.of("id", HoodieSchema.create(HoodieSchemaType.LONG)),
+ HoodieSchemaField.of("choice", unionSchema)
+ ));
+ HoodieSchema requiredSchema = HoodieSchema.createRecord("test_record",
null, null, Collections.singletonList(
+ HoodieSchemaField.of("choice", unionSchema)
+ ));
+
+ HoodieSchema pruned = HoodieSchemaUtils.pruneDataSchema(dataSchema,
requiredSchema, Collections.emptySet());
+
+ assertEquals(1, pruned.getFields().size());
+ assertEquals(unionSchema, pruned.getFields().get(0).schema());
+ }
+
+ @Test
+ void testPruningPreservesRecordWhenRequiredIsMemberUnion() {
+ // HoodieSparkSchemaConverters reads a struct of nullable member0..memberN
fields back as a union, so
+ // the Spark-side required schema can be a union where the data schema
still holds the record it was
+ // written from. The pruner must pass the record through instead of
rejecting the mismatch (#19825).
+ HoodieSchema memberRecord = HoodieSchema.createRecord("choice", null,
null, Arrays.asList(
+ HoodieSchemaField.of("member0",
HoodieSchema.createNullable(HoodieSchemaType.STRING), null, null),
+ HoodieSchemaField.of("member1",
HoodieSchema.createNullable(HoodieSchemaType.INT), null, null)
+ ));
+ HoodieSchema dataSchema = HoodieSchema.createRecord("test_record", null,
null, Collections.singletonList(
+ HoodieSchemaField.of("choice", memberRecord)
+ ));
+ HoodieSchema requiredSchema = HoodieSchema.createRecord("test_record",
null, null, Collections.singletonList(
+ HoodieSchemaField.of("choice", HoodieSchema.createUnion(
+ HoodieSchema.create(HoodieSchemaType.STRING),
+ HoodieSchema.create(HoodieSchemaType.INT)))
+ ));
+
+ HoodieSchema pruned = HoodieSchemaUtils.pruneDataSchema(dataSchema,
requiredSchema, Collections.emptySet());
+
+ assertEquals(memberRecord, pruned.getFields().get(0).schema());
+ }
+
+ @Test
+ void testPruningPreservesMultiBranchUnionWhenRequiredIsOneMember() {
+ // Spark's nested schema pruning can cut the member0..memberN struct a
union is read as down to a
+ // single member, and HoodieSparkSchemaConverters converts that one-member
struct back to a union
+ // over the member's own type. The required schema then presents as a
record, array or map while the
+ // data schema still holds the whole union, which used to be rejected as a
type mismatch (#19825).
+ HoodieSchema branchRecord = HoodieSchema.createRecord("branch_record",
null, null, Arrays.asList(
+ HoodieSchemaField.of("x",
HoodieSchema.create(HoodieSchemaType.STRING), null, null),
+ HoodieSchemaField.of("y",
HoodieSchema.create(HoodieSchemaType.STRING), null, null)
+ ));
+ HoodieSchema prunedBranchRecord =
HoodieSchema.createRecord("branch_record", null, null,
Collections.singletonList(
+ HoodieSchemaField.of("x",
HoodieSchema.create(HoodieSchemaType.STRING), null, null)
+ ));
+ HoodieSchema branchArray =
HoodieSchema.createArray(HoodieSchema.create(HoodieSchemaType.STRING));
+ HoodieSchema branchMap =
HoodieSchema.createMap(HoodieSchema.create(HoodieSchemaType.STRING));
+
+ for (Pair<HoodieSchema, HoodieSchema> branchAndRequired : Arrays.asList(
+ Pair.of(branchRecord, prunedBranchRecord), Pair.of(branchArray,
branchArray), Pair.of(branchMap, branchMap))) {
+ HoodieSchema dataUnion = HoodieSchema.createUnion(
+ HoodieSchema.create(HoodieSchemaType.NULL),
+ HoodieSchema.create(HoodieSchemaType.INT),
+ branchAndRequired.getLeft());
+ HoodieSchema dataSchema = HoodieSchema.createRecord("test_record", null,
null, Collections.singletonList(
+ HoodieSchemaField.of("choice", dataUnion, null, null)
+ ));
+ HoodieSchema requiredSchema = HoodieSchema.createRecord("test_record",
null, null, Collections.singletonList(
+ HoodieSchemaField.of("choice",
HoodieSchema.createNullable(branchAndRequired.getRight()), null, null)
+ ));
+
+ HoodieSchema pruned = HoodieSchemaUtils.pruneDataSchema(dataSchema,
requiredSchema, Collections.emptySet());
+
+ assertEquals(dataUnion, pruned.getFields().get(0).schema());
+ }
+ }
+
@Test
void testPruningPreserveNullable() {
String dataSchemaStr = "{"
@@ -1152,6 +1233,20 @@ public class TestHoodieSchemaUtils {
assertEquals(fullSchema, actual);
}
+ @Test
+ public void testFindNestedFieldThroughMultiBranchUnion() {
+ // A field typed ["null", "string", "int"] has no record to descend into
and used to recurse forever (#19825)
+ HoodieSchema schema = HoodieSchema.createRecord("record", null, null,
false,
+ Collections.singletonList(
+ HoodieSchemaField.of("u", HoodieSchema.createUnion(
+ HoodieSchema.create(HoodieSchemaType.NULL),
+ HoodieSchema.create(HoodieSchemaType.STRING),
+ HoodieSchema.create(HoodieSchemaType.INT)), null, null)
+ ));
+ assertTrue(HoodieSchemaUtils.findNestedField(schema, "u").isPresent());
+ assertFalse(HoodieSchemaUtils.findNestedField(schema, "u.x").isPresent());
+ }
+
@Test
public void testCreateNewSchemaFromFieldsWithReference_NullSchema() {
// This test should throw an IllegalArgumentException
@@ -1361,6 +1456,26 @@ public class TestHoodieSchemaUtils {
HoodieSchemaField.of("arrayfield",
HoodieSchema.createArray(HoodieSchema.createDecimal(10, 6)), null, null)
));
assertTrue(HoodieSchemaUtils.hasDecimalField(recordWithMapAndDecArray));
+ // Unions with two or more non-null branches used to recurse forever
(#19825)
+ HoodieSchema recordWithMultiBranchUnions =
HoodieSchema.createRecord("recordWithMultiBranchUnions", null, null, false,
+ Arrays.asList(
+ HoodieSchemaField.of("nullableStringOrInt",
HoodieSchema.createUnion(
+ HoodieSchema.create(HoodieSchemaType.NULL),
+ HoodieSchema.create(HoodieSchemaType.STRING),
+ HoodieSchema.create(HoodieSchemaType.INT)), null, null),
+ HoodieSchemaField.of("stringOrInt", HoodieSchema.createUnion(
+ HoodieSchema.create(HoodieSchemaType.STRING),
+ HoodieSchema.create(HoodieSchemaType.INT)), null, null)
+ ));
+
assertFalse(HoodieSchemaUtils.hasDecimalField(recordWithMultiBranchUnions));
+ HoodieSchema recordWithDecimalInMultiBranchUnion =
HoodieSchema.createRecord("recordWithDecimalInMultiBranchUnion", null, null,
false,
+ Collections.singletonList(
+ HoodieSchemaField.of("nullableStringOrDecimal",
HoodieSchema.createUnion(
+ HoodieSchema.create(HoodieSchemaType.NULL),
+ HoodieSchema.create(HoodieSchemaType.STRING),
+ HoodieSchema.createDecimal(10, 6)), null, null)
+ ));
+
assertTrue(HoodieSchemaUtils.hasDecimalField(recordWithDecimalInMultiBranchUnion));
}
@Test
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/schema/internal/convert/TestInternalSchemaConverter.java
b/hudi-common/src/test/java/org/apache/hudi/common/schema/internal/convert/TestInternalSchemaConverter.java
index 33b53a512173..256461713678 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/schema/internal/convert/TestInternalSchemaConverter.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/schema/internal/convert/TestInternalSchemaConverter.java
@@ -30,6 +30,7 @@ import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.List;
+import java.util.stream.Collectors;
import static
org.apache.hudi.common.schema.HoodieSchemaTestUtils.createArrayField;
import static
org.apache.hudi.common.schema.HoodieSchemaTestUtils.createMapField;
@@ -209,6 +210,33 @@ public class TestInternalSchemaConverter {
expectedOutput = getDeeplyNestedFieldSchemaExpectedColumnNames();
assertEquals(expectedOutput.size(), fieldNames.size());
assertTrue(fieldNames.containsAll(expectedOutput));
+
+ // A primitive union with two or more non-null branches is a leaf and used
to recurse forever (#19825)
+ HoodieSchema schemaWithMultiBranchUnion =
createRecord("schemaWithMultiBranchUnion",
+ HoodieSchemaField.of("field1", HoodieSchema.createUnion(
+ HoodieSchema.create(HoodieSchemaType.NULL),
+ HoodieSchema.create(HoodieSchemaType.STRING),
+ HoodieSchema.create(HoodieSchemaType.INT)), null, null),
+ createPrimitiveField("field2", HoodieSchemaType.STRING));
+ fieldNames =
InternalSchemaConverter.collectColNamesFromSchema(schemaWithMultiBranchUnion);
+ assertEquals(Arrays.asList("field1", "field2"), fieldNames);
+
+ // visitSchemaToBuildType keeps the first non-null branch and drops the
rest, so a record, array or
+ // map sitting in one of the dropped branches has no ids in the internal
schema. Naming its leaves
+ // here made pruneInternalSchema throw "cannot prune col: field1.x which
does not exist in hudi
+ // table" (#19825).
+ HoodieSchema schemaWithRecordBranchUnion =
createRecord("schemaWithRecordBranchUnion",
+ HoodieSchemaField.of("field1", HoodieSchema.createUnion(
+ HoodieSchema.create(HoodieSchemaType.NULL),
+ HoodieSchema.create(HoodieSchemaType.INT),
+ createRecord("branchRecord", createPrimitiveField("x",
HoodieSchemaType.STRING))), null, null),
+ createPrimitiveField("field2", HoodieSchemaType.STRING));
+ fieldNames =
InternalSchemaConverter.collectColNamesFromSchema(schemaWithRecordBranchUnion);
+ assertEquals(Arrays.asList("field1", "field2"), fieldNames);
+ InternalSchema prunedInternalSchema =
InternalSchemaConverter.pruneHoodieSchemaToInternalSchema(
+ schemaWithRecordBranchUnion,
InternalSchemaConverter.convert(schemaWithRecordBranchUnion));
+ assertEquals(Arrays.asList("field1", "field2"),
+
prunedInternalSchema.getRecord().fields().stream().map(Types.Field::name).collect(Collectors.toList()));
}
@Test
diff --git
a/hudi-common/src/test/java/org/apache/hudi/metadata/TestHoodieTableMetadataUtil.java
b/hudi-common/src/test/java/org/apache/hudi/metadata/TestHoodieTableMetadataUtil.java
index 5b4f921936da..36256f50ae90 100644
---
a/hudi-common/src/test/java/org/apache/hudi/metadata/TestHoodieTableMetadataUtil.java
+++
b/hudi-common/src/test/java/org/apache/hudi/metadata/TestHoodieTableMetadataUtil.java
@@ -412,6 +412,27 @@ class TestHoodieTableMetadataUtil {
}
}
+ @Test
+ void testMultiBranchUnionColumnsAreNotSupportedForColumnStats() {
+ // A column typed ["null", "string", "int"] has no single value type, so
it must be skipped the same
+ // way RECORD/MAP/ARRAY are. Indexing it instead reaches
ValueType#fromSchema and #coerceToComparable,
+ // whose throws are caught by readColumnRangeMetadataFrom and drop the
whole file's stats (#19825).
+ HoodieSchema multiBranchUnion = HoodieSchema.createUnion(
+ HoodieSchema.create(HoodieSchemaType.NULL),
+ HoodieSchema.create(HoodieSchemaType.STRING),
+ HoodieSchema.create(HoodieSchemaType.INT));
+ HoodieSchema nullableString =
HoodieSchema.createNullable(HoodieSchema.create(HoodieSchemaType.STRING));
+
+ for (HoodieIndexVersion indexVersion : new HoodieIndexVersion[]
{HoodieIndexVersion.V1, HoodieIndexVersion.V2}) {
+ for (Option<HoodieRecordType> rt :
Arrays.<Option<HoodieRecordType>>asList(Option.empty(),
Option.of(HoodieRecordType.AVRO), Option.of(HoodieRecordType.SPARK))) {
+
assertFalse(HoodieTableMetadataUtil.isColumnTypeSupported(multiBranchUnion, rt,
indexVersion),
+ "A union with two or more non-null branches must be excluded from
" + indexVersion + " column stats");
+
assertTrue(HoodieTableMetadataUtil.isColumnTypeSupported(nullableString, rt,
indexVersion),
+ "A plain nullable column must stay supported for " + indexVersion
+ " column stats");
+ }
+ }
+ }
+
@Test
void testCreateRecordIndexUpdateMillisOverloadMatchesStringOverload() {
String instantTime = "20260610153045678";
@@ -533,6 +554,15 @@ class TestHoodieTableMetadataUtil {
HoodieSchema.create(HoodieSchemaType.DOUBLE), false));
assertNull(HoodieTableMetadataUtil.coerceToComparable(
HoodieSchema.create(HoodieSchemaType.NULL), "ignored"));
+ assertEquals(1, HoodieTableMetadataUtil.coerceToComparable(
+ HoodieSchema.createNullable(HoodieSchemaType.INT), true));
+ // A union with two or more non-null branches cannot be coerced and used
to recurse forever (#19825)
+ HoodieSchema multiBranchUnion = HoodieSchema.createUnion(
+ HoodieSchema.create(HoodieSchemaType.NULL),
+ HoodieSchema.create(HoodieSchemaType.STRING),
+ HoodieSchema.create(HoodieSchemaType.INT));
+ assertThrows(HoodieNotSupportedException.class,
+ () -> HoodieTableMetadataUtil.coerceToComparable(multiBranchUnion,
"ignored"));
}
@Test
diff --git
a/hudi-common/src/test/java/org/apache/hudi/metadata/stats/TestValueType.java
b/hudi-common/src/test/java/org/apache/hudi/metadata/stats/TestValueType.java
index 333ec2ea1ab6..0cf3d240a6d6 100644
---
a/hudi-common/src/test/java/org/apache/hudi/metadata/stats/TestValueType.java
+++
b/hudi-common/src/test/java/org/apache/hudi/metadata/stats/TestValueType.java
@@ -273,6 +273,12 @@ public class TestValueType {
public void testFromSchemaUnwrapsUnion() {
HoodieSchema nullableInt =
HoodieSchema.createNullable(HoodieSchemaType.INT);
assertEquals(ValueType.INT, ValueType.fromSchema(nullableInt));
+ // A union with two or more non-null branches has no single value type and
used to recurse forever (#19825)
+ HoodieSchema multiBranchUnion = HoodieSchema.createUnion(
+ HoodieSchema.create(HoodieSchemaType.NULL),
+ HoodieSchema.create(HoodieSchemaType.STRING),
+ HoodieSchema.create(HoodieSchemaType.INT));
+ assertThrows(IllegalArgumentException.class, () ->
ValueType.fromSchema(multiBranchUnion));
}
@Test
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
index aba5c31bc1a4..3bc8f90ac55e 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
@@ -51,7 +51,7 @@ import
org.apache.spark.sql.HoodieCatalystExpressionUtils.generateUnsafeProjecti
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{JoinedRow, UnsafeProjection}
-import org.apache.spark.sql.execution.datasources.{OutputWriterFactory,
PartitionedFile, SparkColumnarFileReader}
+import org.apache.spark.sql.execution.datasources.{OutputWriterFactory,
PartitionedFile, SparkColumnarFileReader, SparkSchemaTransformUtils}
import org.apache.spark.sql.execution.datasources.orc.OrcUtils
import org.apache.spark.sql.execution.vectorized.{OffHeapColumnVector,
OnHeapColumnVector}
import org.apache.spark.sql.hudi.MultipleColumnarFileFormatReader
@@ -318,6 +318,20 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
partitionSchema.fields.foreach(f => exclusionFields.add(f.name))
val requestedStructType = StructType(readRequiredSchema.fields ++
partitionSchema.fields.filter(f => mandatoryFields.contains(f.name) &&
!isNestedPartitionField(f.name)))
val requestedSchema = HoodieSchemaUtils.pruneDataSchema(schema,
HoodieSchemaConversionUtils.convertStructTypeToHoodieSchema(requestedStructType,
sanitizedTableName), exclusionFields)
+ // The reader emits requestedSchema -- FileGroupReaderSchemaHandler
projects its merged rows back
+ // down to it -- and that is wider than requestedStructType wherever
pruneDataSchema had to keep a
+ // column whole: a union (a member0..memberN struct on the Spark side), a
BLOB or a VARIANT that
+ // Spark's nested schema pruning asked only some inner fields of. Bind the
output projection to the
+ // emitted shape rather than to what was asked for, so it resolves by name
and drops the rest; a
+ // field the reader does not widen keeps the requested type, which is what
it already had.
+ val readerStructType =
HoodieSchemaConversionUtils.convertHoodieSchemaToStructType(requestedSchema)
+ val requestedFieldsByName = requestedStructType.fields.map(f => f.name ->
f).toMap
+ val projectionInputSchema = StructType(readerStructType.fields.map {
readerField =>
+ requestedFieldsByName.get(readerField.name) match {
+ case Some(f) if
!SparkSchemaTransformUtils.needsNestedPruning(readerField.dataType, f.dataType)
=> f
+ case _ => readerField
+ }
+ })
val dataStructTypeWithMandatoryPartitionFields =
StructType(dataStructType.fields ++ partitionSchema.fields.filter(f =>
mandatoryFields.contains(f.name) && !isNestedPartitionField(f.name)))
val dataSchema = HoodieSchemaUtils.pruneDataSchema(schema,
HoodieSchemaConversionUtils.convertStructTypeToHoodieSchema(dataStructTypeWithMandatoryPartitionFields,
sanitizedTableName), exclusionFields)
@@ -402,7 +416,7 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
// Append partition values to rows and project to output schema
appendPartitionAndProject(
reader.getClosableIterator,
- requestedStructType,
+ projectionInputSchema,
remainingPartitionSchema,
outputSchema,
fileSliceMapping.getPartitionValues,
@@ -493,7 +507,7 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
//some partition fields read from file, some were not
getFixedPartitionValues(partitionValues, partitionSchema,
fixedPartitionIndexes)
}
- val unsafeProjection =
generateUnsafeProjection(StructType(inputSchema.fields ++
partitionSchema.fields), to)
+ val unsafeProjection =
generateOutputProjection(StructType(inputSchema.fields ++
partitionSchema.fields), to)
val joinedRow = new JoinedRow()
makeCloseableFileGroupMappingRecordIterator(iter, d =>
unsafeProjection(joinedRow(d, fixedPartitionValues)))
}
@@ -502,10 +516,26 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
private def projectSchema(iter: ClosableIterator[InternalRow],
from: StructType,
to: StructType): Iterator[InternalRow] = {
- val unsafeProjection = generateUnsafeProjection(from, to)
+ val unsafeProjection = generateOutputProjection(from, to)
makeCloseableFileGroupMappingRecordIterator(iter, d => unsafeProjection(d))
}
+ /**
+ * The scan's output projection. Stays the by-name top-level projection
unless a column comes out of the
+ * reader with nested fields Spark did not ask for (see
`projectionInputSchema` in
+ * buildReaderWithPartitionValues), in which case those are dropped by name
at every depth.
+ */
+ private def generateOutputProjection(from: StructType, to: StructType):
UnsafeProjection = {
+ val hasWiderNestedInput = to.fields.exists { f =>
+ from.getFieldIndex(f.name).exists(i =>
SparkSchemaTransformUtils.needsNestedPruning(from.fields(i).dataType,
f.dataType))
+ }
+ if (hasWiderNestedInput) {
+ SparkSchemaTransformUtils.generateNestedPruningProjection(from, to)
+ } else {
+ generateUnsafeProjection(from, to)
+ }
+ }
+
private def
makeCloseableFileGroupMappingRecordIterator(closeableFileGroupRecordIterator:
ClosableIterator[InternalRow],
mappingFunction:
Function[InternalRow, InternalRow]): Iterator[InternalRow] = {
CloseableIteratorListener.addListener(closeableFileGroupRecordIterator)
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/avro/TestSchemaConverters.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/avro/TestSchemaConverters.scala
index 2c5a448d46ec..5cd911f47f7d 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/avro/TestSchemaConverters.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/avro/TestSchemaConverters.scala
@@ -20,8 +20,9 @@ package org.apache.spark.sql.avro
import org.apache.hudi.HoodieSparkUtils
import org.apache.hudi.SparkAdapterSupport
import org.apache.hudi.avro.model.HoodieMetadataColumnStats
-import org.apache.hudi.common.schema.{HoodieSchema, HoodieSchemaField,
HoodieSchemaType}
+import org.apache.hudi.common.schema.{HoodieSchema, HoodieSchemaField,
HoodieSchemaRepair, HoodieSchemaType, HoodieSchemaUtils}
import org.apache.hudi.common.schema.internal.HoodieSchemaException
+import org.apache.hudi.metadata.stats.ValueType
import org.apache.avro.JsonProperties
import org.apache.spark.sql.types.{ArrayType, DataTypes, FloatType,
MetadataBuilder, StructField, StructType}
@@ -54,6 +55,63 @@ class TestSchemaConverters extends SparkAdapterSupport {
}
}
+ @Test
+ def testMemberStructBecomesMultiBranchUnionThatSchemaWalksCanTraverse():
Unit = {
+ // A struct of nullable member0..memberN fields is spark-avro's encoding
of a complex union and
+ // toHoodieType turns it back into one, so a datasource write can land
["null","string","int"] in a
+ // table schema. The recursive schema walks used to loop forever on that
shape (#19825).
+ val structType = StructType(Seq(
+ StructField("id", DataTypes.LongType, nullable = false),
+ StructField("choice", StructType(Seq(
+ StructField("member0", DataTypes.StringType, nullable = true),
+ StructField("member1", DataTypes.IntegerType, nullable = true))),
nullable = true),
+ StructField("amount", StructType(Seq(
+ StructField("member0", DataTypes.StringType, nullable = true),
+ StructField("member1", DataTypes.createDecimalType(10, 6), nullable =
true))), nullable = true)))
+
+ val hoodieSchema = HoodieSparkSchemaConverters.toHoodieType(structType)
+
+ val choice = hoodieSchema.getField("choice").get().schema()
+ assertEquals(HoodieSchemaType.UNION, choice.getType)
+ assertEquals(3, choice.getTypes.size())
+ // getNonNullType() cannot reduce this union to a single branch, which is
what tripped the walkers
+ assertEquals(HoodieSchemaType.UNION, choice.getNonNullType.getType)
+
+ assertFalse(HoodieSchemaRepair.hasTimestampMillisField(hoodieSchema))
+ assertTrue(HoodieSchemaUtils.hasDecimalField(hoodieSchema))
+ assertTrue(HoodieSchemaUtils.findNestedField(hoodieSchema,
"choice").isPresent)
+ assertFalse(HoodieSchemaUtils.findNestedField(hoodieSchema,
"choice.member0").isPresent)
+ assertThrows(classOf[IllegalArgumentException], () =>
ValueType.fromSchema(choice))
+
+ // and the union still converts back to the member struct it came from
+ assertEquals(structType("choice").dataType,
HoodieSparkSchemaConverters.toSqlType(hoodieSchema)._1.asInstanceOf[StructType]("choice").dataType)
+ }
+
+ @Test
+ def testMemberStructUnionSurvivesReadPathSchemaSteps(): Unit = {
+ // The two schema steps
HoodieFileGroupReaderBasedFileFormat#buildReaderWithPartitionValues takes
+ // before it opens a file, run over a table schema carrying
["null","string","int"]: the walk that
+ // decides logical-timestamp repair, then the prune of the table schema
down to what Spark asked
+ // for. Both used to fail on this shape -- the first by recursing forever,
the second by rejecting
+ // the union outright (#19825).
+ val tableSchema = HoodieSchema.createRecord("test_record", "hoodie.test",
null, util.Arrays.asList(
+ HoodieSchemaField.of("id", HoodieSchema.create(HoodieSchemaType.LONG)),
+ HoodieSchemaField.of("choice", HoodieSchema.createUnion(
+ HoodieSchema.create(HoodieSchemaType.NULL),
+ HoodieSchema.create(HoodieSchemaType.STRING),
+ HoodieSchema.create(HoodieSchemaType.INT)), null, null)))
+
+ assertFalse(HoodieSchemaRepair.hasTimestampMillisField(tableSchema))
+
+ // Spark sees the union as the member struct it round-trips through, and
asks for it back
+ val requestedStructType =
HoodieSparkSchemaConverters.toSqlType(tableSchema)._1.asInstanceOf[StructType]
+ val requestedSchema = HoodieSchemaUtils.pruneDataSchema(
+ tableSchema,
HoodieSparkSchemaConverters.toHoodieType(requestedStructType, recordName =
"test_record", nameSpace = "hoodie.test"), util.Collections.emptySet[String]())
+
+ assertEquals(HoodieSchemaType.UNION,
requestedSchema.getField("choice").get().schema().getType)
+ assertEquals(requestedStructType,
HoodieSparkSchemaConverters.toSqlType(requestedSchema)._1)
+ }
+
@Test
def testSchemaWithBlobsRoundTrip(): Unit = {
val originalSchema = HoodieSchema.createRecord("document", "test", null,
util.Arrays.asList(
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala
index be8cf2eda5a7..3ca08f7eec80 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala
@@ -17,15 +17,27 @@
package org.apache.spark.sql.hudi.common
+import org.apache.hudi.client.SparkRDDWriteClient
+import org.apache.hudi.client.common.HoodieSparkEngineContext
+import org.apache.hudi.common.model.{HoodieAvroPayload, HoodieAvroRecord,
HoodieKey, HoodieRecord, HoodieTableType}
+import org.apache.hudi.common.schema.{HoodieSchema, HoodieSchemaType}
+import org.apache.hudi.common.table.{HoodieTableMetaClient,
TableSchemaResolver}
import org.apache.hudi.common.table.read.CustomPayloadForTesting
+import org.apache.hudi.common.util.{Option => HOption}
import org.apache.hudi.config.HoodieWriteConfig
+import org.apache.hudi.hadoop.fs.HadoopFSUtils
+import org.apache.hudi.testutils.HoodieClientTestUtils.createMetaClient
-import org.apache.spark.sql.DataFrame
+import org.apache.avro.generic.{GenericData, GenericRecord}
+import org.apache.spark.api.java.JavaSparkContext
+import org.apache.spark.sql.{DataFrame, Row}
import org.apache.spark.sql.execution.{FileSourceScanExec, ProjectExec}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{ArrayType, IntegerType, MapType,
StringType, StructField, StructType}
import org.junit.jupiter.api.Assertions.assertEquals
+import scala.collection.JavaConverters._
+
/**
* Verifies nested schema pruning on Hudi tables. The pruning itself is
performed by Spark's
* built-in SchemaPruning rule: the file-group-reader based file format
extends ParquetFileFormat,
@@ -148,6 +160,123 @@ class TestNestedSchemaPruningOptimization extends
HoodieSparkSqlTestBase {
}
}
+ test("Test nested schema pruning through a union column") {
+ withTempDir { tmp =>
+ Seq(HoodieTableType.COPY_ON_WRITE,
HoodieTableType.MERGE_ON_READ).foreach { tableType =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ // A union with two or more non-null branches reaches a table only
from a writer that takes a raw
+ // Avro schema (Streamer, Flink, Kafka Connect, the Java client): a
Spark write round-trips its
+ // writer schema through InternalSchema, which keeps the first branch
only. Spark reads the union
+ // as the struct of nullable member0..memberN fields spark-avro gives
it, and its nested schema
+ // pruning can then ask for a subset of those members, which
pruneDataSchema cannot honour: a
+ // union is kept whole (#19825), so the scan's output projection has
to drop the rest by name.
+ val schema = HoodieSchema.parse(
+
"""{"type":"record","name":"union_record","namespace":"hoodie.test","fields":[
+ |{"name":"id","type":"int"},
+ |{"name":"choice","type":["null","string","int"],"default":null},
+
|{"name":"pick","type":["null","string","int","long"],"default":null},
+
|{"name":"nested","type":["null","int",{"type":"record","name":"nested_branch","fields":[
+ |{"name":"x","type":["null","string"],"default":null},
+
|{"name":"y","type":["null","string"],"default":null}]}],"default":null},
+ |{"name":"ts","type":"long"}]}""".stripMargin)
+ HoodieTableMetaClient.newTableBuilder()
+ .setTableType(tableType)
+ .setTableName(tableName)
+ .setRecordKeyFields("id")
+ .setOrderingFields("ts")
+
.initTable(HadoopFSUtils.getStorageConf(spark.sessionState.newHadoopConf()),
tablePath)
+ val jsc = new JavaSparkContext(spark.sparkContext)
+ val writeConfig =
HoodieWriteConfig.newBuilder().withPath(tablePath).withSchema(schema.toString).forTable(tableName).build()
+ val client = new SparkRDDWriteClient[HoodieAvroPayload](new
HoodieSparkEngineContext(jsc), writeConfig)
+ def nestedBranch(x: String, y: String): GenericRecord = {
+ val branchSchema =
schema.getField("nested").get().schema().getTypes.get(2).getAvroSchema
+ val branchRecord = new GenericData.Record(branchSchema)
+ branchRecord.put("x", x)
+ branchRecord.put("y", y)
+ branchRecord
+ }
+ def record(id: Int, choice: AnyRef, pick: AnyRef, nested: AnyRef, ts:
Long): HoodieRecord[HoodieAvroPayload] = {
+ val avroRecord = new GenericData.Record(schema.getAvroSchema)
+ avroRecord.put("id", Int.box(id))
+ avroRecord.put("choice", choice)
+ avroRecord.put("pick", pick)
+ avroRecord.put("nested", nested)
+ avroRecord.put("ts", Long.box(ts))
+ new HoodieAvroRecord(new HoodieKey(id.toString, ""), new
HoodieAvroPayload(HOption.of[GenericRecord](avroRecord)))
+ }
+ try {
+ val insertInstant = client.startCommit()
+ client.commit(insertInstant, client.insert(jsc.parallelize(Seq(
+ record(1, "a1", Long.box(100L), nestedBranch("n1", "n2"), 1000L),
+ record(2, Int.box(7), "b2", Int.box(9), 1000L)).asJava),
insertInstant))
+ if (tableType == HoodieTableType.MERGE_ON_READ) {
+ // The update writes a log file, so the pruned reads below merge
base and log records
+ val updateInstant = client.startCommit()
+ client.commit(updateInstant, client.upsert(jsc.parallelize(Seq(
+ record(1, "a1", Long.box(100L), nestedBranch("n1", "n2"),
1001L)).asJava), updateInstant))
+ }
+ } finally {
+ client.close()
+ }
+ val choiceSchema = new TableSchemaResolver(createMetaClient(spark,
tablePath)).getTableSchema.getField("choice").get().schema()
+ assertEquals(HoodieSchemaType.UNION, choiceSchema.getType)
+ assertEquals(3, choiceSchema.getTypes.size())
+
+
spark.read.format("hudi").load(tablePath).createOrReplaceTempView(tableName)
+ // One member out of two: Spark prunes the struct down to it while the
reader still emits both
+ val member1DF = spark.sql(s"SELECT id, choice.member1 FROM $tableName")
+ assertEquals(StructType(Seq(StructField("member1", IntegerType,
nullable = true))), prunedStructTypeOf(member1DF, "choice"))
+ checkAnswer(s"SELECT id, choice.member1 FROM $tableName")(Seq(1,
null), Seq(2, 7))
+ checkAnswer(s"SELECT id, choice.member0 FROM $tableName")(Seq(1,
"a1"), Seq(2, null))
+ // Two members out of three, and not the leading ones
+ checkAnswer(s"SELECT id, pick.member0, pick.member2 FROM
$tableName")(Seq(1, null, 100L), Seq(2, "b2", null))
+ checkAnswer(s"SELECT id, pick.member2 FROM $tableName")(Seq(1, 100L),
Seq(2, null))
+ // A record branch: the required schema converts back to a union over
a record, and the projection
+ // has to drop both the other member and the inner field Spark did not
ask for
+ val nestedDF = spark.sql(s"SELECT id, nested.member1.x FROM
$tableName")
+ assertEquals(
+ StructType(Seq(StructField("member1",
StructType(Seq(StructField("x", StringType, nullable = true))), nullable =
true))),
+ prunedStructTypeOf(nestedDF, "nested"))
+ checkAnswer(s"SELECT id, nested.member1.x FROM $tableName")(Seq(1,
"n1"), Seq(2, null))
+ checkAnswer(s"SELECT id, nested.member0 FROM $tableName")(Seq(1,
null), Seq(2, 9))
+ checkAnswer(s"SELECT id, nested.member1 FROM $tableName")(Seq(1,
Row("n1", "n2")), Seq(2, null))
+ // The whole struct is not pruned and comes back as written, next to
the ordering value the log record carries
+ val tsOfFirstRecord = if (tableType == HoodieTableType.MERGE_ON_READ)
1001L else 1000L
+ checkAnswer(s"SELECT id, choice, ts FROM $tableName")(Seq(1, Row("a1",
null), tsOfFirstRecord), Seq(2, Row(null, 7), 1000L))
+ }
+ }
+ }
+
+ test("Test a projection of columns named memberN is not read as a union") {
+ withTempDir { tmp =>
+ Seq("cow", "mor").foreach { tableType =>
+ val tableName = generateTableName
+ val tablePath = s"${tmp.getCanonicalPath}/$tableName"
+
+ // canBeUnion matches a struct whose fields are all nullable and named
member0..memberN. At the
+ // root that is a projection of columns that happen to be named that
way, not a union, and
+ // converting it to one made pruneDataSchema hand the reader the whole
table schema -- so
+ // SELECT member0 came back with _hoodie_commit_time -- or made Avro
reject the duplicate
+ // branch types outright.
+ spark.sql(
+ s"""
+ |CREATE TABLE $tableName USING HUDI
+ |TBLPROPERTIES (type = '$tableType', primaryKey = 'id',
orderingFields = 'ts')
+ |LOCATION '$tablePath'
+ |AS SELECT 1 AS id, 'a1' AS member0, 'b1' AS member1, 123456 AS ts
+ """.stripMargin)
+ // The update writes a log file on MOR, so the reads below go through
the merge path
+ spark.sql(s"UPDATE $tableName SET ts = 123457 WHERE id = 1")
+
+ checkAnswer(s"SELECT member0 FROM $tableName")(Seq("a1"))
+ checkAnswer(s"SELECT member0, member1 FROM $tableName")(Seq("a1",
"b1"))
+ checkAnswer(s"SELECT id, member0 FROM $tableName")(Seq(1, "a1"))
+ }
+ }
+ }
+
test("Test no nested schema pruning when disabled") {
withTempDir { tmp =>
val tableName = generateTableName
