Spark: Support writing shredded variant in Iceberg-Spark by aihuaxu · Pull Request #14297 · apache/iceberg

aihuaxu · 2025-10-11T21:02:14Z

This change adds support for writing shredded variants in the iceberg-spark module, enabling Spark to write shredded variant data into Iceberg tables.

Ideally, this should follow the approach described in the reader/writer API proposal for Iceberg V4, where the execution engine provides the shredded writer schema before creating the Iceberg writer. This design is cleaner, as it delegates schema generation responsibility to the engine.

As an interim solution, this PR implements a writer with lazy initialization for the actual Parquet writer. It buffers a portion of the data first, derives the shredded schema from the buffered records, then initializes the Parquet writer and flushes the buffered data to the file.

The current shredding algorithm is to shred to the most common type for a field.

aihuaxu · 2025-10-15T01:57:15Z

@amogh-jahagirdar @Fokko @huaxingao Can you help take a look at this PR and if we have better approach for this?

aihuaxu · 2025-10-21T04:59:31Z

cc @RussellSpitzer, @pvary and @rdblue Seems it's better to have the implementation with new File Format proposal but want to check if this is acceptable approach as an interim solution or you see a better alternative.

pvary · 2025-10-21T10:01:30Z

@aihuaxu: Don't we want to do the same but instead of wrapping the ParquetWriter, we could wrap the DataWriter. The schema would be created near the SparkWrite.WriterFactory and it would be easier to move to the new API when it is ready. The added benefit would be that when other formats implement the Variant, we could reuse the code.

Would this be prohibitively complex?

huaxingao · 2025-10-21T18:32:50Z

In Spark DSv2, planning/validation happens on the driver. BatchWrite#createBatchWriterFactory runs on the driver and returns a DataWriterFactory that is serialized to executors. That factory must already carry the write schema the executors will use when they create DataWriters.

For shredded variant, we don’t know the shredded schema at planning time. We have to inspect some records to derive it. Doing a read on the driver during createBatchWriterFactory would mean starting a second job inside planning, which is not how DSv2 is intended to work.

Because of that, the current proposed Spark approach is: put the logical variant in the writer factory, on the executor, buffer the first N rows, infer the shredded schema from data, then initialize the concrete writer and flush the buffer. I believe this PR follow the same approach, which seems like a practical solution to me given DSV2's constraints.

pvary · 2025-10-22T08:47:11Z

Thanks for the explanation, @huaxingao! I see several possible workarounds for the DataWriterFactory serialization issue, but I have some more fundamental concerns about the overall approach.
I believe shredding should be driven by future reader requirements rather than by the actual data being written. Ideally, it should remain relatively stable across data files within the same table and originate from a writer job configuration—or even better, from a table-level configuration.

Even if we accept that the written data should dictate the shredding logic, Spark’s implementation—while dependent on input order—is at least somewhat stable. It drops rarely used fields, handles inconsistent types, and limits the number of columns.
I understand this is only a PoC implementation for shredding, but I’m concerned that the current simplifications make it very unstable. If I’m interpreting correctly, the logic infers the type from the first occurrence of each field and creates a column for every field. This could lead to highly inconsistent column layouts within a table, especially in IoT scenarios where multiple sensors produce vastly different data.
Did I miss anything?

aihuaxu · 2025-10-24T16:28:26Z

Thanks @huaxingao and @pvary for reviewing, and thanks to Huaxin for explaining how the writer works in Spark.

Regarding the concern about unstable schemas, Spark's approach makes sense:

If a field appears consistently with a consistent type, create both value and typed_value
If a field appears with inconsistent types, create only value
Drop fields that occur in less than 10% of sampled rows
Cap the total at 300 fields (counting value and typed_value separately)

We could implement similar heuristics. Additionally, making the shredded schema configurable would allow users to choose which fields to shred at write time based on their read patterns.

For this POC, I'd like any feedback on whether there are any significant high-level design options to consider first and if this approach is acceptable. This seems hacky. I may have missed big picture on how the writers work across Spark + Iceberg + Parquet and we may have better way.

github-actions · 2025-11-24T00:19:23Z

This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.

Tishj · 2025-11-30T21:28:05Z

This PR caught my eye, as I've implemented the equivalent in DuckDB: duckdb/duckdb#19336

The PR description doesn't give much away, but I think the approach is similar to the proposed (interim) solution here: buffer the first rowgroup, infer the shredded schema from this, then finalize the file schema and start writing data.

We've opted to create a typed_value even though the type isn't 100% consistent within the buffered data, as long as it's the most common. I think you're losing potential compression by not doing that.

We've also added a copy option to force the shredded schema, for debugging purposes and for power users.

As for DECIMAL, it's kind of a special case in the shredding inference. We only shred on a DECIMAL type if all the decimal values we've seen for a column/field have the same width+scale, if any decimal value differs, DECIMAL won't be considered anymore when determining the shredded type of the column/field

yguy-ryft · 2025-12-24T17:43:26Z

This PR is super exciting!
Does this rely on variant shredding support in Spark? Is it supported in Spark 4.1 already, or planned for future releases?

Regarding the heuristics - I'd like to propose adding table properties as hints for variant shredding.
Similarly to properties used for bloom filters, it could be good to introduce something like write.parquet.variant-shredding-enabled.column.col1, which will hint to the writer that this column is important for shredding.
Many variants have important fields for which shredding should be enforced, and other fields which are less central and can be managed with simpler heuristics.
Would love to hear your thoughts!

aihuaxu · 2026-01-09T19:55:50Z

This PR caught my eye, as I've implemented the equivalent in DuckDB: duckdb/duckdb#19336

The PR description doesn't give much away, but I think the approach is similar to the proposed (interim) solution here: buffer the first rowgroup, infer the shredded schema from this, then finalize the file schema and start writing data.

That is correct.

We've opted to create a typed_value even though the type isn't 100% consistent within the buffered data, as long as it's the most common. I think you're losing potential compression by not doing that.

I'm still trying to improve the heuristics to use the most common one as shredding type rather than the first one and probably cap the number of shredded fields, etc. but it doesn't need 100% consistent type to be shredded.

We've also added a copy option to force the shredded schema, for debugging purposes and for power users.

Yeah. I think that makes sense for advanced user to determine the shredded schema since they may know the read pattern.

As for DECIMAL, it's kind of a special case in the shredding inference. We only shred on a DECIMAL type if all the decimal values we've seen for a column/field have the same width+scale, if any decimal value differs, DECIMAL won't be considered anymore when determining the shredded type of the column/field

Why is DECIMAL special here? If we determine DECIMAL4 to be shredded type, then we may shred as DECIMAL4 or not shred if they cannot fit in DECIMAL4, right?

aihuaxu · 2026-01-09T19:58:25Z

This PR is super exciting! Does this rely on variant shredding support in Spark? Is it supported in Spark 4.1 already, or planned for future releases?

Regarding the heuristics - I'd like to propose adding table properties as hints for variant shredding. Similarly to properties used for bloom filters, it could be good to introduce something like write.parquet.variant-shredding-enabled.column.col1, which will hint to the writer that this column is important for shredding. Many variants have important fields for which shredding should be enforced, and other fields which are less central and can be managed with simpler heuristics. Would love to hear your thoughts!

Yeah. I'm also thinking of that too. Will address that separately. Basically based on read pattern, the user can specify the shredding schema.

gkpanda4

When processing JSON objects containing null field values (e.g., {"field": null}), the variant shredding creates schema columns for these null fields instead of omitting them entirely. This would cause schema bloat.

Adding a null check in ParquetVariantUtil.java:386 in the object() method should fix it.

aihuaxu · 2026-01-15T19:39:27Z

When processing JSON objects containing null field values (e.g., {"field": null}), the variant shredding creates schema columns for these null fields instead of omitting them entirely. This would cause schema bloat.

Adding a null check in ParquetVariantUtil.java:386 in the object() method should fix it.

I addressed this null value check in VariantShreddingAnalyzer.java instead. If it's NULL, then we will not add the shredded field.

pvary · 2026-03-24T12:55:38Z

+    if (delegate != null) {
+      return delegate.length();
+    }
+    return 0L;


Are we sure about the 0 length?

since there is nothing buffered yet, 0 makes sense. Let me know if you prefer a different response

We use this length to decide if we should roll over to next file, right?
Does it cause the file includes more data when it later writes the actual data?

pvary · 2026-03-24T13:51:03Z

+        for (Record row : bufferedRows) {
+          appender.add(row);
+        }
+        return appender;


nit: newline

I figured spotless would catch some of these, but maybe not. I'll fix them in upcoming commits

pvary · 2026-03-24T13:53:51Z

+    Preconditions.checkArgument(
+        bufferRowCount > 0, "bufferRowCount must be > 0, got %s", bufferRowCount);
+    Preconditions.checkNotNull(appenderFactory, "appenderFactory must not be null");
+    Preconditions.checkNotNull(copyFunc, "copyFunc must not be null");


How frequent is the need to copy?
Do we need to get this always, or is it worth to have a non-copy consructor?

pvary · 2026-03-24T13:56:07Z

+        for (Record row : bufferedRows) {
+          appender.add(row);
+        }


Suggested change

for (Record row : bufferedRows) {

appender.add(row);

}

bufferedRows.forEach(appender::add);

pvary · 2026-03-24T14:05:49Z

+    public WriteBuilder withFileSchema(MessageType newFileSchema) {
+      this.fileSchema = newFileSchema;
+      return this;
+    }


Is there an easy way to encode this to the engineSchema?

Or could we just set the createReaderFunction based on the MessageType?

pvary · 2026-03-24T14:06:40Z

+    public DataWriteBuilder withFileSchema(MessageType newFileSchema) {
+      appenderBuilder.withFileSchema(newFileSchema);
+      return this;
+    }
+


Why do we expose this?

pvary · 2026-03-24T14:16:15Z

+    MessageTypeBuilder builder = Types.buildMessage();
+
+    for (Type field : fields) {
+      if (field != null) {


Do we need these null checks?

pvary · 2026-03-24T14:24:35Z

+    Function<List<InternalRow>, FileAppender<InternalRow>> appenderFactory =
+        bufferedRows -> {
+          Preconditions.checkNotNull(bufferedRows, "bufferedRows must not be null");
+          MessageType originalSchema = ParquetSchemaUtil.convert(dataSchema, "table");


This is not the place for things like this.
There should not be any FileFormat related thing here.

ParquetFormatModel should contain this logic as this is parquet specific.
If there are Spark specific parts, that should be a method used to parametrize the ParquetFormatModel, like the WriterBuilderFunction

+1. When shredding is enabled, this skips the FormatModelRegistry / ParquetFormatModel path and constructs the Parquet writer directly, pulling format-specific imports into the format-agnostic SparkFileWriterFactory.

Agree that this is the main part that we can refactor to follow the new FileFormat API.

I'll work on the refactor based on the comments

pvary · 2026-03-24T14:24:50Z

+
+  private boolean shouldUseVariantShredding() {
+    // Variant shredding is currently only supported for Parquet files
+    if (dataFileFormat != FileFormat.PARQUET) {


Will clean this

steveloughran · 2026-03-24T19:57:29Z

@nssalian pointed me at this: I've reviewed it as far as I'm safe to. I am doing benchmarks for read performance #15629 which goes alongside this. It'll show when shedded variant performance equals or exceeds that of unshedded.

Right now the results of #https://github.com/apache/iceberg/issues/15628#issuecomment-4120285243 show that is not yet the case, at least in the test setup

steveloughran · 2026-03-24T19:12:16Z

+    spark.conf().set(SparkSQLProperties.SHRED_VARIANTS, "true");
+
+    String values =
+        "(1, parse_json('{\"age\": \"25\"}')),"


java17 has that """" multiline string thing which is ideal for json like this

Let me do that

steveloughran · 2026-03-24T19:15:02Z

+    MessageType expectedSchema = parquetSchema(address);
+
+    Table table = validationCatalog.loadTable(tableIdent);
+    verifyParquetSchema(table, expectedSchema);


I'd add a test to make sure that row 2 had age of value 30:int, just to make sure that the parser hasn't decided to "be helpful"

steveloughran · 2026-03-24T19:32:30Z

+  }
+
+  @Override
+  public void close() throws IOException {


what happens on a create().close() sequence with no data written? it should be a no-op. Is this tested?

I'll add an empty close test

…eedback

nssalian · 2026-03-30T04:38:35Z

Made some changes based on the previous feedback:

Removed the file schema override and Spark-specific schema visitor. Now all shredding goes through the FormatModel API.
Moved column resolution into the analyzer so engine-specific logic stays in engine subclasses.
Redesigned the buffered appender with internal row replay
Added table property precedence, round-trip tests, and docs.

Benchmarks should be added in a follow up PR, IMHO and any adjustments should be made for inefficiencies.
@pvary @huaxingao @aihuaxu PTAL

steveloughran · 2026-03-30T18:55:24Z

FYI I've been benchmarking parquet row scans with variants, latest results up at
apache/parquet-java#3452 (comment)

As copilot says

Lean schema + shredded = 45% faster than reading all columns. Skipping idstr, varid, and col4 typed columns saves ~590ms per 1M rows.
Lean schema + unshredded = 93% slower. The lean schema requests typed_value.varcategory which does not exist in the unshredded file. Parquet handles the missing columns at every row, which is more expensive than
reading the single binary blob directly.
Schema detection in ReadSupport.init() is essential. Applying containsField("typed_value") to choose between lean and full schema prevents the unshredded penalty while preserving the shredded speedup.

What that means is the automatic shredding is critical, and equally critical, the query engine mustn't try to read files with a shredded schema unless there's shredded data. That's possibly what's been leading to the perf numbers I'm seeing.

Guosmilesmile · 2026-03-31T09:23:13Z

+    for (D row : buffer) {
+      delegate.add(row);
+    }


Suggested change

for (D row : buffer) {

delegate.add(row);

}

buffer.forEach(delegate::add);

Guosmilesmile · 2026-03-31T11:43:50Z

+  public static final int DEFAULT_BUFFER_SIZE = TableProperties.PARQUET_VARIANT_BUFFER_SIZE_DEFAULT;
  private final boolean isBatchReader;
+  private final VariantShreddingAnalyzer<D, S> variantAnalyzer;
+  private final UnaryOperator<D> copyFunc;


I prefer using Function<S, UnaryOperator<D>> copyFuncFactory here, because Flink's RowData does not provide a copy method directly. Instead, it needs to be wrapped with the corresponding type into a serializer to perform the copy.

rowType -> new RowDataSerializer(rowType)::copy)

@pvary WDYT?

Great call out. That should be a doable change. I'll wait for other reviewers to take a look and do all the changes together.

Guosmilesmile · 2026-03-31T11:58:25Z

+  public BufferedFileAppender(
+      int bufferRowCount, Function<List<D>, FileAppender<D>> appenderFactory) {
+    this(bufferRowCount, appenderFactory, UnaryOperator.identity());
+  }


Do we still need this? As we only create BufferedFileAppender in ParquetFormatModel, rather than creating it in the engine as before.

Removing it means anyone using BufferedFileAppender outside of ParquetFormatModel must always provide a copy func even when their rows aren't reused. Happy to remove if that's ok for a future use case. I left it in there to be more flexible.

aihuaxu · 2026-04-05T00:03:20Z

+    if (delegate != null) {
+      return delegate.length();
+    }
+    return 0L;


We use this length to decide if we should roll over to next file, right?
Does it cause the file includes more data when it later writes the actual data?

aihuaxu · 2026-04-05T00:28:53Z

+  public Map<Integer, Type> analyzeVariantColumns(
+      List<T> bufferedRows, Schema icebergSchema, S engineSchema) {
+    Map<Integer, Type> shreddedTypes = Maps.newHashMap();
+    for (org.apache.iceberg.types.Types.NestedField col : icebergSchema.columns()) {


Use imports this class.

aihuaxu · 2026-04-05T00:58:07Z

+      return null;
+    }
+
+    if (rootType == PhysicalType.OBJECT) {


How about Array and primitive types? I think we should handle them the same way?

Good catch. Will fix for array and primitive.

nssalian · 2026-04-08T04:25:38Z

To check if this works, I wrote local write + read tests that produce shredded Parquet files and ran a 200-row stress test covering nested objects, arrays of objects, mixed types, decimal precision/scale, frequency pruning, and nulls. I used the parquet-cli locally to look at the schema of the files and the metadata. I can add it to an integration test in a follow up PR since this PR is already large. @aihuaxu , @pvary, @huaxingao @RussellSpitzer PTAL.

pvary · 2026-04-16T13:00:54Z

+            .overwrite()
+            .build();
+      } catch (IOException e) {
+        throw new org.apache.iceberg.exceptions.RuntimeIOException(e);


pvary · 2026-04-16T13:09:40Z

+    assertThat(actual).hasSize(5);
+    assertThat(actual.get(0).getField("id")).isEqualTo(1L);
+    assertThat(actual.get(4).getField("id")).isEqualTo(5L);


Maybe we could move DataTestHelpers to the core and then use the methods there to check that the results are matching

DataTestHelpers in core would benefit other modules too. Happy to do that as a follow-up to keep this PR scoped to the shredding changes. Let me know what you think.

Wouldn't it be easier the other way around?
Move the DataTestHelpers first, then simplify the code here and continue the review here?

Ok. I'll get that fix in before so it helps here. Will open a PR.

pvary · 2026-04-16T13:16:14Z


    @Override
    public FileAppender<D> build() throws IOException {
+      Preconditions.checkState(content != null, "File content type must be set before building");


Removing setAll and adding this check could be done independently by another PR. They are unrelated changes and could be applied to all FormatModel implementations

pvary · 2026-04-16T13:21:31Z

              (icebergSchema, messageType) ->
                  writerFunction.write(icebergSchema, messageType, engineSchema));
+          if (shreddingEnabled && variantAnalyzer != null && hasVariantColumns(schema)) {
+            return buildShreddedAppender();


I don't really like this return here.
Could we just create a boolean flag and do this next to the other return and wrap the result of the internal.build() if needed?

pvary · 2026-04-16T13:29:04Z

+        }
+      }
+    }
+    return shreddedTypes;


nit: fix missing newlines please

Co-authored-by: Neelesh Salian <n_salian@apple.com>

github-actions Bot added spark parquet labels Oct 11, 2025

aihuaxu force-pushed the spark-write-iceberg-variant branch from 16b7a09 to dc4f72e Compare October 11, 2025 21:03

aihuaxu marked this pull request as ready for review October 11, 2025 21:15

aihuaxu force-pushed the spark-write-iceberg-variant branch 3 times, most recently from 97851f0 to b87e999 Compare October 13, 2025 16:47

huaxingao reviewed Oct 21, 2025

View reviewed changes

Comment thread parquet/src/main/java/org/apache/iceberg/parquet/ParquetWriter.java Outdated

deniskuzZ mentioned this pull request Oct 31, 2025

HIVE-29287: Iceberg: [V3] Variant Shredding support apache/hive#6152

Merged

github-actions Bot added the stale label Nov 24, 2025

github-actions Bot removed the stale label Dec 1, 2025

gkpanda4 reviewed Jan 14, 2026

View reviewed changes

aihuaxu force-pushed the spark-write-iceberg-variant branch 2 times, most recently from 2e81d79 to 7e1b608 Compare January 15, 2026 19:35

aihuaxu force-pushed the spark-write-iceberg-variant branch 4 times, most recently from 7c805f6 to 67dbe97 Compare January 15, 2026 22:50

pvary reviewed Mar 24, 2026

View reviewed changes

Comment thread core/src/main/java/org/apache/iceberg/io/BufferedFileAppender.java

pvary reviewed Mar 24, 2026

View reviewed changes

Comment thread core/src/main/java/org/apache/iceberg/io/BufferedFileAppender.java

pvary reviewed Mar 24, 2026

View reviewed changes

steveloughran mentioned this pull request Mar 24, 2026

Core, Spark: Add JMH benchmarks for Variants #15628

Open

3 tasks

steveloughran reviewed Mar 25, 2026

View reviewed changes

Wire variant shredding write path through FormatModel API as per PR f…

9df8b43

…eedback

github-actions Bot added the docs label Mar 30, 2026

Guosmilesmile reviewed Mar 31, 2026

View reviewed changes

aihuaxu commented Apr 5, 2026

View reviewed changes

pvary reviewed Apr 16, 2026

View reviewed changes

Comment thread parquet/src/main/java/org/apache/iceberg/parquet/ParquetFormatModel.java

pvary reviewed Apr 16, 2026

View reviewed changes

add co-author

8bb90a3

Co-authored-by: Neelesh Salian <n_salian@apple.com>

aihuaxu force-pushed the spark-write-iceberg-variant branch from 9c5355d to 8bb90a3 Compare April 21, 2026 16:39

Conversation

aihuaxu commented Oct 11, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

aihuaxu commented Oct 15, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

aihuaxu commented Oct 21, 2025

Uh oh!

Uh oh!

pvary commented Oct 21, 2025

Uh oh!

huaxingao commented Oct 21, 2025

Uh oh!

pvary commented Oct 22, 2025

Uh oh!

aihuaxu commented Oct 24, 2025

Uh oh!

github-actions Bot commented Nov 24, 2025

Uh oh!

Tishj commented Nov 30, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

yguy-ryft commented Dec 24, 2025

Uh oh!

aihuaxu commented Jan 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

aihuaxu commented Jan 9, 2026

Uh oh!

gkpanda4 left a comment

Choose a reason for hiding this comment

Uh oh!

aihuaxu commented Jan 15, 2026

Uh oh!

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!

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!

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!

steveloughran commented Mar 24, 2026

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

aihuaxu commented Oct 11, 2025 •

edited

Loading

aihuaxu commented Oct 15, 2025 •

edited

Loading

Tishj commented Nov 30, 2025 •

edited

Loading

aihuaxu commented Jan 9, 2026 •

edited

Loading

nssalian commented Apr 8, 2026 •

edited

Loading