Variant Shredding Bug: Unshredded Fields Lost During Partial Shredding (Data Loss)

[dirtysalt]

Apache Iceberg version

1.10.1 (latest release)
Also tested: 1.10.0

Query engine

Other (Java API)

Please describe the bug

Summary

When using variantShreddingFunc to partially shred variant fields (i.e., shredding only a subset of fields), unshredded fields are completely lost during the write operation. The binary value field contains only metadata headers and zeros instead of the actual field values, causing permanent data loss.

Expected Behavior

When a variant has fields {id, name, city} and only id is shredded into typed_value:

• id should be stored in the typed_value group structure
• name and city should be serialized and stored in the binary value field
• On read, the variant should be reconstructed with all three fields intact

Actual Behavior

• Schema generation works correctly: only id appears in typed_value structure
• Binary value field contains only placeholder bytes (zeros), not the actual unshredded field values
• Reading fails with IndexOutOfBoundsException in SerializedObject.initOffsetsAndLengths()
• Unshredded field data is permanently lost

Impact

This is a critical data loss bug that makes partial variant shredding completely unusable. Full shredding (all fields) and no shredding work correctly, but partial shredding loses data.

Reproduction Steps

Test Environment:

• Iceberg version: 1.10.1 (also reproduced on 1.10.0)
• Maven dependency:

  <dependency>
      <groupId>org.apache.iceberg</groupId>
      <artifactId>iceberg-parquet</artifactId>
      <version>1.10.1</version>
  </dependency>

Minimal Reproduction Code:

package com.starrocks.lab;

import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.data.GenericRecord;
import org.apache.iceberg.data.Record;
import org.apache.iceberg.data.parquet.GenericParquetWriter;
import org.apache.iceberg.io.FileAppender;
import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.parquet.Parquet;
import org.apache.iceberg.Schema;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.variants.ShreddedObject;
import org.apache.iceberg.variants.Variant;
import org.apache.iceberg.variants.VariantMetadata;
import org.apache.iceberg.variants.Variants;
import org.apache.parquet.hadoop.ParquetFileReader;
import org.apache.parquet.hadoop.metadata.ParquetMetadata;
import org.apache.parquet.column.page.PageReadStore;
import org.apache.parquet.example.data.Group;
import org.apache.parquet.example.data.simple.convert.GroupRecordConverter;
import org.apache.parquet.io.ColumnIOFactory;
import org.apache.parquet.io.MessageColumnIO;
import org.apache.parquet.io.RecordReader;
import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
import org.apache.parquet.schema.Type;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class MinimalVariantShreddingBugTest {
    
    private static final Schema SCHEMA = new Schema(
            Types.NestedField.optional(1, "data", Types.VariantType.get())
    );
    
    public static void main(String[] args) throws IOException {
        String outputPath = "/tmp/variant-bug-test.parquet";
        File outputFile = new File(outputPath);
        if (outputFile.exists()) {
            outputFile.delete();
        }
        
        // Create variants with 3 fields: id, name, city
        // Shred only 'id', leaving 'name' and 'city' unshredded
        List<Record> records = buildTestRecords();
        
        // Write with partial shredding
        OutputFile out = org.apache.iceberg.Files.localOutput(outputFile);
        Parquet.WriteBuilder builder = Parquet.write(out)
                .schema(SCHEMA)
                .createWriterFunc(GenericParquetWriter::create)
                .variantShreddingFunc(MinimalVariantShreddingBugTest::shreddedType);
        
        try (FileAppender<Record> appender = builder.build()) {
            for (Record r : records) {
                appender.add(r);
            }
        }
        
        System.out.println("Written " + records.size() + " records");
        
        // Verify the binary value field
        verifyParquetFile(outputFile);
    }
    
    /**
     * Shredding function: only 'id' field goes to typed_value.
     * 'name' and 'city' should remain in binary value.
     */
    private static Type shreddedType(int fieldId, String name) {
        return org.apache.parquet.schema.Types.optionalGroup()
                .id(fieldId)
                .addField(shreddedField("id",
                        org.apache.parquet.schema.Types.optional(PrimitiveTypeName.BINARY).named("value"),
                        org.apache.parquet.schema.Types.optional(PrimitiveTypeName.INT64).named("typed_value")))
                .named("typed_value");
    }
    
    private static Type shreddedField(String fieldName, Type valueField, Type typedValueField) {
        return org.apache.parquet.schema.Types.optionalGroup()
                .addField(valueField)
                .addField(typedValueField)
                .named(fieldName);
    }
    
    private static List<Record> buildTestRecords() {
        List<Record> records = new ArrayList<>();
        VariantMetadata metadata = Variants.metadata("id", "name", "city");
        
        for (int i = 0; i < 3; i++) {
            Record rec = GenericRecord.create(SCHEMA.asStruct());
            ShreddedObject obj = Variants.object(metadata);
            obj.put("id", Variants.of(1000L + i));
            obj.put("name", Variants.of("user_" + i));
            obj.put("city", Variants.of("city_" + i));
            
            Variant value = Variant.of(metadata, obj);
            rec.setField("data", value);
            records.add(rec);
        }
        
        return records;
    }
    
    private static void verifyParquetFile(File parquetFile) throws IOException {
        try (ParquetFileReader reader = ParquetFileReader.open(
                new Configuration(), new org.apache.hadoop.fs.Path(parquetFile.getAbsolutePath()))) {
            
            ParquetMetadata metadata = reader.getFooter();
            MessageType schema = metadata.getFileMetaData().getSchema();
            
            PageReadStore pages = reader.readNextRowGroup();
            if (pages != null) {
                MessageColumnIO columnIO = new ColumnIOFactory().getColumnIO(schema);
                RecordReader<Group> recordReader = columnIO.getRecordReader(
                        pages, new GroupRecordConverter(schema));
                
                for (int i = 0; i < pages.getRowCount(); i++) {
                    Group record = recordReader.read();
                    Group dataGroup = record.getGroup("data", 0);
                    
                    if (dataGroup.getFieldRepetitionCount("value") > 0) {
                        byte[] valueBytes = dataGroup.getBinary("value", 0).getBytes();
                        
                        System.out.println("Row " + i + " binary value (hex): " + bytesToHex(valueBytes));
                        
                        boolean hasName = containsString(valueBytes, "user_" + i);
                        boolean hasCity = containsString(valueBytes, "city_" + i);
                        
                        System.out.println("  Contains 'user_" + i + "': " + hasName);
                        System.out.println("  Contains 'city_" + i + "': " + hasCity);
                        
                        if (!hasName || !hasCity) {
                            System.out.println("  BUG: Unshredded fields missing from binary value!");
                        }
                    }
                }
            }
        }
    }
    
    private static String bytesToHex(byte[] bytes) {
        StringBuilder sb = new StringBuilder();
        for (byte b : bytes) {
            sb.append(String.format("%02x", b));
        }
        return sb.toString();
    }
    
    private static boolean containsString(byte[] bytes, String search) {
        byte[] searchBytes = search.getBytes(java.nio.charset.StandardCharsets.UTF_8);
        for (int i = 0; i <= bytes.length - searchBytes.length; i++) {
            boolean found = true;
            for (int j = 0; j < searchBytes.length; j++) {
                if (bytes[i + j] != searchBytes[j]) {
                    found = false;
                    break;
                }
            }
            if (found) {
                return true;
            }
        }
        return false;
    }
}

Expected Output:

Row 0 binary value (hex): [should contain UTF-8 encoded "user_0" and "city_0"]
  Contains 'user_0': true
  Contains 'city_0': true

Actual Output:

Row 0 binary value (hex): 020200000000000000000000000000000000000000
  Contains 'user_0': false
  Contains 'city_0': false
  BUG: Unshredded fields missing from binary value!

Technical Analysis

Binary Value Field Structure:

Byte 0: 02 = header byte
Byte 1: 02 = num_fields (indicates 2 unshredded fields)
Remaining: 00 00 00... = all zeros (no actual field data)

The header correctly indicates 2 unshredded fields exist, but the serialization logic fails to write their actual values.

Root Cause:
The variant writer correctly excludes unshredded fields from the typed_value structure but fails to serialize them into the binary value field. The ShreddedObject serialization logic appears to only handle fields defined in the shredding schema, ignoring the remaining fields entirely.

Affected Components:

• org.apache.iceberg.variants.ShreddedObject serialization
• org.apache.iceberg.parquet.ParquetVariantWriters write path
• Variant value field encoding logic

Workarounds

Until this is fixed, avoid partial shredding:

1. Either shred all fields (full shredding)
2. Or shred no fields (no shredding)

Both approaches work correctly. Only partial shredding exhibits this data loss bug.

Stack Trace (on read attempt)

java.lang.IndexOutOfBoundsException: Index 0 out of bounds for length 0
    at java.base/jdk.internal.util.Preconditions.outOfBounds(Preconditions.java:64)
    at java.base/jdk.internal.util.Preconditions.outOfBoundsCheckIndex(Preconditions.java:70)
    at java.base/jdk.internal.util.Preconditions.checkIndex(Preconditions.java:266)
    at java.base/java.util.Objects.checkIndex(Objects.java:359)
    at org.apache.iceberg.variants.SerializedObject.initOffsetsAndLengths(SerializedObject.java:122)
    at org.apache.iceberg.variants.SerializedObject.<init>(SerializedObject.java:54)

Additional Context

The issue has been verified with:

• Direct Parquet binary inspection
• Multiple test scenarios
• Both Iceberg 1.10.0 and 1.10.1

Willingness to contribute

• I can contribute a fix for this bug independently
• I would be willing to contribute a fix for this bug with guidance from the Iceberg community
• I cannot contribute a fix for this bug at this time

[RussellSpitzer]

#15085

[RussellSpitzer]

Do we know why this test is not covering the issue?

iceberg/parquet/src/test/java/org/apache/iceberg/parquet/TestVariantWriters.java

Lines 215 to 232 in de1eab8

	@ParameterizedTest
	@FieldSource("VARIANTS")
	public void testMixedShredding(Variant variant) throws IOException {
	List<Record> expected =
	IntStream.range(0, VARIANTS.length)
	.mapToObj(i -> RECORD.copy("id", i, "var", VARIANTS[i]))
	.collect(Collectors.toList());
	List<Record> actual =
	writeAndRead((id, name) -> ParquetVariantUtil.toParquetSchema(variant.value()), expected);
	assertThat(actual).hasSameSizeAs(expected);
	for (int i = 0; i < expected.size(); i += 1) {
	InternalTestHelpers.assertEquals(SCHEMA.asStruct(), expected.get(i), actual.get(i));
	}
	}

[RussellSpitzer]

OK I see, the issue is that the tests all use a "complete" Transformation function.

[RussellSpitzer]

I wrote another small test which seems to work, this goes in TestVariantWriters.java

  @org.junit.jupiter.api.Test
  public void testPartialShredding() throws IOException {
    ByteBuffer partialMetadata =
        VariantTestUtil.createMetadata(
            ImmutableList.of("shred_a", "unshred_b", "shred_c", "unshred_d"), true);
    ByteBuffer partialObjectBuffer =
        VariantTestUtil.createObject(
            partialMetadata,
            ImmutableMap.of(
                "shred_a", Variants.of(42),
                "unshred_b", Variants.of("unshredded-b"),
                "shred_c", Variants.of(100),
                "unshred_d", Variants.of("unshredded-d")));

    VariantMetadata metadata = Variants.metadata(partialMetadata);
    VariantObject variantObject = (VariantObject) Variants.value(metadata, partialObjectBuffer);
    Variant variant = Variant.of(metadata, variantObject);

    Record record = RECORD.copy("id", 1, "var", variant);

    VariantShreddingFunction partialShreddingFunc =
        (id, name) -> {
          if (name.equals("var")) {
            GroupType fieldShredA =
                org.apache.parquet.schema.Types.buildGroup(REQUIRED)
                    .optional(BINARY)
                    .named("value")
                    .addField(ParquetVariantUtil.toParquetSchema(variantObject.get("shred_a")))
                    .named("shred_a");

            GroupType fieldShredC =
                org.apache.parquet.schema.Types.buildGroup(REQUIRED)
                    .optional(BINARY)
                    .named("value")
                    .addField(ParquetVariantUtil.toParquetSchema(variantObject.get("shred_c")))
                    .named("shred_c");

            return org.apache.parquet.schema.Types.buildGroup(OPTIONAL)
                .addField(fieldShredA)
                .addField(fieldShredC)
                .named("typed_value");
          }
          return null;
        };

    Record actual = writeAndRead(partialShreddingFunc, record);

    // Verify the record is equivalent
    InternalTestHelpers.assertEquals(SCHEMA.asStruct(), record, actual);

    // Extract the variant from the actual record and verify its structure
    Variant actualVariant = (Variant) actual.getField("var");
    assertThat(actualVariant).isNotNull();

    VariantValue actualValue = actualVariant.value();
    assertThat(actualValue).isInstanceOf(VariantObject.class);

    VariantObject actualObject = (VariantObject) actualValue;

    // Verify that shredded fields "shred_a" and "shred_c" are accessible
    assertThat(actualObject.get("shred_a")).isNotNull();
    assertThat(actualObject.get("shred_a").asPrimitive().get()).isEqualTo(42);
    assertThat(actualObject.get("shred_c")).isNotNull();
    assertThat(actualObject.get("shred_c").asPrimitive().get()).isEqualTo(100);

    // Verify that unshredded fields "unshred_b" and "unshred_d" are accessible
    assertThat(actualObject.get("unshred_b")).isNotNull();
    assertThat(actualObject.get("unshred_b").asPrimitive().get()).isEqualTo("unshredded-b");
    assertThat(actualObject.get("unshred_d")).isNotNull();
    assertThat(actualObject.get("unshred_d").asPrimitive().get()).isEqualTo("unshredded-d");

    // Verify all 4 fields are present in the variant object
    assertThat(actualObject.numFields()).isEqualTo(4);
  }

[RussellSpitzer]

I think there is a bug in the test code you submitted

    /**
     * Shredding function: only 'id' field goes to typed_value.
     * 'name' and 'city' should remain in binary value.
     */
    private static Type shreddedType(int fieldId, String name) {
        return org.apache.parquet.schema.Types.optionalGroup() // I think this should be required, but I don't think it's an issue
                .id(fieldId) /*
                    This is assigning the Variant's field id to the typed value structure within the variant, 
it would end up having the same field ID as the parent variant .. . surprised this isn't throwing an error
                    This line should be dropped */
                .addField(shreddedField("id",
                        org.apache.parquet.schema.Types.optional(PrimitiveTypeName.BINARY).named("value"),
                        org.apache.parquet.schema.Types.optional(PrimitiveTypeName.INT64).named("typed_value")))
                .named("typed_value");
    }
    
    private static Type shreddedField(String fieldName, Type valueField, Type typedValueField) {
        return org.apache.parquet.schema.Types.optionalGroup()
                .addField(valueField)
                .addField(typedValueField)
                .named(fieldName);
    }

[RussellSpitzer]

Looking further I think I found the real issue in the test code (which may also be a library bug?)

            Record rec = GenericRecord.create(SCHEMA.asStruct());
            ShreddedObject obj = Variants.object(metadata);
            obj.put("id", Variants.of(1000L + i));
            obj.put("name", Variants.of("user_" + i));
            obj.put("city", Variants.of("city_" + i));
            
            Variant value = Variant.of(metadata, obj);

The obj.put methods add name and city as "shredded" fields in the object.
So we end up with an object that looks like

<unshredded = null, shredded <id, name, city>>

When the parquet writer "shreds"

iceberg/parquet/src/main/java/org/apache/iceberg/parquet/ParquetVariantWriters.java

Lines 324 to 359 in bfec39f

	public void write(int repetitionLevel, VariantValue value) {
	if (value.type() != PhysicalType.OBJECT) {
	valueWriter.write(repetitionLevel, value);
	// write null for the typed_value group
	for (ParquetValueWriter<?> writer : typedWriters.values()) {
	writeNull(writer, repetitionLevel, typedDefinitionLevel);
	}
	} else {
	VariantObject object = value.asObject();
	ShreddedObject shredded = Variants.object(object);
	for (Map.Entry<String, ParquetValueWriter<VariantValue>> entry : typedWriters.entrySet()) {
	String fieldName = entry.getKey();
	ParquetValueWriter<VariantValue> writer = entry.getValue();
	VariantValue fieldValue = object.get(fieldName);
	if (fieldValue != null) {
	// shredded: suppress the field in the object and write it to the value pair
	shredded.remove(fieldName);
	writer.write(repetitionLevel, fieldValue);
	} else {
	// missing: write null to both value and typed_value
	writeNull(writer, repetitionLevel, fieldDefinitionLevel);
	}
	}
	if (shredded.numFields() > 0) {
	// partially shredded: write the unshredded fields
	valueWriter.write(repetitionLevel, shredded);
	} else {
	// completely shredded: omit the empty value
	writeNull(valueWriter, repetitionLevel, valueDefinitionLevel);
	}
	}
	}

It will not "unshred" the shredded fields in your object just because they are not listed in the shredded fields schema. It instead just extracts any shredded values (line 336) and then copies over the untyped value portion (353). In your test case, this is an empty object.

I'm not sure this is a bug... but maybe we need to document the object better

If instead you use the VariantTestUtil to construct the variant it would work correctly

    private static List<Record> buildTestRecords() {
        List<Record> records = new ArrayList<>();
        ByteBuffer metadataBuffer = VariantTestUtil.createMetadata(List.of("id", "name", "city"), true);
        VariantMetadata metadata = Variants.metadata(metadataBuffer);
        
        for (int i = 0; i < 3; i++) {
            // Create the full object with all fields using VariantTestUtil
            ByteBuffer objectBuffer = VariantTestUtil.createObject(
                metadataBuffer,
                ImmutableMap.of(
                    "id", Variants.of(1000L + i),
                    "name", Variants.of("user_" + i),
                    "city", Variants.of("city_" + i)
                )
            );
            
            // Convert to VariantObject
            VariantObject variantObject = (VariantObject) Variants.value(metadata, objectBuffer);
            Variant value = Variant.of(metadata, variantObject);
            
            Record rec = GenericRecord.create(SCHEMA.asStruct());
            rec.setField("data", value);
            records.add(rec);
        }
        
        return records;
    }

Output from my own test

Row 0:
  metadata: 11030004060a6369747969646e616d65
  metadata (decoded): ......cityidname
  value (hex): 0202000200070e19636974795f3019757365725f30
  value (ASCII dump): [02][02][00][02][00][07][0e][19]city_0[19]user_0
  value contains 'user_0': true
  value contains 'city_0': true
  typed_value/id/typed_value: 1000

Row 1:
  metadata: 11030004060a6369747969646e616d65
  metadata (decoded): ......cityidname
  value (hex): 0202000200070e19636974795f3119757365725f31
  value (ASCII dump): [02][02][00][02][00][07][0e][19]city_1[19]user_1
  value contains 'user_1': true
  value contains 'city_1': true
  typed_value/id/typed_value: 1001

Row 2:
  metadata: 11030004060a6369747969646e616d65
  metadata (decoded): ......cityidname
  value (hex): 0202000200070e19636974795f3219757365725f32
  value (ASCII dump): [02][02][00][02][00][07][0e][19]city_2[19]user_2
  value contains 'user_2': true
  value contains 'city_2': true
  typed_value/id/typed_value: 1002

[RussellSpitzer]

TLDR: For anyone who doesn't want to read all that. There is no Data Loss bug but our API for Variants.object() is a bit confusing if you are using that to build raw variants from scratch and the serialize them out.

[dirtysalt]

@RussellSpitzer thanks for looking into this issue.

I think it works because it uses SerializedObject underneath.

VariantObject variantObject = (VariantObject) Variants.value(metadata, objectBuffer);
---> return VariantValue.from(metadata, value);
---> return SerializedObject.from(metadata, value, header);

But in my code I use ShreddedObject instead, and there is a bug in this class which I propose a fix in this pr https://github.com/apache/iceberg/pull/15087/changes

ShreddedObject obj = Variants.object(metadata);