[SPARK-53659][SQL] Infer Variant shredding schema when writing to Parquet #52406

cashmand · 2025-09-22T13:48:15Z

What changes were proposed in this pull request?

When writing Variant to Parquet, we want the shredding schema to adapt to the data being written on a per-file basis. This PR adds a new output writer that buffers the first few rows before starting the write, then uses the content of those rows to determine a shredding schema, and only then creates the Parquet writer with that schema.

The heuristics for determining the shredding schema are currently fairly simple: if a field appears consistently with a consistent type, we create value and typed_value, and if it appears with an inconsistent type, we only create value. We drop fields that occur in less than 10% of sampled rows, and have an upper bound of 300 total fields (counting value and typed_value separately) to avoid creating excessively wide Parquet schemas, which can cause performance issues.

Why are the changes needed?

Allows Spark to make use of the Variant shredding spec without requiring the user to manually set a shredding schema.

Does this PR introduce any user-facing change?

Only if spark.sql.variant.inferShreddingSchema and spark.sql.variant.writeShredding.enabled are both set to true. They currently false by default.

How was this patch tested?

Unit tests in PR.

Was this patch authored or co-authored using generative AI tooling?

No.

gene-db

@cashmand Thanks for this awesome feature! I left a few comments.

gene-db · 2025-09-29T16:32:52Z

...n/scala/org/apache/spark/sql/execution/datasources/parquet/InferVariantShreddingSchema.scala

+/**
+ *
+ * Infer a schema when there are Variant values in the shredding schema.
+ * Only VariantType values at the top level or nested in struct fields are replaced.


What are they replaced with?

I meant replacing VariantType in the schema with the shredding schema, but I'll change the comment to be are shredded.

gene-db · 2025-09-29T16:33:52Z

...n/scala/org/apache/spark/sql/execution/datasources/parquet/InferVariantShreddingSchema.scala

+ * Infer a schema when there are Variant values in the shredding schema.
+ * Only VariantType values at the top level or nested in struct fields are replaced.
+ * VariantType nested in arrays or maps are not modified.
+ * @param schema The original schema containing VariantType.


Does this require that the schema must contain a VariantType, and does the VariantType have to be a top-level field?

No, it operates on nested struct fields as well, and it's fine to call it if there's no VariantType at all in the schema, it will just be a no-op.

I'll change the description to be The original schema, with no shredding.

gene-db · 2025-09-29T16:37:03Z

...n/scala/org/apache/spark/sql/execution/datasources/parquet/InferVariantShreddingSchema.scala

+      .flatten
+  }
+
+  private def getValueAtPath(s: StructType, row: InternalRow, p: Seq[Int]): Option[VariantVal] = {


NIT:

Suggested change

private def getValueAtPath(s: StructType, row: InternalRow, p: Seq[Int]): Option[VariantVal] = {

private def getValueAtPath(schema: StructType, row: InternalRow, path: Seq[Int]): Option[VariantVal] = {

Also, can we comment on what this will return? It looks it will return None of the field will be null?

gene-db · 2025-09-29T16:47:05Z

...n/scala/org/apache/spark/sql/execution/datasources/parquet/InferVariantShreddingSchema.scala

+  }
+
+  /**
+   * Update each VariantType with its inferred schema.


Does it return a copy of the schema (with the updates), or does it update in place?

The former. I'll change the comment to Return a new schema, with each VariantType replaced its inferred shredding schema.

gene-db

Thanks for this feature!

LGTM

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

cloud-fan · 2025-10-27T03:51:49Z

The last commit just update the config version doc, no need to wait for CI, thanks, merging to master!

pan3793 · 2025-11-10T03:38:25Z

@cloud-fan @gene-db @cashmand, currently, variant is guarded by several internal configs, do you plan to enable variant support in Spark 4.1?

cashmand · 2025-11-10T17:12:50Z

@cloud-fan @gene-db @cashmand, currently, variant is guarded by several internal configs, do you plan to enable variant support in Spark 4.1?

Hi @pan3793, I think the only remaining item is to update the parquet reader and writer to recognize and write the Variant logical type annotation that was released in the latest version of parquet-java, and then we should be able to enable it by default.

pan3793 · 2025-11-10T17:25:29Z

@cashmand, thanks for the information, sounds promising, I hope it can happen in 4.1

…quet ### What changes were proposed in this pull request? When writing Variant to Parquet, we want the shredding schema to adapt to the data being written on a per-file basis. This PR adds a new output writer that buffers the first few rows before starting the write, then uses the content of those rows to determine a shredding schema, and only then creates the Parquet writer with that schema. The heuristics for determining the shredding schema are currently fairly simple: if a field appears consistently with a consistent type, we create `value` and `typed_value`, and if it appears with an inconsistent type, we only create `value`. We drop fields that occur in less than 10% of sampled rows, and have an upper bound of 300 total fields (counting `value` and `typed_value` separately) to avoid creating excessively wide Parquet schemas, which can cause performance issues. ### Why are the changes needed? Allows Spark to make use of the [Variant shredding spec](https://github.com/apache/parquet-format/blob/master/VariantShredding.md) without requiring the user to manually set a shredding schema. ### Does this PR introduce _any_ user-facing change? Only if `spark.sql.variant.inferShreddingSchema` and `spark.sql.variant.writeShredding.enabled` are both set to true. They currently false by default. ### How was this patch tested? Unit tests in PR. ### Was this patch authored or co-authored using generative AI tooling? No. Closes apache#52406 from cashmand/variant_shredding_inference. Lead-authored-by: cashmand <david.cashman@databricks.com> Co-authored-by: Wenchen Fan <cloud0fan@gmail.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

see apache/spark#52406

* Move GlutenStreamingQuerySuite to correct package * Add Spark 4.1 new test suites for Gluten * Enable new and existing Gluten test suites for Spark 4.1 UT * Update workflow trigger paths to exclude Spark 4.0 and 4.1 shims directories for clickhouse backend * Add support for Spark 4.1 in build script * Merge Spark 4.1.0 sql-tests into Gluten Spark 4.1 (three-way merge) Three-way merge performed using Git: - Base: Spark 4.0.1 (29434ea766b) - Left: Spark 4.1.0 (e221b56be7b) - Right: Gluten Spark 4.1 backends-velox Summary: - Auto-merged: 165 files - New tests added: 31 files (collations, edge cases, recursion, spatial, etc.) - Modified tests: 134 files - Deleted tests: 2 files (collations.sql -> split into 4 files, timestamp-ntz.sql) Conflicts resolved: - inputs/timestamp-ntz.sql: Right deleted + Left modified -> DELETED (per resolution rule) New test suites from Spark 4.1.0: - Collations (4 files): aliases, basic, padding-trim, string-functions - Edge cases (6 files): alias-resolution, extract-value, join-resolution, etc. - Advanced features: cte-recursion, generators, kllquantiles, thetasketch, time - Name resolution: order-by-alias, session-variable-precedence, runtime-replaceable - Spatial functions: st-functions (ANSI and non-ANSI variants) - Various resolution edge cases Total files after merge: 671 (up from 613) * Enable additional Spark 4.1 SQL tests by resolving TODOs * Add new Spark 4.1 test files to VeloxSQLQueryTestSettings * [Fix] Replace `RuntimeReplaceable` with its `replacement` to fix UT. see apache/spark#50287 * [4.1.0] Exclude "infer shredding with mixed scale" see apache/spark#52406 * [Fix] Implement Kryo serialization for CachedColumnarBatch see apache/spark#50599 * [4.1.0] Exclude GlutenMapStatusEndToEndSuite and configure parallelism see apache/spark#50230 * [4.1.0] Exclude Spark Structure Steaming tests in Gluten see - apache/spark#52473 - apache/spark#52870 - apache/spark#52891 * [4.1.0] Exclude failing SQL tests on Spark 4.1 * Replace SparkException.require with standard require in ColumnarCachedBatchSerializer to work across different spark versions * [Fix] Replace `RuntimeReplaceable` with its `replacement` to fix UT. see apache/spark#50287 * Exclude Spark 4.0 and 4.1 paths in clickhouse_be_trigger using `!` prefix * [Fix] Update GlutenShowNamespacesParserSuite to use GlutenSQLTestsBaseTrait

Infer shredding for parquet

0076f55

github-actions bot added the SQL label Sep 22, 2025

cashmand changed the title ~~[SPARK-53659] Infer Variant shredding schema when writing to Parquet~~ [SPARK-53659][SQL] Infer Variant shredding schema when writing to Parquet Sep 22, 2025

Cleanup

c492a6c

gene-db suggested changes Sep 29, 2025

View reviewed changes

Code review feedback

761ccad

cashmand requested a review from gene-db October 17, 2025 18:03

huaxingao mentioned this pull request Oct 21, 2025

Spark: Support writing shredded variant in Iceberg-Spark apache/iceberg#14297

Open

gene-db approved these changes Oct 24, 2025

View reviewed changes

cloud-fan reviewed Oct 27, 2025

View reviewed changes

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala Outdated Show resolved Hide resolved

cloud-fan reviewed Oct 27, 2025

View reviewed changes

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala Outdated Show resolved Hide resolved

cloud-fan reviewed Oct 27, 2025

View reviewed changes

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala Outdated Show resolved Hide resolved

Apply suggestions from code review

7044331

cloud-fan closed this in ec944cc Oct 27, 2025

baibaichen added a commit to baibaichen/gluten that referenced this pull request Jan 12, 2026

[4.1.0] Exclude "infer shredding with mixed scale"

9460289

see apache/spark#52406

baibaichen added a commit to baibaichen/gluten that referenced this pull request Jan 13, 2026

[4.1.0] Exclude "infer shredding with mixed scale"

69fd383

see apache/spark#52406

This was referenced Jan 13, 2026

[GLUTEN-11355][UT] Add new Spark 4.1 tests apache/incubator-gluten#11380

Merged

[VL] Track on Spark-4.1.x failed unit tests apache/incubator-gluten#11400

Open

baibaichen added a commit to baibaichen/gluten that referenced this pull request Jan 13, 2026

[4.1.0] Exclude "infer shredding with mixed scale"

f8d0963

see apache/spark#52406

	private def getValueAtPath(s: StructType, row: InternalRow, p: Seq[Int]): Option[VariantVal] = {
	private def getValueAtPath(schema: StructType, row: InternalRow, path: Seq[Int]): Option[VariantVal] = {

[SPARK-53659][SQL] Infer Variant shredding schema when writing to Parquet #52406

[SPARK-53659][SQL] Infer Variant shredding schema when writing to Parquet #52406

Uh oh!

Conversation

cashmand commented Sep 22, 2025

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

Uh oh!

gene-db left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gene-db left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

cloud-fan commented Oct 27, 2025

Uh oh!

pan3793 commented Nov 10, 2025

Uh oh!

cashmand commented Nov 10, 2025

Uh oh!

pan3793 commented Nov 10, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants