wombatu-kun commented on code in PR #19808:
URL: https://github.com/apache/hudi/pull/19808#discussion_r3910318194
##########
hudi-common/src/main/java/org/apache/hudi/common/config/HoodieStorageConfig.java:
##########
@@ -308,22 +310,26 @@ 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, 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 through clustering with it
disabled to return to the "
Review Comment:
"with it disabled" binds to the inference option, but
`HoodieAvroWriteSupport.generateEffectiveSchema` strips `typed_value` only when
`hoodie.parquet.variant.write.shredding.enabled` is false, which is also what
the error message added in this PR tells users. Name the write-shredding config
in the recipe here.
##########
hudi-common/src/main/java/org/apache/hudi/common/config/HoodieStorageConfig.java:
##########
@@ -308,22 +310,26 @@ 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, Hive and Flink readers fail fast
on them, so disable this "
Review Comment:
The reader list here says Spark 4.0, Hive and Flink fail fast but leaves out
Spark 3.x, which this PR itself makes fail fast through
`ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs`. Add Spark 3.x
to that sentence.
##########
hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala:
##########
@@ -199,6 +201,71 @@ class TestParquetSchemaEvolutionUtils {
private def footerOf(column: Type): FileMetaData =
new FileMetaData(Types.buildMessage().addField(column).named("test"), new
HashMap[String, String](), "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 = {
Review Comment:
These pin the schema walk, but nothing pins that the four new call sites
reach it: every Spark 3.x leg of `TestVariantDataType` runs on an unshredded
fixture. `variant_shredded_mixed_cow.zip` is already checked in and used by
`ITTestVariantCrossEngineCompatibility`, so a Spark 3 leg pointing the compat
struct DDL at it would close that gap.
##########
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)
Review Comment:
This covers base files only: `SparkFileFormatInternalRowReaderContext`
routes native and inline parquet log blocks to
`HoodieSparkParquetReader.getUnsafeRowIterator`, which builds its own
`ParquetReader` on the plain `HoodieParquetReadSupport` that `SparkAdapter`
hands Spark 3.x, so a shredded variant in a log block still reads back as a
null value. Worth calling the guard from that log branch too, or having
`BaseSpark3Adapter` override `createParquetReadSupport` the way
`Spark4_0Adapter` does.
##########
hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/hadoop/TestHoodieAvroFileWriterFactoryVariantInference.java:
##########
@@ -54,25 +54,28 @@
/**
* 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 this
gate is reached by every
Review Comment:
`HoodieAvroFileWriterFactory.newParquetFileWriter` requires
`isShreddingProviderAvailable` before it reaches the inferrer gate, and the
only provider ships in hudi-spark4-common, so on Flink, Java and Spark 3.x
classpaths a stock config short-circuits before this gate - which is why the
test names a provider class explicitly. Reword "reached by every engine's Avro
write path with a stock config".
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
