diff --git a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlan.java b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlan.java new file mode 100644 index 000000000000..521dbd5b5552 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlan.java @@ -0,0 +1,304 @@ +/* + * 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.InternalArray; +import org.apache.paimon.data.InternalMap; +import org.apache.paimon.data.columnar.ArrayColumnVector; +import org.apache.paimon.data.columnar.ColumnVector; +import org.apache.paimon.data.columnar.ColumnVectorUtils; +import org.apache.paimon.data.columnar.MapColumnVector; +import org.apache.paimon.data.columnar.RowColumnVector; +import org.apache.paimon.data.columnar.RowToColumnConverter; +import org.apache.paimon.data.columnar.VectorizedColumnBatch; +import org.apache.paimon.data.columnar.heap.CastedMapColumnVector; +import org.apache.paimon.data.columnar.heap.HeapMapVector; +import org.apache.paimon.data.columnar.writable.WritableColumnVector; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.MapType; +import org.apache.paimon.types.RowType; + +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.Map; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** Read plan that rebuilds logical MAP values from shared-shredding physical ROW values. */ +public class MapSharedShreddingReadPlan implements ShreddingReadPlan { + + private final RowType logicalType; + private final RowType physicalType; + private final Map contextByFieldIndex; + + public MapSharedShreddingReadPlan( + RowType logicalType, Map fieldMetas) { + this.logicalType = logicalType; + this.physicalType = MapSharedShreddingUtils.buildPhysicalReadType(logicalType, fieldMetas); + this.contextByFieldIndex = createContexts(logicalType, fieldMetas); + } + + @Override + public RowType logicalRowType() { + return logicalType; + } + + @Override + public RowType physicalRowType() { + return physicalType; + } + + @Override + public boolean isIdentity() { + return contextByFieldIndex.isEmpty(); + } + + @Override + public ShreddingBatchAssembler batchAssembler() { + return new MapSharedShreddingBatchAssembler(); + } + + private static Map createContexts( + RowType logicalType, Map fieldMetas) { + Map contexts = new LinkedHashMap<>(); + for (int i = 0; i < logicalType.getFieldCount(); i++) { + DataField field = logicalType.getFields().get(i); + MapSharedShreddingFieldMeta fieldMeta = fieldMetas.get(field.name()); + if (fieldMeta != null && field.type() instanceof MapType) { + contexts.put(i, new SharedShreddingContext(fieldMeta, field.type())); + } + } + return contexts; + } + + private class MapSharedShreddingBatchAssembler implements ShreddingBatchAssembler { + + @Override + public VectorizedColumnBatch assemble(VectorizedColumnBatch physicalBatch) { + ColumnVector[] logicalVectors = new ColumnVector[logicalType.getFieldCount()]; + for (int i = 0; i < logicalVectors.length; i++) { + SharedShreddingContext context = contextByFieldIndex.get(i); + if (context == null) { + logicalVectors[i] = physicalBatch.columns[i]; + } else { + logicalVectors[i] = + materializeLogicalMapVector( + (RowColumnVector) physicalBatch.columns[i], + context, + physicalBatch.getNumRows()); + } + } + return physicalBatch.copy(logicalVectors); + } + } + + private static CastedMapColumnVector materializeLogicalMapVector( + RowColumnVector physicalVector, SharedShreddingContext context, int rowCount) { + ColumnVector[] physicalChildren = physicalChildren(physicalVector); + int totalElements = + countLogicalMapElements(physicalVector, physicalChildren, context, rowCount); + + WritableColumnVector keyVector = + ColumnVectorUtils.createWritableColumnVector( + totalElements, context.mapType.getKeyType()); + WritableColumnVector valueVector = + ColumnVectorUtils.createWritableColumnVector( + totalElements, context.mapType.getValueType()); + HeapMapVector mapVector = new HeapMapVector(rowCount, keyVector, valueVector); + + int offset = 0; + for (int row = 0; row < rowCount; row++) { + if (physicalVector.isNullAt(row)) { + mapVector.setNullAt(row); + mapVector.putOffsetLength(row, offset, 0); + continue; + } + + int elementCount = + appendLogicalMapElements( + physicalChildren, row, context, keyVector, valueVector); + mapVector.putOffsetLength(row, offset, elementCount); + offset += elementCount; + } + mapVector.addElementsAppended(rowCount); + + return new CastedMapColumnVector( + mapVector, + ColumnVectorUtils.createReadableColumnVectors( + Arrays.asList(context.mapType.getKeyType(), context.mapType.getValueType()), + new WritableColumnVector[] {keyVector, valueVector})); + } + + private static ColumnVector[] physicalChildren(RowColumnVector physicalVector) { + ColumnVector[] physicalChildren = physicalVector.getChildren(); + if (physicalChildren != null) { + return physicalChildren; + } + return physicalVector.getBatch().columns; + } + + private static int countLogicalMapElements( + RowColumnVector physicalVector, + ColumnVector[] physicalChildren, + SharedShreddingContext context, + int rowCount) { + int totalElements = 0; + for (int row = 0; row < rowCount; row++) { + if (!physicalVector.isNullAt(row)) { + totalElements += countLogicalMapElements(physicalChildren, row, context); + } + } + return totalElements; + } + + private static int countLogicalMapElements( + ColumnVector[] physicalChildren, int row, SharedShreddingContext context) { + int count = 0; + InternalArray fieldMapping = fieldMapping(physicalChildren, row, context); + for (int column = 0; column < context.numColumns; column++) { + if (logicalFieldName(fieldMapping, column, context) != null) { + count++; + } + } + + InternalMap overflow = overflowMap(physicalChildren, row, context); + if (overflow != null) { + InternalArray keys = overflow.keyArray(); + for (int i = 0; i < overflow.size(); i++) { + if (context.nameById.containsKey(keys.getInt(i))) { + count++; + } + } + } + return count; + } + + private static int appendLogicalMapElements( + ColumnVector[] physicalChildren, + int row, + SharedShreddingContext context, + WritableColumnVector keyVector, + WritableColumnVector valueVector) { + int count = 0; + InternalArray fieldMapping = fieldMapping(physicalChildren, row, context); + for (int column = 0; column < context.numColumns; column++) { + BinaryString fieldName = logicalFieldName(fieldMapping, column, context); + if (fieldName == null) { + continue; + } + + context.keyConverter.append(fieldName, keyVector); + ColumnVector valueColumn = physicalChildren[column + 1]; + appendValue(context, valueColumn, row, valueVector); + count++; + } + + InternalMap overflow = overflowMap(physicalChildren, row, context); + if (overflow != null) { + InternalArray keys = overflow.keyArray(); + InternalArray values = overflow.valueArray(); + for (int i = 0; i < overflow.size(); i++) { + BinaryString fieldName = context.nameById.get(keys.getInt(i)); + if (fieldName != null) { + context.keyConverter.append(fieldName, keyVector); + appendValue(context, values, i, valueVector); + count++; + } + } + } + return count; + } + + private static void appendValue( + SharedShreddingContext context, + ColumnVector source, + int row, + WritableColumnVector valueVector) { + if (source.isNullAt(row)) { + valueVector.appendNull(); + } else { + context.valueConverter.append(source, row, valueVector); + } + } + + private static void appendValue( + SharedShreddingContext context, + InternalArray source, + int pos, + WritableColumnVector valueVector) { + if (source.isNullAt(pos)) { + valueVector.appendNull(); + } else { + context.valueConverter.append(source, pos, valueVector); + } + } + + private static InternalArray fieldMapping( + ColumnVector[] physicalChildren, int row, SharedShreddingContext context) { + InternalArray fieldMapping = ((ArrayColumnVector) physicalChildren[0]).getArray(row); + checkArgument( + fieldMapping.size() == context.numColumns, + "Shared-shredding field mapping size %s does not match metadata num columns %s.", + fieldMapping.size(), + context.numColumns); + return fieldMapping; + } + + private static BinaryString logicalFieldName( + InternalArray fieldMapping, int column, SharedShreddingContext context) { + int fieldId = fieldMapping.isNullAt(column) ? -1 : fieldMapping.getInt(column); + return fieldId < 0 ? null : context.nameById.get(fieldId); + } + + private static InternalMap overflowMap( + ColumnVector[] physicalChildren, int row, SharedShreddingContext context) { + if (context.overflowPosition >= physicalChildren.length) { + return null; + } + + ColumnVector overflowVector = physicalChildren[context.overflowPosition]; + return overflowVector.isNullAt(row) ? null : ((MapColumnVector) overflowVector).getMap(row); + } + + private static class SharedShreddingContext { + + private final MapType mapType; + private final Map nameById; + private final RowToColumnConverter.ElementConverter keyConverter; + private final RowToColumnConverter.ElementConverter valueConverter; + private final int numColumns; + private final int overflowPosition; + + private SharedShreddingContext(MapSharedShreddingFieldMeta fieldMeta, DataType fieldType) { + this.mapType = (MapType) fieldType; + this.nameById = new LinkedHashMap<>(); + for (Map.Entry entry : fieldMeta.nameToId().entrySet()) { + this.nameById.put(entry.getValue(), BinaryString.fromString(entry.getKey())); + } + this.keyConverter = + RowToColumnConverter.createElementConverter(this.mapType.getKeyType()); + this.valueConverter = + RowToColumnConverter.createElementConverter(this.mapType.getValueType()); + this.numColumns = fieldMeta.numColumns(); + this.overflowPosition = fieldMeta.numColumns() + 1; + } + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlanFactory.java b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlanFactory.java new file mode 100644 index 000000000000..36ee35b8f725 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlanFactory.java @@ -0,0 +1,78 @@ +/* + * 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.format.shredding.ShreddingReadPlanFactory; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.RowType; + +import javax.annotation.Nullable; + +import java.util.LinkedHashMap; +import java.util.Map; + +/** Creates per-file shared-shredding MAP read plans. */ +public class MapSharedShreddingReadPlanFactory implements ShreddingReadPlanFactory { + + private final RowType logicalRowType; + + public MapSharedShreddingReadPlanFactory(RowType logicalRowType) { + this.logicalRowType = logicalRowType; + } + + @Override + public RowType logicalRowType() { + return logicalRowType; + } + + @Override + public boolean shouldCreateReadPlan( + Map> fieldMetadata, @Nullable Object fileSchema) { + for (DataField field : logicalRowType.getFields()) { + Map metadata = fieldMetadata.get(field.name()); + if (MapSharedShreddingUtils.hasShreddingMetadata(metadata)) { + return true; + } + } + return false; + } + + @Override + public ShreddingReadPlan createReadPlan( + Map> fieldMetadata, @Nullable Object fileSchema) { + return new MapSharedShreddingReadPlan( + logicalRowType, readSharedShreddingMetas(logicalRowType, fieldMetadata)); + } + + private static Map readSharedShreddingMetas( + RowType logicalRowType, Map> fieldMetadata) { + Map metas = new LinkedHashMap<>(); + for (Map.Entry> entry : fieldMetadata.entrySet()) { + if (logicalRowType != null && !logicalRowType.containsField(entry.getKey())) { + continue; + } + if (MapSharedShreddingUtils.hasShreddingMetadata(entry.getValue())) { + metas.put( + entry.getKey(), + MapSharedShreddingUtils.deserializeMetadata(entry.getValue())); + } + } + return metas; + } +} 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 82f191f62454..c2526b03bbde 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 @@ -103,6 +103,38 @@ public static RowType logicalToPhysicalSchema( return new RowType(logicalSchema.isNullable(), physicalFields); } + public static RowType buildPhysicalReadType( + RowType logicalReadType, + Map sharedShreddingFieldMetas) { + if (sharedShreddingFieldMetas.isEmpty()) { + return logicalReadType; + } + + List physicalReadFields = new ArrayList<>(); + boolean converted = false; + for (DataField logicalReadField : logicalReadType.getFields()) { + MapSharedShreddingFieldMeta fieldMeta = + sharedShreddingFieldMetas.get(logicalReadField.name()); + if (fieldMeta == null || !(logicalReadField.type() instanceof MapType)) { + physicalReadFields.add(logicalReadField); + continue; + } + + MapType mapType = (MapType) logicalReadField.type(); + DataType physicalType = + buildSpecificPhysicalStructType( + mapType.getValueType(), + fieldMeta.numColumns(), + !fieldMeta.overflowFieldSet().isEmpty()) + .copy(logicalReadField.type().isNullable()); + physicalReadFields.add(logicalReadField.newType(physicalType)); + converted = true; + } + return converted + ? new RowType(logicalReadType.isNullable(), physicalReadFields) + : logicalReadType; + } + public static Map buildColumnToNumColumns( List shreddingFieldNames, CoreOptions options) { Map fieldToNumColumns = new HashMap<>(); @@ -222,6 +254,19 @@ private static RowType buildPhysicalStructType(DataType valueType, int numColumn return builder.build(); } + private static RowType buildSpecificPhysicalStructType( + DataType valueType, int numColumns, boolean includeOverflow) { + RowType.Builder builder = RowType.builder(); + builder.field(MapSharedShreddingDefine.FIELD_MAPPING, new ArrayType(new IntType())); + for (int i = 0; i < numColumns; i++) { + builder.field(MapSharedShreddingDefine.physicalColumnName(i), valueType); + } + if (includeOverflow) { + builder.field(MapSharedShreddingDefine.OVERFLOW, new MapType(new IntType(), valueType)); + } + return builder.build(); + } + private static Map> sortedFieldColumns( Map> fieldToColumns) { Map> result = new TreeMap<>(); diff --git a/paimon-common/src/main/java/org/apache/paimon/format/FileFormat.java b/paimon-common/src/main/java/org/apache/paimon/format/FileFormat.java index 430fc1404cd2..39eee3f1e705 100644 --- a/paimon-common/src/main/java/org/apache/paimon/format/FileFormat.java +++ b/paimon-common/src/main/java/org/apache/paimon/format/FileFormat.java @@ -19,8 +19,13 @@ package org.apache.paimon.format; import org.apache.paimon.CoreOptions; +import org.apache.paimon.data.shredding.MapSharedShreddingContext; import org.apache.paimon.factories.FormatFactoryUtil; import org.apache.paimon.format.FileFormatFactory.FormatContext; +import org.apache.paimon.format.shredding.ShreddingWritePlanFactory; +import org.apache.paimon.format.shredding.ShreddingWritePlanType; +import org.apache.paimon.format.shredding.ShreddingWritePlanWriterFactories; +import org.apache.paimon.format.shredding.ShreddingWritePlanWriterFactory; import org.apache.paimon.options.Options; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.statistics.SimpleColStatsCollector; @@ -28,10 +33,12 @@ import javax.annotation.Nullable; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.Set; import static org.apache.paimon.CoreOptions.normalizeFileFormat; @@ -43,9 +50,15 @@ public abstract class FileFormat { protected String formatIdentifier; + private final Options options; protected FileFormat(String formatIdentifier) { + this(formatIdentifier, new Options()); + } + + protected FileFormat(String formatIdentifier, Options options) { this.formatIdentifier = formatIdentifier; + this.options = options; } public String getFormatIdentifier() { @@ -63,7 +76,39 @@ public abstract FormatReaderFactory createReaderFactory( RowType dataSchemaRowType, RowType projectedRowType, @Nullable List filters); /** Create a {@link FormatWriterFactory} from the type. */ - public abstract FormatWriterFactory createWriterFactory(RowType type); + public FormatWriterFactory createWriterFactory(RowType type) { + return createWriterFactory(type, null); + } + + /** + * Create a {@link FormatWriterFactory} from the type and optional restored MAP shared-shredding + * context. + */ + public final FormatWriterFactory createWriterFactory( + RowType type, @Nullable MapSharedShreddingContext mapSharedShreddingContext) { + ShreddingWritePlanFactory writePlanFactory = + ShreddingWritePlanWriterFactories.createWritePlanFactory( + type, + options, + mapSharedShreddingContext, + supportedShreddingWritePlans(), + formatIdentifier); + FormatWriterFactory rawFactory = createRawWriterFactory(type); + return writePlanFactory == null + ? rawFactory + : new ShreddingWritePlanWriterFactory(rawFactory, writePlanFactory); + } + + /** Create the raw format writer factory without logical-to-physical conversion. */ + protected FormatWriterFactory createRawWriterFactory(RowType type) { + throw new UnsupportedOperationException( + "Raw writer factory is not implemented for file format: " + formatIdentifier); + } + + /** Shredding write plans supported by this format. */ + protected Set supportedShreddingWritePlans() { + return Collections.emptySet(); + } /** Validate data field type supported or not. */ public abstract void validateDataFields(RowType rowType); diff --git a/paimon-common/src/main/java/org/apache/paimon/format/shredding/ShreddingWritePlanType.java b/paimon-common/src/main/java/org/apache/paimon/format/shredding/ShreddingWritePlanType.java new file mode 100644 index 000000000000..a94eb0272df5 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/format/shredding/ShreddingWritePlanType.java @@ -0,0 +1,25 @@ +/* + * 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; + +/** Logical-to-physical shredding write plan types. */ +public enum ShreddingWritePlanType { + VARIANT, + MAP_SHARED_SHREDDING +} 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 index 1fba3ddedd3b..fa69f502404b 100644 --- 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 @@ -18,32 +18,57 @@ 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.format.variant.VariantShreddingWritePlanFactory; +import org.apache.paimon.options.Options; import org.apache.paimon.types.RowType; -import java.util.List; +import javax.annotation.Nullable; -/** Helpers for composing per-file shredding writer factories. */ +import java.util.Set; + +/** Creates the single active shredding write plan factory for a format writer. */ public class ShreddingWritePlanWriterFactories { private ShreddingWritePlanWriterFactories() {} - public static FormatWriterFactory wrapMapSharedShredding( - FormatWriterFactory delegate, RowType rowType, CoreOptions options) { - List shreddingFields = - MapSharedShreddingUtils.detectShreddingColumns(rowType, options); - if (shreddingFields.isEmpty()) { - return delegate; + @Nullable + public static ShreddingWritePlanFactory createWritePlanFactory( + RowType rowType, + Options options, + @Nullable MapSharedShreddingContext mapSharedShreddingContext, + Set supportedTypes, + String formatIdentifier) { + ShreddingWritePlanFactory activeFactory = null; + ShreddingWritePlanType activeType = null; + + VariantShreddingWritePlanFactory variantFactory = + new VariantShreddingWritePlanFactory(rowType, options); + if (variantFactory.shouldCreateWritePlan()) { + activeFactory = variantFactory; + activeType = ShreddingWritePlanType.VARIANT; + } + + if (mapSharedShreddingContext != null && !mapSharedShreddingContext.isEmpty()) { + if (activeFactory != null) { + throw new UnsupportedOperationException( + "Composing multiple active shredding write plans is not supported."); + } + activeFactory = + new MapSharedShreddingWritePlanFactory(rowType, mapSharedShreddingContext); + activeType = ShreddingWritePlanType.MAP_SHARED_SHREDDING; } - MapSharedShreddingContext context = - new MapSharedShreddingContext( - MapSharedShreddingUtils.buildColumnToNumColumns(shreddingFields, options)); - return new ShreddingWritePlanWriterFactory( - delegate, new MapSharedShreddingWritePlanFactory(rowType, context)); + if (activeFactory == null) { + return null; + } + if (!supportedTypes.contains(activeType)) { + throw new UnsupportedOperationException( + String.format( + "File format '%s' does not support %s write plans.", + formatIdentifier, activeType)); + } + return activeFactory; } } 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 index 32be092dcd12..5441eb4a9799 100644 --- 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 @@ -22,80 +22,44 @@ import org.apache.paimon.format.FormatWriter; import org.apache.paimon.format.FormatWriterFactory; import org.apache.paimon.fs.PositionOutputStream; +import org.apache.paimon.utils.Preconditions; 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 { +/** Decorates a raw format writer factory with an active per-file shredding write plan. */ +public class ShreddingWritePlanWriterFactory implements FormatWriterFactory { - private final FormatWriterFactory delegate; + private final SupportsShreddingWritePlan 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()) { - return delegate.create(out, compression); - } + Preconditions.checkArgument( + writePlanFactory.shouldCreateWritePlan(), + "Shredding write plan factory must be active."); if (!(delegate instanceof SupportsShreddingWritePlan)) { throw new UnsupportedOperationException( "Delegate writer factory does not support shredding write plans: " + delegate.getClass().getName()); } + this.delegate = (SupportsShreddingWritePlan) delegate; + this.writePlanFactory = writePlanFactory; + } - SupportsShreddingWritePlan shreddingDelegate = (SupportsShreddingWritePlan) delegate; + @Override + public FormatWriter create(PositionOutputStream out, String compression) throws IOException { if (writePlanFactory.shouldInferWritePlan()) { - return new InferShreddingWritePlanWriter( - shreddingDelegate, writePlanFactory, out, compression); + return new InferShreddingWritePlanWriter(delegate, writePlanFactory, out, compression); } return createWriterWithPlan( - shreddingDelegate, + delegate, 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, diff --git a/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlanTest.java b/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlanTest.java new file mode 100644 index 000000000000..943bf94dbf75 --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingReadPlanTest.java @@ -0,0 +1,185 @@ +/* + * 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.InternalMap; +import org.apache.paimon.data.columnar.BytesColumnVector; +import org.apache.paimon.data.columnar.ColumnVector; +import org.apache.paimon.data.columnar.LongColumnVector; +import org.apache.paimon.data.columnar.MapColumnVector; +import org.apache.paimon.data.columnar.VectorizedColumnBatch; +import org.apache.paimon.data.columnar.heap.HeapArrayVector; +import org.apache.paimon.data.columnar.heap.HeapIntVector; +import org.apache.paimon.data.columnar.heap.HeapLongVector; +import org.apache.paimon.data.columnar.heap.HeapMapVector; +import org.apache.paimon.data.columnar.heap.HeapRowVector; +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 java.util.TreeMap; +import java.util.TreeSet; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link MapSharedShreddingReadPlan}. */ +class MapSharedShreddingReadPlanTest { + + @Test + void testReadProjectedPhysicalRowWithoutOverflowColumn() { + MapSharedShreddingFieldMeta fieldMeta = + new MapSharedShreddingFieldMeta( + nameToId("a", 0, "b", 1), + Collections.emptyMap(), + new TreeSet(), + 3, + 2); + HeapRowVector physicalMap = + rowVector( + fieldMapping(0, -1, 1), longVector(10L), longVector(null), longVector(20L)); + + InternalMap restored = readMap(fieldMeta, physicalMap); + + assertThat(restored.size()).isEqualTo(2); + assertThat(restored.keyArray().getString(0)).isEqualTo(BinaryString.fromString("a")); + assertThat(restored.keyArray().getString(1)).isEqualTo(BinaryString.fromString("b")); + assertThat(restored.valueArray().getLong(0)).isEqualTo(10L); + assertThat(restored.valueArray().getLong(1)).isEqualTo(20L); + } + + @Test + void testReadOverflowOnlyWhenOverflowColumnExists() { + MapSharedShreddingFieldMeta fieldMeta = + new MapSharedShreddingFieldMeta( + nameToId("a", 0, "overflowed", 1), + Collections.emptyMap(), + new TreeSet(Collections.singletonList(1)), + 1, + 1); + HeapRowVector physicalMap = + rowVector(fieldMapping(-1), longVector(null), overflowMap(1, 30L)); + + InternalMap restored = readMap(fieldMeta, physicalMap); + + assertThat(restored.size()).isEqualTo(1); + assertThat(restored.keyArray().getString(0)) + .isEqualTo(BinaryString.fromString("overflowed")); + assertThat(restored.valueArray().getLong(0)).isEqualTo(30L); + } + + @Test + void testAssembledMapVectorExposesKeyValueChildren() { + MapSharedShreddingFieldMeta fieldMeta = + new MapSharedShreddingFieldMeta( + nameToId("a", 0, "b", 1), + Collections.emptyMap(), + new TreeSet(), + 3, + 2); + HeapRowVector physicalMap = + rowVector( + fieldMapping(0, 1, -1), longVector(10L), longVector(null), longVector(20L)); + + MapColumnVector mapVector = assembleMapVector(fieldMeta, physicalMap); + ColumnVector[] children = mapVector.getChildren(); + + assertThat(children).hasSize(2); + assertThat(BinaryString.fromBytes(((BytesColumnVector) children[0]).getBytes(0).getBytes())) + .isEqualTo(BinaryString.fromString("a")); + assertThat(BinaryString.fromBytes(((BytesColumnVector) children[0]).getBytes(1).getBytes())) + .isEqualTo(BinaryString.fromString("b")); + assertThat(((LongColumnVector) children[1]).getLong(0)).isEqualTo(10L); + assertThat(children[1].isNullAt(1)).isTrue(); + assertThat(mapVector.getMap(0).size()).isEqualTo(2); + } + + private static InternalMap readMap( + MapSharedShreddingFieldMeta fieldMeta, HeapRowVector physicalMap) { + return assembleMapVector(fieldMeta, physicalMap).getMap(0); + } + + private static MapColumnVector assembleMapVector( + MapSharedShreddingFieldMeta fieldMeta, HeapRowVector physicalMap) { + Map fieldMetas = new LinkedHashMap<>(); + fieldMetas.put("metrics", fieldMeta); + MapSharedShreddingReadPlan readPlan = + new MapSharedShreddingReadPlan(logicalType(), fieldMetas); + VectorizedColumnBatch physicalBatch = + new VectorizedColumnBatch(new ColumnVector[] {physicalMap}); + physicalBatch.setNumRows(1); + VectorizedColumnBatch logicalBatch = readPlan.batchAssembler().assemble(physicalBatch); + return (MapColumnVector) logicalBatch.columns[0]; + } + + private static RowType logicalType() { + return DataTypes.ROW( + DataTypes.FIELD( + 0, + "metrics", + DataTypes.MAP(DataTypes.STRING().notNull(), DataTypes.BIGINT()))); + } + + private static HeapRowVector rowVector(ColumnVector... children) { + HeapRowVector vector = new HeapRowVector(1, children); + vector.appendRow(); + return vector; + } + + private static HeapArrayVector fieldMapping(int... ids) { + HeapIntVector child = new HeapIntVector(ids.length); + for (int id : ids) { + child.appendInt(id); + } + HeapArrayVector vector = new HeapArrayVector(1, child); + vector.putOffsetLength(0, 0, ids.length); + return vector; + } + + private static HeapLongVector longVector(Long value) { + HeapLongVector vector = new HeapLongVector(1); + if (value == null) { + vector.appendNull(); + } else { + vector.appendLong(value); + } + return vector; + } + + private static HeapMapVector overflowMap(int key, Long value) { + HeapIntVector keys = new HeapIntVector(1); + keys.appendInt(key); + HeapLongVector values = longVector(value); + HeapMapVector vector = new HeapMapVector(1, keys, values); + vector.putOffsetLength(0, 0, 1); + return vector; + } + + private static Map nameToId(Object... pairs) { + Map result = new TreeMap<>(); + for (int i = 0; i < pairs.length; i += 2) { + result.put((String) pairs[i], (Integer) pairs[i + 1]); + } + return result; + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java b/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java index 66dcc75613f6..ff2de0012cdc 100644 --- a/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java +++ b/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java @@ -23,10 +23,12 @@ import org.apache.paimon.compact.CompactManager; import org.apache.paimon.compression.CompressOptions; import org.apache.paimon.data.InternalRow; +import org.apache.paimon.data.shredding.MapSharedShreddingContext; import org.apache.paimon.disk.IOManager; import org.apache.paimon.disk.RowBuffer; import org.apache.paimon.fileindex.FileIndexOptions; import org.apache.paimon.format.FileFormat; +import org.apache.paimon.format.FormatWriterFactory; import org.apache.paimon.fs.FileIO; import org.apache.paimon.io.BundleRecords; import org.apache.paimon.io.CompactIncrement; @@ -98,6 +100,7 @@ public class AppendOnlyWriter implements BatchRecordWriter, MemoryOwner { @Nullable private final IOManager ioManager; private final FileIndexOptions fileIndexOptions; private final MemorySize maxDiskSize; + @Nullable private final MapSharedShreddingContext sharedShreddingContext; @Nullable private CompactDeletionFile compactDeletionFile; private SinkWriter sinkWriter; @@ -132,6 +135,68 @@ public AppendOnlyWriter( boolean dataEvolutionEnabled, @Nullable FileFormat rowSidecarFileFormat, @Nullable BlobFileContext blobContext) { + this( + fileIO, + ioManager, + schemaId, + fileFormat, + vectorFileFormat, + targetFileSize, + blobTargetFileSize, + vectorTargetFileSize, + writeSchema, + writeCols, + maxSequenceNumber, + compactManager, + dataFileRead, + forceCompact, + pathFactory, + increment, + useWriteBuffer, + spillable, + fileCompression, + spillCompression, + statsCollectorFactories, + maxDiskSize, + fileIndexOptions, + asyncFileWrite, + statsDenseStore, + dataEvolutionEnabled, + rowSidecarFileFormat, + blobContext, + null); + } + + public AppendOnlyWriter( + FileIO fileIO, + @Nullable IOManager ioManager, + long schemaId, + FileFormat fileFormat, + @Nullable FileFormat vectorFileFormat, + long targetFileSize, + long blobTargetFileSize, + long vectorTargetFileSize, + RowType writeSchema, + @Nullable List writeCols, + long maxSequenceNumber, + CompactManager compactManager, + IOFunction, RecordReaderIterator> dataFileRead, + boolean forceCompact, + DataFilePathFactory pathFactory, + @Nullable CommitIncrement increment, + boolean useWriteBuffer, + boolean spillable, + String fileCompression, + CompressOptions spillCompression, + StatsCollectorFactories statsCollectorFactories, + MemorySize maxDiskSize, + FileIndexOptions fileIndexOptions, + boolean asyncFileWrite, + boolean statsDenseStore, + boolean dataEvolutionEnabled, + @Nullable FileFormat rowSidecarFileFormat, + @Nullable BlobFileContext blobContext, + @Nullable MapSharedShreddingContext sharedShreddingContext) { this.fileIO = fileIO; this.schemaId = schemaId; this.fileFormat = fileFormat; @@ -162,6 +227,7 @@ public AppendOnlyWriter( this.statsCollectorFactories = statsCollectorFactories; this.maxDiskSize = maxDiskSize; this.fileIndexOptions = fileIndexOptions; + this.sharedShreddingContext = sharedShreddingContext; this.sinkWriter = useWriteBuffer @@ -290,6 +356,10 @@ public void close() throws Exception { } public void toBufferedWriter() throws Exception { + if (sharedShreddingContext != null && !sharedShreddingContext.isEmpty()) { + throw new UnsupportedOperationException( + "MAP shared-shredding does not support force buffer spill yet because it may rewrite data files."); + } if (sinkWriter != null && !sinkWriter.bufferSpillableWriter() && dataFileRead != null) { // fetch the written results List files = sinkWriter.flush(); @@ -313,8 +383,14 @@ public void toBufferedWriter() throws Exception { } private RollingFileWriter createRollingRowWriter() { - if (blobContext != null - || !fieldsInVectorFile(writeSchema, vectorFileFormat != null).isEmpty()) { + boolean hasDedicatedFields = + blobContext != null + || !fieldsInVectorFile(writeSchema, vectorFileFormat != null).isEmpty(); + if (hasDedicatedFields) { + if (sharedShreddingContext != null && !sharedShreddingContext.isEmpty()) { + throw new UnsupportedOperationException( + "MAP shared-shredding does not support blob or vector file writes yet."); + } return new DedicatedFormatRollingFileWriter( fileIO, schemaId, @@ -344,6 +420,7 @@ private RollingFileWriter createRollingRowWriter() { fileCompression, statsCollectorFactories.statsCollectors(writeSchema.getFieldNames()), fileIndexOptions, + writerFactory(), FileSource.APPEND, asyncFileWrite, statsDenseStore, @@ -351,6 +428,12 @@ private RollingFileWriter createRollingRowWriter() { rowSidecarFileFormat); } + private FormatWriterFactory writerFactory() { + return sharedShreddingContext == null + ? fileFormat.createWriterFactory(writeSchema) + : fileFormat.createWriterFactory(writeSchema, sharedShreddingContext); + } + private void trySyncLatestCompaction(boolean blocking) throws ExecutionException, InterruptedException { compactManager diff --git a/paimon-core/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingCoreUtils.java b/paimon-core/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingCoreUtils.java new file mode 100644 index 000000000000..00f4fbd69869 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingCoreUtils.java @@ -0,0 +1,141 @@ +/* + * 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.CoreOptions; +import org.apache.paimon.format.FileFormat; +import org.apache.paimon.format.FileFormatDiscover; +import org.apache.paimon.format.FormatReaderContext; +import org.apache.paimon.format.SupportsFieldMetadata; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.io.DataFilePathFactory; +import org.apache.paimon.types.RowType; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Core utilities for shared-shredding MAP write and restore flows. */ +public class MapSharedShreddingCoreUtils { + + private static final Logger LOG = LoggerFactory.getLogger(MapSharedShreddingCoreUtils.class); + + private MapSharedShreddingCoreUtils() {} + + @Nullable + public static MapSharedShreddingContext createAndRestoreContext( + RowType writeType, + List restoredFiles, + DataFilePathFactory pathFactory, + CoreOptions options, + FileIO fileIO) { + List shreddingFieldNames = + MapSharedShreddingUtils.detectShreddingColumns(writeType, options); + if (shreddingFieldNames.isEmpty()) { + return null; + } + + MapSharedShreddingContext context = + new MapSharedShreddingContext( + MapSharedShreddingUtils.buildColumnToNumColumns( + shreddingFieldNames, options)); + restoreRecentFileStats( + context, + shreddingFieldNames, + restoredFiles, + pathFactory, + fileIO, + FileFormatDiscover.of(options)); + return context; + } + + private static void restoreRecentFileStats( + MapSharedShreddingContext context, + List shreddingFieldNames, + List restoredFiles, + DataFilePathFactory pathFactory, + FileIO fileIO, + FileFormatDiscover fileFormatDiscover) { + Set pendingFieldNames = new HashSet<>(shreddingFieldNames); + if (pendingFieldNames.isEmpty()) { + return; + } + + for (int i = restoredFiles.size() - 1; i >= 0 && !pendingFieldNames.isEmpty(); i--) { + DataFileMeta file = restoredFiles.get(i); + List candidateFields = + candidateFields(file.writeCols(), shreddingFieldNames, pendingFieldNames); + if (candidateFields.isEmpty()) { + continue; + } + + FileFormat fileFormat = fileFormatDiscover.discover(file.fileFormat()); + if (!(fileFormat instanceof SupportsFieldMetadata)) { + continue; + } + try { + Map> fieldMetadata = + ((SupportsFieldMetadata) fileFormat) + .readFieldMetadata( + new FormatReaderContext( + fileIO, pathFactory.toPath(file), file.fileSize())); + for (String fieldName : candidateFields) { + Map metadata = fieldMetadata.get(fieldName); + if (!MapSharedShreddingUtils.hasShreddingMetadata(metadata)) { + continue; + } + + MapSharedShreddingFieldMeta fieldMeta = + MapSharedShreddingUtils.deserializeMetadata(metadata); + context.reportFileStats(fieldName, fieldMeta.maxRowWidth()); + pendingFieldNames.remove(fieldName); + } + } catch (IOException e) { + LOG.warn( + "Failed to restore MAP shared-shredding statistics from file {}. " + + "Skipping this file.", + pathFactory.toPath(file), + e); + } + } + } + + private static List candidateFields( + @Nullable List writeCols, + List shreddingFieldNames, + Set pendingFieldNames) { + List fields = new ArrayList<>(); + for (String fieldName : shreddingFieldNames) { + if (pendingFieldNames.contains(fieldName) + && (writeCols == null || writeCols.contains(fieldName))) { + fields.add(fieldName); + } + } + return fields; + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/io/RowDataRollingFileWriter.java b/paimon-core/src/main/java/org/apache/paimon/io/RowDataRollingFileWriter.java index 0da1badf6229..56cc9975e6ad 100644 --- a/paimon-core/src/main/java/org/apache/paimon/io/RowDataRollingFileWriter.java +++ b/paimon-core/src/main/java/org/apache/paimon/io/RowDataRollingFileWriter.java @@ -21,6 +21,7 @@ import org.apache.paimon.data.InternalRow; import org.apache.paimon.fileindex.FileIndexOptions; import org.apache.paimon.format.FileFormat; +import org.apache.paimon.format.FormatWriterFactory; import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; import org.apache.paimon.manifest.FileSource; @@ -52,6 +53,42 @@ public RowDataRollingFileWriter( boolean statsDenseStore, @Nullable List writeCols, @Nullable FileFormat rowSidecarFormat) { + this( + fileIO, + schemaId, + fileFormat, + targetFileSize, + writeSchema, + pathFactory, + seqNumCounterSupplier, + fileCompression, + statsCollectors, + fileIndexOptions, + null, + fileSource, + asyncFileWrite, + statsDenseStore, + writeCols, + rowSidecarFormat); + } + + public RowDataRollingFileWriter( + FileIO fileIO, + long schemaId, + FileFormat fileFormat, + long targetFileSize, + RowType writeSchema, + DataFilePathFactory pathFactory, + Supplier seqNumCounterSupplier, + String fileCompression, + SimpleColStatsCollector.Factory[] statsCollectors, + FileIndexOptions fileIndexOptions, + @Nullable FormatWriterFactory writerFactory, + FileSource fileSource, + boolean asyncFileWrite, + boolean statsDenseStore, + @Nullable List writeCols, + @Nullable FileFormat rowSidecarFormat) { super( () -> { Path dataPath = pathFactory.newPath(); @@ -61,8 +98,13 @@ public RowDataRollingFileWriter( : new Path(dataPath.getParent(), dataPath.getName() + ".row"); return new RowDataFileWriter( fileIO, - RollingFileWriter.createFileWriterContext( - fileFormat, writeSchema, statsCollectors, fileCompression), + new FileWriterContext( + writerFactory == null + ? fileFormat.createWriterFactory(writeSchema) + : writerFactory, + RollingFileWriter.createStatsProducer( + fileFormat, writeSchema, statsCollectors), + fileCompression), dataPath, writeSchema, schemaId, diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java index 574143192e19..0df63cc6e2db 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java @@ -26,6 +26,9 @@ import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.BlobConsumer; import org.apache.paimon.data.InternalRow; +import org.apache.paimon.data.shredding.MapSharedShreddingContext; +import org.apache.paimon.data.shredding.MapSharedShreddingCoreUtils; +import org.apache.paimon.data.shredding.MapSharedShreddingUtils; import org.apache.paimon.deletionvectors.BucketedDvMaintainer; import org.apache.paimon.deletionvectors.DeletionVector; import org.apache.paimon.fileindex.FileIndexOptions; @@ -33,6 +36,7 @@ import org.apache.paimon.fs.FileIO; import org.apache.paimon.io.BundleRecords; import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.io.DataFilePathFactory; import org.apache.paimon.io.RowDataRollingFileWriter; import org.apache.paimon.manifest.FileSource; import org.apache.paimon.reader.RecordReaderIterator; @@ -127,6 +131,11 @@ protected RecordWriter createWriter( ExecutorService compactExecutor, @Nullable BucketedDvMaintainer dvMaintainer, boolean ignorePreviousFiles) { + DataFilePathFactory dataPathFactory = + pathFactory.createDataFilePathFactory(partition, bucket); + MapSharedShreddingContext sharedShreddingContext = + MapSharedShreddingCoreUtils.createAndRestoreContext( + writeType, restoredFiles, dataPathFactory, options, fileIO); return new AppendOnlyWriter( fileIO, ioManager, @@ -143,7 +152,7 @@ protected RecordWriter createWriter( // it is only for new files, no dv files -> createFilesIterator(partition, bucket, files, null), options.commitForceCompact(), - pathFactory.createDataFilePathFactory(partition, bucket), + dataPathFactory, restoreIncrement, options.useWriteBufferForAppend() || forceBufferSpill, options.writeBufferSpillable() || forceBufferSpill, @@ -156,7 +165,8 @@ protected RecordWriter createWriter( options.statsDenseStore(), options.dataEvolutionEnabled(), rowSidecarFileFormat(), - blobContext); + blobContext, + sharedShreddingContext); } @Override @@ -195,6 +205,7 @@ public List compactRewrite( if (toCompact.isEmpty()) { return Collections.emptyList(); } + checkNoSharedShreddingRewrite("Compaction rewrite"); Exception collectedExceptions = null; RowDataRollingFileWriter rewriter = createRollingFileWriter( @@ -227,6 +238,7 @@ public List compactRewrite( public List clusterRewrite( BinaryRow partition, int bucket, List toCluster) throws Exception { + checkNoSharedShreddingRewrite("Cluster rewrite"); RecordReaderIterator reader = createFilesIterator(partition, bucket, toCluster, null); @@ -263,6 +275,13 @@ public List clusterRewrite( return rewriter.result(); } + private void checkNoSharedShreddingRewrite(String rewriteName) { + if (!MapSharedShreddingUtils.detectShreddingColumns(writeType, options).isEmpty()) { + throw new UnsupportedOperationException( + rewriteName + " is not supported for MAP shared-shredding."); + } + } + private RowDataRollingFileWriter createRollingFileWriter( BinaryRow partition, int bucket, Supplier seqNumCounterSupplier) { return new RowDataRollingFileWriter( diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java index 56b4676dcab5..8d0f7b7e0427 100644 --- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java +++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java @@ -657,7 +657,13 @@ private static void validateMapStorageLayout(TableSchema schema, CoreOptions opt if (!MapSharedShreddingUtils.isShreddingKeyMap(fieldType)) { throw new IllegalArgumentException( String.format( - "Column '%s' is configured with map.storage-layout=shared-shredding but its type is not MAP.", + "Column '%s' is configured with map.storage-layout=shared-shredding but its type is not MAP.", + fieldName)); + } + if (((MapType) fieldType).getKeyType().isNullable()) { + throw new IllegalArgumentException( + String.format( + "Column '%s' is configured with map.storage-layout=shared-shredding but its map key type is nullable.", fieldName)); } options.mapSharedShreddingMaxColumns(fieldName); diff --git a/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java b/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java index 55965234bb59..053682deaccb 100644 --- a/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java @@ -19,21 +19,30 @@ package org.apache.paimon.append; import org.apache.paimon.CoreOptions; +import org.apache.paimon.compact.NoopCompactManager; import org.apache.paimon.compression.CompressOptions; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.BinaryRowWriter; import org.apache.paimon.data.BinaryString; import org.apache.paimon.data.BinaryVector; import org.apache.paimon.data.BlobData; +import org.apache.paimon.data.GenericMap; import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalMap; import org.apache.paimon.data.InternalRow; +import org.apache.paimon.data.shredding.MapSharedShreddingContext; +import org.apache.paimon.data.shredding.MapSharedShreddingCoreUtils; +import org.apache.paimon.data.shredding.MapSharedShreddingFieldMeta; +import org.apache.paimon.data.shredding.MapSharedShreddingUtils; import org.apache.paimon.disk.ChannelWithMeta; import org.apache.paimon.disk.ExternalBuffer; import org.apache.paimon.disk.IOManager; import org.apache.paimon.disk.RowBuffer; import org.apache.paimon.fileindex.FileIndexOptions; import org.apache.paimon.format.FileFormat; +import org.apache.paimon.format.FormatReaderContext; import org.apache.paimon.format.SimpleColStats; +import org.apache.paimon.format.SupportsFieldMetadata; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.io.DataFileMeta; @@ -66,6 +75,8 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import java.io.File; import java.io.IOException; @@ -74,10 +85,13 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.TreeMap; +import java.util.TreeSet; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; import java.util.stream.Collectors; @@ -470,6 +484,232 @@ public void testNoSpillWhenMeetBlobType() throws Exception { writer.close(); } + @ParameterizedTest(name = "{0}") + @ValueSource(strings = {CoreOptions.FILE_FORMAT_PARQUET, CoreOptions.FILE_FORMAT_ORC}) + public void testWriteSharedShreddingWithFileIndex(String fileFormat) throws Exception { + RowType writeType = sharedShreddingTagsWriteType(); + Options rawOptions = sharedShreddingOptions("tags", 3); + rawOptions.setString("file-index.bloom-filter.columns", "id"); + SharedShreddingAppendContext context = + createSharedShreddingAppendContext(fileFormat, writeType, rawOptions); + AppendOnlyWriter writer = + createSharedShreddingAppendWriter( + writeType, context, SCHEMA_ID, -1L, new FileIndexOptions(context.options)); + + writer.write(sharedShreddingRow(1, "a", 10L)); + writer.write(sharedShreddingRow(2, "b", 20L)); + CommitIncrement increment = writer.prepareCommit(true); + writer.close(); + + DataFileMeta file = assertSingleNewFile(increment); + assertThat(file.embeddedIndex() != null || !file.extraFiles().isEmpty()).isTrue(); + assertThat(readSharedShreddingFieldMeta(context, file, "tags")) + .isEqualTo( + sharedShreddingMeta( + nameToId("a", 0, "b", 1), + fieldToColumns(0, columns(0), 1, columns(0)), + overflowFields(), + 3, + 1)); + } + + @ParameterizedTest(name = "{0}") + @ValueSource(strings = {CoreOptions.FILE_FORMAT_PARQUET, CoreOptions.FILE_FORMAT_ORC}) + public void testSharedShreddingMapAllEmptyFirstFile(String fileFormat) throws Exception { + RowType writeType = sharedShreddingTagsWriteType(); + Options rawOptions = sharedShreddingOptions("tags", 3); + SharedShreddingAppendContext context = + createSharedShreddingAppendContext(fileFormat, writeType, rawOptions); + AppendOnlyWriter writer = createSharedShreddingAppendWriter(writeType, context); + + DataFileMeta file = + writeSharedShreddingFile(writer, sharedShreddingRow(1), sharedShreddingRow(2)); + writer.close(); + + assertThat(readSharedShreddingFieldMeta(context, file, "tags")) + .isEqualTo( + sharedShreddingMeta(nameToId(), fieldToColumns(), overflowFields(), 3, 0)); + } + + @ParameterizedTest(name = "{0}") + @ValueSource(strings = {CoreOptions.FILE_FORMAT_PARQUET, CoreOptions.FILE_FORMAT_ORC}) + public void testSharedShreddingMapAllNullThenAllEmptyFiles(String fileFormat) throws Exception { + RowType writeType = sharedShreddingTagsWriteType(); + Options rawOptions = sharedShreddingOptions("tags", 3); + SharedShreddingAppendContext context = + createSharedShreddingAppendContext(fileFormat, writeType, rawOptions); + AppendOnlyWriter writer = createSharedShreddingAppendWriter(writeType, context); + + DataFileMeta nullFile = + writeSharedShreddingFile( + writer, + sharedShreddingLogicalRow(1, (Object) null), + sharedShreddingLogicalRow(2, (Object) null)); + DataFileMeta emptyFile = + writeSharedShreddingFile(writer, sharedShreddingRow(3), sharedShreddingRow(4)); + DataFileMeta valuesFile = + writeSharedShreddingFile( + writer, + sharedShreddingRow(5, "a", null), + sharedShreddingRow(6, "b", null), + sharedShreddingRow(7, "c", 7L, "d", null)); + writer.close(); + + assertThat(readSharedShreddingFieldMeta(context, nullFile, "tags")) + .isEqualTo( + sharedShreddingMeta(nameToId(), fieldToColumns(), overflowFields(), 3, 0)); + assertThat(readSharedShreddingFieldMeta(context, emptyFile, "tags")) + .isEqualTo( + sharedShreddingMeta(nameToId(), fieldToColumns(), overflowFields(), 1, 0)); + assertThat(readSharedShreddingFieldMeta(context, valuesFile, "tags")) + .isEqualTo( + sharedShreddingMeta( + nameToId("a", 0, "b", 1, "c", 2, "d", 3), + fieldToColumns(0, columns(0), 1, columns(0), 2, columns(0)), + overflowFields(3), + 1, + 2)); + } + + @ParameterizedTest(name = "{0}") + @ValueSource(strings = {CoreOptions.FILE_FORMAT_PARQUET, CoreOptions.FILE_FORMAT_ORC}) + public void testSharedShreddingMapDataFileMetaInfo(String fileFormat) throws Exception { + RowType writeType = sharedShreddingTagsWriteType(); + Options rawOptions = sharedShreddingOptions("tags", 3); + rawOptions.setString("metadata.stats-mode", "full"); + SharedShreddingAppendContext context = + createSharedShreddingAppendContext(fileFormat, writeType, rawOptions); + AppendOnlyWriter writer = createSharedShreddingAppendWriter(writeType, context, 5L, 9L); + + DataFileMeta file = + writeSharedShreddingFile( + writer, + sharedShreddingRow(1, "a", 10L, "b", 20L), + sharedShreddingLogicalRow(2, (Object) null), + sharedShreddingRow(3, "a", 40L, "b", 50L, "c", 60L)); + writer.close(); + + assertThat(file.rowCount()).isEqualTo(3L); + assertThat(file.schemaId()).isEqualTo(5L); + assertThat(file.minSequenceNumber()).isEqualTo(10L); + assertThat(file.maxSequenceNumber()).isEqualTo(12L); + assertThat(file.level()).isEqualTo(DataFileMeta.DUMMY_LEVEL); + assertThat(file.minKey()).isEqualTo(EMPTY_ROW); + assertThat(file.maxKey()).isEqualTo(EMPTY_ROW); + assertThat(file.keyStats()).isEqualTo(EMPTY_STATS); + assertThat(file.valueStats().minValues().getInt(0)).isEqualTo(1); + assertThat(file.valueStats().maxValues().getInt(0)).isEqualTo(3); + assertThat(file.valueStats().nullCounts().getLong(0)).isEqualTo(0L); + assertThat(file.valueStats().minValues().isNullAt(1)).isTrue(); + assertThat(file.valueStats().maxValues().isNullAt(1)).isTrue(); + if (CoreOptions.FILE_FORMAT_ORC.equals(fileFormat)) { + assertThat(file.valueStats().nullCounts().getLong(1)).isEqualTo(1L); + } else { + assertThat(file.valueStats().nullCounts().isNullAt(1)).isTrue(); + } + assertThat(file.fileSource()).hasValue(FileSource.APPEND); + assertThat(file.fileFormat()).isEqualTo(fileFormat); + assertThat(readSharedShreddingFieldMeta(context, file, "tags")) + .isEqualTo( + sharedShreddingMeta( + nameToId("a", 0, "b", 1, "c", 2), + fieldToColumns(0, columns(0), 1, columns(1), 2, columns(2)), + overflowFields(), + 3, + 3)); + } + + @ParameterizedTest(name = "{0}") + @ValueSource(strings = {CoreOptions.FILE_FORMAT_PARQUET, CoreOptions.FILE_FORMAT_ORC}) + public void testSharedShreddingMapIgnoresUnreadableRestoredFile(String fileFormat) + throws Exception { + RowType writeType = sharedShreddingTagsWriteType(); + Options rawOptions = sharedShreddingOptions("tags", 3); + SharedShreddingAppendContext context = + createSharedShreddingAppendContext(fileFormat, writeType, rawOptions); + AppendOnlyWriter writer = createSharedShreddingAppendWriter(writeType, context); + + DataFileMeta restoredFile = + writeSharedShreddingFile(writer, sharedShreddingRow(1, "a", 10L)); + writer.close(); + context.fileIO.delete(context.pathFactory.toPath(restoredFile), false); + + MapSharedShreddingContext restoredContext = + MapSharedShreddingCoreUtils.createAndRestoreContext( + writeType, + Collections.singletonList(restoredFile), + context.pathFactory, + context.options, + context.fileIO); + assertThat(restoredContext).isNotNull(); + assertThat(restoredContext.computeNextK()).containsEntry("tags", 3); + + SharedShreddingAppendContext newContext = + new SharedShreddingAppendContext( + context.options, + context.pathFactory, + context.fileIO, + context.format, + restoredContext); + AppendOnlyWriter newWriter = createSharedShreddingAppendWriter(writeType, newContext); + DataFileMeta newFile = writeSharedShreddingFile(newWriter, sharedShreddingRow(2, "b", 20L)); + newWriter.close(); + + assertThat(readSharedShreddingFieldMeta(newContext, newFile, "tags").numColumns()) + .isEqualTo(3); + } + + @ParameterizedTest(name = "{0}") + @ValueSource(strings = {CoreOptions.FILE_FORMAT_PARQUET, CoreOptions.FILE_FORMAT_ORC}) + public void testSharedShreddingMapDoesNotSupportBlob(String fileFormat) throws Exception { + RowType writeType = + DataTypes.ROW( + DataTypes.FIELD(0, "id", DataTypes.INT()), + DataTypes.FIELD( + 1, + "tags", + DataTypes.MAP(DataTypes.STRING().notNull(), DataTypes.BIGINT())), + DataTypes.FIELD(2, "payload", new BlobType())); + Options rawOptions = sharedShreddingOptions("tags", 3); + SharedShreddingAppendContext context = + createSharedShreddingAppendContext(fileFormat, writeType, rawOptions); + BlobFileContext blobContext = + BlobFileContext.create(writeType, new CoreOptions(rawOptions)); + assertThat(blobContext).isNotNull(); + + AppendOnlyWriter writer = + createSharedShreddingAppendWriter( + writeType, context, SCHEMA_ID, -1L, new FileIndexOptions(), blobContext); + + Assertions.assertThatThrownBy( + () -> + writer.write( + sharedShreddingLogicalRow( + 1, + map("a", 10L), + new BlobData(new byte[] {1, 2, 3})))) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining( + "MAP shared-shredding does not support blob or vector file writes yet."); + writer.close(); + } + + @ParameterizedTest(name = "{0}") + @ValueSource(strings = {CoreOptions.FILE_FORMAT_PARQUET, CoreOptions.FILE_FORMAT_ORC}) + public void testSharedShreddingMapDoesNotSupportForceBufferSpill(String fileFormat) + throws Exception { + RowType writeType = sharedShreddingTagsWriteType(); + Options rawOptions = sharedShreddingOptions("tags", 3); + SharedShreddingAppendContext context = + createSharedShreddingAppendContext(fileFormat, writeType, rawOptions); + AppendOnlyWriter writer = createSharedShreddingAppendWriter(writeType, context); + + Assertions.assertThatThrownBy(writer::toBufferedWriter) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("MAP shared-shredding does not support force buffer spill"); + writer.close(); + } + @Test public void testNoBuffer() throws Exception { AppendOnlyWriter writer = createEmptyWriter(Long.MAX_VALUE); @@ -647,6 +887,105 @@ private InternalRow row(int id, String name, String dt) { return GenericRow.of(id, BinaryString.fromString(name), BinaryString.fromString(dt)); } + private RowType sharedShreddingTagsWriteType() { + return DataTypes.ROW( + DataTypes.FIELD(0, "id", DataTypes.INT()), + DataTypes.FIELD( + 1, + "tags", + DataTypes.MAP(DataTypes.STRING().notNull(), DataTypes.BIGINT()))); + } + + private Options sharedShreddingOptions(Object... fieldToMaxColumns) { + Options options = new Options(); + for (int i = 0; i < fieldToMaxColumns.length; i += 2) { + String fieldName = (String) fieldToMaxColumns[i]; + options.setString("fields." + fieldName + ".map.storage-layout", "shared-shredding"); + options.setString( + "fields." + fieldName + ".map.shared-shredding.max-columns", + String.valueOf(fieldToMaxColumns[i + 1])); + } + options.setString("metadata.stats-mode", "none"); + return options; + } + + private InternalRow sharedShreddingRow(int id, Object... keyValues) { + return sharedShreddingLogicalRow(id, map(keyValues)); + } + + private InternalRow sharedShreddingLogicalRow(int id, Object... fields) { + GenericRow row = new GenericRow(fields.length + 1); + row.setField(0, id); + for (int i = 0; i < fields.length; i++) { + row.setField(i + 1, fields[i]); + } + return row; + } + + private InternalMap map(Object... keyValues) { + Map map = new LinkedHashMap<>(); + for (int i = 0; i < keyValues.length; i += 2) { + map.put( + BinaryString.fromString((String) keyValues[i]), + internalValue(keyValues[i + 1])); + } + return new GenericMap(map); + } + + private Object internalValue(Object value) { + if (value instanceof String) { + return BinaryString.fromString((String) value); + } + return value; + } + + private MapSharedShreddingFieldMeta sharedShreddingMeta( + Map nameToId, + Map> fieldToColumns, + Set overflowFields, + int numColumns, + int maxRowWidth) { + return new MapSharedShreddingFieldMeta( + new TreeMap<>(nameToId), + new TreeMap<>(fieldToColumns), + new TreeSet<>(overflowFields), + numColumns, + maxRowWidth); + } + + private Map nameToId(Object... pairs) { + Map result = new TreeMap<>(); + for (int i = 0; i < pairs.length; i += 2) { + result.put((String) pairs[i], (Integer) pairs[i + 1]); + } + return result; + } + + @SuppressWarnings("unchecked") + private Map> fieldToColumns(Object... pairs) { + Map> result = new TreeMap<>(); + for (int i = 0; i < pairs.length; i += 2) { + result.put((Integer) pairs[i], (List) pairs[i + 1]); + } + return result; + } + + private List columns(int... columns) { + List result = new ArrayList<>(); + for (int column : columns) { + result.add(column); + } + return result; + } + + private Set overflowFields(int... fieldIds) { + Set result = new TreeSet<>(); + for (int fieldId : fieldIds) { + result.add(fieldId); + } + return result; + } + private InternalRow rowWithVectors(int id, String name, float[]... vectors) { GenericRow row = new GenericRow(vectors.length + 2); row.setField(0, id); @@ -658,9 +997,13 @@ private InternalRow rowWithVectors(int id, String name, float[]... vectors) { } private DataFilePathFactory createPathFactory() { + return createPathFactory(CoreOptions.FILE_FORMAT_AVRO); + } + + private DataFilePathFactory createPathFactory(String fileFormat) { return new DataFilePathFactory( new Path(tempDir + "/dt=" + PART + "/bucket-0"), - CoreOptions.FILE_FORMAT_AVRO, + fileFormat, CoreOptions.DATA_FILE_PREFIX.defaultValue(), CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue(), CoreOptions.FILE_SUFFIX_INCLUDE_COMPRESSION.defaultValue(), @@ -668,6 +1011,136 @@ private DataFilePathFactory createPathFactory() { null); } + private SharedShreddingAppendContext createSharedShreddingAppendContext( + String fileFormat, RowType writeType, Options rawOptions) { + CoreOptions options = new CoreOptions(rawOptions); + DataFilePathFactory pathFactory = createPathFactory(fileFormat); + LocalFileIO fileIO = LocalFileIO.create(); + MapSharedShreddingContext sharedShreddingContext = + MapSharedShreddingCoreUtils.createAndRestoreContext( + writeType, Collections.emptyList(), pathFactory, options, fileIO); + assertThat(sharedShreddingContext).isNotNull(); + return new SharedShreddingAppendContext( + options, + pathFactory, + fileIO, + FileFormat.fromIdentifier(fileFormat, rawOptions), + sharedShreddingContext); + } + + private AppendOnlyWriter createSharedShreddingAppendWriter( + RowType writeType, SharedShreddingAppendContext context) { + return createSharedShreddingAppendWriter(writeType, context, SCHEMA_ID, -1L); + } + + private AppendOnlyWriter createSharedShreddingAppendWriter( + RowType writeType, + SharedShreddingAppendContext context, + long schemaId, + long maxSequenceNumber) { + return createSharedShreddingAppendWriter( + writeType, context, schemaId, maxSequenceNumber, new FileIndexOptions()); + } + + private AppendOnlyWriter createSharedShreddingAppendWriter( + RowType writeType, + SharedShreddingAppendContext context, + long schemaId, + long maxSequenceNumber, + FileIndexOptions fileIndexOptions) { + return createSharedShreddingAppendWriter( + writeType, context, schemaId, maxSequenceNumber, fileIndexOptions, null); + } + + private AppendOnlyWriter createSharedShreddingAppendWriter( + RowType writeType, + SharedShreddingAppendContext context, + long schemaId, + long maxSequenceNumber, + FileIndexOptions fileIndexOptions, + BlobFileContext blobContext) { + return new AppendOnlyWriter( + context.fileIO, + null, + schemaId, + context.format, + null, + 1024 * 1024L, + 1024 * 1024L, + 1024 * 1024L, + writeType, + null, + maxSequenceNumber, + new NoopCompactManager(), + null, + false, + context.pathFactory, + null, + false, + false, + CoreOptions.FILE_COMPRESSION.defaultValue(), + CompressOptions.defaultOptions(), + new StatsCollectorFactories(context.options), + MemorySize.MAX_VALUE, + fileIndexOptions, + true, + false, + context.options.dataEvolutionEnabled(), + null, + blobContext, + context.sharedShreddingContext); + } + + private DataFileMeta writeSharedShreddingFile(AppendOnlyWriter writer, InternalRow... rows) + throws Exception { + for (InternalRow row : rows) { + writer.write(row); + } + return assertSingleNewFile(writer.prepareCommit(true)); + } + + private DataFileMeta assertSingleNewFile(CommitIncrement increment) { + assertThat(increment.newFilesIncrement().newFiles()).hasSize(1); + assertThat(increment.compactIncrement().isEmpty()).isTrue(); + return increment.newFilesIncrement().newFiles().get(0); + } + + private MapSharedShreddingFieldMeta readSharedShreddingFieldMeta( + SharedShreddingAppendContext context, DataFileMeta file, String fieldName) + throws IOException { + assertThat(context.format).isInstanceOf(SupportsFieldMetadata.class); + Map> fieldMetadata = + ((SupportsFieldMetadata) context.format) + .readFieldMetadata( + new FormatReaderContext( + context.fileIO, + context.pathFactory.toPath(file), + file.fileSize())); + return MapSharedShreddingUtils.deserializeMetadata(fieldMetadata.get(fieldName)); + } + + private static class SharedShreddingAppendContext { + + private final CoreOptions options; + private final DataFilePathFactory pathFactory; + private final LocalFileIO fileIO; + private final FileFormat format; + private final MapSharedShreddingContext sharedShreddingContext; + + private SharedShreddingAppendContext( + CoreOptions options, + DataFilePathFactory pathFactory, + LocalFileIO fileIO, + FileFormat format, + MapSharedShreddingContext sharedShreddingContext) { + this.options = options; + this.pathFactory = pathFactory; + this.fileIO = fileIO; + this.format = format; + this.sharedShreddingContext = sharedShreddingContext; + } + } + private AppendOnlyWriter createEmptyWriter(long targetFileSize) { return createWriter(targetFileSize, false, false, false, Collections.emptyList()).getLeft(); } @@ -828,7 +1301,8 @@ private Pair> createWriterBase( false, options.dataEvolutionEnabled(), null, - BlobFileContext.create(writeSchema, options)); + BlobFileContext.create(writeSchema, options), + null); writer.setMemoryPool( new HeapMemorySegmentPool(options.writeBufferSize(), options.pageSize())); return Pair.of(writer, compactManager.allFiles()); diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java index c1d673b168dd..ce23657eb100 100644 --- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java @@ -217,7 +217,9 @@ public void testMapStorageLayout() { Arrays.asList( new DataField(0, "id", DataTypes.INT()), new DataField( - 1, "metrics", DataTypes.MAP(DataTypes.STRING(), DataTypes.INT())), + 1, + "metrics", + DataTypes.MAP(DataTypes.STRING().notNull(), DataTypes.INT())), new DataField( 2, "codes", DataTypes.MAP(DataTypes.INT(), DataTypes.STRING()))); @@ -295,10 +297,29 @@ public void testMapStorageLayout() { options, ""))) .hasMessageContaining( - "Column 'codes' is configured with map.storage-layout=shared-shredding but its type is not MAP."); + "Column 'codes' is configured with map.storage-layout=shared-shredding but its type is not MAP."); options.remove("fields.codes.map.storage-layout"); options.put("fields.metrics.map.storage-layout", "shared-shredding"); + List nullableKeyFields = + Arrays.asList( + new DataField(0, "id", DataTypes.INT()), + new DataField( + 1, "metrics", DataTypes.MAP(DataTypes.STRING(), DataTypes.INT()))); + assertThatThrownBy( + () -> + validateTableSchema( + new TableSchema( + 1, + nullableKeyFields, + 10, + emptyList(), + emptyList(), + options, + ""))) + .hasMessageContaining( + "Column 'metrics' is configured with map.storage-layout=shared-shredding but its map key type is nullable."); + options.put("fields.metrics.map.shared-shredding.max-columns", "0"); assertThatThrownBy( () -> @@ -1261,7 +1282,8 @@ private void assertMapSharedShreddingValidationFailed( private List mapValueFields(DataType valueType) { return Arrays.asList( new DataField(0, "id", DataTypes.INT()), - new DataField(1, "metrics", DataTypes.MAP(DataTypes.STRING(), valueType))); + new DataField( + 1, "metrics", DataTypes.MAP(DataTypes.STRING().notNull(), valueType))); } private List nestedMapValueFields(DataType valueType) { @@ -1271,14 +1293,15 @@ private List nestedMapValueFields(DataType valueType) { 1, "metrics", DataTypes.MAP( - DataTypes.STRING(), + DataTypes.STRING().notNull(), DataTypes.ROW(DataTypes.FIELD(0, "payload", valueType))))); } private List topLevelPayloadFields(DataType payloadType) { return Arrays.asList( new DataField(0, "id", DataTypes.INT()), - new DataField(1, "metrics", DataTypes.MAP(DataTypes.STRING(), DataTypes.INT())), + new DataField( + 1, "metrics", DataTypes.MAP(DataTypes.STRING().notNull(), DataTypes.INT())), new DataField(2, "payload", payloadType)); } diff --git a/paimon-core/src/test/java/org/apache/paimon/table/MapSharedShreddingTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/MapSharedShreddingTableTest.java new file mode 100644 index 000000000000..1d8982fcdf2f --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/MapSharedShreddingTableTest.java @@ -0,0 +1,1218 @@ +/* + * 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.table; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.Decimal; +import org.apache.paimon.data.GenericArray; +import org.apache.paimon.data.GenericMap; +import org.apache.paimon.data.GenericRow; +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.shredding.MapSharedShreddingFieldMeta; +import org.apache.paimon.data.shredding.MapSharedShreddingUtils; +import org.apache.paimon.format.FileFormat; +import org.apache.paimon.format.FileFormatDiscover; +import org.apache.paimon.format.FormatReaderContext; +import org.apache.paimon.format.SupportsFieldMetadata; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.io.DataFilePathFactory; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.schema.SchemaChange; +import org.apache.paimon.table.sink.BatchTableCommit; +import org.apache.paimon.table.sink.BatchTableWrite; +import org.apache.paimon.table.sink.BatchWriteBuilder; +import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.table.sink.CommitMessageImpl; +import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.table.source.ReadBuilder; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.Range; + +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.math.BigDecimal; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.Comparator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Table-level tests for MAP shared-shredding. */ +public class MapSharedShreddingTableTest extends TableTestBase { + + @ParameterizedTest + @ValueSource(strings = {"orc", "parquet"}) + public void testAppendOnlyTableReadWrite(String format) throws Exception { + Table table = createTable(format, "metrics"); + + write( + table, + GenericRow.of(1, mapOf("a", 11L, "b", 12L, "c", 13L)), + GenericRow.of(2, mapOf()), + GenericRow.of(3, null), + GenericRow.of(4, mapOf("a", null, "b", 42L, "c", null))); + + Map> actual = new LinkedHashMap<>(); + for (InternalRow row : read(table)) { + actual.put(row.getInt(0), row.isNullAt(1) ? null : toJavaMap(row.getMap(1))); + } + + assertThat(actual) + .containsEntry(1, javaMapOf("a", 11L, "b", 12L, "c", 13L)) + .containsEntry(2, javaMapOf()) + .containsEntry(3, null) + .containsEntry(4, javaMapOf("a", null, "b", 42L, "c", null)); + } + + @ParameterizedTest + @ValueSource(strings = {"orc", "parquet"}) + public void testAppendOnlyTableReadWriteWithTwoMapFields(String format) throws Exception { + Table table = createTable(format, "metrics", "labels"); + + write( + table, + GenericRow.of( + 1, + mapOf("a", 11L, "b", 12L, "c", 13L), + mapOf("x", 21L, "y", 22L, "z", 23L)), + GenericRow.of(2, mapOf("a", 31L), mapOf()), + GenericRow.of(3, null, mapOf("x", 41L))); + + Map>> actual = new LinkedHashMap<>(); + for (InternalRow row : read(table)) { + actual.put( + row.getInt(0), + Arrays.asList( + row.isNullAt(1) ? null : toJavaMap(row.getMap(1)), + row.isNullAt(2) ? null : toJavaMap(row.getMap(2)))); + } + + assertThat(actual) + .containsEntry( + 1, + Arrays.asList( + javaMapOf("a", 11L, "b", 12L, "c", 13L), + javaMapOf("x", 21L, "y", 22L, "z", 23L))) + .containsEntry(2, Arrays.asList(javaMapOf("a", 31L), javaMapOf())) + .containsEntry(3, Arrays.asList(null, javaMapOf("x", 41L))); + + Map> projected = new LinkedHashMap<>(); + for (InternalRow row : read(table, new int[] {0, 2})) { + projected.put(row.getInt(0), row.isNullAt(1) ? null : toJavaMap(row.getMap(1))); + } + + assertThat(projected) + .containsEntry(1, javaMapOf("x", 21L, "y", 22L, "z", 23L)) + .containsEntry(2, javaMapOf()) + .containsEntry(3, javaMapOf("x", 41L)); + } + + @ParameterizedTest + @ValueSource(strings = {"orc", "parquet"}) + public void testAppendOnlyTableReadWriteWithComplexValue(String format) throws Exception { + Table table = createComplexValueTable(format); + + write( + table, + GenericRow.of( + 1, + complexMapOf( + "a", + complexValue(11L, stringArray("x", "y"), longMapOf("p", 101L)), + "b", + complexValue(12L, stringArray("z"), longMapOf("q", 102L)), + "c", + complexValue(13L, null, longMapOf("r", null)))), + GenericRow.of( + 2, + complexMapOf( + "single", + complexValue( + null, + stringArray("single-tag"), + longMapOf("k1", 201L, "k2", 202L))))); + + Map> actual = new LinkedHashMap<>(); + for (InternalRow row : read(table)) { + actual.put(row.getInt(0), toJavaComplexMap(row.getMap(1))); + } + + assertThat(actual) + .containsEntry( + 1, + javaComplexMapOf( + "a", + new ComplexValue( + 11L, Arrays.asList("x", "y"), javaMapOf("p", 101L)), + "b", + new ComplexValue( + 12L, Collections.singletonList("z"), javaMapOf("q", 102L)), + "c", + new ComplexValue(13L, null, javaMapOf("r", null)))) + .containsEntry( + 2, + javaComplexMapOf( + "single", + new ComplexValue( + null, + Collections.singletonList("single-tag"), + javaMapOf("k1", 201L, "k2", 202L)))); + } + + @ParameterizedTest + @ValueSource(strings = {"orc", "parquet"}) + public void testAppendOnlyTableReadWriteWithAllSupportedComplexValueTypes(String format) + throws Exception { + Table table = createAllSupportedValueTypesTable(format); + + WideValue fixedValue = + new WideValue( + true, + (byte) 1, + (short) 2, + 3, + 4L, + 5.5f, + 6.25d, + "str", + "varchar", + "char", + new byte[] {1, 2, 3}, + new byte[] {4, 5}, + Decimal.fromBigDecimal(new BigDecimal("12345678.90"), 10, 2), + Decimal.fromBigDecimal(new BigDecimal("123456789012345678.12345"), 23, 5), + 19500, + 12_345, + Timestamp.fromEpochMillis(1_700_000_000_123L), + Timestamp.fromEpochMillis(1_700_000_000_123L, 456_789), + Timestamp.fromEpochMillis(1_700_000_000_123L), + Timestamp.fromEpochMillis(1_700_000_000_123L, 456_000), + Arrays.asList(7, null, 8), + javaMapOf("m1", 10L, "m2", null), + new NestedValue("nested", 99)); + WideValue sparseValue = + new WideValue( + null, null, null, null, null, null, null, null, null, null, null, null, + null, null, null, null, null, null, null, null, null, null, null); + WideValue overflowValue = + new WideValue( + false, + (byte) 9, + (short) 10, + 11, + 12L, + 13.5f, + 14.25d, + "overflow", + "overflow-varchar", + "over", + new byte[] {9, 8, 7}, + new byte[] {6, 5, 4, 3}, + Decimal.fromBigDecimal(new BigDecimal("-1.23"), 10, 2), + Decimal.fromBigDecimal(new BigDecimal("-123456789012345678.12345"), 23, 5), + 19600, + 54_321, + Timestamp.fromEpochMillis(1_800_000_000_321L), + Timestamp.fromEpochMillis(1_800_000_000_321L, 987_654), + Timestamp.fromEpochMillis(1_800_000_000_321L), + Timestamp.fromEpochMillis(1_800_000_000_321L, 987_000), + Collections.singletonList(42), + javaMapOf("om", 100L), + new NestedValue("overflow-nested", -7)); + + write( + table, + GenericRow.of( + 1, + wideMapOf( + "fixed-a", + toWideRow(fixedValue), + "fixed-b", + toWideRow(sparseValue), + "overflow", + toWideRow(overflowValue))), + GenericRow.of(2, null), + GenericRow.of(3, wideMapOf())); + + Map> actual = new LinkedHashMap<>(); + for (InternalRow row : read(table)) { + actual.put(row.getInt(0), row.isNullAt(1) ? null : toJavaWideMap(row.getMap(1))); + } + + assertThat(actual) + .containsEntry( + 1, + javaWideMapOf( + "fixed-a", + fixedValue, + "fixed-b", + sparseValue, + "overflow", + overflowValue)) + .containsEntry(2, null) + .containsEntry(3, Collections.emptyMap()); + } + + @ParameterizedTest + @ValueSource(strings = {"orc", "parquet"}) + public void testRestoreAdaptiveColumnCountFromFileMetadata(String format) throws Exception { + Table table = createTableWithBucket(format, 8, "1", "metrics"); + + write(table, GenericRow.of(1, mapOf("a", 11L))); + write(table, GenericRow.of(2, mapOf("b", 22L))); + + FileStoreTable fileStoreTable = (FileStoreTable) table; + List files = currentDataFiles(fileStoreTable); + files.sort(Comparator.comparingLong(file -> file.dataFile.minSequenceNumber())); + assertThat(files).hasSize(2); + + MapSharedShreddingFieldMeta firstFileMeta = + readSharedShreddingFieldMeta(fileStoreTable, files.get(0), "metrics"); + assertThat(firstFileMeta.numColumns()).isEqualTo(8); + assertThat(firstFileMeta.maxRowWidth()).isEqualTo(1); + + MapSharedShreddingFieldMeta secondFileMeta = + readSharedShreddingFieldMeta(fileStoreTable, files.get(1), "metrics"); + assertThat(secondFileMeta.numColumns()).isEqualTo(1); + assertThat(secondFileMeta.maxRowWidth()).isEqualTo(1); + + Map> actual = new LinkedHashMap<>(); + for (InternalRow row : read(table)) { + actual.put(row.getInt(0), row.isNullAt(1) ? null : toJavaMap(row.getMap(1))); + } + + assertThat(actual) + .containsEntry(1, javaMapOf("a", 11L)) + .containsEntry(2, javaMapOf("b", 22L)); + } + + @ParameterizedTest + @ValueSource(strings = {"orc", "parquet"}) + public void testSwitchMapLayoutAndUseMaxColumnsWithoutMetadata(String format) throws Exception { + Table table = + createTableWithBucket( + format, + 4, + "1", + Arrays.asList("metrics", "labels"), + Arrays.asList("labels")); + + write(table, GenericRow.of(1, mapOf("a", 11L, "b", 12L), mapOf("x", 21L))); + + catalog.alterTable( + identifier(format), + Arrays.asList( + SchemaChange.setOption( + "fields.metrics.map.storage-layout", "shared-shredding"), + SchemaChange.setOption( + "fields.metrics.map.shared-shredding.max-columns", "3"), + SchemaChange.setOption("fields.labels.map.storage-layout", "default")), + false); + table = catalog.getTable(identifier(format)); + + write(table, GenericRow.of(2, mapOf("c", 31L), mapOf("y", 41L, "z", 42L))); + + FileStoreTable fileStoreTable = (FileStoreTable) table; + List files = currentDataFiles(fileStoreTable); + files.sort(Comparator.comparingLong(file -> file.dataFile.minSequenceNumber())); + assertThat(files).hasSize(2); + + MapSharedShreddingFieldMeta metricsMeta = + readSharedShreddingFieldMeta(fileStoreTable, files.get(1), "metrics"); + assertThat(metricsMeta.numColumns()).isEqualTo(3); + assertThat(metricsMeta.maxRowWidth()).isEqualTo(1); + + Map>> actual = new LinkedHashMap<>(); + for (InternalRow row : read(table)) { + actual.put( + row.getInt(0), + Arrays.asList( + row.isNullAt(1) ? null : toJavaMap(row.getMap(1)), + row.isNullAt(2) ? null : toJavaMap(row.getMap(2)))); + } + + assertThat(actual) + .containsEntry(1, Arrays.asList(javaMapOf("a", 11L, "b", 12L), javaMapOf("x", 21L))) + .containsEntry( + 2, Arrays.asList(javaMapOf("c", 31L), javaMapOf("y", 41L, "z", 42L))); + } + + @ParameterizedTest + @ValueSource(strings = {"orc", "parquet"}) + public void testReadSharedShreddingMapAfterRenameColumn(String format) throws Exception { + Table table = createTable(format, "metrics"); + + write( + table, + GenericRow.of(1, mapOf("a", 11L, "b", 12L)), + GenericRow.of(2, mapOf("c", 21L))); + + catalog.alterTable( + identifier(format), + Arrays.asList( + SchemaChange.renameColumn("metrics", "renamed_metrics"), + SchemaChange.removeOption("fields.metrics.map.storage-layout"), + SchemaChange.removeOption( + "fields.metrics.map.shared-shredding.max-columns"), + SchemaChange.setOption( + "fields.renamed_metrics.map.storage-layout", "shared-shredding"), + SchemaChange.setOption( + "fields.renamed_metrics.map.shared-shredding.max-columns", "2")), + false); + table = catalog.getTable(identifier(format)); + + assertThat(table.rowType().getFieldNames()).containsExactly("id", "renamed_metrics"); + + Map> actual = new LinkedHashMap<>(); + for (InternalRow row : read(table)) { + actual.put(row.getInt(0), row.isNullAt(1) ? null : toJavaMap(row.getMap(1))); + } + + assertThat(actual) + .containsEntry(1, javaMapOf("a", 11L, "b", 12L)) + .containsEntry(2, javaMapOf("c", 21L)); + } + + @ParameterizedTest + @ValueSource(strings = {"orc", "parquet"}) + public void testDataEvolutionMergeWithOverwrittenSharedShreddingMaps(String format) + throws Exception { + Table table = createDataEvolutionTable(format, "metrics", "labels"); + RowType rowType = table.rowType(); + + writeWithWriteType( + table, + rowType.project(Arrays.asList("id", "metrics")), + GenericRow.of(1, mapOf("old-a", 11L)), + GenericRow.of(2, mapOf("old-b", 21L))); + writeWithWriteType( + table, + rowType.project(Arrays.asList("id", "labels")), + 0L, + GenericRow.of(101, mapOf("label-a", 101L)), + GenericRow.of(102, mapOf("label-b", 102L))); + writeWithWriteType( + table, + rowType.project(Collections.singletonList("metrics")), + 0L, + GenericRow.of(mapOf("new-a", 111L)), + GenericRow.of(mapOf("new-b", 222L))); + + Map>> actual = new LinkedHashMap<>(); + for (InternalRow row : read(table)) { + actual.put( + row.getInt(0), + Arrays.asList(toJavaMap(row.getMap(1)), toJavaMap(row.getMap(2)))); + } + + assertThat(actual) + .containsEntry( + 101, Arrays.asList(javaMapOf("new-a", 111L), javaMapOf("label-a", 101L))) + .containsEntry( + 102, Arrays.asList(javaMapOf("new-b", 222L), javaMapOf("label-b", 102L))); + + Map>> partialActual = new LinkedHashMap<>(); + ReadBuilder readBuilder = + table.newReadBuilder().withRowRanges(Collections.singletonList(new Range(1L, 1L))); + readBuilder + .newRead() + .createReader(readBuilder.newScan().plan()) + .forEachRemaining( + row -> + partialActual.put( + row.getInt(0), + Arrays.asList( + toJavaMap(row.getMap(1)), + toJavaMap(row.getMap(2))))); + + assertThat(partialActual) + .containsOnlyKeys(102) + .containsEntry( + 102, Arrays.asList(javaMapOf("new-b", 222L), javaMapOf("label-b", 102L))); + } + + private Table createTable(String format, String... sharedShreddingFields) throws Exception { + return createTable(format, 2, sharedShreddingFields); + } + + private Table createTable(String format, int maxColumns, String... sharedShreddingFields) + throws Exception { + return createTableWithBucket(format, maxColumns, "-1", sharedShreddingFields); + } + + private Table createTableWithBucket( + String format, int maxColumns, String bucket, String... sharedShreddingFields) + throws Exception { + return createTableWithBucket( + format, + maxColumns, + bucket, + Arrays.asList(sharedShreddingFields), + Arrays.asList(sharedShreddingFields)); + } + + private Table createTableWithBucket( + String format, + int maxColumns, + String bucket, + List mapFields, + List sharedShreddingFields) + throws Exception { + catalog.createTable( + identifier(format), + schemaWithBucket(format, maxColumns, bucket, mapFields, sharedShreddingFields), + true); + return catalog.getTable(identifier(format)); + } + + private Table createDataEvolutionTable(String format, String... sharedShreddingFields) + throws Exception { + Schema.Builder builder = Schema.newBuilder().column("id", DataTypes.INT()); + for (String field : sharedShreddingFields) { + builder.column(field, DataTypes.MAP(DataTypes.STRING().notNull(), DataTypes.BIGINT())); + } + builder.option("bucket", "-1") + .option("file.format", format) + .option(CoreOptions.WRITE_ONLY.key(), "true") + .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true") + .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true"); + for (String field : sharedShreddingFields) { + builder.option("fields." + field + ".map.storage-layout", "shared-shredding") + .option("fields." + field + ".map.shared-shredding.max-columns", "2"); + } + catalog.createTable(identifier(format), builder.build(), true); + return catalog.getTable(identifier(format)); + } + + private Table createComplexValueTable(String format) throws Exception { + catalog.createTable( + identifier(format), + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column( + "metrics", + DataTypes.MAP( + DataTypes.STRING().notNull(), + DataTypes.ROW( + DataTypes.FIELD(0, "count", DataTypes.BIGINT()), + DataTypes.FIELD( + 1, + "tags", + DataTypes.ARRAY(DataTypes.STRING())), + DataTypes.FIELD( + 2, + "attrs", + DataTypes.MAP( + DataTypes.STRING(), + DataTypes.BIGINT()))))) + .option("bucket", "-1") + .option("file.format", format) + .option(CoreOptions.WRITE_ONLY.key(), "true") + .option("fields.metrics.map.storage-layout", "shared-shredding") + .option("fields.metrics.map.shared-shredding.max-columns", "2") + .build(), + true); + return catalog.getTable(identifier(format)); + } + + private Table createAllSupportedValueTypesTable(String format) throws Exception { + catalog.createTable( + identifier(format), + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column( + "metrics", + DataTypes.MAP( + DataTypes.STRING().notNull(), + DataTypes.ROW( + DataTypes.FIELD(0, "bool", DataTypes.BOOLEAN()), + DataTypes.FIELD(1, "tiny", DataTypes.TINYINT()), + DataTypes.FIELD(2, "small", DataTypes.SMALLINT()), + DataTypes.FIELD(3, "i", DataTypes.INT()), + DataTypes.FIELD(4, "big", DataTypes.BIGINT()), + DataTypes.FIELD(5, "f", DataTypes.FLOAT()), + DataTypes.FIELD(6, "d", DataTypes.DOUBLE()), + DataTypes.FIELD(7, "s", DataTypes.STRING()), + DataTypes.FIELD( + 8, "varchar_value", DataTypes.VARCHAR(32)), + DataTypes.FIELD(9, "char_value", DataTypes.CHAR(6)), + DataTypes.FIELD(10, "bin", DataTypes.BINARY(3)), + DataTypes.FIELD( + 11, "var_bin", DataTypes.VARBINARY(8)), + DataTypes.FIELD( + 12, + "compact_decimal", + DataTypes.DECIMAL(10, 2)), + DataTypes.FIELD( + 13, + "large_decimal", + DataTypes.DECIMAL(23, 5)), + DataTypes.FIELD(14, "date", DataTypes.DATE()), + DataTypes.FIELD(15, "time", DataTypes.TIME()), + DataTypes.FIELD(16, "ts3", DataTypes.TIMESTAMP(3)), + DataTypes.FIELD(17, "ts9", DataTypes.TIMESTAMP(9)), + DataTypes.FIELD( + 18, + "ts_ltz3", + DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE( + 3)), + DataTypes.FIELD( + 19, + "ts_ltz6", + DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE( + 6)), + DataTypes.FIELD( + 20, + "ints", + DataTypes.ARRAY(DataTypes.INT())), + DataTypes.FIELD( + 21, + "attrs", + DataTypes.MAP( + DataTypes.STRING(), + DataTypes.BIGINT())), + DataTypes.FIELD( + 22, + "nested", + DataTypes.ROW( + DataTypes.FIELD( + 0, + "name", + DataTypes.STRING()), + DataTypes.FIELD( + 1, + "score", + DataTypes.INT())))))) + .option("bucket", "-1") + .option("file.format", format) + .option(CoreOptions.WRITE_ONLY.key(), "true") + .option("fields.metrics.map.storage-layout", "shared-shredding") + .option("fields.metrics.map.shared-shredding.max-columns", "2") + .build(), + true); + return catalog.getTable(identifier(format)); + } + + private void writeWithWriteType(Table table, RowType writeType, InternalRow... rows) + throws Exception { + writeWithWriteType(table, writeType, null, rows); + } + + private void writeWithWriteType( + Table table, RowType writeType, Long firstRowId, InternalRow... rows) throws Exception { + BatchWriteBuilder builder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = builder.newWrite().withWriteType(writeType); + BatchTableCommit commit = builder.newCommit()) { + for (InternalRow row : rows) { + write.write(row); + } + List commitMessages = write.prepareCommit(); + if (firstRowId != null) { + setFirstRowId(commitMessages, firstRowId); + } + commit.commit(commitMessages); + } + } + + private void setFirstRowId(List commitMessages, long firstRowId) { + for (CommitMessage message : commitMessages) { + CommitMessageImpl commitMessage = (CommitMessageImpl) message; + List newFiles = + new ArrayList<>(commitMessage.newFilesIncrement().newFiles()); + commitMessage.newFilesIncrement().newFiles().clear(); + for (DataFileMeta newFile : newFiles) { + commitMessage + .newFilesIncrement() + .newFiles() + .add(newFile.assignFirstRowId(firstRowId)); + } + } + } + + private Schema schema(String format, String... sharedShreddingFields) { + return schema(format, 2, sharedShreddingFields); + } + + private Schema schema(String format, int maxColumns, String... sharedShreddingFields) { + return schemaWithBucket(format, maxColumns, "-1", sharedShreddingFields); + } + + private Schema schemaWithBucket( + String format, int maxColumns, String bucket, String... sharedShreddingFields) { + return schemaWithBucket( + format, + maxColumns, + bucket, + Arrays.asList(sharedShreddingFields), + Arrays.asList(sharedShreddingFields)); + } + + private Schema schemaWithBucket( + String format, + int maxColumns, + String bucket, + List mapFields, + List sharedShreddingFields) { + Schema.Builder builder = Schema.newBuilder().column("id", DataTypes.INT()); + for (String field : mapFields) { + builder.column(field, DataTypes.MAP(DataTypes.STRING().notNull(), DataTypes.BIGINT())); + } + builder.option("bucket", bucket) + .option("file.format", format) + .option(CoreOptions.WRITE_ONLY.key(), "true"); + if (!"-1".equals(bucket)) { + builder.option("bucket-key", "id"); + } + for (String field : sharedShreddingFields) { + builder.option("fields." + field + ".map.storage-layout", "shared-shredding") + .option( + "fields." + field + ".map.shared-shredding.max-columns", + String.valueOf(maxColumns)); + } + return builder.build(); + } + + private List currentDataFiles(FileStoreTable table) throws Exception { + List files = new ArrayList<>(); + for (DataSplit split : table.newSnapshotReader().read().dataSplits()) { + for (DataFileMeta dataFile : split.dataFiles()) { + files.add(new DataFileWithSplit(split.partition(), split.bucket(), dataFile)); + } + } + return files; + } + + private MapSharedShreddingFieldMeta readSharedShreddingFieldMeta( + FileStoreTable table, DataFileWithSplit file, String fieldName) throws Exception { + DataFilePathFactory pathFactory = + table.store().pathFactory().createDataFilePathFactory(file.partition, file.bucket); + FileFormat fileFormat = + FileFormatDiscover.of(new CoreOptions(table.options())) + .discover(file.dataFile.fileFormat()); + Map> fieldMetadata = + ((SupportsFieldMetadata) fileFormat) + .readFieldMetadata( + new FormatReaderContext( + table.fileIO(), + pathFactory.toPath(file.dataFile), + file.dataFile.fileSize())); + return MapSharedShreddingUtils.deserializeMetadata(fieldMetadata.get(fieldName)); + } + + private GenericMap mapOf(Object... entries) { + Map map = new LinkedHashMap<>(); + for (int i = 0; i < entries.length; i += 2) { + map.put(BinaryString.fromString((String) entries[i]), (Long) entries[i + 1]); + } + return new GenericMap(map); + } + + private GenericMap complexMapOf(Object... entries) { + Map map = new LinkedHashMap<>(); + for (int i = 0; i < entries.length; i += 2) { + map.put(BinaryString.fromString((String) entries[i]), (GenericRow) entries[i + 1]); + } + return new GenericMap(map); + } + + private GenericMap wideMapOf(Object... entries) { + Map map = new LinkedHashMap<>(); + for (int i = 0; i < entries.length; i += 2) { + map.put(BinaryString.fromString((String) entries[i]), (GenericRow) entries[i + 1]); + } + return new GenericMap(map); + } + + private GenericRow complexValue(Long count, GenericArray tags, GenericMap attrs) { + return GenericRow.of(count, tags, attrs); + } + + private GenericRow toWideRow(WideValue value) { + return GenericRow.of( + value.bool, + value.tiny, + value.small, + value.i, + value.big, + value.f, + value.d, + value.s == null ? null : BinaryString.fromString(value.s), + value.varcharValue == null ? null : BinaryString.fromString(value.varcharValue), + value.charValue == null ? null : BinaryString.fromString(value.charValue), + value.bin, + value.varBin, + value.compactDecimal, + value.largeDecimal, + value.date, + value.time, + value.ts3, + value.ts9, + value.tsLtz3, + value.tsLtz6, + value.ints == null ? null : new GenericArray(value.ints.toArray(new Integer[0])), + value.attrs == null ? null : longMapFromJava(value.attrs), + value.nested == null + ? null + : GenericRow.of( + BinaryString.fromString(value.nested.name), value.nested.score)); + } + + private GenericArray stringArray(String... values) { + BinaryString[] strings = new BinaryString[values.length]; + for (int i = 0; i < values.length; i++) { + strings[i] = BinaryString.fromString(values[i]); + } + return new GenericArray(strings); + } + + private GenericMap longMapOf(Object... entries) { + Map map = new LinkedHashMap<>(); + for (int i = 0; i < entries.length; i += 2) { + map.put(BinaryString.fromString((String) entries[i]), (Long) entries[i + 1]); + } + return new GenericMap(map); + } + + private GenericMap longMapFromJava(Map values) { + Map map = new LinkedHashMap<>(); + for (Map.Entry entry : values.entrySet()) { + map.put(BinaryString.fromString(entry.getKey()), entry.getValue()); + } + return new GenericMap(map); + } + + private Map toJavaMap(InternalMap map) { + Map result = new LinkedHashMap<>(); + for (int i = 0; i < map.size(); i++) { + result.put( + map.keyArray().getString(i).toString(), + map.valueArray().isNullAt(i) ? null : map.valueArray().getLong(i)); + } + return result; + } + + private Map javaMapOf(Object... entries) { + Map map = new LinkedHashMap<>(); + for (int i = 0; i < entries.length; i += 2) { + map.put((String) entries[i], (Long) entries[i + 1]); + } + return map; + } + + private Map toJavaComplexMap(InternalMap map) { + Map result = new LinkedHashMap<>(); + InternalArray keys = map.keyArray(); + InternalArray values = map.valueArray(); + for (int i = 0; i < map.size(); i++) { + result.put( + keys.getString(i).toString(), + values.isNullAt(i) ? null : toJavaComplexValue(values.getRow(i, 3))); + } + return result; + } + + private ComplexValue toJavaComplexValue(InternalRow row) { + return new ComplexValue( + row.isNullAt(0) ? null : row.getLong(0), + row.isNullAt(1) ? null : toJavaStringList(row.getArray(1)), + row.isNullAt(2) ? null : toJavaLongMap(row.getMap(2))); + } + + private List toJavaStringList(InternalArray array) { + List result = new ArrayList<>(); + for (int i = 0; i < array.size(); i++) { + result.add(array.isNullAt(i) ? null : array.getString(i).toString()); + } + return result; + } + + private Map toJavaLongMap(InternalMap map) { + Map result = new LinkedHashMap<>(); + InternalArray keys = map.keyArray(); + InternalArray values = map.valueArray(); + for (int i = 0; i < map.size(); i++) { + result.put(keys.getString(i).toString(), values.isNullAt(i) ? null : values.getLong(i)); + } + return result; + } + + private Map javaComplexMapOf(Object... entries) { + Map map = new LinkedHashMap<>(); + for (int i = 0; i < entries.length; i += 2) { + map.put((String) entries[i], (ComplexValue) entries[i + 1]); + } + return map; + } + + private Map toJavaWideMap(InternalMap map) { + Map result = new LinkedHashMap<>(); + InternalArray keys = map.keyArray(); + InternalArray values = map.valueArray(); + for (int i = 0; i < map.size(); i++) { + result.put( + keys.getString(i).toString(), + values.isNullAt(i) ? null : toJavaWideValue(values.getRow(i, 23))); + } + return result; + } + + private WideValue toJavaWideValue(InternalRow row) { + return new WideValue( + row.isNullAt(0) ? null : row.getBoolean(0), + row.isNullAt(1) ? null : row.getByte(1), + row.isNullAt(2) ? null : row.getShort(2), + row.isNullAt(3) ? null : row.getInt(3), + row.isNullAt(4) ? null : row.getLong(4), + row.isNullAt(5) ? null : row.getFloat(5), + row.isNullAt(6) ? null : row.getDouble(6), + row.isNullAt(7) ? null : row.getString(7).toString(), + row.isNullAt(8) ? null : row.getString(8).toString(), + row.isNullAt(9) ? null : row.getString(9).toString(), + row.isNullAt(10) ? null : row.getBinary(10), + row.isNullAt(11) ? null : row.getBinary(11), + row.isNullAt(12) ? null : row.getDecimal(12, 10, 2), + row.isNullAt(13) ? null : row.getDecimal(13, 23, 5), + row.isNullAt(14) ? null : row.getInt(14), + row.isNullAt(15) ? null : row.getInt(15), + row.isNullAt(16) ? null : row.getTimestamp(16, 3), + row.isNullAt(17) ? null : row.getTimestamp(17, 9), + row.isNullAt(18) ? null : row.getTimestamp(18, 3), + row.isNullAt(19) ? null : row.getTimestamp(19, 6), + row.isNullAt(20) ? null : toJavaIntList(row.getArray(20)), + row.isNullAt(21) ? null : toJavaLongMap(row.getMap(21)), + row.isNullAt(22) ? null : toJavaNestedValue(row.getRow(22, 2))); + } + + private List toJavaIntList(InternalArray array) { + List result = new ArrayList<>(); + for (int i = 0; i < array.size(); i++) { + result.add(array.isNullAt(i) ? null : array.getInt(i)); + } + return result; + } + + private NestedValue toJavaNestedValue(InternalRow row) { + return new NestedValue( + row.isNullAt(0) ? null : row.getString(0).toString(), + row.isNullAt(1) ? null : row.getInt(1)); + } + + private Map javaWideMapOf(Object... entries) { + Map map = new LinkedHashMap<>(); + for (int i = 0; i < entries.length; i += 2) { + map.put((String) entries[i], (WideValue) entries[i + 1]); + } + return map; + } + + @Override + protected Schema schemaDefault() { + return schema("parquet", "metrics"); + } + + private static class DataFileWithSplit { + + private final BinaryRow partition; + private final int bucket; + private final DataFileMeta dataFile; + + private DataFileWithSplit(BinaryRow partition, int bucket, DataFileMeta dataFile) { + this.partition = partition; + this.bucket = bucket; + this.dataFile = dataFile; + } + } + + private static class ComplexValue { + + private final Long count; + private final List tags; + private final Map attrs; + + private ComplexValue(Long count, List tags, Map attrs) { + this.count = count; + this.tags = tags; + this.attrs = attrs; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof ComplexValue)) { + return false; + } + ComplexValue that = (ComplexValue) o; + return java.util.Objects.equals(count, that.count) + && java.util.Objects.equals(tags, that.tags) + && java.util.Objects.equals(attrs, that.attrs); + } + + @Override + public int hashCode() { + return java.util.Objects.hash(count, tags, attrs); + } + + @Override + public String toString() { + return "ComplexValue{" + "count=" + count + ", tags=" + tags + ", attrs=" + attrs + '}'; + } + } + + private static class NestedValue { + + private final String name; + private final Integer score; + + private NestedValue(String name, Integer score) { + this.name = name; + this.score = score; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof NestedValue)) { + return false; + } + NestedValue that = (NestedValue) o; + return java.util.Objects.equals(name, that.name) + && java.util.Objects.equals(score, that.score); + } + + @Override + public int hashCode() { + return java.util.Objects.hash(name, score); + } + + @Override + public String toString() { + return "NestedValue{" + "name='" + name + '\'' + ", score=" + score + '}'; + } + } + + private static class WideValue { + + private final Boolean bool; + private final Byte tiny; + private final Short small; + private final Integer i; + private final Long big; + private final Float f; + private final Double d; + private final String s; + private final String varcharValue; + private final String charValue; + private final byte[] bin; + private final byte[] varBin; + private final Decimal compactDecimal; + private final Decimal largeDecimal; + private final Integer date; + private final Integer time; + private final Timestamp ts3; + private final Timestamp ts9; + private final Timestamp tsLtz3; + private final Timestamp tsLtz6; + private final List ints; + private final Map attrs; + private final NestedValue nested; + + private WideValue( + Boolean bool, + Byte tiny, + Short small, + Integer i, + Long big, + Float f, + Double d, + String s, + String varcharValue, + String charValue, + byte[] bin, + byte[] varBin, + Decimal compactDecimal, + Decimal largeDecimal, + Integer date, + Integer time, + Timestamp ts3, + Timestamp ts9, + Timestamp tsLtz3, + Timestamp tsLtz6, + List ints, + Map attrs, + NestedValue nested) { + this.bool = bool; + this.tiny = tiny; + this.small = small; + this.i = i; + this.big = big; + this.f = f; + this.d = d; + this.s = s; + this.varcharValue = varcharValue; + this.charValue = charValue; + this.bin = bin; + this.varBin = varBin; + this.compactDecimal = compactDecimal; + this.largeDecimal = largeDecimal; + this.date = date; + this.time = time; + this.ts3 = ts3; + this.ts9 = ts9; + this.tsLtz3 = tsLtz3; + this.tsLtz6 = tsLtz6; + this.ints = ints; + this.attrs = attrs; + this.nested = nested; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof WideValue)) { + return false; + } + WideValue wideValue = (WideValue) o; + return java.util.Objects.equals(bool, wideValue.bool) + && java.util.Objects.equals(tiny, wideValue.tiny) + && java.util.Objects.equals(small, wideValue.small) + && java.util.Objects.equals(i, wideValue.i) + && java.util.Objects.equals(big, wideValue.big) + && java.util.Objects.equals(f, wideValue.f) + && java.util.Objects.equals(d, wideValue.d) + && java.util.Objects.equals(s, wideValue.s) + && java.util.Objects.equals(varcharValue, wideValue.varcharValue) + && java.util.Objects.equals(charValue, wideValue.charValue) + && Arrays.equals(bin, wideValue.bin) + && Arrays.equals(varBin, wideValue.varBin) + && java.util.Objects.equals(compactDecimal, wideValue.compactDecimal) + && java.util.Objects.equals(largeDecimal, wideValue.largeDecimal) + && java.util.Objects.equals(date, wideValue.date) + && java.util.Objects.equals(time, wideValue.time) + && java.util.Objects.equals(ts3, wideValue.ts3) + && java.util.Objects.equals(ts9, wideValue.ts9) + && java.util.Objects.equals(tsLtz3, wideValue.tsLtz3) + && java.util.Objects.equals(tsLtz6, wideValue.tsLtz6) + && java.util.Objects.equals(ints, wideValue.ints) + && java.util.Objects.equals(attrs, wideValue.attrs) + && java.util.Objects.equals(nested, wideValue.nested); + } + + @Override + public int hashCode() { + int result = + java.util.Objects.hash( + bool, + tiny, + small, + i, + big, + f, + d, + s, + varcharValue, + charValue, + compactDecimal, + largeDecimal, + date, + time, + ts3, + ts9, + tsLtz3, + tsLtz6, + ints, + attrs, + nested); + result = 31 * result + Arrays.hashCode(bin); + result = 31 * result + Arrays.hashCode(varBin); + return result; + } + + @Override + public String toString() { + return "WideValue{" + + "bool=" + + bool + + ", tiny=" + + tiny + + ", small=" + + small + + ", i=" + + i + + ", big=" + + big + + ", f=" + + f + + ", d=" + + d + + ", s='" + + s + + '\'' + + ", varcharValue='" + + varcharValue + + '\'' + + ", charValue='" + + charValue + + '\'' + + ", bin=" + + Arrays.toString(bin) + + ", varBin=" + + Arrays.toString(varBin) + + ", compactDecimal=" + + compactDecimal + + ", largeDecimal=" + + largeDecimal + + ", date=" + + date + + ", time=" + + time + + ", ts3=" + + ts3 + + ", ts9=" + + ts9 + + ", tsLtz3=" + + tsLtz3 + + ", tsLtz6=" + + tsLtz6 + + ", ints=" + + ints + + ", attrs=" + + attrs + + ", nested=" + + nested + + '}'; + } + } +} 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 77790dd95fc1..b20cc44c8b7e 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 @@ -31,6 +31,7 @@ import org.apache.paimon.format.orc.filter.OrcSimpleStatsExtractor; import org.apache.paimon.format.orc.writer.RowDataVectorizer; import org.apache.paimon.format.orc.writer.Vectorizer; +import org.apache.paimon.format.shredding.ShreddingWritePlanType; import org.apache.paimon.options.MemorySize; import org.apache.paimon.options.Options; import org.apache.paimon.predicate.Predicate; @@ -52,10 +53,12 @@ import java.io.IOException; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Optional; import java.util.Properties; +import java.util.Set; import java.util.stream.Collectors; import static org.apache.paimon.CoreOptions.DELETION_VECTORS_ENABLED; @@ -67,6 +70,9 @@ public class OrcFileFormat extends FileFormat implements SupportsFieldMetadata { public static final String IDENTIFIER = "orc"; + private static final Set SUPPORTED_SHREDDING_WRITE_PLANS = + Collections.singleton(ShreddingWritePlanType.MAP_SHARED_SHREDDING); + private final Properties orcProperties; private final org.apache.hadoop.conf.Configuration readerConf; private final org.apache.hadoop.conf.Configuration writerConf; @@ -77,7 +83,7 @@ public class OrcFileFormat extends FileFormat implements SupportsFieldMetadata { private final boolean legacyTimestampLtzType; public OrcFileFormat(FormatContext formatContext) { - super(IDENTIFIER); + super(IDENTIFIER, formatContext.options()); this.orcProperties = getOrcProperties(formatContext.options(), formatContext); this.readerConf = new org.apache.hadoop.conf.Configuration(false); this.orcProperties.forEach((k, v) -> readerConf.set(k.toString(), v.toString())); @@ -162,7 +168,7 @@ public void validateDataFields(RowType rowType) { * @return The factory of the writer */ @Override - public FormatWriterFactory createWriterFactory(RowType type) { + protected FormatWriterFactory createRawWriterFactory(RowType type) { RowType refinedType = (RowType) refineDataType(type); TypeDescription typeDescription = OrcTypeUtil.convertToOrcSchema(refinedType); Vectorizer vectorizer = @@ -178,6 +184,11 @@ public FormatWriterFactory createWriterFactory(RowType type) { legacyTimestampLtzType); } + @Override + protected Set supportedShreddingWritePlans() { + return SUPPORTED_SHREDDING_WRITE_PLANS; + } + private Properties getOrcProperties(Options options, FormatContext formatContext) { Properties orcProperties = new Properties(); orcProperties.putAll(getIdentifierPrefixOptions(options).toMap()); diff --git a/paimon-format/src/main/java/org/apache/paimon/format/orc/OrcReaderFactory.java b/paimon-format/src/main/java/org/apache/paimon/format/orc/OrcReaderFactory.java index 1809bc654e9d..05ebc5da2206 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/orc/OrcReaderFactory.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/orc/OrcReaderFactory.java @@ -24,11 +24,16 @@ import org.apache.paimon.data.columnar.ColumnarRowIterator; import org.apache.paimon.data.columnar.VectorizedColumnBatch; import org.apache.paimon.data.columnar.VectorizedRowIterator; +import org.apache.paimon.data.shredding.MapSharedShreddingReadPlanFactory; +import org.apache.paimon.data.shredding.ShreddingReadPlan; import org.apache.paimon.format.FormatMetadataUtils; import org.apache.paimon.format.FormatReaderFactory; import org.apache.paimon.format.OrcFormatReaderContext; import org.apache.paimon.format.fs.HadoopReadOnlyFileSystem; import org.apache.paimon.format.orc.filter.OrcFilters; +import org.apache.paimon.format.shredding.ShreddingFormatReader; +import org.apache.paimon.format.shredding.ShreddingReadPlanFactories; +import org.apache.paimon.format.shredding.ShreddingReadPlanFactory; import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; import org.apache.paimon.reader.FileRecordReader; @@ -99,27 +104,52 @@ public OrcReaderFactory( // ------------------------------------------------------------------------ @Override - public OrcVectorizedReader createReader(FormatReaderFactory.Context context) + public FileRecordReader createReader(FormatReaderFactory.Context context) throws IOException { int poolSize = context instanceof OrcFormatReaderContext ? ((OrcFormatReaderContext) context).poolSize() : 1; - Pool poolOfBatches = - createPoolOfBatches(context.filePath(), poolSize, context.fileIO()); - - OrcRecordReader orcReader = - createRecordReader( - hadoopConfig, - schema, - conjunctPredicates, - context.fileIO(), - context.filePath(), - 0, - context.fileSize(), - context.selection(), - deletionVectorsEnabled); - return new OrcVectorizedReader(orcReader, poolOfBatches); + org.apache.orc.Reader fileReader = + createReader( + hadoopConfig, context.fileIO(), context.filePath(), context.selection()); + try { + ShreddingReadPlan readPlan = + ShreddingReadPlanFactories.createReadPlan( + tableType, + readFieldMetadata(fileReader), + null, + shreddingReadPlanFactories(tableType)); + RowType physicalReadType = readPlan.physicalRowType(); + TypeDescription physicalReadSchema = + readPlan.isIdentity() ? schema : convertToOrcSchema(physicalReadType); + Pool poolOfBatches = + createPoolOfBatches( + context.filePath(), + poolSize, + context.fileIO(), + physicalReadSchema, + physicalReadType); + + OrcRecordReader orcReader = + createRecordReader( + hadoopConfig, + fileReader, + physicalReadSchema, + conjunctPredicates, + 0, + context.fileSize(), + context.selection(), + deletionVectorsEnabled); + OrcVectorizedReader orcVectorizedReader = + new OrcVectorizedReader(orcReader, poolOfBatches); + return readPlan.isIdentity() + ? orcVectorizedReader + : new ShreddingFormatReader(orcVectorizedReader, readPlan); + } catch (IOException | RuntimeException e) { + IOUtils.closeQuietly(fileReader); + throw e; + } } /** @@ -132,11 +162,20 @@ public OrcReaderBatch createReaderBatch( VectorizedRowBatch orcBatch, Pool.Recycler recycler, FileIO fileIO) { - List tableFieldNames = tableType.getFieldNames(); - List tableFieldTypes = tableType.getFieldTypes(); + return createReaderBatch(filePath, orcBatch, recycler, fileIO, tableType); + } + + private OrcReaderBatch createReaderBatch( + Path filePath, + VectorizedRowBatch orcBatch, + Pool.Recycler recycler, + FileIO fileIO, + RowType readType) { + List tableFieldNames = readType.getFieldNames(); + List tableFieldTypes = readType.getFieldTypes(); // create and initialize the row batch - ColumnVector[] vectors = new ColumnVector[tableType.getFieldCount()]; + ColumnVector[] vectors = new ColumnVector[readType.getFieldCount()]; for (int i = 0; i < vectors.length; i++) { String name = tableFieldNames.get(i); DataType type = tableFieldTypes.get(i); @@ -153,14 +192,30 @@ public OrcReaderBatch createReaderBatch( // ------------------------------------------------------------------------ + private static List shreddingReadPlanFactories(RowType readType) { + return Collections.singletonList( + new MapSharedShreddingReadPlanFactory(readType)); + } + + // ------------------------------------------------------------------------ + private Pool createPoolOfBatches(Path filePath, int numBatches, FileIO fileIO) { + return createPoolOfBatches(filePath, numBatches, fileIO, schema, tableType); + } + + private Pool createPoolOfBatches( + Path filePath, + int numBatches, + FileIO fileIO, + TypeDescription readSchema, + RowType readType) { final Pool pool = new Pool<>(numBatches); for (int i = 0; i < numBatches; i++) { final VectorizedRowBatch orcBatch = - createBatchWrapper(schema, Math.max(1, batchSize / numBatches)); + createBatchWrapper(readSchema, Math.max(1, batchSize / numBatches)); final OrcReaderBatch batch = - createReaderBatch(filePath, orcBatch, pool.recycler(), fileIO); + createReaderBatch(filePath, orcBatch, pool.recycler(), fileIO, readType); pool.add(batch); } @@ -297,6 +352,32 @@ private static OrcRecordReader createRecordReader( boolean deletionVectorsEnabled) throws IOException { org.apache.orc.Reader orcReader = createReader(conf, fileIO, path, selection); + try { + return createRecordReader( + conf, + orcReader, + schema, + conjunctPredicates, + splitStart, + splitLength, + selection, + deletionVectorsEnabled); + } catch (IOException | RuntimeException e) { + IOUtils.closeQuietly(orcReader); + throw e; + } + } + + private static OrcRecordReader createRecordReader( + org.apache.hadoop.conf.Configuration conf, + org.apache.orc.Reader orcReader, + TypeDescription schema, + List conjunctPredicates, + long splitStart, + long splitLength, + @Nullable RoaringBitmap32 selection, + boolean deletionVectorsEnabled) + throws IOException { try { // get offset and length for the stripes that start in the split Pair offsetAndLength = 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 2380a6cf7acb..63649fbea852 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 @@ -26,8 +26,7 @@ import org.apache.paimon.format.SimpleStatsExtractor; import org.apache.paimon.format.SupportsFieldMetadata; import org.apache.paimon.format.parquet.writer.RowDataParquetBuilder; -import org.apache.paimon.format.shredding.ShreddingWritePlanWriterFactory; -import org.apache.paimon.format.variant.VariantShreddingWritePlanFactory; +import org.apache.paimon.format.shredding.ShreddingWritePlanType; import org.apache.paimon.options.CatalogOptions; import org.apache.paimon.options.MemorySize; import org.apache.paimon.options.Options; @@ -43,23 +42,30 @@ import javax.annotation.Nullable; import java.io.IOException; +import java.util.Collections; +import java.util.EnumSet; import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.Set; import static org.apache.paimon.format.parquet.ParquetFileFormatFactory.IDENTIFIER; /** Parquet {@link FileFormat}. */ public class ParquetFileFormat extends FileFormat implements SupportsFieldMetadata { - private final FormatContext formatContext; + private static final Set SUPPORTED_SHREDDING_WRITE_PLANS = + Collections.unmodifiableSet( + EnumSet.of( + ShreddingWritePlanType.VARIANT, + ShreddingWritePlanType.MAP_SHARED_SHREDDING)); + private final Options options; private final int readBatchSize; public ParquetFileFormat(FormatContext formatContext) { - super(IDENTIFIER); + super(IDENTIFIER, formatContext.options()); - this.formatContext = formatContext; this.options = getParquetConfiguration(formatContext); this.readBatchSize = formatContext.readBatchSize(); } @@ -99,11 +105,13 @@ public Map> readFieldMetadata(FormatReaderFactory.Co } @Override - public FormatWriterFactory createWriterFactory(RowType type) { - ParquetWriterFactory baseFactory = - new ParquetWriterFactory(new RowDataParquetBuilder(type, options)); - return new ShreddingWritePlanWriterFactory( - baseFactory, new VariantShreddingWritePlanFactory(type, formatContext.options())); + protected FormatWriterFactory createRawWriterFactory(RowType type) { + return new ParquetWriterFactory(new RowDataParquetBuilder(type, options)); + } + + @Override + protected Set supportedShreddingWritePlans() { + return SUPPORTED_SHREDDING_WRITE_PLANS; } @Override diff --git a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java index eb8c01e983b6..a94e068efd4d 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetReaderFactory.java @@ -22,6 +22,7 @@ import org.apache.paimon.data.InternalRow; import org.apache.paimon.data.columnar.VectorizedColumnBatch; import org.apache.paimon.data.columnar.writable.WritableColumnVector; +import org.apache.paimon.data.shredding.MapSharedShreddingReadPlanFactory; import org.apache.paimon.data.shredding.ShreddingReadPlan; import org.apache.paimon.format.FormatMetadataUtils; import org.apache.paimon.format.FormatReaderFactory; @@ -60,6 +61,7 @@ import java.io.IOException; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Map; @@ -130,7 +132,7 @@ public FileRecordReader createReader(FormatReaderFactory.Context co ShreddingReadPlan readPlan = ShreddingReadPlanFactories.createReadPlan( readType, - Collections.emptyMap(), + readFieldMetadata(reader), fileSchema, shreddingReadPlanFactories(readType)); DataField[] physicalReadFields = readFields(readPlan.physicalRowType()); @@ -181,7 +183,8 @@ private RequestedSchema getOrCreateRequestedSchema(MessageType fileSchema) { } private List shreddingReadPlanFactories(RowType readType) { - return Collections.singletonList( + return Arrays.asList( + new MapSharedShreddingReadPlanFactory(readType), new VariantShreddingReadPlanFactory(readType, caseSensitive)); } 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 index 32f513a5e4be..a6b79ba2a3c5 100644 --- 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 @@ -22,6 +22,7 @@ import org.apache.paimon.data.BinaryString; import org.apache.paimon.data.GenericMap; import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.shredding.MapSharedShreddingContext; import org.apache.paimon.data.shredding.MapSharedShreddingFieldMeta; import org.apache.paimon.data.shredding.MapSharedShreddingUtils; import org.apache.paimon.data.shredding.MapShreddingDefine; @@ -53,6 +54,7 @@ import java.util.UUID; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests for format integration of {@link ShreddingWritePlanWriterFactory}. */ class ShreddingWritePlanFormatTest { @@ -60,10 +62,10 @@ class ShreddingWritePlanFormatTest { @TempDir java.nio.file.Path tempDir; @Test - void testParquetWritesMapSharedShreddingMetadataThroughVariantWrapper() throws Exception { + void testParquetWritesMapSharedShreddingMetadata() throws Exception { + Options options = mapSharedShreddingOptions(); FileFormat format = - new ParquetFileFormat( - new FileFormatFactory.FormatContext(new Options(), 1024, 1024)); + new ParquetFileFormat(new FileFormatFactory.FormatContext(options, 1024, 1024)); Map> fieldMetadata = writeAndReadFieldMetadata(format, "parquet", "none"); @@ -77,8 +79,9 @@ void testParquetWritesMapSharedShreddingMetadataThroughVariantWrapper() throws E @Test void testOrcWritesMapSharedShreddingMetadata() throws Exception { + Options options = mapSharedShreddingOptions(); FileFormat format = - new OrcFileFormat(new FileFormatFactory.FormatContext(new Options(), 1024, 1024)); + new OrcFileFormat(new FileFormatFactory.FormatContext(options, 1024, 1024)); Map> fieldMetadata = writeAndReadFieldMetadata(format, "orc", "none"); @@ -89,6 +92,41 @@ void testOrcWritesMapSharedShreddingMetadata() throws Exception { .containsEntry(OrcTypeUtil.PAIMON_ORC_FIELD_ID_KEY, "1"); } + @Test + void testOrcRejectsVariantShreddingWritePlan() { + Options options = new Options(); + options.set(CoreOptions.VARIANT_INFER_SHREDDING_SCHEMA, true); + FileFormat format = + new OrcFileFormat(new FileFormatFactory.FormatContext(options, 1024, 1024)); + RowType rowType = DataTypes.ROW(DataTypes.FIELD(0, "v", DataTypes.VARIANT())); + + assertThatThrownBy(() -> format.createWriterFactory(rowType)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("File format 'orc' does not support VARIANT write plans"); + } + + @Test + void testRejectsMultipleActiveWritePlans() { + Options options = mapSharedShreddingOptions(); + options.set(CoreOptions.VARIANT_INFER_SHREDDING_SCHEMA, true); + FileFormat format = + new ParquetFileFormat(new FileFormatFactory.FormatContext(options, 1024, 1024)); + RowType rowType = + DataTypes.ROW( + DataTypes.FIELD( + 0, "tags", DataTypes.MAP(DataTypes.STRING(), DataTypes.BIGINT())), + DataTypes.FIELD(1, "v", DataTypes.VARIANT())); + + assertThatThrownBy( + () -> + format.createWriterFactory( + rowType, + new MapSharedShreddingContext( + Collections.singletonMap("tags", 2)))) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("Composing multiple active shredding write plans is not supported."); + } + private Map> writeAndReadFieldMetadata( FileFormat format, String extension, String compression) throws IOException { FileIO fileIO = LocalFileIO.create(); @@ -96,8 +134,9 @@ private Map> writeAndReadFieldMetadata( RowType rowType = logicalRowType(); FormatWriterFactory writerFactory = - ShreddingWritePlanWriterFactories.wrapMapSharedShredding( - format.createWriterFactory(rowType), rowType, mapSharedShreddingOptions()); + format.createWriterFactory( + rowType, + new MapSharedShreddingContext(Collections.singletonMap("tags", 2))); PositionOutputStream out = fileIO.newOutputStream(file, false); FormatWriter writer = writerFactory.create(out, compression); writer.addElement(GenericRow.of(1, stringKeyMap("a", 10L, "b", 20L, "c", 30L))); @@ -115,11 +154,11 @@ private static RowType logicalRowType() { DataTypes.FIELD(1, "tags", DataTypes.MAP(DataTypes.STRING(), DataTypes.BIGINT()))); } - private static CoreOptions mapSharedShreddingOptions() { + private static Options 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); + return options; } private static void assertMapSharedShreddingMetadata(