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 ef834187ce47 feat(variant): infer shredding schemas by default on
Spark 4.1+ (#19808)
ef834187ce47 is described below
commit ef834187ce47eb235110f35160ccc961e577fca9
Author: voonhous <[email protected]>
AuthorDate: Wed Sep 2 18:19:36 2026 +0800
feat(variant): infer shredding schemas by default on Spark 4.1+ (#19808)
hoodie.parquet.variant.shredding.schema.inference.enabled now
defaults to true, matching Spark 4.1's own
spark.sql.variant.inferShreddingSchema and writeShredding defaults.
The key is new in 1.3.0, so no released table changes behavior on
upgrade; flipping it after a release is what would have.
Only writers with a Spark 4.1+ inferrer on the classpath change.
Spark 3.x, Spark 4.0, Flink and Java keep writing unshredded. The
schema is inferred per file from a sample of its first records
(4096 rows / 64MB), for top-level variant columns only, in base
files and in native parquet log files on table version 10+; Avro
log blocks stay unshredded and shred at compaction. Shredded-ness
is therefore per file, never a table-level property, and mixed
layouts within a table are supported.
Shredded files can only be read back by Spark 4.1+. Spark 4.0,
Hive and Flink already failed fast; Spark 3.x now does too. Its
documented way to read a variant table declares the column as
struct<value: binary, metadata: binary>, and parquet matched that
request against a shredded group by name, leaving typed_value
behind and returning a null value for every shredded row.
ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs
rejects that read from Spark33/34/35ParquetReader and the legacy
file format for base files, and from the new
Spark3HoodieParquetReadSupport, returned by BaseSpark3Adapter, for
log blocks, which HoodieSparkParquetReader reads through the read
support instead. The anchor is two-sided: on the request, binary
members named metadata and value and nothing else - a subset too,
since nested schema pruning narrows SELECT v.value to one member -
resolved case-insensitively because the Spark 3.x readers force
spark.sql.caseSensitive=false; on the file, typed_value beside a
binary metadata, the anchor the Hive and Spark 4.0 guards use. A
plain user struct, an unshredded file and a query that does not
project the column are untouched.
The row-writer factory now orders its inference gates by cost -
static inferrer lookup, StructType scan for a top-level variant,
then the config-schema parse - so a classpath that cannot infer,
and any table without a variant, no longer pays an uncached parse
of hoodie.avro.schema per file handle.
Config docs: the reader list names Spark 3.x, and the un-shred
recipe names hoodie.parquet.variant.write.shredding.enabled=false,
the key that actually strips typed_value from a schema read back
off shredded files. Opt out with either that key or the inference
key on tables other engines read; clustering with it off rewrites
already shredded files unshredded.
Tests: TestVariantSchemaUtils and the Avro writer factory pin the
default; TestParquetSchemaEvolutionUtils covers the schema walk
(nesting, pruning, case, exemptions); TestSpark3HoodieParquetReadSupport
covers the read-support wiring over a hand-built InitContext;
TestVariantDataType gains a Spark 3.x fail-fast leg over the mixed
shredded fixture, full and pruned, and its COW inference leg now
relies on the default.
---
.../row/HoodieInternalRowFileWriterFactory.java | 26 +++-
.../hudi/common/config/HoodieStorageConfig.java | 36 +++---
.../hudi/common/avro/TestVariantSchemaUtils.java | 16 ++-
...oodieAvroFileWriterFactoryVariantInference.java | 16 ++-
.../parquet/ParquetSchemaEvolutionUtils.scala | 103 +++++++++++++++-
.../parquet/TestParquetSchemaEvolutionUtils.scala | 132 ++++++++++++++++++++-
.../sql/hudi/dml/schema/TestVariantDataType.scala | 49 +++++++-
.../spark/sql/adapter/BaseSpark3Adapter.scala | 19 ++-
.../parquet/Spark3HoodieParquetReadSupport.scala | 61 ++++++++++
.../Spark3LegacyHoodieParquetFileFormat.scala | 8 ++
.../TestSpark3HoodieParquetReadSupport.scala | 89 ++++++++++++++
.../datasources/parquet/Spark33ParquetReader.scala | 5 +
.../datasources/parquet/Spark34ParquetReader.scala | 5 +
.../datasources/parquet/Spark35ParquetReader.scala | 5 +
14 files changed, 533 insertions(+), 37 deletions(-)
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieInternalRowFileWriterFactory.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieInternalRowFileWriterFactory.java
index b5a3ab365133..7d7a76fe6957 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieInternalRowFileWriterFactory.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieInternalRowFileWriterFactory.java
@@ -18,6 +18,7 @@
package org.apache.hudi.io.storage.row;
+import org.apache.hudi.SparkAdapterSupport$;
import org.apache.hudi.common.avro.VariantSchemaUtils;
import org.apache.hudi.common.avro.VariantShreddingRuntime;
import org.apache.hudi.common.avro.VariantShreddingSchemaInferrer;
@@ -36,6 +37,7 @@ import
org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
import org.apache.hudi.table.HoodieTable;
import org.apache.hadoop.conf.Configuration;
+import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import java.io.IOException;
@@ -87,11 +89,15 @@ public class HoodieInternalRowFileWriterFactory {
// The row write support resolves its HoodieSchema from the config
(hoodie.write.schema /
// hoodie.avro.schema), so inferable columns are detected on that same
config schema and the
// deferred creation splices a copied config; the StructType argument
stays original.
- // (getInferableVariantColumnsFromConfig only parses the schema once the
flag gates pass.)
- List<String> inferableColumns =
VariantSchemaUtils.getInferableVariantColumnsFromConfig(writeConfig);
- if (!inferableColumns.isEmpty()) {
- Option<VariantShreddingSchemaInferrer> inferrer =
VariantShreddingRuntime.lookupInferrer();
- if (inferrer.isPresent()) {
+ // Gates are ordered by cost, because with inference on by default the
flag gates inside
+ // getInferableVariantColumnsFromConfig pass on every table and the
config-schema parse behind
+ // them is uncached and per file handle: the inferrer lookup is a static
field, so a classpath
+ // that cannot infer (Spark 3.x, 4.0) pays nothing here; the StructType
scan is a handful of type
+ // checks; only a table with a top-level variant then pays the parse.
+ Option<VariantShreddingSchemaInferrer> inferrer =
VariantShreddingRuntime.lookupInferrer();
+ if (inferrer.isPresent() && hasTopLevelVariant(structType)) {
+ List<String> inferableColumns =
VariantSchemaUtils.getInferableVariantColumnsFromConfig(writeConfig);
+ if (!inferableColumns.isEmpty()) {
return new VariantShreddingInferenceInternalRowFileWriter(
inferableColumns,
VariantShreddingInferenceInternalRowFileWriter.resolveOrdinals(structType,
inferableColumns),
@@ -104,6 +110,16 @@ public class HoodieInternalRowFileWriterFactory {
return createParquetInternalRowFileWriter(path, table, writeConfig,
writeConfig, structType, bloomFilterOpt);
}
+ /** Whether any top-level column is a variant; inference applies to no other
position. */
+ private static boolean hasTopLevelVariant(StructType structType) {
+ for (StructField field : structType.fields()) {
+ if
(SparkAdapterSupport$.MODULE$.sparkAdapter().isVariantType(field.dataType())) {
+ return true;
+ }
+ }
+ return false;
+ }
+
private static HoodieInternalRowFileWriter
createParquetInternalRowFileWriter(StoragePath path,
HoodieTable table,
HoodieWriteConfig writeConfig,
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieStorageConfig.java
b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieStorageConfig.java
index 391ea62a571e..5e753b1f737c 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieStorageConfig.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieStorageConfig.java
@@ -266,8 +266,10 @@ public class HoodieStorageConfig extends HoodieConfig {
.defaultValue(true)
.sinceVersion("1.1.0")
.withDocumentation("Controls whether variant columns are written in
shredded format. "
- + "When enabled (default), variant columns with shredding
information in the schema will be written "
- + "in shredded format with typed_value columns. When disabled,
variant columns are always written "
+ + "When enabled (default), variant columns are written in shredded
format with typed_value "
+ + "columns when the write schema carries shredding information or
when "
+ + "hoodie.parquet.variant.shredding.schema.inference.enabled (on by
default) infers one for "
+ + "a top-level column. When disabled, variant columns are always
written "
+ "in unshredded format regardless of the schema. "
+ "Equivalent to Spark's spark.sql.variant.writeShredding.enabled.");
@@ -308,22 +310,28 @@ public class HoodieStorageConfig extends HoodieConfig {
public static final ConfigProperty<Boolean>
PARQUET_VARIANT_SHREDDING_SCHEMA_INFERENCE_ENABLED = ConfigProperty
.key("hoodie.parquet.variant.shredding.schema.inference.enabled")
- .defaultValue(false)
+ .defaultValue(true)
.sinceVersion("1.3.0")
- .withDocumentation("When enabled, the shredding schema for variant
columns without an explicit "
- + "typed_value in the write schema is inferred automatically per
parquet file from a sample of "
- + "the records written to that file, mirroring Spark 4.1's "
- + "spark.sql.variant.inferShreddingSchema. Requires Spark 4.1+ on
the writer classpath; "
- + "writes stay unshredded otherwise (Spark 4.0, Flink, Java
engines). Applies to every "
- + "parquet file the writer produces: base files and, on table
version 10+, the native "
+ .withDocumentation("Infers the shredding schema of variant columns that
have no explicit "
+ + "typed_value in the write schema, per parquet file, from a sample
of the records written "
+ + "to that file, mirroring Spark 4.1's
spark.sql.variant.inferShreddingSchema (also on by "
+ + "default there). Takes effect only when a Spark 4.1+ writer is on
the classpath; other "
+ + "writers (Spark 3.x, Spark 4.0, Flink, Java) ignore it and write
unshredded. Applies to "
+ + "every parquet file the writer produces: base files and, on table
version 10+, the native "
+ "parquet log files of MOR tables (each infers its own schema).
Data blocks inside "
+ "Avro-format log files, whether Avro or parquet
(hoodie.logfile.data.block.format), stay "
+ "unshredded and shred at compaction. Applies to top-level variant
columns only; a variant "
- + "nested inside a struct, array or map stays unshredded. This is a
write config rather than "
- + "a table config: SQL DML and procedures called by table name pick
it up from the table's "
- + "catalog properties, while path-based procedures, the DataSource
writer and the streamer "
- + "must be handed it explicitly. Up to 4096 records or 64MB are
buffered per "
- + "open file writer before the writer is created, on top of
parquet's own row-group "
+ + "nested inside a struct, array or map stays unshredded. Shredded
files can only be read "
+ + "back by Spark 4.1+: Spark 4.0, Spark 3.x, Hive and Flink readers
fail fast on them, so "
+ + "disable this option (or
hoodie.parquet.variant.write.shredding.enabled) on tables those "
+ + "engines read, and rewrite already shredded files by clustering
with "
+ + "hoodie.parquet.variant.write.shredding.enabled=false - that key,
not this one, is what "
+ + "strips typed_value from a schema read back off shredded files -
to return to the "
+ + "unshredded layout. This is a write config rather than a table
config: SQL DML and "
+ + "procedures called by table name pick it up from the table's
catalog properties, while "
+ + "path-based procedures, the DataSource writer and the streamer
must be handed it "
+ + "explicitly when a non-default value is wanted. Up to 4096 records
or 64MB are buffered "
+ + "per open file writer before the writer is created, on top of
parquet's own row-group "
+ "buffer, so size executor memory for concurrently open handles
accordingly. Ignored when "
+ "hoodie.parquet.variant.force.shredding.schema.for.test is set,
when write shredding "
+ "is disabled, or when the table has a schema-on-read internal
schema "
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/avro/TestVariantSchemaUtils.java
b/hudi-common/src/test/java/org/apache/hudi/common/avro/TestVariantSchemaUtils.java
index eb3d00cceadb..ecb8d379e934 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/avro/TestVariantSchemaUtils.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/avro/TestVariantSchemaUtils.java
@@ -92,8 +92,16 @@ public class TestVariantSchemaUtils {
public void testGetInferableVariantColumnsGating() {
HoodieSchema schema = schemaWithVariants();
- // Inference disabled (default)
- assertTrue(VariantSchemaUtils.getInferableVariantColumns(new
HoodieConfig(), schema).isEmpty());
+ // Inference enabled (the default since #19690)
+
assertTrue(HoodieStorageConfig.PARQUET_VARIANT_SHREDDING_SCHEMA_INFERENCE_ENABLED.defaultValue(),
+ "#19690 pins inference on by default");
+ assertEquals(Arrays.asList("v1", "v2"),
+ VariantSchemaUtils.getInferableVariantColumns(new HoodieConfig(),
schema));
+
+ // Inference disabled explicitly
+ HoodieConfig inferenceOff = new HoodieConfig();
+
inferenceOff.setValue(HoodieStorageConfig.PARQUET_VARIANT_SHREDDING_SCHEMA_INFERENCE_ENABLED,
"false");
+ assertTrue(VariantSchemaUtils.getInferableVariantColumns(inferenceOff,
schema).isEmpty());
// Write shredding disabled
HoodieConfig shreddingOff = inferenceEnabledConfig();
@@ -156,9 +164,11 @@ public class TestVariantSchemaUtils {
writeOnly.setValue("hoodie.write.schema", schemaString);
assertEquals(Arrays.asList("v1", "v2"),
VariantSchemaUtils.getInferableVariantColumnsFromConfig(writeOnly));
- // No schema in the config, or inference disabled: nothing, and no schema
parse attempted.
+ // No schema in the config, or inference disabled: nothing, and with
inference disabled no
+ // schema parse is attempted at all (the unparseable schema below would
throw otherwise).
assertTrue(VariantSchemaUtils.getInferableVariantColumnsFromConfig(inferenceEnabledConfig()).isEmpty());
HoodieConfig disabled = new HoodieConfig();
+
disabled.setValue(HoodieStorageConfig.PARQUET_VARIANT_SHREDDING_SCHEMA_INFERENCE_ENABLED,
"false");
disabled.setValue("hoodie.avro.schema", "not a schema");
assertTrue(VariantSchemaUtils.getInferableVariantColumnsFromConfig(disabled).isEmpty());
}
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieAvroFileWriterFactoryVariantInference.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieAvroFileWriterFactoryVariantInference.java
index 161f537dd101..d7f63b5ec6b0 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieAvroFileWriterFactoryVariantInference.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieAvroFileWriterFactoryVariantInference.java
@@ -54,9 +54,14 @@ import static org.junit.jupiter.api.Assumptions.assumeFalse;
/**
* Pins the no-inferrer degradation of shredding-schema inference in
- * {@link HoodieAvroFileWriterFactory}: this module's classpath carries no
Spark version module,
- * so {@link VariantShreddingRuntime#lookupInferrer()} is empty here, which is
what engines without
- * Spark 4.1+ (Flink, Java, Spark 4.0) see in production.
+ * {@link HoodieAvroFileWriterFactory}. Inference is on by default, so the
inferrer gate decides what
+ * a Spark 4.0 classpath writes: the shredding provider ships in
hudi-spark4-common and is therefore
+ * present there, which carries the factory past its provider gate and up to
this one. Flink, Java
+ * and Spark 3.x classpaths carry no provider either and stop one gate
earlier, which is why the test
+ * below names a provider class explicitly - it isolates the inferrer gate
rather than passing
+ * through whichever gate happens to fire first. This module's classpath
carries no Spark version
+ * module, so {@link VariantShreddingRuntime#lookupInferrer()} is empty here,
and a write must
+ * degrade to the plain unshredded writer.
*/
public class TestHoodieAvroFileWriterFactoryVariantInference {
@@ -64,7 +69,7 @@ public class TestHoodieAvroFileWriterFactoryVariantInference {
java.nio.file.Path tmpDir;
@Test
- public void testInferenceFlagWithoutInferrerWritesPlainUnshreddedFile()
throws Exception {
+ public void testDefaultInferenceWithoutInferrerWritesPlainUnshreddedFile()
throws Exception {
assumeFalse(VariantShreddingRuntime.lookupInferrer().isPresent(),
"this test pins the fallback for classpaths without a shredding-schema
inferrer");
@@ -72,7 +77,8 @@ public class TestHoodieAvroFileWriterFactoryVariantInference {
HoodieSchemaField.of("id",
HoodieSchema.create(HoodieSchemaType.STRING)),
HoodieSchemaField.of("v",
HoodieSchema.createNullable(HoodieSchema.createVariant()))));
HoodieConfig config = new HoodieConfig();
-
config.setValue(HoodieStorageConfig.PARQUET_VARIANT_SHREDDING_SCHEMA_INFERENCE_ENABLED,
"true");
+
assertTrue(config.getBooleanOrDefault(HoodieStorageConfig.PARQUET_VARIANT_SHREDDING_SCHEMA_INFERENCE_ENABLED),
+ "inference is on by default since #19690; this test pins what that
default does without an inferrer");
config.setValue(HoodieStorageConfig.PARQUET_COMPRESSION_CODEC_NAME,
"zstd");
// Name a provider explicitly: the factory also declines when no shredding
provider is available,
// and this module ships none, so without this the inferrer gate (the one
under test) would never
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala
index 113d7fba1358..ad67f5cd007c 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala
@@ -20,6 +20,7 @@ package org.apache.spark.sql.execution.datasources.parquet
import org.apache.hudi.SparkAdapterSupport
import org.apache.hudi.client.utils.SparkInternalSchemaConverter
import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.schema.HoodieSchema
import org.apache.hudi.common.schema.internal.{InternalSchema, Type =>
InternalType}
import org.apache.hudi.common.schema.internal.Types
import org.apache.hudi.common.schema.internal.action.InternalSchemaMerger
@@ -36,13 +37,14 @@ import org.apache.hudi.hadoop.fs.HadoopFSUtils
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import org.apache.parquet.hadoop.metadata.FileMetaData
-import org.apache.parquet.schema.{Type => ParquetType}
+import org.apache.parquet.schema.{GroupType, MessageType, Type => ParquetType}
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
import org.apache.spark.sql.HoodieSchemaUtils
import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
UnsafeProjection}
import org.apache.spark.sql.execution.datasources.SparkSchemaTransformUtils
import
org.apache.spark.sql.execution.datasources.parquet.ParquetSchemaEvolutionUtils.pruneInternalSchema
import org.apache.spark.sql.sources._
-import org.apache.spark.sql.types.{ArrayType, AtomicType, DataType, MapType,
StructType}
+import org.apache.spark.sql.types.{ArrayType, AtomicType, BinaryType,
DataType, MapType, StructType}
import java.time.ZoneId
@@ -271,6 +273,103 @@ object ParquetSchemaEvolutionUtils {
}
}
+ /**
+ * Fails the read when a column requested as the unshredded variant struct
sits over a parquet
+ * group that carries typed_value. This is the shape Spark 3.x readers use
for a variant column:
+ * Spark 3.x has no VariantType, so the table's own schema does not convert
(see
+ * BaseSpark3Adapter) and the documented way to read such a table is to
declare the column as
+ * struct<value: binary, metadata: binary> - the same shape Hive sync
writes to the
+ * metastore. Parquet reconciles requested against file fields by name, so
without this guard a
+ * shredded group's typed_value is simply not projected and the rows come
back with a null
+ * `value`: the payload is dropped silently. Reconstruction is not an option
on Spark 3.x, whose
+ * classpath carries no VariantShreddingProvider (the only implementation
ships in spark4-common),
+ * so the read fails instead, as it already does on Spark 4.0, Flink and
Hive.
+ *
+ * The anchor is two-sided: the requested side must be binary members named
`metadata` and
+ * `value`, either or both and nothing else (a struct carrying any further
member is a plain user
+ * struct, exempt here as it is in the sibling Hive and Spark 4.0 guards),
and the file must carry
+ * typed_value at that same path. A column the query does not project is
never walked, so a read
+ * that does not touch the variant keeps working, as does an unshredded file.
+ */
+ def validateNoShreddedVariantStructs(requiredSchema: StructType,
fileParquetSchema: MessageType): Unit = {
+ requiredSchema.fields.foreach { field =>
+ parquetFieldIgnoreCase(fileParquetSchema, field.name)
+ .foreach(validateNoShreddedVariantStruct(field.dataType, _,
field.name))
+ }
+ }
+
+ /**
+ * The file field a requested name resolves to. Case-insensitive on purpose:
the Spark 3.x
+ * readers this guards force spark.sql.caseSensitive=false
(SparkParquetReaderBase.read), so a
+ * column declared `V` or a member declared `Value` still lands on the
file's lower-case group,
+ * and a guard that only matched exactly would let that request straight
through to the
+ * null-value read.
+ */
+ private def parquetFieldIgnoreCase(group: GroupType, name: String):
Option[ParquetType] =
+ group.getFields.find(_.getName.equalsIgnoreCase(name))
+
+ private def validateNoShreddedVariantStruct(dataType: DataType, parquetType:
ParquetType, path: String): Unit = {
+ if (!parquetType.isPrimitive) {
+ val group = parquetType.asGroupType()
+ dataType match {
+ case struct: StructType if isUnshreddedVariantStruct(struct) =>
+ if (isShreddedVariantGroup(group)) {
+ throw new HoodieException(String.format(
+ "Column '%s' is a shredded variant (typed_value present)
requested as its unshredded "
+ + "struct shape; Spark 3.x cannot reconstruct shredded
variants, and reading it "
+ + "here would return a null value for every shredded row. Read
the table with "
+ + "Spark 4.1+, or rewrite it unshredded (e.g. cluster with "
+ + "hoodie.parquet.variant.write.shredding.enabled=false).",
path))
+ }
+ case struct: StructType =>
+ struct.fields.foreach { field =>
+ parquetFieldIgnoreCase(group, field.name)
+ .foreach(validateNoShreddedVariantStruct(field.dataType, _,
concatPath(path, field.name)))
+ }
+ case array: ArrayType =>
+
parquetListElement(group).foreach(validateNoShreddedVariantStruct(array.elementType,
_, concatPath(path, "element")))
+ case map: MapType =>
+
parquetMapValue(group).foreach(validateNoShreddedVariantStruct(map.valueType,
_, concatPath(path, "value")))
+ case _ =>
+ }
+ }
+ }
+
+ /**
+ * Whether a file group is a shredded variant: typed_value next to a binary
metadata, the two
+ * members every shredded variant group carries (value is optional under the
spec). The same
+ * file-side anchor as the sibling Hive and Spark 4.0 guards, and needed for
the same reason the
+ * requested side is exact: once pruning has narrowed a request to a lone
`value`, only the
+ * file can tell a variant apart from a user struct that merely holds a
typed_value member.
+ */
+ private def isShreddedVariantGroup(group: GroupType): Boolean = {
+ group.containsField(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD) &&
+ group.containsField(HoodieSchema.Variant.VARIANT_METADATA_FIELD) && {
+ val metadata =
group.getType(HoodieSchema.Variant.VARIANT_METADATA_FIELD)
+ metadata.isPrimitive &&
metadata.asPrimitiveType().getPrimitiveTypeName == PrimitiveTypeName.BINARY
+ }
+ }
+
+ /**
+ * Whether `struct` is the unshredded variant shape: a non-empty set of the
binary members a
+ * variant group carries, in any order. A subset counts because Spark's
nested schema pruning
+ * narrows the request to the leaves a query touches - `SELECT v.value`
reaches this guard as a
+ * one-member struct (see TestNestedSchemaPruningOptimization for the
pruning itself), and that
+ * single member is exactly the one a shredded file would return null for.
+ *
+ * Any member outside those two names exempts the struct, so a plain user
struct is untouched, as
+ * is the {metadata, value, typed_value} shape whose caller already sees the
shredded layout and
+ * is reading it deliberately - though pruning that shape down to `value`
alone does land here,
+ * since nothing then distinguishes it from the variant request this guards.
+ */
+ private def isUnshreddedVariantStruct(struct: StructType): Boolean = {
+ // Names compared case-insensitively for the same reason
parquetFieldIgnoreCase resolves them so.
+ struct.fields.nonEmpty && struct.fields.forall(field =>
+ field.dataType == BinaryType
+ &&
(field.name.equalsIgnoreCase(HoodieSchema.Variant.VARIANT_METADATA_FIELD)
+ ||
field.name.equalsIgnoreCase(HoodieSchema.Variant.VARIANT_VALUE_FIELD)))
+ }
+
/**
* The dotted path of the first PushVariantIntoScan rewrite struct in the
schema, if any (see
* SparkInternalSchemaConverter.isVariantRewriteStruct for the marker).
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala
b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala
index f462633f7347..4321728c4cc3 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala
@@ -25,17 +25,20 @@ import
org.apache.hudi.common.schema.internal.convert.InternalSchemaConverter
import org.apache.hudi.exception.HoodieException
import org.apache.parquet.hadoop.metadata.FileMetaData
-import org.apache.parquet.schema.{Type, Types}
+import org.apache.parquet.schema.{MessageType, Type, Types}
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
import
org.apache.spark.sql.execution.datasources.parquet.VariantParquetTestFixtures.{shreddedVariant,
stringKeyMap, threeLevelList, twoLevelList, unshreddedVariant}
-import org.apache.spark.sql.types.{BinaryType, MetadataBuilder, StructField,
StructType}
+import org.apache.spark.sql.types.{ArrayType, BinaryType, IntegerType,
MapType, MetadataBuilder, StringType, StructField, StructType}
import org.junit.jupiter.api.{Assertions, Test}
import java.util.{Arrays, Collections, HashMap}
/**
- * Unit tests for [[ParquetSchemaEvolutionUtils.validateNoShreddedVariants]],
the schema-on-read
- * guard that fails a read the merged internal-schema request would otherwise
serve with the
- * typed_value clipped away. No SparkSession: the guard is a pure schema walk.
+ * Unit tests for the two shredded-variant read guards of
[[ParquetSchemaEvolutionUtils]]:
+ * validateNoShreddedVariants, the schema-on-read guard that fails a read the
merged
+ * internal-schema request would otherwise serve with the typed_value clipped
away, and
+ * validateNoShreddedVariantStructs, the Spark 3.x guard for a variant
requested as its
+ * unshredded struct shape. No SparkSession: both guards are pure schema walks.
*/
class TestParquetSchemaEvolutionUtils {
@@ -199,6 +202,125 @@ class TestParquetSchemaEvolutionUtils {
private def footerOf(column: Type): FileMetaData =
new FileMetaData(Types.buildMessage().addField(column).named("test"), new
HashMap[String, String](), "test")
+ /** The parquet schema of a file of one top-level column. */
+ private def schemaOf(column: Type): MessageType =
Types.buildMessage().addField(column).named("test")
+
+ /**
+ * The Spark 3.x shape: no VariantType there, so a variant column is
declared as
+ * struct<value: binary, metadata: binary> (the shape Hive sync also
writes). Either member
+ * order is the same column, and the unshredded twin of the same file must
still read.
+ */
+ @Test
+ def testValidateNoShreddedVariantStructsRejectsTopLevelShreddedVariant():
Unit = {
+ Seq(
+ ("metadata first", new StructType().add("metadata",
BinaryType).add("value", BinaryType)),
+ ("value first", new StructType().add("value",
BinaryType).add("metadata", BinaryType))
+ ).foreach { case (order, variant) =>
+ val requiredSchema = new StructType().add("v", variant)
+ val failure = Assertions.assertThrows(classOf[HoodieException], () =>
+
ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs(requiredSchema,
schemaOf(shreddedVariant("v"))))
+ Assertions.assertTrue(
+ failure.getMessage.contains("shredded variant") &&
failure.getMessage.contains("'v'"),
+ s"The $order error must name the shredded variant column, got:
${failure.getMessage}")
+
+
ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs(requiredSchema,
schemaOf(unshreddedVariant("v")))
+ }
+ }
+
+ /** The walk has to reach a variant below a struct, a list element and a map
value. */
+ @Test
+ def testValidateNoShreddedVariantStructsRejectsNestedShreddedVariant(): Unit
= {
+ Seq(
+ ("struct", new StructType().add("s", new StructType().add("inner",
variantStruct)),
+
schemaOf(Types.optionalGroup().addField(shreddedVariant("inner")).named("s")),
"'s.inner'"),
+ ("list", new StructType().add("v", ArrayType(variantStruct)),
+ schemaOf(threeLevelList("v", shreddedVariant("element"))),
"'v.element'"),
+ ("map", new StructType().add("v", MapType(StringType, variantStruct)),
+ schemaOf(stringKeyMap("v", shreddedVariant("value"))), "'v.value'")
+ ).foreach { case (leg, requiredSchema, footer, path) =>
+ val failure = Assertions.assertThrows(classOf[HoodieException], () =>
+
ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs(requiredSchema,
footer))
+ Assertions.assertTrue(failure.getMessage.contains(path),
+ s"The $leg error must name $path, got: ${failure.getMessage}")
+ }
+ }
+
+ /**
+ * The requested side must be the variant shape exactly, so a user struct
that merely contains
+ * those two names, or carries them with another type, reads the same file
untouched - as does a
+ * column the file does not hold at all. Each leg fails the test by throwing.
+ */
+ @Test
+ def testValidateNoShreddedVariantStructsLeavesOtherRequestsAlone(): Unit = {
+ Seq(
+ new StructType().add("metadata", BinaryType).add("value",
BinaryType).add("extra", BinaryType),
+ new StructType().add("metadata", BinaryType).add("value", IntegerType),
+ new StructType().add("a", BinaryType).add("b", BinaryType)
+ ).foreach { requested =>
+ ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs(
+ new StructType().add("v", requested), schemaOf(shreddedVariant("v")))
+ }
+
+ // A column added after the file was written has no footer field to walk.
+ ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs(
+ new StructType().add("added", variantStruct),
schemaOf(shreddedVariant("v")))
+
+ // The file side is anchored too: a user struct that merely holds a
typed_value member has no
+ // binary metadata beside it, so even a request pruned down to `value`
alone - which on the
+ // requested side is indistinguishable from a variant - is left to read.
+ val userStructWithTypedValue = Types.optionalGroup()
+ .addField(Types.optional(PrimitiveTypeName.BINARY).named("value"))
+ .addField(Types.optional(PrimitiveTypeName.INT32).named("typed_value"))
+ .named("v")
+ ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs(
+ new StructType().add("v", new StructType().add("value", BinaryType)),
schemaOf(userStructWithTypedValue))
+ }
+
+ /**
+ * Nested schema pruning narrows the request to the leaves a query touches,
so `SELECT v.value`
+ * arrives here as a one-member struct. That member is exactly the one a
shredded file returns
+ * null for, so a subset has to be rejected too.
+ */
+ @Test
+ def testValidateNoShreddedVariantStructsRejectsPrunedVariantStruct(): Unit =
{
+ Seq("value", "metadata").foreach { member =>
+ val requiredSchema = new StructType().add("v", new
StructType().add(member, BinaryType))
+ val failure = Assertions.assertThrows(classOf[HoodieException], () =>
+
ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs(requiredSchema,
schemaOf(shreddedVariant("v"))))
+ Assertions.assertTrue(failure.getMessage.contains("'v'"),
+ s"The pruned-to-$member error must name the column, got:
${failure.getMessage}")
+
+
ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs(requiredSchema,
schemaOf(unshreddedVariant("v")))
+ }
+ }
+
+ /**
+ * The Spark 3.x readers force spark.sql.caseSensitive=false, so a request
spelled in another case
+ * still resolves onto the file's lower-case group - at the column, at the
member, and below a
+ * struct. The guard has to follow the same resolution or the mixed-case
spelling reads past it.
+ */
+ @Test
+ def testValidateNoShreddedVariantStructsIgnoresCase(): Unit = {
+ val mixedCaseVariant = new StructType().add("Value",
BinaryType).add("Metadata", BinaryType)
+ Seq(
+ ("column", new StructType().add("V", mixedCaseVariant),
schemaOf(shreddedVariant("v")), "'V'"),
+ ("nested", new StructType().add("S", new StructType().add("Inner",
mixedCaseVariant)),
+
schemaOf(Types.optionalGroup().addField(shreddedVariant("inner")).named("s")),
"'S.Inner'")
+ ).foreach { case (leg, requiredSchema, fileSchema, path) =>
+ val failure = Assertions.assertThrows(classOf[HoodieException], () =>
+
ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs(requiredSchema,
fileSchema))
+ Assertions.assertTrue(failure.getMessage.contains(path),
+ s"The mixed-case $leg error must name $path as requested, got:
${failure.getMessage}")
+ }
+
+ ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs(
+ new StructType().add("V", mixedCaseVariant),
schemaOf(unshreddedVariant("v")))
+ }
+
+ /** How a variant column is declared on Spark 3.x, which has no VariantType.
*/
+ private def variantStruct: StructType =
+ new StructType().add("value", BinaryType).add("metadata", BinaryType)
+
/**
* What a 2-level repeated group wraps here: a single struct field "e"
holding the shredded
* variant "inner". The repeated group is itself the element record, so
without the name arms
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
index 7644c232d61b..f89b51cfce65 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
@@ -1090,6 +1090,49 @@ class TestVariantDataType extends HoodieSparkSqlTestBase
with VariantShreddingTe
}
}
+ test(s"Test Backward Compatibility: Shredded Variant Table Read Fails Fast
in Spark 3.x") {
+ // The compat recipe above (declare the variant column as its two binary
members) returns a null
+ // `value` for every shredded row, because parquet matches the request
against the file by name
+ // and simply leaves typed_value behind. Spark 3.x cannot reconstruct one
- no
+ // VariantShreddingProvider on its classpath - so the read must fail
instead, through
+ // ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs. This is
the leg that pins the
+ // guard's call sites; the schema walk itself is covered by
TestParquetSchemaEvolutionUtils.
+ assume(HoodieSparkUtils.isSpark3, "This test verifies Spark 3.x rejects
shredded Variant tables")
+
+ withTempDir { tmpDir =>
+ // The fixture's first file group is shredded and its second is not, so
the failure has to come
+ // from the shredded file rather than from the table being uniformly
unreadable.
+ HoodieTestUtils.extractZipToDirectory(
+ "variant_backward_compat/variant_shredded_mixed_cow.zip",
tmpDir.toPath, getClass)
+ val tablePath =
tmpDir.toPath.resolve("variant_shredded_mixed_cow").toString
+ val tableName = generateTableName
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | v struct<value: binary, metadata: binary>,
+ | ts long
+ |) using hudi
+ |location '$tablePath'
+ |tblproperties (
+ | primaryKey = 'id',
+ | tableType = 'cow',
+ | preCombineField = 'ts'
+ |)
+ """.stripMargin)
+
+ Seq("select id, v from %s order by id", "select id, v.value from %s
order by id").foreach { query =>
+ val failure =
intercept[Exception](spark.sql(query.format(tableName)).collect())
+ // Assert the guard's own message, not just any failure naming the
column: the second query
+ // is pruned to a single member, which must land on the guard rather
than read past it.
+ val messages =
Iterator.iterate[Throwable](failure)(_.getCause).takeWhile(_ != null)
+ .map(t => if (t.getMessage == null) "" else t.getMessage)
+ assert(messages.exists(m => m.contains("shredded variant") &&
m.contains("'v'")),
+ s"The read must fail with the shredded-variant guard, got:
${failure}")
+ }
+ }
+ }
+
/**
* Helper method to verify backward compatibility of reading Spark 4.0
Variant tables in Spark 3.x
*/
@@ -1289,6 +1332,9 @@ class TestVariantDataType extends HoodieSparkSqlTestBase
with VariantShreddingTe
withRecordType()(withTempDir { tmp =>
val tableName = generateTableName
+ // Deliberately no
hoodie.parquet.variant.shredding.schema.inference.enabled here: this leg
+ // relies on the default (on since #19690) so the flip is pinned end to
end - the inferred
+ // footers asserted below would stop appearing if the default went back
to false.
spark.sql(
s"""
|create table $tableName (
@@ -1302,8 +1348,7 @@ class TestVariantDataType extends HoodieSparkSqlTestBase
with VariantShreddingTe
| tblproperties (
| primaryKey = 'id',
| type = 'cow',
- | preCombineField = 'ts',
- | hoodie.parquet.variant.shredding.schema.inference.enabled =
'true'
+ | preCombineField = 'ts'
| )
""".stripMargin)
diff --git
a/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark3Adapter.scala
b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark3Adapter.scala
index 90be1df5068d..92010be7f336 100644
---
a/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark3Adapter.scala
+++
b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark3Adapter.scala
@@ -23,9 +23,11 @@ import org.apache.hudi.common.model.FileSlice
import org.apache.hudi.common.schema.HoodieSchema
import org.apache.hudi.common.table.HoodieTableMetaClient
import org.apache.hudi.common.table.cdc.HoodieCDCFileSplit
+import org.apache.hudi.common.util.{Option => HOption}
import org.apache.hudi.common.util.JsonUtils
import org.apache.hudi.spark.internal.ReflectUtil
+import org.apache.parquet.schema.MessageType
import org.apache.parquet.schema.Type
import org.apache.parquet.schema.Type.Repetition
import org.apache.spark.api.java.JavaSparkContext
@@ -42,9 +44,10 @@ import
org.apache.spark.sql.catalyst.planning.PhysicalOperation
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.trees.Origin
import org.apache.spark.sql.catalyst.util.DateFormatter
+import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec
import org.apache.spark.sql.execution.{PartitionedFileUtil, QueryExecution,
SQLExecution}
import org.apache.spark.sql.execution.datasources._
-import org.apache.spark.sql.execution.datasources.parquet.HoodieFormatTrait
+import org.apache.spark.sql.execution.datasources.parquet.{HoodieFormatTrait,
HoodieParquetReadSupport, Spark3HoodieParquetReadSupport}
import org.apache.spark.sql.execution.streaming.MemoryStream
import org.apache.spark.sql.hudi.{HoodieMemoryStream, SparkAdapter}
import org.apache.spark.sql.sources.{BaseRelation, Filter}
@@ -220,6 +223,20 @@ abstract class BaseSpark3Adapter extends SparkAdapter with
Logging {
): Type = {
throw new UnsupportedOperationException("Spark 3.x does not support
VariantType")
}
+ override def createParquetReadSupport(convertTz: Option[java.time.ZoneId],
+ enableVectorizedReader: Boolean,
+ enableTimestampFieldRepair: Boolean,
+ datetimeRebaseSpec: RebaseSpec,
+ tableSchemaOpt: HOption[MessageType])
+ : HoodieParquetReadSupport = {
+ // Spark 3.x reads a variant as its unshredded struct shape and cannot
reconstruct a shredded
+ // one; the subclass rejects such a file rather than returning null
values. Needed on this route
+ // specifically because log blocks are read through the read support
rather than through the
+ // per-version parquet reader that guards base files.
+ new Spark3HoodieParquetReadSupport(convertTz, enableVectorizedReader,
enableTimestampFieldRepair,
+ datetimeRebaseSpec, getRebaseSpec("LEGACY"), tableSchemaOpt)
+ }
+
override def isVariantShreddingStruct(structType: StructType): Boolean = {
// Spark 3.x does not support Variant shredding
false
diff --git
a/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark3HoodieParquetReadSupport.scala
b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark3HoodieParquetReadSupport.scala
new file mode 100644
index 000000000000..0c11f31f6321
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark3HoodieParquetReadSupport.scala
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.parquet
+
+import org.apache.hudi.common.util.{Option => HOption}
+
+import org.apache.parquet.hadoop.api.InitContext
+import org.apache.parquet.hadoop.api.ReadSupport.ReadContext
+import org.apache.parquet.schema.MessageType
+import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec
+import org.apache.spark.sql.types.StructType
+
+import java.time.ZoneId
+
+/**
+ * The Spark 3.x [[HoodieParquetReadSupport]], rejecting a shredded variant
the request cannot
+ * reconstruct. Mirrors
[[org.apache.spark.sql.adapter.BaseSpark3Adapter#createParquetReadSupport]]'s
+ * Spark 4.0 sibling, which rejects at the same point for the same reason.
+ *
+ * The per-version parquet readers guard base-file reads, but they are not the
only route: log
+ * blocks - native parquet log files and the inline blocks of an avro log file
- are read by
+ * {@code HoodieSparkParquetReader.getUnsafeRowIterator}, which builds a
{@code ParquetReader} on
+ * this read support instead. A shredded variant in a log block therefore only
meets a guard here.
+ */
+class Spark3HoodieParquetReadSupport(convertTz: Option[ZoneId],
+ enableVectorizedReader: Boolean,
+ enableTimestampFieldRepair: Boolean,
+ datetimeRebaseSpec: RebaseSpec,
+ int96RebaseSpec: RebaseSpec,
+ tableSchemaOpt: HOption[MessageType] =
HOption.empty())
+ extends HoodieParquetReadSupport(
+ convertTz, enableVectorizedReader, enableTimestampFieldRepair,
+ datetimeRebaseSpec, int96RebaseSpec, tableSchemaOpt) {
+
+ override def init(context: InitContext): ReadContext = {
+ val readContext = super.init(context)
+ // Anchored on the catalyst request and the file schema, not on the
requested parquet schema:
+ // a Spark 3.x read asks for the variant's binary members alone, so the
requested schema has
+ // already had typed_value clipped away by the time it gets here and only
the file can show
+ // that the column is shredded.
+
Option(context.getConfiguration.get(ParquetReadSupport.SPARK_ROW_REQUESTED_SCHEMA))
+ .map(StructType.fromString)
+ .foreach(ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs(_,
context.getFileSchema))
+ readContext
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark3LegacyHoodieParquetFileFormat.scala
b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark3LegacyHoodieParquetFileFormat.scala
index 1e4411943e0e..e84201f1a33c 100644
---
a/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark3LegacyHoodieParquetFileFormat.scala
+++
b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark3LegacyHoodieParquetFileFormat.scala
@@ -247,6 +247,14 @@ abstract class
Spark3LegacyHoodieParquetFileFormat(shouldAppendPartitionValues:
// Clone new conf
val hadoopAttemptConf = new
Configuration(broadcastedHadoopConf.value.value)
+ // A variant column is declared as its unshredded struct shape on Spark
3.x (no VariantType);
+ // reject a file that shreds it before either branch below, so the read
fails naming the
+ // column instead of projecting the group by name and returning a null
value for every
+ // shredded row. Gated like the schema-on-read guard: an empty
projection (count(*)) reads no
+ // column data and must not pay a footer read.
+ if (requiredSchema.nonEmpty) {
+
ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs(requiredSchema,
footerFileMetaData.getSchema)
+ }
val typeChangeInfos: java.util.Map[Integer, Pair[DataType, DataType]] =
if (shouldUseInternalSchema) {
// Same guard as ParquetSchemaEvolutionUtils.getHadoopConfClone:
schema-on-read cannot
// reconstruct shredded variants, so fail loudly instead of silently
dropping typed_value.
diff --git
a/hudi-spark-datasource/hudi-spark3-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestSpark3HoodieParquetReadSupport.scala
b/hudi-spark-datasource/hudi-spark3-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestSpark3HoodieParquetReadSupport.scala
new file mode 100644
index 000000000000..1edd02fb3cb8
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark3-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestSpark3HoodieParquetReadSupport.scala
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.parquet
+
+import org.apache.hudi.common.util.{Option => HOption}
+import org.apache.hudi.exception.HoodieException
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.parquet.hadoop.api.InitContext
+import org.apache.parquet.schema.{MessageType, Type, Types}
+import org.apache.spark.sql.execution.datasources.DataSourceUtils
+import
org.apache.spark.sql.execution.datasources.parquet.VariantParquetTestFixtures.{shreddedVariant,
unshreddedVariant}
+import org.apache.spark.sql.types.{BinaryType, StructType}
+import org.junit.jupiter.api.{Assertions, Test}
+
+import java.util.{HashMap => JHashMap, Set => JSet}
+
+/**
+ * Pins the wiring of [[Spark3HoodieParquetReadSupport]] over a hand-built
{@link InitContext}: that
+ * `init` reads the catalyst request from SPARK_ROW_REQUESTED_SCHEMA and hands
the guard the file
+ * schema. The schema walk itself is covered by
TestParquetSchemaEvolutionUtils; a slip in either
+ * wire here would not fail there, it would fall straight back to the
null-value read this class
+ * exists to stop. No SparkSession: Spark's own `init` only needs the
requested schema in the conf.
+ */
+class TestSpark3HoodieParquetReadSupport {
+
+ /** A variant column the way a Spark 3.x reader declares it: its two binary
members. */
+ private val variantRequestedAsStruct =
+ new StructType().add("v", new StructType().add("value",
BinaryType).add("metadata", BinaryType))
+
+ @Test
+ def testInitRejectsShreddedFileForVariantRequestedAsStruct(): Unit = {
+ val failure = Assertions.assertThrows(classOf[HoodieException], () =>
+ readSupport().init(initContext(variantRequestedAsStruct,
fileOf(shreddedVariant("v")))))
+ Assertions.assertTrue(
+ failure.getMessage.contains("shredded variant") &&
failure.getMessage.contains("'v'"),
+ s"init must fail with the guard's own message naming the column, got:
${failure.getMessage}")
+ }
+
+ @Test
+ def testInitReadsUnshreddedFileForVariantRequestedAsStruct(): Unit = {
+ val context = readSupport().init(initContext(variantRequestedAsStruct,
fileOf(unshreddedVariant("v"))))
+ Assertions.assertTrue(context.getRequestedSchema.containsField("v"),
+ "an unshredded file must initialise as before, with the column in the
requested schema")
+ }
+
+ /**
+ * Why `init` anchors on the file schema: Spark clips the requested parquet
schema to the catalyst
+ * request, and a two-member request leaves no typed_value in it. Anchoring
on the requested schema
+ * instead would therefore never fire - which is what the base class,
guard-free, demonstrates.
+ */
+ @Test
+ def testClippedRequestCarriesNoTypedValue(): Unit = {
+ val base = new HoodieParquetReadSupport(None, false, false, rebaseSpec,
rebaseSpec, HOption.empty())
+ val context = base.init(initContext(variantRequestedAsStruct,
fileOf(shreddedVariant("v"))))
+ val requested = context.getRequestedSchema
+
Assertions.assertFalse(requested.getType(requested.getFieldIndex("v")).asGroupType().containsField("typed_value"),
+ "the clipped request must not carry typed_value, so only the file schema
can show the shredding")
+ }
+
+ private def readSupport(): Spark3HoodieParquetReadSupport =
+ new Spark3HoodieParquetReadSupport(None, false, false, rebaseSpec,
rebaseSpec, HOption.empty())
+
+ /** A CORRECTED rebase spec built the version-neutral way the readers
themselves use. */
+ private def rebaseSpec = DataSourceUtils.datetimeRebaseSpec(_ => null,
"CORRECTED")
+
+ private def initContext(requested: StructType, fileSchema: MessageType):
InitContext = {
+ val conf = new Configuration(false)
+ conf.set(ParquetReadSupport.SPARK_ROW_REQUESTED_SCHEMA, requested.json)
+ new InitContext(conf, new JHashMap[String, JSet[String]](), fileSchema)
+ }
+
+ private def fileOf(column: Type): MessageType =
Types.buildMessage().addField(column).named("file")
+}
diff --git
a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark33ParquetReader.scala
b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark33ParquetReader.scala
index becf3911c426..e5f94df47866 100644
---
a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark33ParquetReader.scala
+++
b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark33ParquetReader.scala
@@ -113,6 +113,11 @@ class Spark33ParquetReader(enableVectorizedReader: Boolean,
}
lazy val footerFileMetaData = fileFooter.getFileMetaData
+ // A variant column is declared as its unshredded struct shape on Spark
3.x (no VariantType);
+ // reject a file that shreds it here, before the reader is built, so the
read fails naming the
+ // column instead of projecting the group by name and returning a null
value for every
+ // shredded row.
+
ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs(requiredSchema,
footerFileMetaData.getSchema)
val datetimeRebaseSpec = DataSourceUtils.datetimeRebaseSpec(
footerFileMetaData.getKeyValueMetaData.get,
datetimeRebaseModeInRead)
diff --git
a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark34ParquetReader.scala
b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark34ParquetReader.scala
index 9f09d03dba07..f350afacfecf 100644
---
a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark34ParquetReader.scala
+++
b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark34ParquetReader.scala
@@ -110,6 +110,11 @@ class Spark34ParquetReader(enableVectorizedReader: Boolean,
}
lazy val footerFileMetaData = fileFooter.getFileMetaData
+ // A variant column is declared as its unshredded struct shape on Spark
3.x (no VariantType);
+ // reject a file that shreds it here, before the reader is built, so the
read fails naming the
+ // column instead of projecting the group by name and returning a null
value for every
+ // shredded row.
+
ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs(requiredSchema,
footerFileMetaData.getSchema)
val datetimeRebaseSpec = DataSourceUtils.datetimeRebaseSpec(
footerFileMetaData.getKeyValueMetaData.get,
datetimeRebaseModeInRead)
diff --git
a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark35ParquetReader.scala
b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark35ParquetReader.scala
index e91d01d07f22..31955d13dbdf 100644
---
a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark35ParquetReader.scala
+++
b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark35ParquetReader.scala
@@ -117,6 +117,11 @@ class Spark35ParquetReader(enableVectorizedReader: Boolean,
}
val footerFileMetaData = fileFooter.getFileMetaData
+ // A variant column is declared as its unshredded struct shape on Spark
3.x (no VariantType);
+ // reject a file that shreds it here, before the reader is built, so the
read fails naming the
+ // column instead of projecting the group by name and returning a null
value for every
+ // shredded row.
+
ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs(requiredSchema,
footerFileMetaData.getSchema)
val datetimeRebaseSpec = DataSourceUtils.datetimeRebaseSpec(
footerFileMetaData.getKeyValueMetaData.get,
datetimeRebaseModeInRead)
