This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new eea31a30ac [core] Refactor shredding write plan framework (#8379)
eea31a30ac is described below
commit eea31a30ac78a713c749d58df5e80fbee808eda3
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Jun 30 10:47:53 2026 +0800
[core] Refactor shredding write plan framework (#8379)
Refactor Variant shredding writes onto a generic per-file shredding
write plan framework, and prepare the same framework for MAP
shared-shredding writes.
---
.../data/shredding/MapSharedShreddingUtils.java | 21 +-
.../shredding/MapSharedShreddingWritePlan.java | 85 ++++++++
.../MapSharedShreddingWritePlanFactory.java | 63 ++++++
.../paimon/data/shredding/ShreddingWritePlan.java | 51 +++++
.../data/variant/VariantShreddingWritePlan.java | 221 +++++++++++++++++++++
.../InferShreddingWritePlanWriter.java} | 141 ++++++-------
.../format/shredding/ShreddingFormatWriter.java | 119 +++++++++++
.../shredding/ShreddingWritePlanFactory.java | 39 ++++
.../ShreddingWritePlanWriterFactories.java | 49 +++++
.../shredding/ShreddingWritePlanWriterFactory.java | 107 ++++++++++
.../SupportsShreddingWritePlan.java} | 31 +--
.../variant/VariantInferenceWriterFactory.java | 64 ------
....java => VariantShreddingWritePlanFactory.java} | 78 ++++++--
.../shredding/MapSharedShreddingUtilsTest.java | 37 ++--
.../shredding/MapSharedShreddingWritePlanTest.java | 102 ++++++++++
.../variant/VariantShreddingWritePlanTest.java | 104 ++++++++++
.../shredding/ShreddingFormatWriterTest.java | 113 +++++++++++
.../apache/paimon/format/orc/OrcFileFormat.java | 7 +-
.../apache/paimon/format/orc/OrcWriterFactory.java | 76 +++++--
.../paimon/format/parquet/ParquetFileFormat.java | 9 +-
.../format/parquet/ParquetWriterFactory.java | 33 ++-
.../apache/paimon/format/parquet/VariantUtils.java | 43 ----
.../parquet/writer/ParquetRowDataBuilder.java | 24 +--
.../parquet/writer/ParquetRowDataWriter.java | 45 +----
.../parquet/writer/RowDataParquetBuilder.java | 17 +-
.../reader/FileTypeNotMatchReadTypeTest.java | 8 +-
.../reader/ParquetRowDataBuilderForTest.java | 4 +-
.../writer/InferVariantShreddingWriteTest.java | 4 +-
.../shredding/ShreddingWritePlanFormatTest.java | 161 +++++++++++++++
.../apache/paimon/spark/sql/VariantTestBase.scala | 21 +-
30 files changed, 1518 insertions(+), 359 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingUtils.java
b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingUtils.java
index 6f0cd879c5..851edb01d8 100644
---
a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingUtils.java
+++
b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingUtils.java
@@ -113,7 +113,7 @@ public class MapSharedShreddingUtils {
public static void serializeMetadata(
MapSharedShreddingFieldMeta fieldMeta,
- String compression,
+ @Nullable String compression,
Map<String, String> metadata) {
metadata.put(
MapShreddingDefine.STORAGE_LAYOUT,
@@ -142,7 +142,7 @@ public class MapSharedShreddingUtils {
}
public static MapSharedShreddingFieldMeta deserializeMetadata(
- @Nullable Map<String, String> metadata, String compression) {
+ @Nullable Map<String, String> metadata, @Nullable String
compression) {
if (!hasShreddingMetadata(metadata)) {
throw new IllegalArgumentException(
"metadata is null or storage layout is not
shared-shredding");
@@ -235,7 +235,11 @@ public class MapSharedShreddingUtils {
}
}
- private static byte[] compress(byte[] input, String compression) {
+ private static byte[] compress(byte[] input, @Nullable String compression)
{
+ if (isNoCompression(compression)) {
+ return input;
+ }
+
BlockCompressionFactory factory =
BlockCompressionFactory.create(new
CompressOptions(compression, 1));
if (factory == null) {
@@ -247,7 +251,12 @@ public class MapSharedShreddingUtils {
return Arrays.copyOf(output, actualSize);
}
- private static byte[] decompress(byte[] input, int originalLength, String
compression) {
+ private static byte[] decompress(
+ byte[] input, int originalLength, @Nullable String compression) {
+ if (isNoCompression(compression)) {
+ return input;
+ }
+
BlockCompressionFactory factory =
BlockCompressionFactory.create(new
CompressOptions(compression, 1));
if (factory == null) {
@@ -259,6 +268,10 @@ public class MapSharedShreddingUtils {
return Arrays.copyOf(output, actualSize);
}
+ private static boolean isNoCompression(@Nullable String compression) {
+ return compression == null || "none".equalsIgnoreCase(compression);
+ }
+
private static String bytesToString(byte[] bytes) {
return new String(bytes, StandardCharsets.ISO_8859_1);
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlan.java
b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlan.java
new file mode 100644
index 0000000000..0c84a33169
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlan.java
@@ -0,0 +1,85 @@
+/*
+ * 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.paimon.data.shredding;
+
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.types.RowType;
+
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/** A physical write plan for the shared-shredding MAP layout. */
+public class MapSharedShreddingWritePlan implements ShreddingWritePlan {
+
+ private final RowType logicalRowType;
+ private final MapSharedShreddingRowConverter converter;
+ @Nullable private final MapSharedShreddingContext context;
+
+ @Nullable private Map<String, Map<String, String>> fieldMetadata;
+
+ public MapSharedShreddingWritePlan(
+ RowType logicalRowType,
+ Map<String, Integer> fieldToNumColumns,
+ @Nullable MapSharedShreddingContext context) {
+ this.logicalRowType = logicalRowType;
+ this.converter = new MapSharedShreddingRowConverter(logicalRowType,
fieldToNumColumns);
+ this.context = context;
+ }
+
+ @Override
+ public RowType logicalRowType() {
+ return logicalRowType;
+ }
+
+ @Override
+ public RowType physicalRowType() {
+ return converter.physicalType();
+ }
+
+ @Override
+ public InternalRow toPhysicalRow(InternalRow row) {
+ return converter.convert(row);
+ }
+
+ @Override
+ public Map<String, Map<String, String>> fieldMetadata(String compression) {
+ if (fieldMetadata == null) {
+ fieldMetadata = buildFieldMetadata(compression == null ? "none" :
compression);
+ }
+ return fieldMetadata;
+ }
+
+ private Map<String, Map<String, String>> buildFieldMetadata(String
compression) {
+ Map<String, Map<String, String>> metadata = new LinkedHashMap<>();
+ for (String fieldName : converter.shreddingFieldNames()) {
+ MapSharedShreddingFieldMeta fieldMeta =
converter.buildFieldMeta(fieldName);
+ Map<String, String> fieldMetadata = new LinkedHashMap<>();
+ MapSharedShreddingUtils.serializeMetadata(fieldMeta, compression,
fieldMetadata);
+ metadata.put(fieldName,
Collections.unmodifiableMap(fieldMetadata));
+
+ if (context != null) {
+ context.reportFileStats(fieldName, fieldMeta.maxRowWidth());
+ }
+ }
+ return Collections.unmodifiableMap(metadata);
+ }
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlanFactory.java
b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlanFactory.java
new file mode 100644
index 0000000000..6e500f1e91
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlanFactory.java
@@ -0,0 +1,63 @@
+/*
+ * 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.paimon.data.shredding;
+
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.format.shredding.ShreddingWritePlanFactory;
+import org.apache.paimon.types.RowType;
+
+import java.util.List;
+
+/** Creates per-file shared-shredding MAP write plans. */
+public class MapSharedShreddingWritePlanFactory implements
ShreddingWritePlanFactory {
+
+ private final RowType logicalRowType;
+ private final MapSharedShreddingContext context;
+
+ public MapSharedShreddingWritePlanFactory(
+ RowType logicalRowType, MapSharedShreddingContext context) {
+ this.logicalRowType = logicalRowType;
+ this.context = context;
+ }
+
+ @Override
+ public RowType logicalRowType() {
+ return logicalRowType;
+ }
+
+ @Override
+ public boolean shouldCreateWritePlan() {
+ return !context.isEmpty();
+ }
+
+ @Override
+ public boolean shouldInferWritePlan() {
+ return false;
+ }
+
+ @Override
+ public int inferBufferRowCount() {
+ return 0;
+ }
+
+ @Override
+ public ShreddingWritePlan createWritePlan(List<InternalRow> sampleRows) {
+ return new MapSharedShreddingWritePlan(logicalRowType,
context.computeNextK(), context);
+ }
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/data/shredding/ShreddingWritePlan.java
b/paimon-common/src/main/java/org/apache/paimon/data/shredding/ShreddingWritePlan.java
new file mode 100644
index 0000000000..ac94be1964
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/data/shredding/ShreddingWritePlan.java
@@ -0,0 +1,51 @@
+/*
+ * 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.paimon.data.shredding;
+
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.types.RowType;
+
+import java.util.Collections;
+import java.util.Map;
+
+/**
+ * A per-file physical layout plan for shredded fields.
+ *
+ * <p>The logical row type is the table-visible schema. The physical row type
is the schema written
+ * into the data file. Implementations convert logical rows to physical rows
before handing them to
+ * a file format writer.
+ */
+public interface ShreddingWritePlan {
+
+ RowType logicalRowType();
+
+ RowType physicalRowType();
+
+ InternalRow toPhysicalRow(InternalRow row);
+
+ /**
+ * Returns top-level field metadata to persist with the physical row type.
+ *
+ * <p>The outer map is keyed by physical top-level field name. The inner
map contains metadata
+ * for that field. Formats decide how to encode this metadata into the
actual file footer.
+ */
+ default Map<String, Map<String, String>> fieldMetadata(String compression)
{
+ return Collections.emptyMap();
+ }
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/data/variant/VariantShreddingWritePlan.java
b/paimon-common/src/main/java/org/apache/paimon/data/variant/VariantShreddingWritePlan.java
new file mode 100644
index 0000000000..27ffa8247a
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/data/variant/VariantShreddingWritePlan.java
@@ -0,0 +1,221 @@
+/*
+ * 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.paimon.data.variant;
+
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.shredding.ShreddingWritePlan;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.VariantType;
+import org.apache.paimon.utils.Preconditions;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/** A physical write plan for Variant shredding. */
+public class VariantShreddingWritePlan implements ShreddingWritePlan {
+
+ private final RowType logicalRowType;
+ private final RowType physicalRowType;
+ private final RowConverter rowConverter;
+
+ public VariantShreddingWritePlan(RowType logicalRowType, RowType
physicalRowType) {
+ this.logicalRowType = logicalRowType;
+ this.physicalRowType = physicalRowType;
+ this.rowConverter = RowConverter.create(logicalRowType,
physicalRowType);
+ }
+
+ public static VariantShreddingWritePlan fromConfiguredSchema(
+ RowType logicalRowType, RowType configuredShreddingSchema) {
+ return new VariantShreddingWritePlan(
+ logicalRowType,
+ configuredPhysicalRowType(logicalRowType,
configuredShreddingSchema));
+ }
+
+ @Override
+ public RowType logicalRowType() {
+ return logicalRowType;
+ }
+
+ @Override
+ public RowType physicalRowType() {
+ return physicalRowType;
+ }
+
+ @Override
+ public InternalRow toPhysicalRow(InternalRow row) {
+ return rowConverter.convert(row);
+ }
+
+ private static RowType configuredPhysicalRowType(
+ RowType logicalRowType, RowType configuredShreddingSchema) {
+ List<DataField> fields = new ArrayList<>();
+ for (DataField field : logicalRowType.getFields()) {
+ if (field.type() instanceof VariantType
+ && configuredShreddingSchema.containsField(field.name())) {
+ RowType fieldShreddingSchema =
+ PaimonShreddingUtils.variantShreddingSchema(
+
configuredShreddingSchema.getField(field.name()).type());
+ fields.add(field.newType(fieldShreddingSchema));
+ } else {
+ fields.add(field);
+ }
+ }
+ return new RowType(logicalRowType.isNullable(), fields);
+ }
+
+ private interface FieldConverter {
+
+ Object convert(InternalRow row);
+
+ boolean isIdentity();
+ }
+
+ private static class RowConverter {
+
+ private final FieldConverter[] fieldConverters;
+ private final boolean identity;
+
+ private RowConverter(FieldConverter[] fieldConverters, boolean
identity) {
+ this.fieldConverters = fieldConverters;
+ this.identity = identity;
+ }
+
+ private static RowConverter create(RowType logicalRowType, RowType
physicalRowType) {
+ Preconditions.checkArgument(
+ logicalRowType.getFieldCount() ==
physicalRowType.getFieldCount(),
+ "Logical and physical row types should have the same field
count.");
+
+ FieldConverter[] converters = new
FieldConverter[logicalRowType.getFieldCount()];
+ boolean identity = true;
+ for (int i = 0; i < converters.length; i++) {
+ DataType logicalType = logicalRowType.getTypeAt(i);
+ DataType physicalType = physicalRowType.getTypeAt(i);
+ converters[i] = createFieldConverter(i, logicalType,
physicalType);
+ identity = identity && converters[i].isIdentity();
+ }
+ return new RowConverter(converters, identity);
+ }
+
+ private InternalRow convert(InternalRow row) {
+ if (identity) {
+ return row;
+ }
+
+ GenericRow physicalRow = new GenericRow(row.getRowKind(),
fieldConverters.length);
+ for (int i = 0; i < fieldConverters.length; i++) {
+ physicalRow.setField(i, fieldConverters[i].convert(row));
+ }
+ return physicalRow;
+ }
+
+ private static FieldConverter createFieldConverter(
+ int fieldIndex, DataType logicalType, DataType physicalType) {
+ if (logicalType instanceof VariantType && physicalType instanceof
RowType) {
+ return new VariantFieldConverter(
+ fieldIndex,
+ PaimonShreddingUtils.buildVariantSchema((RowType)
physicalType));
+ }
+
+ if (logicalType instanceof RowType && physicalType instanceof
RowType) {
+ RowConverter nestedConverter =
+ RowConverter.create((RowType) logicalType, (RowType)
physicalType);
+ if (!nestedConverter.identity) {
+ return new RowFieldConverter(
+ fieldIndex, (RowType) logicalType,
nestedConverter);
+ }
+ }
+
+ return new IdentityFieldConverter(fieldIndex, logicalType);
+ }
+ }
+
+ private static class IdentityFieldConverter implements FieldConverter {
+
+ private final InternalRow.FieldGetter fieldGetter;
+
+ private IdentityFieldConverter(int fieldIndex, DataType logicalType) {
+ this.fieldGetter = InternalRow.createFieldGetter(logicalType,
fieldIndex);
+ }
+
+ @Override
+ public Object convert(InternalRow row) {
+ return fieldGetter.getFieldOrNull(row);
+ }
+
+ @Override
+ public boolean isIdentity() {
+ return true;
+ }
+ }
+
+ private static class RowFieldConverter implements FieldConverter {
+
+ private final int fieldIndex;
+ private final int fieldCount;
+ private final RowConverter rowConverter;
+
+ private RowFieldConverter(
+ int fieldIndex, RowType logicalRowType, RowConverter
rowConverter) {
+ this.fieldIndex = fieldIndex;
+ this.fieldCount = logicalRowType.getFieldCount();
+ this.rowConverter = rowConverter;
+ }
+
+ @Override
+ public Object convert(InternalRow row) {
+ if (row.isNullAt(fieldIndex)) {
+ return null;
+ }
+ return rowConverter.convert(row.getRow(fieldIndex, fieldCount));
+ }
+
+ @Override
+ public boolean isIdentity() {
+ return false;
+ }
+ }
+
+ private static class VariantFieldConverter implements FieldConverter {
+
+ private final int fieldIndex;
+ private final VariantSchema variantSchema;
+
+ private VariantFieldConverter(int fieldIndex, VariantSchema
variantSchema) {
+ this.fieldIndex = fieldIndex;
+ this.variantSchema = variantSchema;
+ }
+
+ @Override
+ public Object convert(InternalRow row) {
+ if (row.isNullAt(fieldIndex)) {
+ return null;
+ }
+ return PaimonShreddingUtils.castShredded(
+ (GenericVariant) row.getVariant(fieldIndex),
variantSchema);
+ }
+
+ @Override
+ public boolean isIdentity() {
+ return false;
+ }
+ }
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/format/variant/InferVariantShreddingWriter.java
b/paimon-common/src/main/java/org/apache/paimon/format/shredding/InferShreddingWritePlanWriter.java
similarity index 51%
rename from
paimon-common/src/main/java/org/apache/paimon/format/variant/InferVariantShreddingWriter.java
rename to
paimon-common/src/main/java/org/apache/paimon/format/shredding/InferShreddingWritePlanWriter.java
index 077a8d3da1..5f497d2d92 100644
---
a/paimon-common/src/main/java/org/apache/paimon/format/variant/InferVariantShreddingWriter.java
+++
b/paimon-common/src/main/java/org/apache/paimon/format/shredding/InferShreddingWritePlanWriter.java
@@ -16,58 +16,46 @@
* limitations under the License.
*/
-package org.apache.paimon.format.variant;
+package org.apache.paimon.format.shredding;
import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.data.variant.InferVariantShreddingSchema;
+import org.apache.paimon.data.shredding.ShreddingWritePlan;
import org.apache.paimon.format.BundleFormatWriter;
import org.apache.paimon.format.FormatWriter;
import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.io.BundleRecords;
-import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.InternalRowUtils;
import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
-/**
- * A generic writer that infers the shredding schema from buffered rows before
writing.
- *
- * <p>This writer buffers rows up to a threshold, infers the optimal schema
from them, then writes
- * all data using the inferred schema. It works with any format that
implements {@link
- * SupportsVariantInference}.
- */
-public class InferVariantShreddingWriter implements BundleFormatWriter {
+/** Buffers initial rows, infers a per-file shredding write plan, and writes
physical rows. */
+public class InferShreddingWritePlanWriter implements BundleFormatWriter {
- private final SupportsVariantInference writerFactory;
- private final RowType rowType;
- private final InferVariantShreddingSchema shreddingSchemaInfer;
- private final int maxBufferRow;
+ private final SupportsShreddingWritePlan writerFactory;
+ private final ShreddingWritePlanFactory writePlanFactory;
private final PositionOutputStream out;
private final String compression;
private final List<InternalRow> bufferedRows;
private final List<BundleRecords> bufferedBundles;
- private FormatWriter actualWriter;
- private boolean schemaFinalized = false;
+ @Nullable private FormatWriter actualWriter;
+ private boolean planFinalized = false;
private long totalBufferedRowCount = 0;
- public InferVariantShreddingWriter(
- SupportsVariantInference writerFactory,
- RowType rowType,
- InferVariantShreddingSchema shreddingSchemaInfer,
- int maxBufferRow,
+ public InferShreddingWritePlanWriter(
+ SupportsShreddingWritePlan writerFactory,
+ ShreddingWritePlanFactory writePlanFactory,
PositionOutputStream out,
String compression) {
this.writerFactory = writerFactory;
- this.rowType = rowType;
- this.shreddingSchemaInfer = shreddingSchemaInfer;
- this.maxBufferRow = maxBufferRow;
+ this.writePlanFactory = writePlanFactory;
this.out = out;
this.compression = compression;
this.bufferedRows = new ArrayList<>();
@@ -76,58 +64,56 @@ public class InferVariantShreddingWriter implements
BundleFormatWriter {
@Override
public void addElement(InternalRow row) throws IOException {
- if (!schemaFinalized) {
- bufferedRows.add(InternalRowUtils.copyInternalRow(row, rowType));
+ if (!planFinalized) {
+ bufferedRows.add(
+ InternalRowUtils.copyInternalRow(row,
writePlanFactory.logicalRowType()));
totalBufferedRowCount++;
- if (totalBufferedRowCount >= maxBufferRow) {
- finalizeSchemaAndFlush();
+ if (totalBufferedRowCount >=
writePlanFactory.inferBufferRowCount()) {
+ finalizePlanAndFlush();
}
- } else {
- actualWriter.addElement(row);
+ return;
}
+
+ actualWriter.addElement(row);
}
@Override
public void writeBundle(BundleRecords bundle) throws IOException {
- if (!schemaFinalized) {
+ if (!planFinalized) {
final List<InternalRow> rows = new ArrayList<>();
- bundle.forEach(row ->
rows.add(InternalRowUtils.copyInternalRow(row, rowType)));
- BundleRecords copiedBundle =
- new BundleRecords() {
- @Override
- @Nonnull
- public Iterator<InternalRow> iterator() {
- return rows.iterator();
- }
-
- @Override
- public long rowCount() {
- return rows.size();
- }
- };
- bufferedBundles.add(copiedBundle);
+ for (InternalRow row : bundle) {
+ rows.add(InternalRowUtils.copyInternalRow(row,
writePlanFactory.logicalRowType()));
+ }
+ bufferedBundles.add(new CopiedBundleRecords(rows));
totalBufferedRowCount += bundle.rowCount();
- if (totalBufferedRowCount >= maxBufferRow) {
- finalizeSchemaAndFlush();
+ if (totalBufferedRowCount >=
writePlanFactory.inferBufferRowCount()) {
+ finalizePlanAndFlush();
}
- } else {
- ((BundleFormatWriter) actualWriter).writeBundle(bundle);
+ return;
}
+
+ ((BundleFormatWriter) actualWriter).writeBundle(bundle);
}
@Override
public boolean reachTargetSize(boolean suggestedCheck, long targetSize)
throws IOException {
- if (!schemaFinalized) {
+ if (!planFinalized) {
return false;
}
return actualWriter.reachTargetSize(suggestedCheck, targetSize);
}
+ @Nullable
+ @Override
+ public Object writerMetadata() {
+ return actualWriter == null ? null : actualWriter.writerMetadata();
+ }
+
@Override
public void close() throws IOException {
try {
- if (!schemaFinalized) {
- finalizeSchemaAndFlush();
+ if (!planFinalized) {
+ finalizePlanAndFlush();
}
} finally {
if (actualWriter != null) {
@@ -136,11 +122,12 @@ public class InferVariantShreddingWriter implements
BundleFormatWriter {
}
}
- private void finalizeSchemaAndFlush() throws IOException {
- RowType inferredShreddingSchema =
shreddingSchemaInfer.inferSchema(collectAllRows());
+ private void finalizePlanAndFlush() throws IOException {
+ ShreddingWritePlan writePlan =
writePlanFactory.createWritePlan(collectAllRows());
actualWriter =
- writerFactory.createWithShreddingSchema(out, compression,
inferredShreddingSchema);
- schemaFinalized = true;
+ ShreddingWritePlanWriterFactory.createWriterWithPlan(
+ writerFactory, out, compression, writePlan);
+ planFinalized = true;
if (!bufferedBundles.isEmpty()) {
BundleFormatWriter bundleWriter = (BundleFormatWriter)
actualWriter;
@@ -157,16 +144,36 @@ public class InferVariantShreddingWriter implements
BundleFormatWriter {
}
private List<InternalRow> collectAllRows() {
- if (!bufferedBundles.isEmpty()) {
- List<InternalRow> allRows = new ArrayList<>();
- for (BundleRecords bundle : bufferedBundles) {
- for (InternalRow row : bundle) {
- allRows.add(row);
- }
- }
- return allRows;
- } else {
+ if (bufferedBundles.isEmpty()) {
return bufferedRows;
}
+
+ List<InternalRow> allRows = new ArrayList<>();
+ for (BundleRecords bundle : bufferedBundles) {
+ for (InternalRow row : bundle) {
+ allRows.add(row);
+ }
+ }
+ return allRows;
+ }
+
+ private static class CopiedBundleRecords implements BundleRecords {
+
+ private final List<InternalRow> rows;
+
+ private CopiedBundleRecords(List<InternalRow> rows) {
+ this.rows = rows;
+ }
+
+ @Override
+ @Nonnull
+ public Iterator<InternalRow> iterator() {
+ return rows.iterator();
+ }
+
+ @Override
+ public long rowCount() {
+ return rows.size();
+ }
}
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/format/shredding/ShreddingFormatWriter.java
b/paimon-common/src/main/java/org/apache/paimon/format/shredding/ShreddingFormatWriter.java
new file mode 100644
index 0000000000..504852819f
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/format/shredding/ShreddingFormatWriter.java
@@ -0,0 +1,119 @@
+/*
+ * 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.paimon.format.shredding;
+
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.shredding.ShreddingWritePlan;
+import org.apache.paimon.format.BundleFormatWriter;
+import org.apache.paimon.format.FormatWriter;
+import org.apache.paimon.io.BundleRecords;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+/** A writer wrapper that converts logical rows to the physical row layout
before writing. */
+public class ShreddingFormatWriter implements BundleFormatWriter {
+
+ private final FormatWriter delegate;
+ private final SupportsShreddingWritePlan writerFactory;
+ private final ShreddingWritePlan writePlan;
+ private final String compression;
+
+ public ShreddingFormatWriter(
+ FormatWriter delegate,
+ SupportsShreddingWritePlan writerFactory,
+ ShreddingWritePlan writePlan,
+ String compression) {
+ this.delegate = delegate;
+ this.writerFactory = writerFactory;
+ this.writePlan = writePlan;
+ this.compression = compression;
+ }
+
+ @Override
+ public void addElement(InternalRow element) throws IOException {
+ delegate.addElement(writePlan.toPhysicalRow(element));
+ }
+
+ @Override
+ public void writeBundle(BundleRecords bundle) throws IOException {
+ if (delegate instanceof BundleFormatWriter) {
+ ((BundleFormatWriter) delegate).writeBundle(new
PhysicalBundleRecords(bundle));
+ return;
+ }
+
+ for (InternalRow row : bundle) {
+ addElement(row);
+ }
+ }
+
+ @Override
+ public boolean reachTargetSize(boolean suggestedCheck, long targetSize)
throws IOException {
+ return delegate.reachTargetSize(suggestedCheck, targetSize);
+ }
+
+ @Nullable
+ @Override
+ public Object writerMetadata() {
+ return delegate.writerMetadata();
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+ writerFactory.commitShreddingMetadata(delegate, writePlan,
compression);
+ } finally {
+ delegate.close();
+ }
+ }
+
+ private class PhysicalBundleRecords implements BundleRecords {
+
+ private final BundleRecords delegateBundle;
+
+ private PhysicalBundleRecords(BundleRecords delegateBundle) {
+ this.delegateBundle = delegateBundle;
+ }
+
+ @Override
+ @Nonnull
+ public Iterator<InternalRow> iterator() {
+ final Iterator<InternalRow> iterator = delegateBundle.iterator();
+ return new Iterator<InternalRow>() {
+ @Override
+ public boolean hasNext() {
+ return iterator.hasNext();
+ }
+
+ @Override
+ public InternalRow next() {
+ return writePlan.toPhysicalRow(iterator.next());
+ }
+ };
+ }
+
+ @Override
+ public long rowCount() {
+ return delegateBundle.rowCount();
+ }
+ }
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/format/shredding/ShreddingWritePlanFactory.java
b/paimon-common/src/main/java/org/apache/paimon/format/shredding/ShreddingWritePlanFactory.java
new file mode 100644
index 0000000000..5adb313fe6
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/format/shredding/ShreddingWritePlanFactory.java
@@ -0,0 +1,39 @@
+/*
+ * 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.paimon.format.shredding;
+
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.shredding.ShreddingWritePlan;
+import org.apache.paimon.types.RowType;
+
+import java.util.List;
+
+/** Creates per-file shredding write plans. */
+public interface ShreddingWritePlanFactory {
+
+ RowType logicalRowType();
+
+ boolean shouldCreateWritePlan();
+
+ boolean shouldInferWritePlan();
+
+ int inferBufferRowCount();
+
+ ShreddingWritePlan createWritePlan(List<InternalRow> sampleRows);
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/format/shredding/ShreddingWritePlanWriterFactories.java
b/paimon-common/src/main/java/org/apache/paimon/format/shredding/ShreddingWritePlanWriterFactories.java
new file mode 100644
index 0000000000..1fba3ddedd
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/format/shredding/ShreddingWritePlanWriterFactories.java
@@ -0,0 +1,49 @@
+/*
+ * 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.paimon.format.shredding;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.shredding.MapSharedShreddingContext;
+import org.apache.paimon.data.shredding.MapSharedShreddingUtils;
+import org.apache.paimon.data.shredding.MapSharedShreddingWritePlanFactory;
+import org.apache.paimon.format.FormatWriterFactory;
+import org.apache.paimon.types.RowType;
+
+import java.util.List;
+
+/** Helpers for composing per-file shredding writer factories. */
+public class ShreddingWritePlanWriterFactories {
+
+ private ShreddingWritePlanWriterFactories() {}
+
+ public static FormatWriterFactory wrapMapSharedShredding(
+ FormatWriterFactory delegate, RowType rowType, CoreOptions
options) {
+ List<String> shreddingFields =
+ MapSharedShreddingUtils.detectShreddingColumns(rowType,
options);
+ if (shreddingFields.isEmpty()) {
+ return delegate;
+ }
+
+ MapSharedShreddingContext context =
+ new MapSharedShreddingContext(
+
MapSharedShreddingUtils.buildColumnToNumColumns(shreddingFields, options));
+ return new ShreddingWritePlanWriterFactory(
+ delegate, new MapSharedShreddingWritePlanFactory(rowType,
context));
+ }
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/format/shredding/ShreddingWritePlanWriterFactory.java
b/paimon-common/src/main/java/org/apache/paimon/format/shredding/ShreddingWritePlanWriterFactory.java
new file mode 100644
index 0000000000..b3f6b2d75e
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/format/shredding/ShreddingWritePlanWriterFactory.java
@@ -0,0 +1,107 @@
+/*
+ * 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.paimon.format.shredding;
+
+import org.apache.paimon.data.shredding.ShreddingWritePlan;
+import org.apache.paimon.format.FormatWriter;
+import org.apache.paimon.format.FormatWriterFactory;
+import org.apache.paimon.fs.PositionOutputStream;
+
+import java.io.IOException;
+import java.util.Collections;
+
+/** Decorates a format writer factory with optional per-file shredding write
plans. */
+public class ShreddingWritePlanWriterFactory
+ implements FormatWriterFactory, SupportsShreddingWritePlan {
+
+ private final FormatWriterFactory delegate;
+ private final ShreddingWritePlanFactory writePlanFactory;
+
+ public ShreddingWritePlanWriterFactory(
+ FormatWriterFactory delegate, ShreddingWritePlanFactory
writePlanFactory) {
+ this.delegate = delegate;
+ this.writePlanFactory = writePlanFactory;
+ }
+
+ @Override
+ public FormatWriter create(PositionOutputStream out, String compression)
throws IOException {
+ if (!writePlanFactory.shouldCreateWritePlan()
+ || !(delegate instanceof SupportsShreddingWritePlan)) {
+ return delegate.create(out, compression);
+ }
+
+ SupportsShreddingWritePlan shreddingDelegate =
(SupportsShreddingWritePlan) delegate;
+ if (writePlanFactory.shouldInferWritePlan()) {
+ return new InferShreddingWritePlanWriter(
+ shreddingDelegate, writePlanFactory, out, compression);
+ }
+
+ return createWriterWithPlan(
+ shreddingDelegate,
+ out,
+ compression,
+ writePlanFactory.createWritePlan(Collections.emptyList()));
+ }
+
+ @Override
+ public FormatWriter createWithShreddingWritePlan(
+ PositionOutputStream out, String compression, ShreddingWritePlan
writePlan)
+ throws IOException {
+ if (writePlanFactory.shouldCreateWritePlan()) {
+ throw new UnsupportedOperationException(
+ "Composing multiple active shredding write plans is not
supported.");
+ }
+
+ return shreddingDelegate().createWithShreddingWritePlan(out,
compression, writePlan);
+ }
+
+ @Override
+ public void commitShreddingMetadata(
+ FormatWriter writer, ShreddingWritePlan writePlan, String
compression)
+ throws IOException {
+ if (writePlanFactory.shouldCreateWritePlan()) {
+ throw new UnsupportedOperationException(
+ "Composing multiple active shredding write plans is not
supported.");
+ }
+
+ shreddingDelegate().commitShreddingMetadata(writer, writePlan,
compression);
+ }
+
+ private SupportsShreddingWritePlan shreddingDelegate() {
+ if (!(delegate instanceof SupportsShreddingWritePlan)) {
+ throw new UnsupportedOperationException(
+ "Delegate writer factory does not support shredding write
plans: "
+ + delegate.getClass().getName());
+ }
+ return (SupportsShreddingWritePlan) delegate;
+ }
+
+ static FormatWriter createWriterWithPlan(
+ SupportsShreddingWritePlan delegate,
+ PositionOutputStream out,
+ String compression,
+ ShreddingWritePlan writePlan)
+ throws IOException {
+ return new ShreddingFormatWriter(
+ delegate.createWithShreddingWritePlan(out, compression,
writePlan),
+ delegate,
+ writePlan,
+ compression);
+ }
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/format/variant/SupportsVariantInference.java
b/paimon-common/src/main/java/org/apache/paimon/format/shredding/SupportsShreddingWritePlan.java
similarity index 50%
rename from
paimon-common/src/main/java/org/apache/paimon/format/variant/SupportsVariantInference.java
rename to
paimon-common/src/main/java/org/apache/paimon/format/shredding/SupportsShreddingWritePlan.java
index 1fe404080e..f635eaae02 100644
---
a/paimon-common/src/main/java/org/apache/paimon/format/variant/SupportsVariantInference.java
+++
b/paimon-common/src/main/java/org/apache/paimon/format/shredding/SupportsShreddingWritePlan.java
@@ -16,33 +16,22 @@
* limitations under the License.
*/
-package org.apache.paimon.format.variant;
+package org.apache.paimon.format.shredding;
+import org.apache.paimon.data.shredding.ShreddingWritePlan;
import org.apache.paimon.format.FormatWriter;
import org.apache.paimon.fs.PositionOutputStream;
-import org.apache.paimon.types.RowType;
import java.io.IOException;
-/**
- * Interface for FormatWriterFactory implementations that support variant
schema inference.
- *
- * <p>Writers implementing this interface can dynamically update their schema
based on inferred
- * variant shredding schemas.
- */
-public interface SupportsVariantInference {
+/** Format writer factories that can create a writer for a shredded physical
row layout. */
+public interface SupportsShreddingWritePlan {
- /**
- * Create the writer with the inferred shredding schema using the same
output stream and
- * compression settings.
- *
- * @param out The output stream to write to
- * @param compression The compression codec
- * @param inferredShreddingSchema The inferred shredding schema for
variant fields
- * @return A new FormatWriter configured with the inferred schema
- * @throws IOException If the writer cannot be created
- */
- FormatWriter createWithShreddingSchema(
- PositionOutputStream out, String compression, RowType
inferredShreddingSchema)
+ FormatWriter createWithShreddingWritePlan(
+ PositionOutputStream out, String compression, ShreddingWritePlan
writePlan)
throws IOException;
+
+ default void commitShreddingMetadata(
+ FormatWriter writer, ShreddingWritePlan writePlan, String
compression)
+ throws IOException {}
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/format/variant/VariantInferenceWriterFactory.java
b/paimon-common/src/main/java/org/apache/paimon/format/variant/VariantInferenceWriterFactory.java
deleted file mode 100644
index cbbd3c5e19..0000000000
---
a/paimon-common/src/main/java/org/apache/paimon/format/variant/VariantInferenceWriterFactory.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * 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.paimon.format.variant;
-
-import org.apache.paimon.format.FormatWriter;
-import org.apache.paimon.format.FormatWriterFactory;
-import org.apache.paimon.fs.PositionOutputStream;
-
-import java.io.IOException;
-
-/**
- * A decorator factory that adds variant schema inference capability to any
{@link
- * FormatWriterFactory}.
- *
- * <p>This factory wraps an existing FormatWriterFactory and automatically
enables variant schema
- * inference if the delegate factory supports it (implements {@link
SupportsVariantInference}) and
- * the configuration enables inference.
- */
-public class VariantInferenceWriterFactory implements FormatWriterFactory {
-
- private final FormatWriterFactory delegate;
- private final VariantInferenceConfig config;
-
- public VariantInferenceWriterFactory(
- FormatWriterFactory delegate, VariantInferenceConfig config) {
- this.delegate = delegate;
- this.config = config;
- }
-
- @Override
- public FormatWriter create(PositionOutputStream out, String compression)
throws IOException {
- if (!config.shouldEnableInference()) {
- return delegate.create(out, compression);
- }
-
- if (!(delegate instanceof SupportsVariantInference)) {
- return delegate.create(out, compression);
- }
-
- return new InferVariantShreddingWriter(
- (SupportsVariantInference) delegate,
- config.rowType(),
- config.createInferrer(),
- config.getMaxBufferRow(),
- out,
- compression);
- }
-}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/format/variant/VariantInferenceConfig.java
b/paimon-common/src/main/java/org/apache/paimon/format/variant/VariantShreddingWritePlanFactory.java
similarity index 52%
rename from
paimon-common/src/main/java/org/apache/paimon/format/variant/VariantInferenceConfig.java
rename to
paimon-common/src/main/java/org/apache/paimon/format/variant/VariantShreddingWritePlanFactory.java
index 991aa864e8..75e4c77e68 100644
---
a/paimon-common/src/main/java/org/apache/paimon/format/variant/VariantInferenceConfig.java
+++
b/paimon-common/src/main/java/org/apache/paimon/format/variant/VariantShreddingWritePlanFactory.java
@@ -19,26 +19,44 @@
package org.apache.paimon.format.variant;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.shredding.ShreddingWritePlan;
import org.apache.paimon.data.variant.InferVariantShreddingSchema;
+import org.apache.paimon.data.variant.VariantShreddingWritePlan;
+import org.apache.paimon.format.shredding.ShreddingWritePlanFactory;
import org.apache.paimon.options.Options;
import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
import org.apache.paimon.types.RowType;
import org.apache.paimon.types.VariantType;
+import org.apache.paimon.utils.JsonSerdeUtil;
-/** Variant schema inference configuration. */
-public class VariantInferenceConfig {
+import java.util.List;
+
+/** Creates Variant shredding write plans from configured or inferred schemas.
*/
+public class VariantShreddingWritePlanFactory implements
ShreddingWritePlanFactory {
private final RowType rowType;
private final Options options;
- public VariantInferenceConfig(RowType rowType, Options options) {
+ public VariantShreddingWritePlanFactory(RowType rowType, Options options) {
this.rowType = rowType;
this.options = options;
}
- /** Determines whether variant schema inference should be enabled. */
- public boolean shouldEnableInference() {
- if (options.contains(CoreOptions.VARIANT_SHREDDING_SCHEMA)) {
+ @Override
+ public RowType logicalRowType() {
+ return rowType;
+ }
+
+ @Override
+ public boolean shouldCreateWritePlan() {
+ return hasConfiguredShreddingSchema() || shouldInferWritePlan();
+ }
+
+ @Override
+ public boolean shouldInferWritePlan() {
+ if (hasConfiguredShreddingSchema()) {
return false;
}
@@ -49,17 +67,32 @@ public class VariantInferenceConfig {
return containsVariantFields(rowType);
}
- private boolean containsVariantFields(RowType rowType) {
- for (DataField field : rowType.getFields()) {
- if (field.type() instanceof VariantType) {
- return true;
- }
+ @Override
+ public int inferBufferRowCount() {
+ return options.get(CoreOptions.VARIANT_SHREDDING_MAX_INFER_BUFFER_ROW);
+ }
+
+ @Override
+ public ShreddingWritePlan createWritePlan(List<InternalRow> sampleRows) {
+ if (hasConfiguredShreddingSchema()) {
+ return VariantShreddingWritePlan.fromConfiguredSchema(
+ rowType, configuredShreddingSchema());
}
- return false;
+
+ RowType physicalRowType = createInferrer().inferSchema(sampleRows);
+ return new VariantShreddingWritePlan(rowType, physicalRowType);
}
- /** Create a schema inferrer. */
- public InferVariantShreddingSchema createInferrer() {
+ private boolean hasConfiguredShreddingSchema() {
+ return options.contains(CoreOptions.VARIANT_SHREDDING_SCHEMA);
+ }
+
+ private RowType configuredShreddingSchema() {
+ String shreddingSchema =
options.get(CoreOptions.VARIANT_SHREDDING_SCHEMA);
+ return (RowType) JsonSerdeUtil.fromJson(shreddingSchema,
DataType.class);
+ }
+
+ private InferVariantShreddingSchema createInferrer() {
return new InferVariantShreddingSchema(
rowType,
options.get(CoreOptions.VARIANT_SHREDDING_MAX_SCHEMA_WIDTH),
@@ -67,12 +100,15 @@ public class VariantInferenceConfig {
options.get(CoreOptions.VARIANT_SHREDDING_MIN_FIELD_CARDINALITY_RATIO));
}
- /** Get the maximum number of rows to buffer for inference. */
- public int getMaxBufferRow() {
- return options.get(CoreOptions.VARIANT_SHREDDING_MAX_INFER_BUFFER_ROW);
- }
-
- public RowType rowType() {
- return rowType;
+ private boolean containsVariantFields(RowType rowType) {
+ for (DataField field : rowType.getFields()) {
+ if (field.type() instanceof VariantType) {
+ return true;
+ }
+ if (field.type() instanceof RowType &&
containsVariantFields((RowType) field.type())) {
+ return true;
+ }
+ }
+ return false;
}
}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingUtilsTest.java
b/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingUtilsTest.java
index 29314c6a00..e9d8d03f09 100644
---
a/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingUtilsTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingUtilsTest.java
@@ -203,24 +203,27 @@ class MapSharedShreddingUtilsTest {
MapSharedShreddingFieldMeta original =
new MapSharedShreddingFieldMeta(nameToId, fieldToColumns,
overflowSet, 3, 2);
- Map<String, String> metadata = new HashMap<>();
- MapSharedShreddingUtils.serializeMetadata(original, "none", metadata);
-
-
assertThat(MapSharedShreddingUtils.hasShreddingMetadata(metadata)).isTrue();
-
assertThat(metadata.get(MapShreddingDefine.STORAGE_LAYOUT)).isEqualTo("shared-shredding");
-
assertThat(metadata.get(MapSharedShreddingDefine.VERSION)).isEqualTo("1");
-
assertThat(metadata.get(MapSharedShreddingDefine.NUM_COLUMNS)).isEqualTo("3");
-
assertThat(metadata.get(MapSharedShreddingDefine.MAX_ROW_WIDTH)).isEqualTo("2");
-
String expectedDict = "{\"age\":0,\"name\":1}";
-
assertThat(metadata.get(MapSharedShreddingDefine.FIELD_DICT)).isEqualTo(expectedDict);
-
assertThat(metadata.get(MapSharedShreddingDefine.FIELD_DICT_ORIGINAL_SIZE))
- .isEqualTo(String.valueOf(expectedDict.length()));
- assertThat(metadata.get(MapSharedShreddingDefine.FIELD_COLUMNS))
- .isEqualTo("{\"0\":[0],\"1\":[1,2]}");
-
assertThat(metadata.get(MapSharedShreddingDefine.OVERFLOW_SET)).isEqualTo("[1,5]");
- assertThat(MapSharedShreddingUtils.deserializeMetadata(metadata,
"none"))
- .isEqualTo(original);
+ for (String compression : Arrays.<String>asList(null, "none", "NONE"))
{
+ Map<String, String> metadata = new HashMap<>();
+ MapSharedShreddingUtils.serializeMetadata(original, compression,
metadata);
+
+
assertThat(MapSharedShreddingUtils.hasShreddingMetadata(metadata)).isTrue();
+ assertThat(metadata.get(MapShreddingDefine.STORAGE_LAYOUT))
+ .isEqualTo("shared-shredding");
+
assertThat(metadata.get(MapSharedShreddingDefine.VERSION)).isEqualTo("1");
+
assertThat(metadata.get(MapSharedShreddingDefine.NUM_COLUMNS)).isEqualTo("3");
+
assertThat(metadata.get(MapSharedShreddingDefine.MAX_ROW_WIDTH)).isEqualTo("2");
+
+
assertThat(metadata.get(MapSharedShreddingDefine.FIELD_DICT)).isEqualTo(expectedDict);
+
assertThat(metadata.get(MapSharedShreddingDefine.FIELD_DICT_ORIGINAL_SIZE))
+ .isEqualTo(String.valueOf(expectedDict.length()));
+ assertThat(metadata.get(MapSharedShreddingDefine.FIELD_COLUMNS))
+ .isEqualTo("{\"0\":[0],\"1\":[1,2]}");
+
assertThat(metadata.get(MapSharedShreddingDefine.OVERFLOW_SET)).isEqualTo("[1,5]");
+ assertThat(MapSharedShreddingUtils.deserializeMetadata(metadata,
compression))
+ .isEqualTo(original);
+ }
}
@Test
diff --git
a/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlanTest.java
b/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlanTest.java
new file mode 100644
index 0000000000..69010c2594
--- /dev/null
+++
b/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlanTest.java
@@ -0,0 +1,102 @@
+/*
+ * 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.paimon.data.shredding;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericMap;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link MapSharedShreddingWritePlan}. */
+class MapSharedShreddingWritePlanTest {
+
+ @Test
+ void testConvertAndBuildFieldMetadata() {
+ RowType logicalType =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "id", DataTypes.INT()),
+ DataTypes.FIELD(
+ 1, "tags", DataTypes.MAP(DataTypes.STRING(),
DataTypes.BIGINT())));
+ MapSharedShreddingContext context =
+ new MapSharedShreddingContext(Collections.singletonMap("tags",
4));
+ MapSharedShreddingWritePlan writePlan =
+ new MapSharedShreddingWritePlan(logicalType,
context.computeNextK(), context);
+
+ InternalRow physicalRow =
+ writePlan.toPhysicalRow(
+ GenericRow.of(1, stringKeyMap("a", 10L, "b", 20L, "c",
30L)));
+ InternalRow physicalTags = physicalRow.getRow(1, 6);
+
+ assertThat(writePlan.logicalRowType()).isEqualTo(logicalType);
+ assertThat(writePlan.physicalRowType().getFieldCount()).isEqualTo(2);
+ assertThat(physicalRow.getInt(0)).isEqualTo(1);
+ assertThat(physicalTags.getArray(0).toIntArray()).containsExactly(0,
1, 2, -1);
+ assertThat(physicalTags.getLong(1)).isEqualTo(10L);
+ assertThat(physicalTags.getLong(2)).isEqualTo(20L);
+ assertThat(physicalTags.getLong(3)).isEqualTo(30L);
+ assertThat(physicalTags.isNullAt(4)).isTrue();
+ assertThat(physicalTags.isNullAt(5)).isTrue();
+
+ Map<String, Map<String, String>> fieldMetadata =
writePlan.fieldMetadata("none");
+ assertThat(fieldMetadata).containsOnlyKeys("tags");
+ MapSharedShreddingFieldMeta fieldMeta =
+
MapSharedShreddingUtils.deserializeMetadata(fieldMetadata.get("tags"), "none");
+ assertThat(fieldMeta.nameToId()).containsEntry("a",
0).containsEntry("b", 1);
+ assertThat(fieldMeta.numColumns()).isEqualTo(4);
+ assertThat(fieldMeta.maxRowWidth()).isEqualTo(3);
+ assertThat(context.computeNextK()).containsEntry("tags", 3);
+ }
+
+ @Test
+ void testFactoryCreatesPlanFromContext() {
+ RowType logicalType =
+ DataTypes.ROW(
+ DataTypes.FIELD(
+ 0, "tags", DataTypes.MAP(DataTypes.STRING(),
DataTypes.INT())));
+ MapSharedShreddingContext context =
+ new MapSharedShreddingContext(Collections.singletonMap("tags",
2));
+ MapSharedShreddingWritePlanFactory factory =
+ new MapSharedShreddingWritePlanFactory(logicalType, context);
+
+ assertThat(factory.shouldCreateWritePlan()).isTrue();
+ assertThat(factory.shouldInferWritePlan()).isFalse();
+
assertThat(factory.createWritePlan(Collections.emptyList()).physicalRowType())
+ .isEqualTo(
+ MapSharedShreddingUtils.logicalToPhysicalSchema(
+ logicalType, Collections.singletonMap("tags",
2)));
+ }
+
+ private static GenericMap stringKeyMap(Object... keyValues) {
+ Map<Object, Object> values = new LinkedHashMap<>();
+ for (int i = 0; i < keyValues.length; i += 2) {
+ values.put(BinaryString.fromString((String) keyValues[i]),
keyValues[i + 1]);
+ }
+ return new GenericMap(values);
+ }
+}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/data/variant/VariantShreddingWritePlanTest.java
b/paimon-common/src/test/java/org/apache/paimon/data/variant/VariantShreddingWritePlanTest.java
new file mode 100644
index 0000000000..ae5a52a9b7
--- /dev/null
+++
b/paimon-common/src/test/java/org/apache/paimon/data/variant/VariantShreddingWritePlanTest.java
@@ -0,0 +1,104 @@
+/*
+ * 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.paimon.data.variant;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.VariantType;
+
+import org.junit.jupiter.api.Test;
+
+import static
org.apache.paimon.data.variant.PaimonShreddingUtils.assembleVariant;
+import static
org.apache.paimon.data.variant.PaimonShreddingUtils.buildVariantSchema;
+import static
org.apache.paimon.data.variant.PaimonShreddingUtils.variantShreddingSchema;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link VariantShreddingWritePlan}. */
+class VariantShreddingWritePlanTest {
+
+ @Test
+ void testConfiguredSchemaConvertsTopLevelVariant() {
+ RowType logicalType =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "id", DataTypes.INT()),
+ DataTypes.FIELD(1, "v", DataTypes.VARIANT()));
+ RowType configuredSchema =
+ DataTypes.ROW(
+ DataTypes.FIELD(
+ 1,
+ "v",
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "age",
DataTypes.BIGINT()),
+ DataTypes.FIELD(1, "name",
DataTypes.STRING()))));
+
+ VariantShreddingWritePlan writePlan =
+ VariantShreddingWritePlan.fromConfiguredSchema(logicalType,
configuredSchema);
+
+
assertThat(writePlan.physicalRowType().getTypeAt(1)).isInstanceOf(RowType.class);
+
assertThat(writePlan.physicalRowType().getTypeAt(1)).isNotInstanceOf(VariantType.class);
+
+ GenericVariant variant =
GenericVariant.fromJson("{\"age\":30,\"name\":\"Alice\"}");
+ InternalRow physicalRow = writePlan.toPhysicalRow(GenericRow.of(1,
variant));
+ RowType physicalVariantType = (RowType)
writePlan.physicalRowType().getTypeAt(1);
+ InternalRow shreddedVariant = physicalRow.getRow(1,
physicalVariantType.getFieldCount());
+
+ assertThat(physicalRow.getInt(0)).isEqualTo(1);
+ assertThat(
+ assembleVariant(shreddedVariant,
buildVariantSchema(physicalVariantType))
+ .toJson())
+ .isEqualTo(variant.toJson());
+ }
+
+ @Test
+ void testNestedVariantConvertsRecursively() {
+ RowType nestedLogicalType =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "payload", DataTypes.VARIANT()),
+ DataTypes.FIELD(1, "name", DataTypes.STRING()));
+ RowType logicalType = DataTypes.ROW(DataTypes.FIELD(0, "nested",
nestedLogicalType));
+
+ RowType payloadPhysicalType =
+ variantShreddingSchema(DataTypes.ROW(DataTypes.FIELD(0,
"score", DataTypes.INT())));
+ RowType nestedPhysicalType =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "payload", payloadPhysicalType),
+ DataTypes.FIELD(1, "name", DataTypes.STRING()));
+ RowType physicalType = DataTypes.ROW(DataTypes.FIELD(0, "nested",
nestedPhysicalType));
+
+ VariantShreddingWritePlan writePlan =
+ new VariantShreddingWritePlan(logicalType, physicalType);
+ GenericVariant variant = GenericVariant.fromJson("{\"score\":98}");
+ InternalRow physicalRow =
+ writePlan.toPhysicalRow(
+ GenericRow.of(
+ GenericRow.of(variant,
BinaryString.fromString("attempt-1"))));
+
+ InternalRow nestedRow = physicalRow.getRow(0,
nestedPhysicalType.getFieldCount());
+ InternalRow shreddedVariant = nestedRow.getRow(0,
payloadPhysicalType.getFieldCount());
+
+
assertThat(nestedRow.getString(1)).isEqualTo(BinaryString.fromString("attempt-1"));
+ assertThat(
+ assembleVariant(shreddedVariant,
buildVariantSchema(payloadPhysicalType))
+ .toJson())
+ .isEqualTo(variant.toJson());
+ }
+}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/format/shredding/ShreddingFormatWriterTest.java
b/paimon-common/src/test/java/org/apache/paimon/format/shredding/ShreddingFormatWriterTest.java
new file mode 100644
index 0000000000..8345d2ae41
--- /dev/null
+++
b/paimon-common/src/test/java/org/apache/paimon/format/shredding/ShreddingFormatWriterTest.java
@@ -0,0 +1,113 @@
+/*
+ * 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.paimon.format.shredding;
+
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.shredding.ShreddingWritePlan;
+import org.apache.paimon.format.FormatWriter;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link ShreddingFormatWriter}. */
+class ShreddingFormatWriterTest {
+
+ @Test
+ void testCloseDelegateWhenCommitMetadataFails() {
+ IOException failure = new IOException("metadata failed");
+ TestingFormatWriter delegate = new TestingFormatWriter();
+ ShreddingFormatWriter writer =
+ new ShreddingFormatWriter(
+ delegate,
+ new ThrowingMetadataFactory(failure),
+ NoOpWritePlan.INSTANCE,
+ "none");
+
+ assertThatThrownBy(writer::close).isSameAs(failure);
+ assertThat(delegate.closed).isTrue();
+ }
+
+ private static class TestingFormatWriter implements FormatWriter {
+
+ private boolean closed;
+
+ @Override
+ public void addElement(InternalRow element) {}
+
+ @Override
+ public boolean reachTargetSize(boolean suggestedCheck, long
targetSize) {
+ return false;
+ }
+
+ @Override
+ public void close() {
+ closed = true;
+ }
+ }
+
+ private static class ThrowingMetadataFactory implements
SupportsShreddingWritePlan {
+
+ private final IOException failure;
+
+ private ThrowingMetadataFactory(IOException failure) {
+ this.failure = failure;
+ }
+
+ @Override
+ public FormatWriter createWithShreddingWritePlan(
+ PositionOutputStream out, String compression,
ShreddingWritePlan writePlan) {
+ return new TestingFormatWriter();
+ }
+
+ @Override
+ public void commitShreddingMetadata(
+ FormatWriter writer, ShreddingWritePlan writePlan, String
compression)
+ throws IOException {
+ throw failure;
+ }
+ }
+
+ private enum NoOpWritePlan implements ShreddingWritePlan {
+ INSTANCE;
+
+ private final RowType rowType = new RowType(Collections.emptyList());
+
+ @Override
+ public RowType logicalRowType() {
+ return rowType;
+ }
+
+ @Override
+ public RowType physicalRowType() {
+ return rowType;
+ }
+
+ @Override
+ public InternalRow toPhysicalRow(InternalRow row) {
+ return row;
+ }
+ }
+}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/orc/OrcFileFormat.java
b/paimon-format/src/main/java/org/apache/paimon/format/orc/OrcFileFormat.java
index 01c2fa9517..5ac85779ab 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/orc/OrcFileFormat.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/orc/OrcFileFormat.java
@@ -154,7 +154,12 @@ public class OrcFileFormat extends FileFormat {
typeDescription, refinedType.getFields(),
legacyTimestampLtzType);
return new OrcWriterFactory(
- vectorizer, orcProperties, writerConf, writeBatchSize,
writeBatchMemory);
+ vectorizer,
+ orcProperties,
+ writerConf,
+ writeBatchSize,
+ writeBatchMemory,
+ legacyTimestampLtzType);
}
private Properties getOrcProperties(Options options, FormatContext
formatContext) {
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/orc/OrcWriterFactory.java
b/paimon-format/src/main/java/org/apache/paimon/format/orc/OrcWriterFactory.java
index 8be5798e2f..948182825a 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/orc/OrcWriterFactory.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/orc/OrcWriterFactory.java
@@ -20,12 +20,18 @@ package org.apache.paimon.format.orc;
import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.shredding.ShreddingWritePlan;
+import org.apache.paimon.format.FormatMetadataUtils;
import org.apache.paimon.format.FormatWriter;
import org.apache.paimon.format.FormatWriterFactory;
+import org.apache.paimon.format.SupportsWriterMetadata;
import org.apache.paimon.format.orc.writer.OrcBulkWriter;
+import org.apache.paimon.format.orc.writer.RowDataVectorizer;
import org.apache.paimon.format.orc.writer.Vectorizer;
+import org.apache.paimon.format.shredding.SupportsShreddingWritePlan;
import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.options.MemorySize;
+import org.apache.paimon.types.RowType;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
@@ -50,11 +56,12 @@ import static
org.apache.paimon.utils.Preconditions.checkNotNull;
* Vectorizer} implementation to convert the element into an {@link
* org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch}.
*/
-public class OrcWriterFactory implements FormatWriterFactory {
+public class OrcWriterFactory implements FormatWriterFactory,
SupportsShreddingWritePlan {
private final Vectorizer<InternalRow> vectorizer;
private final Properties writerProperties;
private final Map<String, String> confMap;
+ private final boolean legacyTimestampLtzType;
private OrcFile.WriterOptions writerOptions;
private final int writeBatchSize;
@@ -68,26 +75,20 @@ public class OrcWriterFactory implements
FormatWriterFactory {
*/
@VisibleForTesting
OrcWriterFactory(Vectorizer<InternalRow> vectorizer) {
- this(vectorizer, new Properties(), new Configuration(false), 1024,
MemorySize.ZERO);
+ this(vectorizer, new Properties(), new Configuration(false), 1024,
MemorySize.ZERO, false);
}
- /**
- * Creates a new OrcBulkWriterFactory using the provided Vectorizer,
Hadoop Configuration, ORC
- * writer properties.
- *
- * @param vectorizer The vectorizer implementation to convert input record
to a
- * VectorizerRowBatch.
- * @param writerProperties Properties that can be used in ORC
WriterOptions.
- */
public OrcWriterFactory(
Vectorizer<InternalRow> vectorizer,
Properties writerProperties,
Configuration configuration,
int writeBatchSize,
- MemorySize writeBatchMemory) {
+ MemorySize writeBatchMemory,
+ boolean legacyTimestampLtzType) {
this.vectorizer = checkNotNull(vectorizer);
this.writerProperties = checkNotNull(writerProperties);
this.confMap = new HashMap<>();
+ this.legacyTimestampLtzType = legacyTimestampLtzType;
// Todo: Replace the Map based approach with a better approach
for (Map.Entry<String, String> entry : configuration) {
@@ -128,18 +129,59 @@ public class OrcWriterFactory implements
FormatWriterFactory {
writeBatchMemory);
}
+ @Override
+ public FormatWriter createWithShreddingWritePlan(
+ PositionOutputStream out, String compression, ShreddingWritePlan
writePlan)
+ throws IOException {
+ RowType refinedType = (RowType)
OrcFileFormat.refineDataType(writePlan.physicalRowType());
+ Vectorizer<InternalRow> physicalVectorizer =
+ new RowDataVectorizer(
+ OrcTypeUtil.convertToOrcSchema(refinedType),
+ refinedType.getFields(),
+ legacyTimestampLtzType);
+ return new OrcWriterFactory(
+ physicalVectorizer,
+ writerProperties,
+ configuration(),
+ writeBatchSize,
+ writeBatchMemory,
+ legacyTimestampLtzType)
+ .create(out, compression);
+ }
+
+ @Override
+ public void commitShreddingMetadata(
+ FormatWriter writer, ShreddingWritePlan writePlan, String
compression) {
+ Map<String, Map<String, String>> fieldMetadata =
writePlan.fieldMetadata(compression);
+ if (fieldMetadata.isEmpty()) {
+ return;
+ }
+
+ Map<String, byte[]> metadata = new HashMap<>();
+ metadata.put(
+ FormatMetadataUtils.ARROW_SCHEMA_METADATA_KEY,
+ FormatMetadataUtils.buildArrowSchemaMetadata(
+ writePlan.physicalRowType(),
+ fieldMetadata,
+ OrcTypeUtil.PAIMON_ORC_FIELD_ID_KEY));
+ ((SupportsWriterMetadata) writer).addMetadata(metadata);
+ }
+
@VisibleForTesting
protected OrcFile.WriterOptions getWriterOptions() {
if (null == writerOptions) {
- Configuration conf = new ThreadLocalClassLoaderConfiguration();
- for (Map.Entry<String, String> entry : confMap.entrySet()) {
- conf.set(entry.getKey(), entry.getValue());
- }
-
- writerOptions = OrcFile.writerOptions(writerProperties, conf);
+ writerOptions = OrcFile.writerOptions(writerProperties,
configuration());
writerOptions.setSchema(this.vectorizer.getSchema());
}
return writerOptions;
}
+
+ private Configuration configuration() {
+ Configuration conf = new ThreadLocalClassLoaderConfiguration();
+ for (Map.Entry<String, String> entry : confMap.entrySet()) {
+ conf.set(entry.getKey(), entry.getValue());
+ }
+ return conf;
+ }
}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetFileFormat.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetFileFormat.java
index 66ab9565e7..7231be2de7 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetFileFormat.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetFileFormat.java
@@ -25,8 +25,8 @@ import org.apache.paimon.format.FormatReaderFactory;
import org.apache.paimon.format.FormatWriterFactory;
import org.apache.paimon.format.SimpleStatsExtractor;
import org.apache.paimon.format.parquet.writer.RowDataParquetBuilder;
-import org.apache.paimon.format.variant.VariantInferenceConfig;
-import org.apache.paimon.format.variant.VariantInferenceWriterFactory;
+import org.apache.paimon.format.shredding.ShreddingWritePlanWriterFactory;
+import org.apache.paimon.format.variant.VariantShreddingWritePlanFactory;
import org.apache.paimon.options.CatalogOptions;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
@@ -77,9 +77,8 @@ public class ParquetFileFormat extends FileFormat {
public FormatWriterFactory createWriterFactory(RowType type) {
ParquetWriterFactory baseFactory =
new ParquetWriterFactory(new RowDataParquetBuilder(type,
options));
- // Wrap with variant inference decorator
- return new VariantInferenceWriterFactory(
- baseFactory, new VariantInferenceConfig(type,
formatContext.options()));
+ return new ShreddingWritePlanWriterFactory(
+ baseFactory, new VariantShreddingWritePlanFactory(type,
formatContext.options()));
}
@Override
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetWriterFactory.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetWriterFactory.java
index a26a0f3d1a..69e6d98678 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetWriterFactory.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetWriterFactory.java
@@ -19,18 +19,20 @@
package org.apache.paimon.format.parquet;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.shredding.ShreddingWritePlan;
+import org.apache.paimon.format.FormatMetadataUtils;
import org.apache.paimon.format.FormatWriter;
import org.apache.paimon.format.FormatWriterFactory;
import org.apache.paimon.format.HadoopCompressionType;
+import org.apache.paimon.format.SupportsWriterMetadata;
import org.apache.paimon.format.parquet.writer.MetadataParquetBuilder;
import org.apache.paimon.format.parquet.writer.ParquetBuilder;
import org.apache.paimon.format.parquet.writer.ParquetBulkWriter;
import org.apache.paimon.format.parquet.writer.ParquetMetadataBulkWriter;
import org.apache.paimon.format.parquet.writer.RowDataParquetBuilder;
import org.apache.paimon.format.parquet.writer.StreamOutputFile;
-import org.apache.paimon.format.variant.SupportsVariantInference;
+import org.apache.paimon.format.shredding.SupportsShreddingWritePlan;
import org.apache.paimon.fs.PositionOutputStream;
-import org.apache.paimon.types.RowType;
import org.apache.parquet.hadoop.ParquetWriter;
import org.apache.parquet.io.OutputFile;
@@ -40,7 +42,7 @@ import java.util.HashMap;
import java.util.Map;
/** A factory that creates a Parquet {@link FormatWriter}. */
-public class ParquetWriterFactory implements FormatWriterFactory,
SupportsVariantInference {
+public class ParquetWriterFactory implements FormatWriterFactory,
SupportsShreddingWritePlan {
/** The builder to construct the ParquetWriter. */
private final ParquetBuilder<InternalRow> writerBuilder;
@@ -71,8 +73,8 @@ public class ParquetWriterFactory implements
FormatWriterFactory, SupportsVarian
}
@Override
- public FormatWriter createWithShreddingSchema(
- PositionOutputStream stream, String compression, RowType
inferredShreddingSchema)
+ public FormatWriter createWithShreddingWritePlan(
+ PositionOutputStream stream, String compression,
ShreddingWritePlan writePlan)
throws IOException {
final OutputFile out = new StreamOutputFile(stream);
if (HadoopCompressionType.NONE.value().equals(compression)) {
@@ -80,11 +82,28 @@ public class ParquetWriterFactory implements
FormatWriterFactory, SupportsVarian
}
RowDataParquetBuilder newBuilder =
- ((RowDataParquetBuilder) writerBuilder)
- .withShreddingSchemas(inferredShreddingSchema);
+ ((RowDataParquetBuilder)
writerBuilder).withRowType(writePlan.physicalRowType());
return createMetadataWriter(newBuilder, out, compression);
}
+ @Override
+ public void commitShreddingMetadata(
+ FormatWriter writer, ShreddingWritePlan writePlan, String
compression) {
+ Map<String, Map<String, String>> fieldMetadata =
writePlan.fieldMetadata(compression);
+ if (fieldMetadata.isEmpty()) {
+ return;
+ }
+
+ Map<String, byte[]> metadata = new HashMap<>();
+ metadata.put(
+ FormatMetadataUtils.ARROW_SCHEMA_METADATA_KEY,
+ FormatMetadataUtils.buildArrowSchemaMetadata(
+ writePlan.physicalRowType(),
+ fieldMetadata,
+ FormatMetadataUtils.PARQUET_FIELD_ID_KEY));
+ ((SupportsWriterMetadata) writer).addMetadata(metadata);
+ }
+
private FormatWriter createMetadataWriter(
MetadataParquetBuilder<InternalRow> builder, OutputFile out,
String compression)
throws IOException {
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/VariantUtils.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/VariantUtils.java
index 7fec5aee9f..2d3ed0e292 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/VariantUtils.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/VariantUtils.java
@@ -18,20 +18,13 @@
package org.apache.paimon.format.parquet;
-import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.variant.PaimonShreddingUtils;
-import org.apache.paimon.options.Options;
import org.apache.paimon.types.DataField;
-import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
-import org.apache.paimon.types.VariantType;
-import org.apache.paimon.utils.JsonSerdeUtil;
import org.apache.parquet.schema.Type;
-import javax.annotation.Nullable;
-
import java.util.ArrayList;
import java.util.List;
@@ -54,40 +47,4 @@ public class VariantUtils {
return new RowType(dataFields);
}
}
-
- /** For writer, extract shredding schemas from conf. */
- @Nullable
- public static RowType shreddingSchemasFromOptions(Options options) {
- if (!options.contains(CoreOptions.VARIANT_SHREDDING_SCHEMA)) {
- return null;
- }
-
- String shreddingSchema =
options.get(CoreOptions.VARIANT_SHREDDING_SCHEMA);
- RowType rowType = (RowType) JsonSerdeUtil.fromJson(shreddingSchema,
DataType.class);
- ArrayList<DataField> fields = new ArrayList<>();
- for (DataField field : rowType.getFields()) {
-
fields.add(field.newType(PaimonShreddingUtils.variantShreddingSchema(field.type())));
- }
- return new RowType(fields);
- }
-
- public static RowType replaceWithShreddingType(
- RowType rowType, @Nullable RowType shreddingSchemas) {
- if (shreddingSchemas == null) {
- return rowType;
- }
-
- List<DataField> newFields = new ArrayList<>();
- for (DataField field : rowType.getFields()) {
- // todo: support nested variant.
- if (field.type() instanceof VariantType
- && shreddingSchemas.containsField(field.name())) {
- RowType shreddingSchema = (RowType)
shreddingSchemas.getField(field.name()).type();
- newFields.add(field.newType(shreddingSchema));
- } else {
- newFields.add(field);
- }
- }
- return new RowType(rowType.isNullable(), newFields);
- }
}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/ParquetRowDataBuilder.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/ParquetRowDataBuilder.java
index 60fd1097fe..8271d589c9 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/ParquetRowDataBuilder.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/ParquetRowDataBuilder.java
@@ -20,7 +20,6 @@ package org.apache.paimon.format.parquet.writer;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.format.FormatMetadataUtils;
-import org.apache.paimon.format.parquet.VariantUtils;
import org.apache.paimon.types.RowType;
import org.apache.hadoop.conf.Configuration;
@@ -31,8 +30,6 @@ import org.apache.parquet.io.OutputFile;
import org.apache.parquet.io.api.RecordConsumer;
import org.apache.parquet.schema.MessageType;
-import javax.annotation.Nullable;
-
import java.util.HashMap;
import java.util.Map;
import java.util.function.Supplier;
@@ -44,14 +41,11 @@ public class ParquetRowDataBuilder
extends ParquetWriter.Builder<InternalRow, ParquetRowDataBuilder> {
private final RowType rowType;
- @Nullable private final RowType shreddingSchemas;
private Supplier<Map<String, byte[]>> metadataSupplier;
- public ParquetRowDataBuilder(
- OutputFile path, RowType rowType, @Nullable RowType
shreddingSchemas) {
+ public ParquetRowDataBuilder(OutputFile path, RowType rowType) {
super(path);
this.rowType = rowType;
- this.shreddingSchemas = shreddingSchemas;
this.metadataSupplier = HashMap::new;
}
@@ -68,23 +62,15 @@ public class ParquetRowDataBuilder
@Override
protected WriteSupport<InternalRow> getWriteSupport(Configuration conf) {
- return new ParquetWriteSupport(conf);
+ return new ParquetWriteSupport();
}
private class ParquetWriteSupport extends WriteSupport<InternalRow> {
- private final Configuration conf;
- private final MessageType schema;
+ private final MessageType schema =
convertToParquetMessageType(rowType);
private ParquetRowDataWriter writer;
- private ParquetWriteSupport(Configuration conf) {
- this.conf = conf;
- this.schema =
- convertToParquetMessageType(
- VariantUtils.replaceWithShreddingType(rowType,
shreddingSchemas));
- }
-
@Override
public WriteContext init(Configuration configuration) {
return new WriteContext(schema, new HashMap<>());
@@ -92,9 +78,7 @@ public class ParquetRowDataBuilder
@Override
public void prepareForWrite(RecordConsumer recordConsumer) {
- this.writer =
- new ParquetRowDataWriter(
- recordConsumer, rowType, schema, conf,
shreddingSchemas);
+ this.writer = new ParquetRowDataWriter(recordConsumer, rowType,
schema);
}
@Override
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/ParquetRowDataWriter.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/ParquetRowDataWriter.java
index 6a77b6cf20..a0e197e1da 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/ParquetRowDataWriter.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/ParquetRowDataWriter.java
@@ -25,10 +25,7 @@ import org.apache.paimon.data.InternalArray;
import org.apache.paimon.data.InternalMap;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.Timestamp;
-import org.apache.paimon.data.variant.GenericVariant;
-import org.apache.paimon.data.variant.PaimonShreddingUtils;
import org.apache.paimon.data.variant.Variant;
-import org.apache.paimon.data.variant.VariantSchema;
import org.apache.paimon.format.parquet.ParquetSchemaConverter;
import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.DataType;
@@ -42,7 +39,6 @@ import org.apache.paimon.types.TimestampType;
import org.apache.paimon.types.VariantType;
import org.apache.paimon.types.VectorType;
-import org.apache.hadoop.conf.Configuration;
import org.apache.parquet.io.api.Binary;
import org.apache.parquet.io.api.RecordConsumer;
import org.apache.parquet.schema.GroupType;
@@ -68,20 +64,11 @@ public class ParquetRowDataWriter {
public static final long NANOS_PER_MILLISECOND =
TimeUnit.MILLISECONDS.toNanos(1);
public static final long NANOS_PER_SECOND = TimeUnit.SECONDS.toNanos(1);
- private final Configuration conf;
private final RowWriter rowWriter;
private final RecordConsumer recordConsumer;
- @Nullable private final RowType shreddingSchemas;
-
- public ParquetRowDataWriter(
- RecordConsumer recordConsumer,
- RowType rowType,
- GroupType schema,
- Configuration conf,
- @Nullable RowType shreddingSchemas) {
- this.conf = conf;
+
+ public ParquetRowDataWriter(RecordConsumer recordConsumer, RowType
rowType, GroupType schema) {
this.recordConsumer = recordConsumer;
- this.shreddingSchemas = shreddingSchemas;
this.rowWriter = new RowWriter(rowType, schema);
}
@@ -159,11 +146,7 @@ public class ParquetRowDataWriter {
} else if (t instanceof RowType && type instanceof GroupType) {
return new RowWriter((RowType) t, groupType);
} else if (t instanceof VariantType && type instanceof GroupType) {
- RowType shreddingSchema =
- shreddingSchemas != null &&
shreddingSchemas.containsField(type.getName())
- ? (RowType)
shreddingSchemas.getField(type.getName()).type()
- : null;
- return new VariantWriter(groupType, shreddingSchema);
+ return new VariantWriter();
} else {
throw new UnsupportedOperationException("Unsupported type: " +
type);
}
@@ -637,19 +620,6 @@ public class ParquetRowDataWriter {
private class VariantWriter implements FieldWriter {
- @Nullable private final VariantSchema variantSchema;
- @Nullable private final RowWriter shreddedVariantWriter;
-
- public VariantWriter(GroupType groupType, @Nullable RowType
shreddingSchema) {
- if (shreddingSchema != null) {
- variantSchema =
PaimonShreddingUtils.buildVariantSchema(shreddingSchema);
- shreddedVariantWriter = new RowWriter(shreddingSchema,
groupType);
- } else {
- variantSchema = null;
- shreddedVariantWriter = null;
- }
- }
-
@Override
public void write(InternalRow row, int ordinal) {
writeVariant(row.getVariant(ordinal));
@@ -661,15 +631,6 @@ public class ParquetRowDataWriter {
}
private void writeVariant(Variant variant) {
- if (shreddedVariantWriter != null) {
- recordConsumer.startGroup();
- InternalRow shreddedVariant =
- PaimonShreddingUtils.castShredded((GenericVariant)
variant, variantSchema);
- shreddedVariantWriter.write(shreddedVariant);
- recordConsumer.endGroup();
- return;
- }
-
recordConsumer.startGroup();
recordConsumer.startField(Variant.VALUE, 0);
recordConsumer.addBinary(Binary.fromReusedByteArray(variant.value()));
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/RowDataParquetBuilder.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/RowDataParquetBuilder.java
index 6239fc546b..10adbfe35b 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/RowDataParquetBuilder.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/RowDataParquetBuilder.java
@@ -21,7 +21,6 @@ package org.apache.paimon.format.parquet.writer;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.format.HadoopCompressionType;
import org.apache.paimon.format.parquet.ColumnConfigParser;
-import org.apache.paimon.format.parquet.VariantUtils;
import org.apache.paimon.options.Options;
import org.apache.paimon.types.RowType;
@@ -32,8 +31,6 @@ import org.apache.parquet.hadoop.ParquetWriter;
import org.apache.parquet.hadoop.metadata.CompressionCodecName;
import org.apache.parquet.io.OutputFile;
-import javax.annotation.Nullable;
-
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
@@ -44,18 +41,20 @@ public class RowDataParquetBuilder implements
MetadataParquetBuilder<InternalRow
private final RowType rowType;
private final Configuration conf;
- @Nullable private RowType shreddingSchemas;
public RowDataParquetBuilder(RowType rowType, Options options) {
this.rowType = rowType;
this.conf = new Configuration(false);
- this.shreddingSchemas =
VariantUtils.shreddingSchemasFromOptions(options);
options.toMap().forEach(conf::set);
}
- public RowDataParquetBuilder withShreddingSchemas(RowType
shreddingSchemas) {
- this.shreddingSchemas = shreddingSchemas;
- return this;
+ private RowDataParquetBuilder(RowType rowType, Configuration conf) {
+ this.rowType = rowType;
+ this.conf = conf;
+ }
+
+ public RowDataParquetBuilder withRowType(RowType rowType) {
+ return new RowDataParquetBuilder(rowType, conf);
}
@Override
@@ -69,7 +68,7 @@ public class RowDataParquetBuilder implements
MetadataParquetBuilder<InternalRow
OutputFile out, String compression, Supplier<Map<String, byte[]>>
metadataSupplier)
throws IOException {
ParquetRowDataBuilder builder =
- new ParquetRowDataBuilder(out, rowType, shreddingSchemas)
+ new ParquetRowDataBuilder(out, rowType)
.withMetadataSupplier(metadataSupplier)
.withConf(conf)
.withCompressionCodec(getCompressionCodec(getCompression(compression)))
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/FileTypeNotMatchReadTypeTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/FileTypeNotMatchReadTypeTest.java
index c1d148116c..7afcbacc9d 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/FileTypeNotMatchReadTypeTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/FileTypeNotMatchReadTypeTest.java
@@ -90,9 +90,7 @@ public class FileTypeNotMatchReadTypeTest {
ParquetRowDataBuilder parquetRowDataBuilder =
new ParquetRowDataBuilder(
- new LocalOutputFile(new
File(fileWholePath).toPath()),
- rowTypeWrite,
- null);
+ new LocalOutputFile(new
File(fileWholePath).toPath()), rowTypeWrite);
ParquetWriter<InternalRow> parquetWriter =
parquetRowDataBuilder.build();
Timestamp timestamp = Timestamp.now();
@@ -131,9 +129,7 @@ public class FileTypeNotMatchReadTypeTest {
ParquetRowDataBuilder parquetRowDataBuilder =
new ParquetRowDataBuilder(
- new LocalOutputFile(new
File(fileWholePath).toPath()),
- rowTypeWrite,
- null);
+ new LocalOutputFile(new
File(fileWholePath).toPath()), rowTypeWrite);
ParquetWriter<InternalRow> parquetWriter =
parquetRowDataBuilder.build();
Decimal decimal =
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/ParquetRowDataBuilderForTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/ParquetRowDataBuilderForTest.java
index cd51332383..f4b0de67af 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/ParquetRowDataBuilderForTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/ParquetRowDataBuilderForTest.java
@@ -65,9 +65,7 @@ public class ParquetRowDataBuilderForTest
@Override
public void prepareForWrite(RecordConsumer recordConsumer) {
- this.writer =
- new ParquetRowDataWriter(
- recordConsumer, rowType, schema, new
Configuration(), null);
+ this.writer = new ParquetRowDataWriter(recordConsumer, rowType,
schema);
}
@Override
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/writer/InferVariantShreddingWriteTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/writer/InferVariantShreddingWriteTest.java
index 4dc7c2b273..8c8cf5bd23 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/writer/InferVariantShreddingWriteTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/writer/InferVariantShreddingWriteTest.java
@@ -33,7 +33,7 @@ import org.apache.paimon.format.SupportsDirectWrite;
import org.apache.paimon.format.parquet.ParquetFileFormat;
import org.apache.paimon.format.parquet.ParquetUtil;
import org.apache.paimon.format.parquet.VariantUtils;
-import org.apache.paimon.format.variant.InferVariantShreddingWriter;
+import org.apache.paimon.format.shredding.InferShreddingWritePlanWriter;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.PositionOutputStream;
@@ -63,7 +63,7 @@ import java.util.UUID;
import static
org.apache.paimon.data.variant.PaimonShreddingUtils.variantShreddingSchema;
import static org.assertj.core.api.Assertions.assertThat;
-/** Test for {@link InferVariantShreddingWriter}. */
+/** Test for {@link InferShreddingWritePlanWriter}. */
public class InferVariantShreddingWriteTest {
@TempDir java.nio.file.Path tempPath;
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/shredding/ShreddingWritePlanFormatTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/shredding/ShreddingWritePlanFormatTest.java
new file mode 100644
index 0000000000..dfc8a485ab
--- /dev/null
+++
b/paimon-format/src/test/java/org/apache/paimon/format/shredding/ShreddingWritePlanFormatTest.java
@@ -0,0 +1,161 @@
+/*
+ * 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.paimon.format.shredding;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericMap;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.shredding.MapSharedShreddingFieldMeta;
+import org.apache.paimon.data.shredding.MapSharedShreddingUtils;
+import org.apache.paimon.data.shredding.MapShreddingDefine;
+import org.apache.paimon.format.FileFormat;
+import org.apache.paimon.format.FileFormatFactory;
+import org.apache.paimon.format.FormatMetadataUtils;
+import org.apache.paimon.format.FormatReaderContext;
+import org.apache.paimon.format.FormatWriter;
+import org.apache.paimon.format.FormatWriterFactory;
+import org.apache.paimon.format.SupportsReaderFieldMetadata;
+import org.apache.paimon.format.orc.OrcFileFormat;
+import org.apache.paimon.format.orc.OrcTypeUtil;
+import org.apache.paimon.format.parquet.ParquetFileFormat;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.reader.FileRecordReader;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for format integration of {@link ShreddingWritePlanWriterFactory}. */
+class ShreddingWritePlanFormatTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void testParquetWritesMapSharedShreddingMetadataThroughVariantWrapper()
throws Exception {
+ FileFormat format =
+ new ParquetFileFormat(
+ new FileFormatFactory.FormatContext(new Options(),
1024, 1024));
+
+ Map<String, Map<String, String>> fieldMetadata =
+ writeAndReadFieldMetadata(format, "parquet", "none");
+
+ assertMapSharedShreddingMetadata(fieldMetadata, "none");
+ assertThat(fieldMetadata.get("id"))
+ .containsEntry(FormatMetadataUtils.PARQUET_FIELD_ID_KEY, "0");
+ assertThat(fieldMetadata.get("tags"))
+ .containsEntry(FormatMetadataUtils.PARQUET_FIELD_ID_KEY, "1");
+ }
+
+ @Test
+ void testOrcWritesMapSharedShreddingMetadata() throws Exception {
+ FileFormat format =
+ new OrcFileFormat(new FileFormatFactory.FormatContext(new
Options(), 1024, 1024));
+
+ Map<String, Map<String, String>> fieldMetadata =
+ writeAndReadFieldMetadata(format, "orc", "none");
+
+ assertMapSharedShreddingMetadata(fieldMetadata, "none");
+
assertThat(fieldMetadata.get("id")).containsEntry(OrcTypeUtil.PAIMON_ORC_FIELD_ID_KEY,
"0");
+ assertThat(fieldMetadata.get("tags"))
+ .containsEntry(OrcTypeUtil.PAIMON_ORC_FIELD_ID_KEY, "1");
+ }
+
+ private Map<String, Map<String, String>> writeAndReadFieldMetadata(
+ FileFormat format, String extension, String compression) throws
IOException {
+ FileIO fileIO = LocalFileIO.create();
+ Path file = new Path(tempDir.toString(), UUID.randomUUID() + "." +
extension);
+ RowType rowType = logicalRowType();
+ FormatWriterFactory writerFactory =
+ ShreddingWritePlanWriterFactories.wrapMapSharedShredding(
+ format.createWriterFactory(rowType), rowType,
mapSharedShreddingOptions());
+
+ PositionOutputStream out = fileIO.newOutputStream(file, false);
+ FormatWriter writer = writerFactory.create(out, compression);
+ writer.addElement(GenericRow.of(1, stringKeyMap("a", 10L, "b", 20L,
"c", 30L)));
+ writer.close();
+ out.close();
+
+ RowType emptyRowType = new RowType(Collections.emptyList());
+ FormatReaderContext readerContext =
+ new FormatReaderContext(fileIO, file,
fileIO.getFileSize(file));
+ try (FileRecordReader<InternalRow> reader =
+ format.createReaderFactory(emptyRowType, emptyRowType,
Collections.emptyList())
+ .createReader(readerContext)) {
+ return ((SupportsReaderFieldMetadata) reader).readFieldMetadata();
+ }
+ }
+
+ private static RowType logicalRowType() {
+ return DataTypes.ROW(
+ DataTypes.FIELD(0, "id", DataTypes.INT()),
+ DataTypes.FIELD(1, "tags", DataTypes.MAP(DataTypes.STRING(),
DataTypes.BIGINT())));
+ }
+
+ private static CoreOptions mapSharedShreddingOptions() {
+ Options options = new Options();
+ options.setString("fields.tags.map.storage-layout",
"shared-shredding");
+ options.setString("fields.tags.map.shared-shredding.max-columns", "2");
+ return new CoreOptions(options);
+ }
+
+ private static void assertMapSharedShreddingMetadata(
+ Map<String, Map<String, String>> fieldMetadata, String
compression) {
+ assertThat(fieldMetadata).containsKey("tags");
+ assertThat(fieldMetadata.get("tags"))
+ .containsEntry(
+ MapShreddingDefine.STORAGE_LAYOUT,
+ MapShreddingDefine.STORAGE_LAYOUT_SHARED_SHREDDING);
+
+ MapSharedShreddingFieldMeta fieldMeta =
+
MapSharedShreddingUtils.deserializeMetadata(fieldMetadata.get("tags"),
compression);
+ assertThat(fieldMeta.nameToId())
+ .containsEntry("a", 0)
+ .containsEntry("b", 1)
+ .containsEntry("c", 2);
+ assertThat(fieldMeta.fieldToColumns())
+ .containsEntry(0, Collections.singletonList(0))
+ .containsEntry(1, Collections.singletonList(1));
+ assertThat(fieldMeta.overflowFieldSet()).containsExactly(2);
+ assertThat(fieldMeta.numColumns()).isEqualTo(2);
+ assertThat(fieldMeta.maxRowWidth()).isEqualTo(3);
+ }
+
+ private static GenericMap stringKeyMap(Object... keyValues) {
+ Map<Object, Object> values = new LinkedHashMap<>();
+ for (int i = 0; i < keyValues.length; i += 2) {
+ values.put(BinaryString.fromString((String) keyValues[i]),
keyValues[i + 1]);
+ }
+ return new GenericMap(values);
+ }
+}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/VariantTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/VariantTestBase.scala
index 34c48a5ba4..4cb3837071 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/VariantTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/VariantTestBase.scala
@@ -922,14 +922,13 @@ abstract class VariantTestBase extends
PaimonSparkTestBase {
| (2, struct('Bob', parse_json('{"score":88,"grade":"B"}')))
| """.stripMargin)
- checkAnswer(
- sql("SELECT * FROM T ORDER BY id"),
- sql("""
- |SELECT 1, struct('Alice', parse_json('{"score":95,"grade":"A"}'))
- |UNION ALL
- |SELECT 2, struct('Bob', parse_json('{"score":88,"grade":"B"}'))
- |""".stripMargin)
- )
+ val result = sql("SELECT * FROM T ORDER BY id")
+ assert(result.collect().length == 2)
+ assert(result.schema.fieldNames.toSeq == Seq("id", "data"))
+ val dataSchema = result.schema("data").dataType.asInstanceOf[StructType]
+ assert(dataSchema.fieldNames.toSeq == Seq("name", "v"))
+ assert(isVariantType(dataSchema("v").dataType))
+
checkAnswer(
sql(
"SELECT id, data.name, variant_get(data.v, '$.score', 'int'),
variant_get(data.v, '$.grade', 'string') FROM T ORDER BY id"),
@@ -1010,10 +1009,12 @@ abstract class VariantTestBase extends
PaimonSparkTestBase {
checkAnswer(
sql("SELECT * FROM T"),
sql("""
- |SELECT 1, struct(
+ |SELECT 1 AS id, named_struct(
+ | 'tags',
| array(parse_json('{"tag":"important","priority":1}'),
parse_json('{"tag":"urgent","priority":2}')),
+ | 'attrs',
| map('color', parse_json('{"r":255,"g":0,"b":0}'), 'size',
parse_json('{"width":100,"height":200}'))
- |)
+ |) AS data
|""".stripMargin)
)
checkAnswer(
