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 2024/10/23 13:41:22 UTC
(paimon) branch master updated: [parquet] Add type id to parquet files (#4362)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 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 20790af0b [parquet] Add type id to parquet files (#4362)
20790af0b is described below
commit 20790af0b966a0914f92d6da58cf2d0655e0a61d
Author: tsreaper <ts...@gmail.com>
AuthorDate: Wed Oct 23 21:41:16 2024 +0800
[parquet] Add type id to parquet files (#4362)
---
.../{SystemFields.java => SpecialFields.java} | 52 +++++++++++++++++-
.../main/java/org/apache/paimon/types/RowType.java | 4 +-
.../src/main/java/org/apache/paimon/KeyValue.java | 6 +-
.../paimon/iceberg/metadata/IcebergDataField.java | 4 +-
.../paimon/iceberg/metadata/IcebergSchema.java | 19 +------
.../org/apache/paimon/schema/SchemaValidation.java | 4 +-
.../apache/paimon/table/PrimaryKeyTableUtils.java | 4 +-
.../apache/paimon/table/system/AuditLogTable.java | 6 +-
.../paimon/utils/StatsCollectorFactories.java | 4 +-
.../org/apache/paimon/TestKeyValueGenerator.java | 8 +--
.../apache/paimon/table/SchemaEvolutionTest.java | 4 +-
.../paimon/flink/sink/CommitterOperatorTest.java | 2 +-
.../format/parquet/ParquetSchemaConverter.java | 52 +++++++++++++++---
.../format/parquet/ParquetReadWriteTest.java | 64 ++++++++++++++++++++++
14 files changed, 183 insertions(+), 50 deletions(-)
diff --git a/paimon-common/src/main/java/org/apache/paimon/table/SystemFields.java b/paimon-common/src/main/java/org/apache/paimon/table/SpecialFields.java
similarity index 53%
rename from paimon-common/src/main/java/org/apache/paimon/table/SystemFields.java
rename to paimon-common/src/main/java/org/apache/paimon/table/SpecialFields.java
index 155683981..36f22a310 100644
--- a/paimon-common/src/main/java/org/apache/paimon/table/SystemFields.java
+++ b/paimon-common/src/main/java/org/apache/paimon/table/SpecialFields.java
@@ -26,8 +26,38 @@ import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
-/** System fields. */
-public class SystemFields {
+/**
+ * Special fields in a {@link org.apache.paimon.types.RowType} with specific field ids.
+ *
+ * <p><b>System fields</b>:
+ *
+ * <ul>
+ * <li><code>_KEY_<key-field></code>: Keys of a key-value. ID = 1073741823 + <code>
+ * (field-id)
+ * </code>.
+ * <li><code>_SEQUENCE_NUMBER</code>: Sequence number of a key-value. ID = 2147483646.
+ * <li><code>_VALUE_KIND</code>: Type of a key-value. See {@link org.apache.paimon.types.RowKind}.
+ * ID = 2147483645.
+ * <li><code>_LEVEL</code>: Which LSM tree level does this key-value stay in. ID = 2147483644.
+ * <li><code>rowkind</code>: THw rowkind field in audit-log system tables. ID = 2147483643.
+ * </ul>
+ *
+ * <p><b>Structured type fields</b>:
+ *
+ * <p>These ids are mainly used as field ids in parquet files, so compute engines can read a field
+ * directly by id. These ids are not stored in {@link org.apache.paimon.types.DataField}.
+ *
+ * <ul>
+ * <li>Array element field: ID = 536870911 + <code>(array-field-id)</code>.
+ * <li>Map key field: ID = 536870911 + <code>(array-field-id)</code>.
+ * <li>Map value field: ID = 536870911 - <code>(array-field-id)</code>.
+ * </ul>
+ */
+public class SpecialFields {
+
+ // ----------------------------------------------------------------------------------------
+ // System fields
+ // ----------------------------------------------------------------------------------------
public static final int SYSTEM_FIELD_ID_START = Integer.MAX_VALUE / 2;
@@ -59,4 +89,22 @@ public class SystemFields {
public static boolean isSystemField(String field) {
return field.startsWith(KEY_FIELD_PREFIX) || SYSTEM_FIELD_NAMES.contains(field);
}
+
+ // ----------------------------------------------------------------------------------------
+ // Structured type fields
+ // ----------------------------------------------------------------------------------------
+
+ public static final int STRUCTURED_TYPE_FIELD_ID_BASE = Integer.MAX_VALUE / 4;
+
+ public static int getArrayElementFieldId(int arrayFieldId) {
+ return STRUCTURED_TYPE_FIELD_ID_BASE + arrayFieldId;
+ }
+
+ public static int getMapKeyFieldId(int mapFieldId) {
+ return STRUCTURED_TYPE_FIELD_ID_BASE + mapFieldId;
+ }
+
+ public static int getMapValueFieldId(int mapFieldId) {
+ return STRUCTURED_TYPE_FIELD_ID_BASE - mapFieldId;
+ }
}
diff --git a/paimon-common/src/main/java/org/apache/paimon/types/RowType.java b/paimon-common/src/main/java/org/apache/paimon/types/RowType.java
index 9dc89117f..80e41be71 100644
--- a/paimon-common/src/main/java/org/apache/paimon/types/RowType.java
+++ b/paimon-common/src/main/java/org/apache/paimon/types/RowType.java
@@ -20,7 +20,7 @@ package org.apache.paimon.types;
import org.apache.paimon.annotation.Public;
import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.table.SystemFields;
+import org.apache.paimon.table.SpecialFields;
import org.apache.paimon.utils.Preconditions;
import org.apache.paimon.utils.StringUtils;
@@ -332,7 +332,7 @@ public final class RowType extends DataType {
Set<Integer> fieldIds = new HashSet<>();
new RowType(fields).collectFieldIds(fieldIds);
return fieldIds.stream()
- .filter(i -> !SystemFields.isSystemField(i))
+ .filter(i -> !SpecialFields.isSystemField(i))
.max(Integer::compareTo)
.orElse(-1);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/KeyValue.java b/paimon-core/src/main/java/org/apache/paimon/KeyValue.java
index f2a6c0bde..36ac88996 100644
--- a/paimon-core/src/main/java/org/apache/paimon/KeyValue.java
+++ b/paimon-core/src/main/java/org/apache/paimon/KeyValue.java
@@ -31,9 +31,9 @@ import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
-import static org.apache.paimon.table.SystemFields.LEVEL;
-import static org.apache.paimon.table.SystemFields.SEQUENCE_NUMBER;
-import static org.apache.paimon.table.SystemFields.VALUE_KIND;
+import static org.apache.paimon.table.SpecialFields.LEVEL;
+import static org.apache.paimon.table.SpecialFields.SEQUENCE_NUMBER;
+import static org.apache.paimon.table.SpecialFields.VALUE_KIND;
/**
* A key value, including user key, sequence number, value kind and value. This object can be
diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java
index c4505bbdd..93cb2ab6d 100644
--- a/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java
+++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergDataField.java
@@ -64,9 +64,9 @@ public class IcebergDataField {
@JsonProperty(FIELD_DOC)
private final String doc;
- public IcebergDataField(DataField dataField, int bias) {
+ public IcebergDataField(DataField dataField) {
this(
- dataField.id() + bias,
+ dataField.id(),
dataField.name(),
!dataField.type().isNullable(),
toTypeString(dataField.type()),
diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergSchema.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergSchema.java
index 3cdd53a79..ff28a3bfd 100644
--- a/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergSchema.java
+++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergSchema.java
@@ -18,7 +18,6 @@
package org.apache.paimon.iceberg.metadata;
-import org.apache.paimon.CoreOptions;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
@@ -53,26 +52,10 @@ public class IcebergSchema {
private final List<IcebergDataField> fields;
public static IcebergSchema create(TableSchema tableSchema) {
- int bias;
- if (new CoreOptions(tableSchema.options())
- .formatType()
- .equals(CoreOptions.FILE_FORMAT_PARQUET)) {
- if (tableSchema.primaryKeys().isEmpty()) {
- // ParquetSchemaUtil.addFallbackIds starts enumerating id from 1 instead of 0
- bias = 1;
- } else {
- // data files start with trimmed primary keys + sequence number + value kind
- // also ParquetSchemaUtil.addFallbackIds starts enumerating id from 1 instead of 0
- bias = tableSchema.trimmedPrimaryKeys().size() + 3;
- }
- } else {
- bias = 0;
- }
-
return new IcebergSchema(
(int) tableSchema.id(),
tableSchema.fields().stream()
- .map(f -> new IcebergDataField(f, bias))
+ .map(IcebergDataField::new)
.collect(Collectors.toList()));
}
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 20a8787a4..bb9b440ce 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
@@ -68,8 +68,8 @@ import static org.apache.paimon.CoreOptions.SNAPSHOT_NUM_RETAINED_MAX;
import static org.apache.paimon.CoreOptions.SNAPSHOT_NUM_RETAINED_MIN;
import static org.apache.paimon.CoreOptions.STREAMING_READ_OVERWRITE;
import static org.apache.paimon.mergetree.compact.PartialUpdateMergeFunction.SEQUENCE_GROUP;
-import static org.apache.paimon.table.SystemFields.KEY_FIELD_PREFIX;
-import static org.apache.paimon.table.SystemFields.SYSTEM_FIELD_NAMES;
+import static org.apache.paimon.table.SpecialFields.KEY_FIELD_PREFIX;
+import static org.apache.paimon.table.SpecialFields.SYSTEM_FIELD_NAMES;
import static org.apache.paimon.types.DataTypeRoot.ARRAY;
import static org.apache.paimon.types.DataTypeRoot.MAP;
import static org.apache.paimon.types.DataTypeRoot.MULTISET;
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyTableUtils.java b/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyTableUtils.java
index 58cde07c3..d156d23a9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyTableUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyTableUtils.java
@@ -34,8 +34,8 @@ import org.apache.paimon.types.RowType;
import java.util.List;
import java.util.stream.Collectors;
-import static org.apache.paimon.table.SystemFields.KEY_FIELD_ID_START;
-import static org.apache.paimon.table.SystemFields.KEY_FIELD_PREFIX;
+import static org.apache.paimon.table.SpecialFields.KEY_FIELD_ID_START;
+import static org.apache.paimon.table.SpecialFields.KEY_FIELD_PREFIX;
/** Utils for creating changelog table with primary keys. */
public class PrimaryKeyTableUtils {
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
index bbbea7441..e0acd9fb3 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
@@ -42,7 +42,7 @@ import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.DataTable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.ReadonlyTable;
-import org.apache.paimon.table.SystemFields;
+import org.apache.paimon.table.SpecialFields;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.DataTableScan;
import org.apache.paimon.table.source.InnerTableRead;
@@ -137,7 +137,7 @@ public class AuditLogTable implements DataTable, ReadonlyTable {
@Override
public RowType rowType() {
List<DataField> fields = new ArrayList<>();
- fields.add(SystemFields.ROW_KIND);
+ fields.add(SpecialFields.ROW_KIND);
fields.addAll(wrapped.rowType().getFields());
return new RowType(fields);
}
@@ -566,7 +566,7 @@ public class AuditLogTable implements DataTable, ReadonlyTable {
boolean rowKindAppeared = false;
for (int i = 0; i < fields.size(); i++) {
String fieldName = fields.get(i).name();
- if (fieldName.equals(SystemFields.ROW_KIND.name())) {
+ if (fieldName.equals(SpecialFields.ROW_KIND.name())) {
rowKindAppeared = true;
readProjection[i] = -1;
} else {
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/StatsCollectorFactories.java b/paimon-core/src/main/java/org/apache/paimon/utils/StatsCollectorFactories.java
index ed23a246a..de94b2e23 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/StatsCollectorFactories.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/StatsCollectorFactories.java
@@ -22,7 +22,7 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.options.Options;
import org.apache.paimon.statistics.SimpleColStatsCollector;
import org.apache.paimon.statistics.TruncateSimpleColStatsCollector;
-import org.apache.paimon.table.SystemFields;
+import org.apache.paimon.table.SpecialFields;
import java.util.List;
@@ -47,7 +47,7 @@ public class StatsCollectorFactories {
.noDefaultValue());
if (fieldMode != null) {
modes[i] = SimpleColStatsCollector.from(fieldMode);
- } else if (SystemFields.isSystemField(field)) {
+ } else if (SpecialFields.isSystemField(field)) {
modes[i] = () -> new TruncateSimpleColStatsCollector(128);
} else {
modes[i] = SimpleColStatsCollector.from(cfg.get(CoreOptions.METADATA_STATS_MODE));
diff --git a/paimon-core/src/test/java/org/apache/paimon/TestKeyValueGenerator.java b/paimon-core/src/test/java/org/apache/paimon/TestKeyValueGenerator.java
index 72e826db7..657c79135 100644
--- a/paimon-core/src/test/java/org/apache/paimon/TestKeyValueGenerator.java
+++ b/paimon-core/src/test/java/org/apache/paimon/TestKeyValueGenerator.java
@@ -30,7 +30,7 @@ import org.apache.paimon.schema.KeyValueFieldsExtractor;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.SchemaEvolutionTableTestBase;
-import org.apache.paimon.table.SystemFields;
+import org.apache.paimon.table.SpecialFields;
import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.BigIntType;
import org.apache.paimon.types.DataField;
@@ -95,9 +95,9 @@ public class TestKeyValueGenerator {
public static final RowType KEY_TYPE =
RowType.of(
new DataField(
- 2 + SystemFields.KEY_FIELD_ID_START, "key_shopId", new IntType(false)),
+ 2 + SpecialFields.KEY_FIELD_ID_START, "key_shopId", new IntType(false)),
new DataField(
- 3 + SystemFields.KEY_FIELD_ID_START,
+ 3 + SpecialFields.KEY_FIELD_ID_START,
"key_orderId",
new BigIntType(false)));
@@ -393,7 +393,7 @@ public class TestKeyValueGenerator {
.map(
f ->
new DataField(
- f.id() + SystemFields.KEY_FIELD_ID_START,
+ f.id() + SpecialFields.KEY_FIELD_ID_START,
"key_" + f.name(),
f.type(),
f.description()))
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/SchemaEvolutionTest.java b/paimon-core/src/test/java/org/apache/paimon/table/SchemaEvolutionTest.java
index 2b20b25c0..ad5c61779 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/SchemaEvolutionTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/SchemaEvolutionTest.java
@@ -57,8 +57,8 @@ import java.util.Map;
import java.util.UUID;
import java.util.function.Consumer;
-import static org.apache.paimon.table.SystemFields.KEY_FIELD_PREFIX;
-import static org.apache.paimon.table.SystemFields.SYSTEM_FIELD_NAMES;
+import static org.apache.paimon.table.SpecialFields.KEY_FIELD_PREFIX;
+import static org.apache.paimon.table.SpecialFields.SYSTEM_FIELD_NAMES;
import static org.apache.paimon.testutils.assertj.PaimonAssertions.anyCauseMatches;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTest.java
index c96db63cb..668d65123 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTest.java
@@ -595,7 +595,7 @@ public class CommitterOperatorTest extends CommitterOperatorTestBase {
table, commit, Committer.createContext("", metricGroup, true, false, null));
committer.commit(Collections.singletonList(manifestCommittable));
CommitterMetrics metrics = committer.getCommitterMetrics();
- assertThat(metrics.getNumBytesOutCounter().getCount()).isEqualTo(529);
+ assertThat(metrics.getNumBytesOutCounter().getCount()).isEqualTo(533);
assertThat(metrics.getNumRecordsOutCounter().getCount()).isEqualTo(2);
committer.close();
}
diff --git a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetSchemaConverter.java b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetSchemaConverter.java
index 43489f9d7..5cf81c375 100644
--- a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetSchemaConverter.java
+++ b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetSchemaConverter.java
@@ -18,8 +18,11 @@
package org.apache.paimon.format.parquet;
+import org.apache.paimon.table.SpecialFields;
import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypeRoot;
import org.apache.paimon.types.DecimalType;
import org.apache.paimon.types.IntType;
import org.apache.paimon.types.LocalZonedTimestampType;
@@ -53,11 +56,7 @@ public class ParquetSchemaConverter {
static final String LIST_ELEMENT_NAME = "element";
public static MessageType convertToParquetMessageType(String name, RowType rowType) {
- Type[] types = new Type[rowType.getFieldCount()];
- for (int i = 0; i < rowType.getFieldCount(); i++) {
- types[i] = convertToParquetType(rowType.getFieldNames().get(i), rowType.getTypeAt(i));
- }
- return new MessageType(name, types);
+ return new MessageType(name, convertToParquetTypes(rowType));
}
public static Type convertToParquetType(String name, DataType type) {
@@ -197,8 +196,47 @@ public class ParquetSchemaConverter {
private static List<Type> convertToParquetTypes(RowType rowType) {
List<Type> types = new ArrayList<>(rowType.getFieldCount());
- for (int i = 0; i < rowType.getFieldCount(); i++) {
- types.add(convertToParquetType(rowType.getFieldNames().get(i), rowType.getTypeAt(i)));
+ for (DataField field : rowType.getFields()) {
+ Type parquetType = convertToParquetType(field.name(), field.type());
+ Type typeWithId = parquetType.withId(field.id());
+ if (field.type().getTypeRoot() == DataTypeRoot.ARRAY) {
+ GroupType groupType = (GroupType) parquetType;
+ GroupType wrapperType = (GroupType) groupType.getFields().get(0);
+ Type elementTypeWithId =
+ wrapperType
+ .getFields()
+ .get(0)
+ .withId(SpecialFields.getArrayElementFieldId(field.id()));
+ typeWithId =
+ ConversionPatterns.listOfElements(
+ groupType.getRepetition(),
+ groupType.getName(),
+ elementTypeWithId)
+ .withId(field.id());
+ } else if (field.type().getTypeRoot() == DataTypeRoot.MAP
+ || field.type().getTypeRoot() == DataTypeRoot.MULTISET) {
+ GroupType groupType = (GroupType) parquetType;
+ GroupType wrapperType = (GroupType) groupType.getFields().get(0);
+ Type keyTypeWithId =
+ wrapperType
+ .getFields()
+ .get(0)
+ .withId(SpecialFields.getMapKeyFieldId(field.id()));
+ Type valueTypeWithId =
+ wrapperType
+ .getFields()
+ .get(1)
+ .withId(SpecialFields.getMapValueFieldId(field.id()));
+ typeWithId =
+ ConversionPatterns.mapType(
+ groupType.getRepetition(),
+ groupType.getName(),
+ MAP_REPEATED_NAME,
+ keyTypeWithId,
+ valueTypeWithId)
+ .withId(field.id());
+ }
+ types.add(typeWithId);
}
return types;
}
diff --git a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetReadWriteTest.java b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetReadWriteTest.java
index 0ccf3fe30..099811d98 100644
--- a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetReadWriteTest.java
+++ b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetReadWriteTest.java
@@ -40,6 +40,7 @@ import org.apache.paimon.types.BigIntType;
import org.apache.paimon.types.BooleanType;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.DecimalType;
import org.apache.paimon.types.DoubleType;
import org.apache.paimon.types.FloatType;
@@ -61,7 +62,13 @@ import org.apache.parquet.hadoop.ParquetFileWriter;
import org.apache.parquet.hadoop.ParquetWriter;
import org.apache.parquet.hadoop.example.ExampleParquetWriter;
import org.apache.parquet.hadoop.util.HadoopOutputFile;
+import org.apache.parquet.schema.ConversionPatterns;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.Type;
+import org.apache.parquet.schema.Types;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.RepeatedTest;
import org.junit.jupiter.api.Test;
@@ -89,6 +96,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -470,6 +478,62 @@ public class ParquetReadWriteTest {
.isInstanceOf(RuntimeException.class);
}
+ @Test
+ public void testConvertToParquetTypeWithId() {
+ List<DataField> nestedFields =
+ Arrays.asList(
+ new DataField(3, "v1", DataTypes.INT()),
+ new DataField(4, "v2", DataTypes.STRING()));
+ List<DataField> fields =
+ Arrays.asList(
+ new DataField(0, "a", DataTypes.INT()),
+ new DataField(1, "b", DataTypes.ARRAY(DataTypes.STRING())),
+ new DataField(
+ 2, "c", DataTypes.MAP(DataTypes.INT(), new RowType(nestedFields))));
+ RowType rowType = new RowType(fields);
+
+ int baseId = 536870911;
+ Type mapValueType =
+ new GroupType(
+ Type.Repetition.OPTIONAL,
+ "value",
+ Types.primitive(INT32, Type.Repetition.OPTIONAL)
+ .named("v1")
+ .withId(3),
+ Types.primitive(
+ PrimitiveType.PrimitiveTypeName.BINARY,
+ Type.Repetition.OPTIONAL)
+ .as(LogicalTypeAnnotation.stringType())
+ .named("v2")
+ .withId(4))
+ .withId(baseId - 2);
+ Type expected =
+ new MessageType(
+ "table",
+ Types.primitive(INT32, Type.Repetition.OPTIONAL).named("a").withId(0),
+ ConversionPatterns.listOfElements(
+ Type.Repetition.OPTIONAL,
+ "b",
+ Types.primitive(
+ PrimitiveType.PrimitiveTypeName.BINARY,
+ Type.Repetition.OPTIONAL)
+ .as(LogicalTypeAnnotation.stringType())
+ .named("element")
+ .withId(baseId + 1))
+ .withId(1),
+ ConversionPatterns.mapType(
+ Type.Repetition.OPTIONAL,
+ "c",
+ "key_value",
+ Types.primitive(INT32, Type.Repetition.REQUIRED)
+ .named("key")
+ .withId(baseId + 2),
+ mapValueType)
+ .withId(2));
+ Type actual = ParquetSchemaConverter.convertToParquetMessageType("table", rowType);
+ assertThat(actual).isEqualTo(expected);
+ }
+
private void innerTestTypes(File folder, List<Integer> records, int rowGroupSize)
throws IOException {
List<InternalRow> rows = records.stream().map(this::newRow).collect(Collectors.toList());