You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/07/17 03:17:41 UTC
(paimon) branch master updated: [core] Support append read/write for MAP shared-shredding (#8524)
This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new a26016e999 [core] Support append read/write for MAP shared-shredding (#8524)
a26016e999 is described below
commit a26016e99977e34affdca0a156b13c2736afffc8
Author: lxy <38...@users.noreply.github.com>
AuthorDate: Fri Jul 17 11:17:35 2026 +0800
[core] Support append read/write for MAP shared-shredding (#8524)
This is a subtask of [MAP shared-shredding
support](https://cwiki.apache.org/confluence/display/PAIMON/PIP-43%3A+Columnar+Storage+Optimization+for+MAP+Type+in+Paimon).
Add append table read/write support for MAP shared-shredding.
This wires MAP shared-shredding into the append write path and
ORC/Parquet read path, keeps the format layer working with physical read
rows, and converts them back to logical MAP rows through the shredding
read plan. It also adds end-to-end coverage for basic read/write,
multiple shared-shredding MAP fields, projection, data evolution,
metadata recovery, null values, overflow values, and complex value
types.
This PR may look somewhat large at first glance, but more than 1600
lines are tests, and the amount of production code added is relatively
limited.
---
.../data/shredding/MapSharedShreddingReadPlan.java | 304 +++++
.../MapSharedShreddingReadPlanFactory.java | 78 ++
.../data/shredding/MapSharedShreddingUtils.java | 45 +
.../shredding/MapSharedShreddingWritePlan.java | 11 +-
.../MapSharedShreddingWritePlanFactory.java | 55 +-
.../format/shredding/ShreddingWritePlanType.java | 25 +
.../ShreddingWritePlanWriterFactories.java | 58 +-
.../shredding/ShreddingWritePlanWriterFactory.java | 64 +-
.../shredding/MapSharedShreddingReadPlanTest.java | 185 +++
.../shredding/MapSharedShreddingWritePlanTest.java | 56 +-
.../org/apache/paimon/append/AppendOnlyWriter.java | 6 +-
.../paimon/operation/BaseAppendFileStoreWrite.java | 15 +-
.../org/apache/paimon/schema/SchemaValidation.java | 8 +-
.../apache/paimon/append/AppendOnlyWriterTest.java | 359 +++++-
.../apache/paimon/schema/SchemaValidationTest.java | 33 +-
.../paimon/table/MapSharedShreddingTableTest.java | 1222 ++++++++++++++++++++
.../apache/paimon/format/orc/OrcFileFormat.java | 33 +-
.../apache/paimon/format/orc/OrcReaderFactory.java | 123 +-
.../paimon/format/parquet/ParquetFileFormat.java | 23 +-
.../format/parquet/ParquetReaderFactory.java | 7 +-
.../shredding/ShreddingWritePlanFormatTest.java | 62 +-
21 files changed, 2614 insertions(+), 158 deletions(-)
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 0000000000..521dbd5b55
--- /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<Integer, SharedShreddingContext> contextByFieldIndex;
+
+ public MapSharedShreddingReadPlan(
+ RowType logicalType, Map<String, MapSharedShreddingFieldMeta> 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<Integer, SharedShreddingContext> createContexts(
+ RowType logicalType, Map<String, MapSharedShreddingFieldMeta> fieldMetas) {
+ Map<Integer, SharedShreddingContext> 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<Integer, BinaryString> 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<String, Integer> 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 0000000000..36ee35b8f7
--- /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<String, Map<String, String>> fieldMetadata, @Nullable Object fileSchema) {
+ for (DataField field : logicalRowType.getFields()) {
+ Map<String, String> metadata = fieldMetadata.get(field.name());
+ if (MapSharedShreddingUtils.hasShreddingMetadata(metadata)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public ShreddingReadPlan createReadPlan(
+ Map<String, Map<String, String>> fieldMetadata, @Nullable Object fileSchema) {
+ return new MapSharedShreddingReadPlan(
+ logicalRowType, readSharedShreddingMetas(logicalRowType, fieldMetadata));
+ }
+
+ private static Map<String, MapSharedShreddingFieldMeta> readSharedShreddingMetas(
+ RowType logicalRowType, Map<String, Map<String, String>> fieldMetadata) {
+ Map<String, MapSharedShreddingFieldMeta> metas = new LinkedHashMap<>();
+ for (Map.Entry<String, Map<String, String>> 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 82f191f624..c2526b03bb 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 class MapSharedShreddingUtils {
return new RowType(logicalSchema.isNullable(), physicalFields);
}
+ public static RowType buildPhysicalReadType(
+ RowType logicalReadType,
+ Map<String, MapSharedShreddingFieldMeta> sharedShreddingFieldMetas) {
+ if (sharedShreddingFieldMetas.isEmpty()) {
+ return logicalReadType;
+ }
+
+ List<DataField> 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<String, Integer> buildColumnToNumColumns(
List<String> shreddingFieldNames, CoreOptions options) {
Map<String, Integer> fieldToNumColumns = new HashMap<>();
@@ -222,6 +254,19 @@ public class MapSharedShreddingUtils {
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<Integer, List<Integer>> sortedFieldColumns(
Map<Integer, List<Integer>> fieldToColumns) {
Map<Integer, List<Integer>> result = new TreeMap<>();
diff --git a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlan.java b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlan.java
index 0c84a33169..af0f614906 100644
--- a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlan.java
+++ b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlan.java
@@ -32,17 +32,12 @@ public class MapSharedShreddingWritePlan implements ShreddingWritePlan {
private final RowType logicalRowType;
private final MapSharedShreddingRowConverter converter;
- @Nullable private final MapSharedShreddingContext context;
-
@Nullable private Map<String, Map<String, String>> fieldMetadata;
public MapSharedShreddingWritePlan(
- RowType logicalRowType,
- Map<String, Integer> fieldToNumColumns,
- @Nullable MapSharedShreddingContext context) {
+ RowType logicalRowType, Map<String, Integer> fieldToNumColumns) {
this.logicalRowType = logicalRowType;
this.converter = new MapSharedShreddingRowConverter(logicalRowType, fieldToNumColumns);
- this.context = context;
}
@Override
@@ -75,10 +70,6 @@ public class MapSharedShreddingWritePlan implements ShreddingWritePlan {
Map<String, String> fieldMetadata = new LinkedHashMap<>();
MapSharedShreddingUtils.serializeMetadata(fieldMeta, compression, fieldMetadata);
metadata.put(fieldName, Collections.unmodifiableMap(fieldMetadata));
-
- if (context != null) {
- context.reportFileStats(fieldName, fieldMeta.maxRowWidth());
- }
}
return Collections.unmodifiableMap(metadata);
}
diff --git a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlanFactory.java b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlanFactory.java
index 6e500f1e91..24c47207fa 100644
--- a/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlanFactory.java
+++ b/paimon-common/src/main/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlanFactory.java
@@ -18,22 +18,39 @@
package org.apache.paimon.data.shredding;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.InternalMap;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.format.shredding.ShreddingWritePlanFactory;
+import org.apache.paimon.options.Options;
import org.apache.paimon.types.RowType;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
/** Creates per-file shared-shredding MAP write plans. */
public class MapSharedShreddingWritePlanFactory implements ShreddingWritePlanFactory {
+ private static final int INFER_BUFFER_ROW_COUNT = 1;
+
private final RowType logicalRowType;
- private final MapSharedShreddingContext context;
+ private final Map<String, Integer> fieldToMaxColumns;
+ private final Map<String, Integer> fieldToPosition;
- public MapSharedShreddingWritePlanFactory(
- RowType logicalRowType, MapSharedShreddingContext context) {
+ public MapSharedShreddingWritePlanFactory(RowType logicalRowType, Options options) {
this.logicalRowType = logicalRowType;
- this.context = context;
+ CoreOptions coreOptions = new CoreOptions(options);
+ List<String> shreddingFields =
+ MapSharedShreddingUtils.detectShreddingColumns(logicalRowType, coreOptions);
+ this.fieldToMaxColumns =
+ MapSharedShreddingUtils.buildColumnToNumColumns(shreddingFields, coreOptions);
+ this.fieldToPosition = new LinkedHashMap<>();
+ for (String field : shreddingFields) {
+ fieldToPosition.put(field, logicalRowType.getFieldIndex(field));
+ }
}
@Override
@@ -43,21 +60,43 @@ public class MapSharedShreddingWritePlanFactory implements ShreddingWritePlanFac
@Override
public boolean shouldCreateWritePlan() {
- return !context.isEmpty();
+ return !fieldToMaxColumns.isEmpty();
}
@Override
public boolean shouldInferWritePlan() {
- return false;
+ return shouldCreateWritePlan();
}
@Override
public int inferBufferRowCount() {
- return 0;
+ return INFER_BUFFER_ROW_COUNT;
}
@Override
public ShreddingWritePlan createWritePlan(List<InternalRow> sampleRows) {
- return new MapSharedShreddingWritePlan(logicalRowType, context.computeNextK(), context);
+ checkArgument(shouldCreateWritePlan(), "MAP shared-shredding write plan is not active.");
+
+ Map<String, Integer> fieldToNumColumns = new LinkedHashMap<>();
+ for (Map.Entry<String, Integer> entry : fieldToMaxColumns.entrySet()) {
+ int maxColumns = entry.getValue();
+ int numColumns = maxColumns;
+ if (!sampleRows.isEmpty()) {
+ int fieldPosition = fieldToPosition.get(entry.getKey());
+ int maxRowWidth = 0;
+ int sampleCount = Math.min(sampleRows.size(), INFER_BUFFER_ROW_COUNT);
+ for (int i = 0; i < sampleCount; i++) {
+ InternalRow row = sampleRows.get(i);
+ InternalMap map =
+ row.isNullAt(fieldPosition) ? null : row.getMap(fieldPosition);
+ maxRowWidth = Math.max(maxRowWidth, map == null ? 0 : map.size());
+ }
+ numColumns = Math.max(1, Math.min(maxRowWidth, maxColumns));
+ }
+ fieldToNumColumns.put(entry.getKey(), numColumns);
+ }
+
+ // TODO: Infer the column count from recent file metadata instead of current-file samples.
+ return new MapSharedShreddingWritePlan(logicalRowType, fieldToNumColumns);
}
}
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 0000000000..a94eb0272d
--- /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 1fba3ddedd..a83e78e33f 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,56 @@
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<String> shreddingFields =
- MapSharedShreddingUtils.detectShreddingColumns(rowType, options);
- if (shreddingFields.isEmpty()) {
- return delegate;
+ @Nullable
+ public static ShreddingWritePlanFactory createWritePlanFactory(
+ RowType rowType,
+ Options options,
+ Set<ShreddingWritePlanType> supportedTypes,
+ String formatIdentifier) {
+ ShreddingWritePlanFactory activeFactory = null;
+ ShreddingWritePlanType activeType = null;
+
+ VariantShreddingWritePlanFactory variantFactory =
+ new VariantShreddingWritePlanFactory(rowType, options);
+ if (variantFactory.shouldCreateWritePlan()) {
+ activeFactory = variantFactory;
+ activeType = ShreddingWritePlanType.VARIANT;
+ }
+
+ MapSharedShreddingWritePlanFactory mapFactory =
+ new MapSharedShreddingWritePlanFactory(rowType, options);
+ if (mapFactory.shouldCreateWritePlan()) {
+ if (activeFactory != null) {
+ throw new UnsupportedOperationException(
+ "Composing multiple active shredding write plans is not supported.");
+ }
+ activeFactory = mapFactory;
+ 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 32be092dcd..5441eb4a97 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.data.shredding.ShreddingWritePlan;
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 0000000000..943bf94dbf
--- /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<Integer>(),
+ 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<Integer>(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<Integer>(),
+ 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<String, MapSharedShreddingFieldMeta> 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<String, Integer> nameToId(Object... pairs) {
+ Map<String, Integer> 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-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlanTest.java b/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlanTest.java
index 2cfde7d648..ccc323bf54 100644
--- a/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlanTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/data/shredding/MapSharedShreddingWritePlanTest.java
@@ -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.InternalRow;
+import org.apache.paimon.options.Options;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
@@ -43,10 +44,8 @@ class MapSharedShreddingWritePlanTest {
DataTypes.FIELD(0, "id", DataTypes.INT()),
DataTypes.FIELD(
1, "tags", DataTypes.MAP(DataTypes.STRING(), DataTypes.BIGINT())));
- MapSharedShreddingContext context =
- new MapSharedShreddingContext(Collections.singletonMap("tags", 4));
MapSharedShreddingWritePlan writePlan =
- new MapSharedShreddingWritePlan(logicalType, context.computeNextK(), context);
+ new MapSharedShreddingWritePlan(logicalType, Collections.singletonMap("tags", 4));
InternalRow physicalRow =
writePlan.toPhysicalRow(
@@ -69,31 +68,64 @@ class MapSharedShreddingWritePlanTest {
.containsEntry(MapSharedShreddingDefine.FIELD_DICT_COMPRESSION, "none");
MapSharedShreddingFieldMeta fieldMeta =
MapSharedShreddingUtils.deserializeMetadata(fieldMetadata.get("tags"));
- assertThat(fieldMeta.nameToId()).containsEntry("a", 0).containsEntry("b", 1);
+ assertThat(fieldMeta.nameToId())
+ .containsEntry("a", 0)
+ .containsEntry("b", 1)
+ .containsEntry("c", 2);
assertThat(fieldMeta.numColumns()).isEqualTo(4);
assertThat(fieldMeta.maxRowWidth()).isEqualTo(3);
- assertThat(context.computeNextK()).containsEntry("tags", 3);
}
@Test
- void testFactoryCreatesPlanFromContext() {
+ void testFactoryInfersColumnCountFromFirstRow() {
RowType logicalType =
DataTypes.ROW(
DataTypes.FIELD(
0, "tags", DataTypes.MAP(DataTypes.STRING(), DataTypes.INT())));
- MapSharedShreddingContext context =
- new MapSharedShreddingContext(Collections.singletonMap("tags", 2));
- MapSharedShreddingWritePlanFactory factory =
- new MapSharedShreddingWritePlanFactory(logicalType, context);
+ MapSharedShreddingWritePlanFactory factory = createFactory(logicalType, 4);
assertThat(factory.shouldCreateWritePlan()).isTrue();
- assertThat(factory.shouldInferWritePlan()).isFalse();
- assertThat(factory.createWritePlan(Collections.emptyList()).physicalRowType())
+ assertThat(factory.shouldInferWritePlan()).isTrue();
+ assertThat(factory.inferBufferRowCount()).isEqualTo(1);
+ assertThat(
+ factory.createWritePlan(
+ Collections.singletonList(
+ GenericRow.of(
+ stringKeyMap("a", 1, "b", 2, "c", 3))))
+ .physicalRowType())
+ .isEqualTo(
+ MapSharedShreddingUtils.logicalToPhysicalSchema(
+ logicalType, Collections.singletonMap("tags", 3)));
+ }
+
+ @Test
+ void testFactoryCapsInferredColumnCountAtMaxColumns() {
+ RowType logicalType =
+ DataTypes.ROW(
+ DataTypes.FIELD(
+ 0, "tags", DataTypes.MAP(DataTypes.STRING(), DataTypes.INT())));
+ MapSharedShreddingWritePlanFactory factory = createFactory(logicalType, 2);
+
+ assertThat(
+ factory.createWritePlan(
+ Collections.singletonList(
+ GenericRow.of(
+ stringKeyMap("a", 1, "b", 2, "c", 3))))
+ .physicalRowType())
.isEqualTo(
MapSharedShreddingUtils.logicalToPhysicalSchema(
logicalType, Collections.singletonMap("tags", 2)));
}
+ private static MapSharedShreddingWritePlanFactory createFactory(
+ RowType logicalType, int maxColumns) {
+ Options options = new Options();
+ options.setString("fields.tags.map.storage-layout", "shared-shredding");
+ options.setString(
+ "fields.tags.map.shared-shredding.max-columns", String.valueOf(maxColumns));
+ return new MapSharedShreddingWritePlanFactory(logicalType, options);
+ }
+
private static GenericMap stringKeyMap(Object... keyValues) {
Map<Object, Object> values = new LinkedHashMap<>();
for (int i = 0; i < keyValues.length; i += 2) {
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 66dcc75613..ea81dea50f 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
@@ -313,8 +313,10 @@ public class AppendOnlyWriter implements BatchRecordWriter, MemoryOwner {
}
private RollingFileWriter<InternalRow, DataFileMeta> createRollingRowWriter() {
- if (blobContext != null
- || !fieldsInVectorFile(writeSchema, vectorFileFormat != null).isEmpty()) {
+ boolean hasDedicatedFields =
+ blobContext != null
+ || !fieldsInVectorFile(writeSchema, vectorFileFormat != null).isEmpty();
+ if (hasDedicatedFields) {
return new DedicatedFormatRollingFileWriter(
fileIO,
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 43e83333ee..901fbe2fe0 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,7 @@ import org.apache.paimon.compact.CompactManager;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BlobConsumer;
import org.apache.paimon.data.InternalRow;
+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 +34,7 @@ import org.apache.paimon.format.FileFormat;
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.metrics.MetricRegistry;
@@ -148,6 +150,8 @@ public abstract class BaseAppendFileStoreWrite extends MemoryFileStoreWrite<Inte
ExecutorService compactExecutor,
@Nullable BucketedDvMaintainer dvMaintainer,
boolean ignorePreviousFiles) {
+ DataFilePathFactory dataPathFactory =
+ pathFactory.createDataFilePathFactory(partition, bucket);
return new AppendOnlyWriter(
fileIO,
ioManager,
@@ -164,7 +168,7 @@ public abstract class BaseAppendFileStoreWrite extends MemoryFileStoreWrite<Inte
// 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,
@@ -224,6 +228,7 @@ public abstract class BaseAppendFileStoreWrite extends MemoryFileStoreWrite<Inte
if (toCompact.isEmpty()) {
return Collections.emptyList();
}
+ checkNoSharedShreddingRewrite("Compaction rewrite");
Exception collectedExceptions = null;
RowDataRollingFileWriter rewriter =
createRollingFileWriter(
@@ -256,6 +261,7 @@ public abstract class BaseAppendFileStoreWrite extends MemoryFileStoreWrite<Inte
public List<DataFileMeta> clusterRewrite(
BinaryRow partition, int bucket, List<DataFileMeta> toCluster) throws Exception {
+ checkNoSharedShreddingRewrite("Cluster rewrite");
RecordReaderIterator<InternalRow> reader =
createFilesIterator(partition, bucket, toCluster, null);
@@ -292,6 +298,13 @@ public abstract class BaseAppendFileStoreWrite extends MemoryFileStoreWrite<Inte
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<LongCounter> 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 7c743f416e..4b81ba6ae4 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
@@ -667,7 +667,13 @@ public class SchemaValidation {
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<STRING, T>.",
+ "Column '%s' is configured with map.storage-layout=shared-shredding but its type is not MAP<STRING NOT NULL, T>.",
+ 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 55965234bb..8354945364 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,28 @@
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.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 +73,8 @@ import org.assertj.core.api.Assertions;
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 +83,13 @@ import java.util.ArrayList;
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 +482,141 @@ public class AppendOnlyWriterTest {
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, 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(),
+ 1,
+ 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, 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(), 1, 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, 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(), 1, 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, 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)),
+ overflowFields(2),
+ 2,
+ 3));
+ }
+
@Test
public void testNoBuffer() throws Exception {
AppendOnlyWriter writer = createEmptyWriter(Long.MAX_VALUE);
@@ -647,6 +794,105 @@ public class AppendOnlyWriterTest {
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<BinaryString, Object> 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<String, Integer> nameToId,
+ Map<Integer, List<Integer>> fieldToColumns,
+ Set<Integer> overflowFields,
+ int numColumns,
+ int maxRowWidth) {
+ return new MapSharedShreddingFieldMeta(
+ new TreeMap<>(nameToId),
+ new TreeMap<>(fieldToColumns),
+ new TreeSet<>(overflowFields),
+ numColumns,
+ maxRowWidth);
+ }
+
+ private Map<String, Integer> nameToId(Object... pairs) {
+ Map<String, Integer> 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<Integer, List<Integer>> fieldToColumns(Object... pairs) {
+ Map<Integer, List<Integer>> result = new TreeMap<>();
+ for (int i = 0; i < pairs.length; i += 2) {
+ result.put((Integer) pairs[i], (List<Integer>) pairs[i + 1]);
+ }
+ return result;
+ }
+
+ private List<Integer> columns(int... columns) {
+ List<Integer> result = new ArrayList<>();
+ for (int column : columns) {
+ result.add(column);
+ }
+ return result;
+ }
+
+ private Set<Integer> overflowFields(int... fieldIds) {
+ Set<Integer> 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 +904,13 @@ public class AppendOnlyWriterTest {
}
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 +918,113 @@ public class AppendOnlyWriterTest {
null);
}
+ private SharedShreddingAppendContext createSharedShreddingAppendContext(
+ String fileFormat, Options rawOptions) {
+ CoreOptions options = new CoreOptions(rawOptions);
+ DataFilePathFactory pathFactory = createPathFactory(fileFormat);
+ LocalFileIO fileIO = LocalFileIO.create();
+ return new SharedShreddingAppendContext(
+ options, pathFactory, fileIO, FileFormat.fromIdentifier(fileFormat, rawOptions));
+ }
+
+ 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 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,
+ null);
+ }
+
+ 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<String, Map<String, String>> 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 SharedShreddingAppendContext(
+ CoreOptions options,
+ DataFilePathFactory pathFactory,
+ LocalFileIO fileIO,
+ FileFormat format) {
+ this.options = options;
+ this.pathFactory = pathFactory;
+ this.fileIO = fileIO;
+ this.format = format;
+ }
+ }
+
private AppendOnlyWriter createEmptyWriter(long targetFileSize) {
return createWriter(targetFileSize, false, false, false, Collections.emptyList()).getLeft();
}
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 d13ae60bff..12608d1720 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
@@ -387,7 +387,9 @@ class SchemaValidationTest {
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())));
@@ -465,10 +467,29 @@ class SchemaValidationTest {
options,
"")))
.hasMessageContaining(
- "Column 'codes' is configured with map.storage-layout=shared-shredding but its type is not MAP<STRING, T>.");
+ "Column 'codes' is configured with map.storage-layout=shared-shredding but its type is not MAP<STRING NOT NULL, T>.");
options.remove("fields.codes.map.storage-layout");
options.put("fields.metrics.map.storage-layout", "shared-shredding");
+ List<DataField> 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(
() ->
@@ -1431,7 +1452,8 @@ class SchemaValidationTest {
private List<DataField> 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<DataField> nestedMapValueFields(DataType valueType) {
@@ -1441,14 +1463,15 @@ class SchemaValidationTest {
1,
"metrics",
DataTypes.MAP(
- DataTypes.STRING(),
+ DataTypes.STRING().notNull(),
DataTypes.ROW(DataTypes.FIELD(0, "payload", valueType)))));
}
private List<DataField> 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 0000000000..66e166c7da
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/table/MapSharedShreddingTableTest.java
@@ -0,0 +1,1222 @@
+/*
+ * 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<Integer, Map<String, Long>> 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<Integer, List<Map<String, Long>>> 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<Integer, Map<String, Long>> 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<Integer, Map<String, ComplexValue>> 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<Integer, Map<String, WideValue>> 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 testInferColumnCountFromFirstRowOfEachFile(String format) throws Exception {
+ Table table = createTableWithBucket(format, 8, "1", "metrics");
+
+ write(
+ table,
+ GenericRow.of(1, mapOf("a", 11L, "b", 12L)),
+ GenericRow.of(2, mapOf("c", 21L, "d", 22L, "e", 23L)));
+ write(table, GenericRow.of(3, mapOf("f", 31L)));
+
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ List<DataFileWithSplit> 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(2);
+ assertThat(firstFileMeta.maxRowWidth()).isEqualTo(3);
+
+ MapSharedShreddingFieldMeta secondFileMeta =
+ readSharedShreddingFieldMeta(fileStoreTable, files.get(1), "metrics");
+ assertThat(secondFileMeta.numColumns()).isEqualTo(1);
+ assertThat(secondFileMeta.maxRowWidth()).isEqualTo(1);
+
+ Map<Integer, Map<String, Long>> 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, "d", 22L, "e", 23L))
+ .containsEntry(3, javaMapOf("f", 31L));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"orc", "parquet"})
+ public void testSwitchMapLayoutAndInferColumns(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<DataFileWithSplit> 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(1);
+ assertThat(metricsMeta.maxRowWidth()).isEqualTo(1);
+
+ Map<Integer, List<Map<String, Long>>> 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<Integer, Map<String, Long>> 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<Integer, List<Map<String, Long>>> 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<Integer, List<Map<String, Long>>> 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<String> mapFields,
+ List<String> 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<CommitMessage> commitMessages = write.prepareCommit();
+ if (firstRowId != null) {
+ setFirstRowId(commitMessages, firstRowId);
+ }
+ commit.commit(commitMessages);
+ }
+ }
+
+ private void setFirstRowId(List<CommitMessage> commitMessages, long firstRowId) {
+ for (CommitMessage message : commitMessages) {
+ CommitMessageImpl commitMessage = (CommitMessageImpl) message;
+ List<DataFileMeta> 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<String> mapFields,
+ List<String> 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<DataFileWithSplit> currentDataFiles(FileStoreTable table) throws Exception {
+ List<DataFileWithSplit> 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<String, Map<String, String>> 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<BinaryString, Long> 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<BinaryString, GenericRow> 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<BinaryString, GenericRow> 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<BinaryString, Long> 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<String, Long> values) {
+ Map<BinaryString, Long> map = new LinkedHashMap<>();
+ for (Map.Entry<String, Long> entry : values.entrySet()) {
+ map.put(BinaryString.fromString(entry.getKey()), entry.getValue());
+ }
+ return new GenericMap(map);
+ }
+
+ private Map<String, Long> toJavaMap(InternalMap map) {
+ Map<String, Long> 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<String, Long> javaMapOf(Object... entries) {
+ Map<String, Long> 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<String, ComplexValue> toJavaComplexMap(InternalMap map) {
+ Map<String, ComplexValue> 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<String> toJavaStringList(InternalArray array) {
+ List<String> 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<String, Long> toJavaLongMap(InternalMap map) {
+ Map<String, Long> 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<String, ComplexValue> javaComplexMapOf(Object... entries) {
+ Map<String, ComplexValue> 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<String, WideValue> toJavaWideMap(InternalMap map) {
+ Map<String, WideValue> 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<Integer> toJavaIntList(InternalArray array) {
+ List<Integer> 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<String, WideValue> javaWideMapOf(Object... entries) {
+ Map<String, WideValue> 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<String> tags;
+ private final Map<String, Long> attrs;
+
+ private ComplexValue(Long count, List<String> tags, Map<String, Long> 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<Integer> ints;
+ private final Map<String, Long> 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<Integer> ints,
+ Map<String, Long> 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 77790dd95f..19d7dcdd81 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,10 @@ import org.apache.paimon.format.orc.filter.OrcPredicateFunctionVisitor;
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.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.MemorySize;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Predicate;
@@ -52,10 +56,12 @@ import javax.annotation.concurrent.ThreadSafe;
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,7 +73,11 @@ public class OrcFileFormat extends FileFormat implements SupportsFieldMetadata {
public static final String IDENTIFIER = "orc";
+ private static final Set<ShreddingWritePlanType> SUPPORTED_SHREDDING_WRITE_PLANS =
+ Collections.singleton(ShreddingWritePlanType.MAP_SHARED_SHREDDING);
+
private final Properties orcProperties;
+ private final Options options;
private final org.apache.hadoop.conf.Configuration readerConf;
private final org.apache.hadoop.conf.Configuration writerConf;
private final int readBatchSize;
@@ -78,6 +88,7 @@ public class OrcFileFormat extends FileFormat implements SupportsFieldMetadata {
public OrcFileFormat(FormatContext formatContext) {
super(IDENTIFIER);
+ this.options = 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()));
@@ -163,19 +174,27 @@ public class OrcFileFormat extends FileFormat implements SupportsFieldMetadata {
*/
@Override
public FormatWriterFactory createWriterFactory(RowType type) {
+ ShreddingWritePlanFactory writePlanFactory =
+ ShreddingWritePlanWriterFactories.createWritePlanFactory(
+ type, options, SUPPORTED_SHREDDING_WRITE_PLANS, IDENTIFIER);
+
RowType refinedType = (RowType) refineDataType(type);
TypeDescription typeDescription = OrcTypeUtil.convertToOrcSchema(refinedType);
Vectorizer<InternalRow> vectorizer =
new RowDataVectorizer(
typeDescription, refinedType.getFields(), legacyTimestampLtzType);
- return new OrcWriterFactory(
- vectorizer,
- orcProperties,
- writerConf,
- writeBatchSize,
- writeBatchMemory,
- legacyTimestampLtzType);
+ FormatWriterFactory rawFactory =
+ new OrcWriterFactory(
+ vectorizer,
+ orcProperties,
+ writerConf,
+ writeBatchSize,
+ writeBatchMemory,
+ legacyTimestampLtzType);
+ return writePlanFactory == null
+ ? rawFactory
+ : new ShreddingWritePlanWriterFactory(rawFactory, writePlanFactory);
}
private Properties getOrcProperties(Options options, FormatContext formatContext) {
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 1809bc654e..05ebc5da22 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.ColumnarRow;
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 class OrcReaderFactory implements FormatReaderFactory {
// ------------------------------------------------------------------------
@Override
- public OrcVectorizedReader createReader(FormatReaderFactory.Context context)
+ public FileRecordReader<InternalRow> createReader(FormatReaderFactory.Context context)
throws IOException {
int poolSize =
context instanceof OrcFormatReaderContext
? ((OrcFormatReaderContext) context).poolSize()
: 1;
- Pool<OrcReaderBatch> 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<OrcReaderBatch> 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 class OrcReaderFactory implements FormatReaderFactory {
VectorizedRowBatch orcBatch,
Pool.Recycler<OrcReaderBatch> recycler,
FileIO fileIO) {
- List<String> tableFieldNames = tableType.getFieldNames();
- List<DataType> tableFieldTypes = tableType.getFieldTypes();
+ return createReaderBatch(filePath, orcBatch, recycler, fileIO, tableType);
+ }
+
+ private OrcReaderBatch createReaderBatch(
+ Path filePath,
+ VectorizedRowBatch orcBatch,
+ Pool.Recycler<OrcReaderBatch> recycler,
+ FileIO fileIO,
+ RowType readType) {
+ List<String> tableFieldNames = readType.getFieldNames();
+ List<DataType> 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 class OrcReaderFactory implements FormatReaderFactory {
// ------------------------------------------------------------------------
+ private static List<ShreddingReadPlanFactory> shreddingReadPlanFactories(RowType readType) {
+ return Collections.<ShreddingReadPlanFactory>singletonList(
+ new MapSharedShreddingReadPlanFactory(readType));
+ }
+
+ // ------------------------------------------------------------------------
+
private Pool<OrcReaderBatch> createPoolOfBatches(Path filePath, int numBatches, FileIO fileIO) {
+ return createPoolOfBatches(filePath, numBatches, fileIO, schema, tableType);
+ }
+
+ private Pool<OrcReaderBatch> createPoolOfBatches(
+ Path filePath,
+ int numBatches,
+ FileIO fileIO,
+ TypeDescription readSchema,
+ RowType readType) {
final Pool<OrcReaderBatch> 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 @@ public class OrcReaderFactory implements FormatReaderFactory {
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<OrcFilters.Predicate> 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<Long, Long> 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 2380a6cf7a..8093e714b0 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,10 @@ import org.apache.paimon.format.FormatWriterFactory;
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.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.format.variant.VariantShreddingWritePlanFactory;
import org.apache.paimon.options.CatalogOptions;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
@@ -43,15 +45,24 @@ import org.apache.parquet.hadoop.ParquetOutputFormat;
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 static final Set<ShreddingWritePlanType> SUPPORTED_SHREDDING_WRITE_PLANS =
+ Collections.unmodifiableSet(
+ EnumSet.of(
+ ShreddingWritePlanType.VARIANT,
+ ShreddingWritePlanType.MAP_SHARED_SHREDDING));
+
private final FormatContext formatContext;
private final Options options;
private final int readBatchSize;
@@ -100,10 +111,14 @@ public class ParquetFileFormat extends FileFormat implements SupportsFieldMetada
@Override
public FormatWriterFactory createWriterFactory(RowType type) {
- ParquetWriterFactory baseFactory =
+ FormatWriterFactory rawFactory =
new ParquetWriterFactory(new RowDataParquetBuilder(type, options));
- return new ShreddingWritePlanWriterFactory(
- baseFactory, new VariantShreddingWritePlanFactory(type, formatContext.options()));
+ ShreddingWritePlanFactory writePlanFactory =
+ ShreddingWritePlanWriterFactories.createWritePlanFactory(
+ type, formatContext.options(), SUPPORTED_SHREDDING_WRITE_PLANS, IDENTIFIER);
+ return writePlanFactory == null
+ ? rawFactory
+ : new ShreddingWritePlanWriterFactory(rawFactory, writePlanFactory);
}
@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 eb8c01e983..a94e068efd 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.annotation.VisibleForTesting;
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 javax.annotation.Nullable;
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 class ParquetReaderFactory implements FormatReaderFactory {
ShreddingReadPlan readPlan =
ShreddingReadPlanFactories.createReadPlan(
readType,
- Collections.emptyMap(),
+ readFieldMetadata(reader),
fileSchema,
shreddingReadPlanFactories(readType));
DataField[] physicalReadFields = readFields(readPlan.physicalRowType());
@@ -181,7 +183,8 @@ public class ParquetReaderFactory implements FormatReaderFactory {
}
private List<ShreddingReadPlanFactory> shreddingReadPlanFactories(RowType readType) {
- return Collections.<ShreddingReadPlanFactory>singletonList(
+ return Arrays.<ShreddingReadPlanFactory>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 32f513a5e4..b6ed75e53f 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
@@ -53,6 +53,7 @@ import java.util.Map;
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 +61,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<String, Map<String, String>> fieldMetadata =
writeAndReadFieldMetadata(format, "parquet", "none");
@@ -77,8 +78,9 @@ class ShreddingWritePlanFormatTest {
@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<String, Map<String, String>> fieldMetadata =
writeAndReadFieldMetadata(format, "orc", "none");
@@ -89,15 +91,43 @@ class ShreddingWritePlanFormatTest {
.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))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage("Composing multiple active shredding write plans is not supported.");
+ }
+
private Map<String, Map<String, String>> writeAndReadFieldMetadata(
FileFormat format, String extension, String compression) throws IOException {
FileIO fileIO = LocalFileIO.create();
Path file = new Path(tempDir.toString(), UUID.randomUUID() + "." + extension);
RowType rowType = logicalRowType();
- FormatWriterFactory writerFactory =
- ShreddingWritePlanWriterFactories.wrapMapSharedShredding(
- format.createWriterFactory(rowType), rowType, mapSharedShreddingOptions());
+ FormatWriterFactory writerFactory = format.createWriterFactory(rowType);
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 +145,11 @@ class ShreddingWritePlanFormatTest {
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(
@@ -132,14 +162,12 @@ class ShreddingWritePlanFormatTest {
MapSharedShreddingFieldMeta fieldMeta =
MapSharedShreddingUtils.deserializeMetadata(fieldMetadata.get("tags"), compression);
- assertThat(fieldMeta.nameToId())
- .containsEntry("a", 0)
- .containsEntry("b", 1)
- .containsEntry("c", 2);
- assertThat(fieldMeta.fieldToColumns())
- .containsEntry(0, Collections.singletonList(0))
- .containsEntry(1, Collections.singletonList(1));
- assertThat(fieldMeta.overflowFieldSet()).containsExactly(2);
+ assertThat(fieldMeta.nameToId()).containsOnlyKeys("a", "b", "c");
+ assertThat(fieldMeta.fieldToColumns()).hasSize(2);
+ assertThat(fieldMeta.fieldToColumns().values())
+ .containsExactlyInAnyOrder(
+ Collections.singletonList(0), Collections.singletonList(1));
+ assertThat(fieldMeta.overflowFieldSet()).hasSize(1);
assertThat(fieldMeta.numColumns()).isEqualTo(2);
assertThat(fieldMeta.maxRowWidth()).isEqualTo(3);
}