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 7e22e56330 [format] Support format metadata for ORC and Parquet (#8347)
7e22e56330 is described below
commit 7e22e56330af73c3de7747f760fb6a4a8af42d1d
Author: lxy <[email protected]>
AuthorDate: Thu Jun 25 10:42:19 2026 +0800
[format] Support format metadata for ORC and Parquet (#8347)
This is a sub-PR for [shared-shredding MAP write
support](https://cwiki.apache.org/confluence/display/PAIMON/PIP-43%3A+Columnar+Storage+Optimization+for+MAP+Type+in+Paimon).
Shared-shredding needs to attach file-level and per-field metadata, such
as field dictionaries and physical column mappings, before ORC/Parquet
files are finalized. This PR adds a format-level metadata write/read
path so upper layers can pass metadata to format writers and read back
per-field metadata from file schemas.
The per-field metadata is stored through Arrow IPC schema metadata under
the standard `ARROW:schema` key, matching the Arrow schema metadata
representation used by Arrow Parquet/ORC integrations. To avoid
introducing Arrow runtime dependencies into `paimon-format` and
`paimon-bundle`, this PR implements a small Arrow schema metadata
encoder/decoder in `paimon-format` based on the Arrow IPC FlatBuffers
layout. RowType-to-Arrow field conversion is kept in sync with
`paimon-arrow`'s `ArrowUtils`, and compatibility is covered in
`paimon-arrow` tests.
---
paimon-arrow/pom.xml | 7 +
.../ArrowSchemaMetadataCompatibilityTest.java | 113 ++++
paimon-format/pom.xml | 16 +
.../apache/paimon/format/ArrowSchemaMetadata.java | 691 +++++++++++++++++++++
.../apache/paimon/format/FormatMetadataUtils.java | 102 +++
.../paimon/format/SupportsReaderFieldMetadata.java | 35 ++
.../paimon/format/SupportsWriterMetadata.java | 33 +
.../apache/paimon/format/orc/OrcReaderFactory.java | 64 +-
.../paimon/format/orc/writer/OrcBulkWriter.java | 29 +-
.../format/parquet/ParquetWriterFactory.java | 25 +-
.../reader/VectorizedParquetRecordReader.java | 25 +-
.../parquet/writer/MetadataParquetBuilder.java | 37 ++
.../parquet/writer/ParquetMetadataBulkWriter.java | 60 ++
.../parquet/writer/ParquetRowDataBuilder.java | 18 +
.../parquet/writer/RowDataParquetBuilder.java | 13 +-
.../paimon/format/FormatMetadataUtilsTest.java | 148 +++++
.../paimon/format/orc/OrcFormatReadWriteTest.java | 67 ++
.../format/parquet/ParquetFormatReadWriteTest.java | 68 ++
18 files changed, 1536 insertions(+), 15 deletions(-)
diff --git a/paimon-arrow/pom.xml b/paimon-arrow/pom.xml
index 8adc682015..7472dd7b0b 100644
--- a/paimon-arrow/pom.xml
+++ b/paimon-arrow/pom.xml
@@ -41,6 +41,13 @@ under the License.
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-format</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>org.apache.arrow</groupId>
<artifactId>arrow-vector</artifactId>
diff --git
a/paimon-arrow/src/test/java/org/apache/paimon/arrow/ArrowSchemaMetadataCompatibilityTest.java
b/paimon-arrow/src/test/java/org/apache/paimon/arrow/ArrowSchemaMetadataCompatibilityTest.java
new file mode 100644
index 0000000000..165383bc34
--- /dev/null
+++
b/paimon-arrow/src/test/java/org/apache/paimon/arrow/ArrowSchemaMetadataCompatibilityTest.java
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.arrow;
+
+import org.apache.paimon.format.FormatMetadataUtils;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.FieldType;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Compatibility tests between Arrow Java schema serialization and format
metadata utilities. */
+public class ArrowSchemaMetadataCompatibilityTest {
+
+ @Test
+ public void testFormatMetadataCanBeReadByArrowJava() {
+ RowType rowType = rowType();
+ Map<String, String> tagsMetadata = tagsMetadata();
+ Map<String, Map<String, String>> fieldMetadata = new LinkedHashMap<>();
+ fieldMetadata.put("tags", tagsMetadata);
+
+ byte[] schemaBytes =
+ FormatMetadataUtils.buildArrowSchemaMetadata(
+ rowType, fieldMetadata,
FormatMetadataUtils.PARQUET_FIELD_ID_KEY);
+ Schema schema =
Schema.deserializeMessage(ByteBuffer.wrap(schemaBytes));
+
+
assertThat(schema.getFields()).extracting(Field::getName).containsExactly("id",
"tags");
+ assertThat(schema.findField("id").getMetadata())
+ .containsEntry(ArrowUtils.PARQUET_FIELD_ID, "0");
+ assertThat(schema.findField("tags").getMetadata())
+ .containsEntry(ArrowUtils.PARQUET_FIELD_ID, "1")
+ .containsAllEntriesOf(tagsMetadata);
+ }
+
+ @Test
+ public void testArrowJavaSchemaCanBeReadByFormatMetadata() {
+ RowType rowType = rowType();
+ Map<String, String> tagsMetadata = tagsMetadata();
+ List<Field> fields = new ArrayList<>();
+ for (DataField field : rowType.getFields()) {
+ Field arrowField = ArrowUtils.toArrowField(field.name(),
field.id(), field.type(), 0);
+ if ("tags".equals(field.name())) {
+ arrowField = withMetadata(arrowField, tagsMetadata);
+ }
+ fields.add(arrowField);
+ }
+ byte[] schemaBytes = new Schema(fields).serializeAsMessage();
+
+ Map<String, Map<String, String>> metadata =
+ FormatMetadataUtils.readFieldMetadata(schemaBytes);
+
+ assertThat(metadata).containsOnlyKeys("id", "tags");
+
assertThat(metadata.get("id")).containsEntry(ArrowUtils.PARQUET_FIELD_ID, "0");
+ assertThat(metadata.get("tags"))
+ .containsEntry(ArrowUtils.PARQUET_FIELD_ID, "1")
+ .containsAllEntriesOf(tagsMetadata);
+ }
+
+ private static RowType rowType() {
+ return DataTypes.ROW(
+ DataTypes.FIELD(0, "id", DataTypes.INT()),
+ DataTypes.FIELD(1, "tags", DataTypes.MAP(DataTypes.STRING(),
DataTypes.INT())));
+ }
+
+ private static Map<String, String> tagsMetadata() {
+ Map<String, String> metadata = new LinkedHashMap<>();
+ metadata.put("paimon.test.tags", "enabled");
+ metadata.put("paimon.test.version", "1");
+ return metadata;
+ }
+
+ private static Field withMetadata(Field field, Map<String, String>
metadata) {
+ Map<String, String> merged = new LinkedHashMap<>();
+ merged.putAll(metadata);
+ merged.putAll(field.getMetadata());
+ FieldType fieldType = field.getFieldType();
+ return new Field(
+ field.getName(),
+ new FieldType(
+ fieldType.isNullable(),
+ fieldType.getType(),
+ fieldType.getDictionary(),
+ merged),
+ field.getChildren());
+ }
+}
diff --git a/paimon-format/pom.xml b/paimon-format/pom.xml
index b4136797b8..867831fe02 100644
--- a/paimon-format/pom.xml
+++ b/paimon-format/pom.xml
@@ -37,6 +37,7 @@ under the License.
<commons.lang3.version>3.18.0</commons.lang3.version>
<storage-api.version>2.8.1</storage-api.version>
<commons.io.version>2.16.1</commons.io.version>
+ <flatbuffers.version>23.5.26</flatbuffers.version>
</properties>
<dependencies>
@@ -53,6 +54,12 @@ under the License.
<version>${snappy.version}</version>
</dependency>
+ <dependency>
+ <groupId>com.google.flatbuffers</groupId>
+ <artifactId>flatbuffers-java</artifactId>
+ <version>${flatbuffers.version}</version>
+ </dependency>
+
<dependency>
<groupId>com.github.luben</groupId>
<artifactId>zstd-jni</artifactId>
@@ -353,6 +360,9 @@ under the License.
<include>commons-pool:commons-pool</include>
<include>org.locationtech.jts:jts-core</include>
+ <!-- Arrow schema metadata encoding -->
+
<include>com.google.flatbuffers:flatbuffers-java</include>
+
<!-- compress -->
<include>com.github.luben:zstd-jni</include>
</includes>
@@ -417,6 +427,12 @@ under the License.
<shadedPattern>org.apache.paimon.shade.org.locationtech.jts</shadedPattern>
</relocation>
+ <!-- Relocate Arrow schema metadata encoding
dependency. -->
+ <relocation>
+ <pattern>com.google.flatbuffers</pattern>
+
<shadedPattern>org.apache.paimon.shade.com.google.flatbuffers</shadedPattern>
+ </relocation>
+
<!-- Relocate Common. -->
<relocation>
<pattern>org.apache.commons</pattern>
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/ArrowSchemaMetadata.java
b/paimon-format/src/main/java/org/apache/paimon/format/ArrowSchemaMetadata.java
new file mode 100644
index 0000000000..371d6d8593
--- /dev/null
+++
b/paimon-format/src/main/java/org/apache/paimon/format/ArrowSchemaMetadata.java
@@ -0,0 +1,691 @@
+/*
+ * 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;
+
+import org.apache.paimon.data.variant.Variant;
+import org.apache.paimon.table.SpecialFields;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.BigIntType;
+import org.apache.paimon.types.BinaryType;
+import org.apache.paimon.types.BlobType;
+import org.apache.paimon.types.BooleanType;
+import org.apache.paimon.types.CharType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypeVisitor;
+import org.apache.paimon.types.DateType;
+import org.apache.paimon.types.DecimalType;
+import org.apache.paimon.types.DoubleType;
+import org.apache.paimon.types.FloatType;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.LocalZonedTimestampType;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.MultisetType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.SmallIntType;
+import org.apache.paimon.types.TimeType;
+import org.apache.paimon.types.TimestampType;
+import org.apache.paimon.types.TinyIntType;
+import org.apache.paimon.types.VarBinaryType;
+import org.apache.paimon.types.VarCharType;
+import org.apache.paimon.types.VariantType;
+import org.apache.paimon.types.VectorType;
+
+import com.google.flatbuffers.FlatBufferBuilder;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Minimal Arrow IPC schema metadata encoder and decoder used by format
metadata.
+ *
+ * <p>NOTE: The RowType-to-Arrow-field conversion in this class is copied from
{@code
+ * org.apache.paimon.arrow.ArrowUtils} and must be kept in sync with that
class. The IPC
+ * serialization code is a minimal implementation of the public Arrow IPC
FlatBuffers layout used by
+ * {@code ARROW:schema}. It implements only the subset needed by Paimon field
metadata so that
+ * {@code paimon-format} can stay compatible with Arrow metadata without
depending on the Arrow
+ * runtime.
+ */
+class ArrowSchemaMetadata {
+
+ private static final String LIST_DATA_VECTOR_NAME = "$data$";
+ private static final String MAP_DATA_VECTOR_NAME = "entries";
+ private static final String MAP_KEY_NAME = "key";
+ private static final String MAP_VALUE_NAME = "value";
+
+ private static final int CONTINUATION_TOKEN = -1;
+
+ private static final short METADATA_VERSION_V5 = 4;
+ private static final short ENDIANNESS_LITTLE = 0;
+
+ private static final byte MESSAGE_HEADER_SCHEMA = 1;
+
+ private static final byte TYPE_NULL = 1;
+ private static final byte TYPE_INT = 2;
+ private static final byte TYPE_FLOATING_POINT = 3;
+ private static final byte TYPE_BINARY = 4;
+ private static final byte TYPE_UTF8 = 5;
+ private static final byte TYPE_BOOL = 6;
+ private static final byte TYPE_DECIMAL = 7;
+ private static final byte TYPE_DATE = 8;
+ private static final byte TYPE_TIME = 9;
+ private static final byte TYPE_TIMESTAMP = 10;
+ private static final byte TYPE_LIST = 12;
+ private static final byte TYPE_STRUCT = 13;
+ private static final byte TYPE_FIXED_SIZE_LIST = 16;
+ private static final byte TYPE_MAP = 17;
+
+ private static final short PRECISION_SINGLE = 1;
+ private static final short PRECISION_DOUBLE = 2;
+
+ private static final short DATE_UNIT_DAY = 0;
+
+ private static final short TIME_UNIT_SECOND = 0;
+ private static final short TIME_UNIT_MILLISECOND = 1;
+ private static final short TIME_UNIT_MICROSECOND = 2;
+ private static final short TIME_UNIT_NANOSECOND = 3;
+
+ private ArrowSchemaMetadata() {}
+
+ static byte[] serialize(
+ RowType rowType, Map<String, Map<String, String>> fieldMetadata,
String fieldIdKey) {
+ FlatBufferBuilder builder = new FlatBufferBuilder();
+ int schemaOffset =
+ buildSchema(
+ builder,
+ rowType.getFields().stream()
+ .map(
+ field ->
+ withMetadata(
+ toArrowField(
+ field.name(),
+ field.id(),
+ field.type(),
+ 0,
+ fieldIdKey),
+
fieldMetadata.get(field.name())))
+ .collect(Collectors.toList()));
+ int messageOffset =
+ buildMessage(builder, MESSAGE_HEADER_SCHEMA, schemaOffset, 0L,
METADATA_VERSION_V5);
+ builder.finish(messageOffset);
+ return withIpcMessagePrefix(builder.sizedByteArray());
+ }
+
+ /** Reads top-level field metadata from serialized Arrow IPC
schema-message bytes. */
+ static Map<String, Map<String, String>> readFieldMetadata(byte[] bytes) {
+ ByteBuffer buffer = metadataBuffer(bytes);
+ int messageTable = rootTable(buffer);
+ if (readByte(buffer, messageTable, 1, (byte) 0) !=
MESSAGE_HEADER_SCHEMA) {
+ return Collections.emptyMap();
+ }
+ int schemaTable = readTableUnion(buffer, messageTable, 2);
+ if (schemaTable == 0) {
+ return Collections.emptyMap();
+ }
+ int fieldsVector = readVector(buffer, schemaTable, 1);
+ Map<String, Map<String, String>> result = new LinkedHashMap<>();
+ for (int i = 0; i < vectorLength(buffer, fieldsVector); i++) {
+ int fieldTable = vectorTable(buffer, fieldsVector, i);
+ Map<String, String> metadata = readMetadata(buffer, fieldTable);
+ if (!metadata.isEmpty()) {
+ result.put(
+ readString(buffer, fieldTable, 0),
Collections.unmodifiableMap(metadata));
+ }
+ }
+ return Collections.unmodifiableMap(result);
+ }
+
+ private static int buildSchema(FlatBufferBuilder builder, List<ArrowField>
fields) {
+ int[] fieldOffsets = new int[fields.size()];
+ for (int i = 0; i < fields.size(); i++) {
+ fieldOffsets[i] = buildField(builder, fields.get(i));
+ }
+ int fieldsVector = createVectorOfTables(builder, fieldOffsets);
+
+ builder.startTable(4);
+ builder.addShort(0, ENDIANNESS_LITTLE, 0);
+ builder.addOffset(1, fieldsVector, 0);
+ return builder.endTable();
+ }
+
+ private static int buildField(FlatBufferBuilder builder, ArrowField field)
{
+ int name = builder.createString(field.name);
+ int type = buildType(builder, field.type);
+ int[] childOffsets = new int[field.children.size()];
+ for (int i = 0; i < field.children.size(); i++) {
+ childOffsets[i] = buildField(builder, field.children.get(i));
+ }
+ int children = createVectorOfTables(builder, childOffsets);
+ int metadata = buildMetadata(builder, field.metadata);
+
+ builder.startTable(7);
+ builder.addOffset(0, name, 0);
+ builder.addBoolean(1, field.nullable, false);
+ builder.addByte(2, field.type.typeType, 0);
+ builder.addOffset(3, type, 0);
+ builder.addOffset(5, children, 0);
+ builder.addOffset(6, metadata, 0);
+ return builder.endTable();
+ }
+
+ private static int buildMetadata(FlatBufferBuilder builder, Map<String,
String> metadata) {
+ int[] offsets = new int[metadata.size()];
+ int index = 0;
+ for (Map.Entry<String, String> entry : metadata.entrySet()) {
+ int key = builder.createString(entry.getKey());
+ int value = builder.createString(entry.getValue());
+ builder.startTable(2);
+ builder.addOffset(0, key, 0);
+ builder.addOffset(1, value, 0);
+ offsets[index++] = builder.endTable();
+ }
+ return createVectorOfTables(builder, offsets);
+ }
+
+ private static int buildMessage(
+ FlatBufferBuilder builder,
+ byte headerType,
+ int header,
+ long bodyLength,
+ short version) {
+ builder.startTable(5);
+ builder.addShort(0, version, 0);
+ builder.addByte(1, headerType, 0);
+ builder.addOffset(2, header, 0);
+ builder.addLong(3, bodyLength, 0L);
+ return builder.endTable();
+ }
+
+ private static int buildType(FlatBufferBuilder builder, ArrowTypeInfo
type) {
+ switch (type.typeType) {
+ case TYPE_NULL:
+ case TYPE_BINARY:
+ case TYPE_UTF8:
+ case TYPE_BOOL:
+ case TYPE_LIST:
+ case TYPE_STRUCT:
+ builder.startTable(0);
+ return builder.endTable();
+ case TYPE_INT:
+ builder.startTable(2);
+ builder.addInt(0, type.bitWidth, 0);
+ builder.addBoolean(1, type.signed, false);
+ return builder.endTable();
+ case TYPE_FLOATING_POINT:
+ builder.startTable(1);
+ builder.addShort(0, type.precision, 0);
+ return builder.endTable();
+ case TYPE_DECIMAL:
+ builder.startTable(3);
+ builder.addInt(0, type.precisionValue, 0);
+ builder.addInt(1, type.scale, 0);
+ builder.addInt(2, type.bitWidth, 0);
+ return builder.endTable();
+ case TYPE_DATE:
+ builder.startTable(1);
+ builder.addShort(0, DATE_UNIT_DAY, 0);
+ return builder.endTable();
+ case TYPE_TIME:
+ builder.startTable(2);
+ builder.addShort(0, type.unit, 0);
+ builder.addInt(1, type.bitWidth, 0);
+ return builder.endTable();
+ case TYPE_TIMESTAMP:
+ int timezone = type.timezone == null ? 0 :
builder.createString(type.timezone);
+ builder.startTable(2);
+ builder.addShort(0, type.unit, 0);
+ if (timezone != 0) {
+ builder.addOffset(1, timezone, 0);
+ }
+ return builder.endTable();
+ case TYPE_FIXED_SIZE_LIST:
+ builder.startTable(1);
+ builder.addInt(0, type.listSize, 0);
+ return builder.endTable();
+ case TYPE_MAP:
+ builder.startTable(1);
+ builder.addBoolean(0, false, false);
+ return builder.endTable();
+ default:
+ throw new UnsupportedOperationException("Unsupported Arrow
type " + type.typeType);
+ }
+ }
+
+ private static int createVectorOfTables(FlatBufferBuilder builder, int[]
offsets) {
+ builder.startVector(4, offsets.length, 4);
+ for (int i = offsets.length - 1; i >= 0; i--) {
+ builder.addOffset(offsets[i]);
+ }
+ return builder.endVector();
+ }
+
+ private static byte[] withIpcMessagePrefix(byte[] metadata) {
+ int paddedLength = align(metadata.length, 8);
+ ByteBuffer buffer = ByteBuffer.allocate(8 +
paddedLength).order(ByteOrder.LITTLE_ENDIAN);
+ buffer.putInt(CONTINUATION_TOKEN);
+ buffer.putInt(paddedLength);
+ buffer.put(metadata);
+ return buffer.array();
+ }
+
+ private static int align(int value, int alignment) {
+ int remainder = value % alignment;
+ return remainder == 0 ? value : value + alignment - remainder;
+ }
+
+ private static ByteBuffer metadataBuffer(byte[] bytes) {
+ ByteBuffer buffer =
ByteBuffer.wrap(bytes).order(ByteOrder.LITTLE_ENDIAN);
+ if (buffer.remaining() >= 8 && buffer.getInt(0) == CONTINUATION_TOKEN)
{
+ int metadataLength = buffer.getInt(4);
+ ByteBuffer metadata =
buffer.duplicate().order(ByteOrder.LITTLE_ENDIAN);
+ metadata.position(8);
+ metadata.limit(8 + metadataLength);
+ return metadata.slice().order(ByteOrder.LITTLE_ENDIAN);
+ }
+ return buffer;
+ }
+
+ private static int rootTable(ByteBuffer buffer) {
+ return buffer.position() + buffer.getInt(buffer.position());
+ }
+
+ private static int vtable(ByteBuffer buffer, int table) {
+ return table - buffer.getInt(table);
+ }
+
+ private static int fieldOffset(ByteBuffer buffer, int table, int field) {
+ int vtable = vtable(buffer, table);
+ int offset = 4 + field * 2;
+ return offset < buffer.getShort(vtable) ? buffer.getShort(vtable +
offset) : 0;
+ }
+
+ private static byte readByte(ByteBuffer buffer, int table, int field, byte
defaultValue) {
+ int offset = fieldOffset(buffer, table, field);
+ return offset == 0 ? defaultValue : buffer.get(table + offset);
+ }
+
+ private static int readVector(ByteBuffer buffer, int table, int field) {
+ int offset = fieldOffset(buffer, table, field);
+ if (offset == 0) {
+ return 0;
+ }
+ int vectorOffset = table + offset;
+ return vectorOffset + buffer.getInt(vectorOffset);
+ }
+
+ private static int readTableUnion(ByteBuffer buffer, int table, int field)
{
+ int offset = fieldOffset(buffer, table, field);
+ if (offset == 0) {
+ return 0;
+ }
+ int unionOffset = table + offset;
+ return unionOffset + buffer.getInt(unionOffset);
+ }
+
+ private static int vectorLength(ByteBuffer buffer, int vector) {
+ return vector == 0 ? 0 : buffer.getInt(vector);
+ }
+
+ private static int vectorTable(ByteBuffer buffer, int vector, int index) {
+ int element = vector + 4 + index * 4;
+ return element + buffer.getInt(element);
+ }
+
+ private static String readString(ByteBuffer buffer, int table, int field) {
+ int offset = fieldOffset(buffer, table, field);
+ if (offset == 0) {
+ return null;
+ }
+ int stringOffset = table + offset;
+ int string = stringOffset + buffer.getInt(stringOffset);
+ int length = buffer.getInt(string);
+ byte[] bytes = new byte[length];
+ ByteBuffer duplicate = buffer.duplicate();
+ duplicate.position(string + 4);
+ duplicate.get(bytes);
+ return new String(bytes, java.nio.charset.StandardCharsets.UTF_8);
+ }
+
+ private static Map<String, String> readMetadata(ByteBuffer buffer, int
fieldTable) {
+ int metadataVector = readVector(buffer, fieldTable, 6);
+ Map<String, String> metadata = new LinkedHashMap<>();
+ for (int i = 0; i < vectorLength(buffer, metadataVector); i++) {
+ int keyValue = vectorTable(buffer, metadataVector, i);
+ metadata.put(readString(buffer, keyValue, 0), readString(buffer,
keyValue, 1));
+ }
+ return metadata;
+ }
+
+ private static ArrowField withMetadata(ArrowField field, Map<String,
String> metadata) {
+ if (metadata == null || metadata.isEmpty()) {
+ return field;
+ }
+ Map<String, String> result = new LinkedHashMap<>();
+ result.putAll(metadata);
+ result.putAll(field.metadata);
+ return new ArrowField(field.name, field.nullable, field.type,
field.children, result);
+ }
+
+ private static ArrowField toArrowField(
+ String fieldName, int fieldId, DataType dataType, int depth,
String fieldIdKey) {
+ ArrowTypeInfo type = dataType.accept(ArrowFieldTypeVisitor.INSTANCE);
+ Map<String, String> metadata = fieldIdMetadata(fieldId, fieldIdKey);
+ List<ArrowField> children = Collections.emptyList();
+ if (dataType instanceof ArrayType || dataType instanceof VectorType) {
+ DataType elementType =
+ dataType instanceof VectorType
+ ? ((VectorType) dataType).getElementType()
+ : ((ArrayType) dataType).getElementType();
+ ArrowField field =
+ toArrowField(
+ LIST_DATA_VECTOR_NAME, fieldId, elementType, depth
+ 1, fieldIdKey);
+ if (fieldIdKey != null) {
+ field =
+ field.withMetadata(
+ Collections.singletonMap(
+ fieldIdKey,
+ String.valueOf(
+
SpecialFields.getArrayElementFieldId(
+ fieldId, depth + 1))));
+ }
+ children = Collections.singletonList(field);
+ } else if (dataType instanceof MapType) {
+ children =
+ Collections.singletonList(
+ toArrowMapEntryField(fieldId, (MapType) dataType,
depth, fieldIdKey));
+ } else if (dataType instanceof VariantType) {
+ children =
+ Arrays.asList(
+ new ArrowField(
+ Variant.VALUE,
+ false,
+ ArrowTypeInfo.simple(TYPE_BINARY),
+ Collections.emptyList(),
+ Collections.emptyMap()),
+ new ArrowField(
+ Variant.METADATA,
+ false,
+ ArrowTypeInfo.simple(TYPE_BINARY),
+ Collections.emptyList(),
+ Collections.emptyMap()));
+ } else if (dataType instanceof RowType) {
+ RowType rowType = (RowType) dataType;
+ List<ArrowField> rowChildren = new ArrayList<>();
+ for (DataField field : rowType.getFields()) {
+ rowChildren.add(
+ toArrowField(field.name(), field.id(), field.type(),
0, fieldIdKey));
+ }
+ children = rowChildren;
+ }
+ return new ArrowField(fieldName, dataType.isNullable(), type,
children, metadata);
+ }
+
+ private static ArrowField toArrowMapEntryField(
+ int fieldId, MapType mapType, int depth, String fieldIdKey) {
+ ArrowField keyField =
+ toArrowField(
+ MAP_KEY_NAME,
+ fieldId,
+ mapType.getKeyType().notNull(),
+ depth + 1,
+ fieldIdKey);
+ if (fieldIdKey != null) {
+ keyField =
+ keyField.withMetadata(
+ Collections.singletonMap(
+ fieldIdKey,
+ String.valueOf(
+
SpecialFields.getMapKeyFieldId(fieldId, depth + 1))));
+ }
+
+ ArrowField valueField =
+ toArrowField(
+ MAP_VALUE_NAME, fieldId, mapType.getValueType(), depth
+ 1, fieldIdKey);
+ if (fieldIdKey != null) {
+ valueField =
+ valueField.withMetadata(
+ Collections.singletonMap(
+ fieldIdKey,
+ String.valueOf(
+
SpecialFields.getMapValueFieldId(fieldId, depth + 1))));
+ }
+
+ return new ArrowField(
+ MAP_DATA_VECTOR_NAME,
+ false,
+ ArrowTypeInfo.simple(TYPE_STRUCT),
+ Arrays.asList(keyField, valueField),
+ fieldIdMetadata(fieldId, fieldIdKey));
+ }
+
+ private static Map<String, String> fieldIdMetadata(int fieldId, String
fieldIdKey) {
+ return fieldIdKey == null
+ ? Collections.emptyMap()
+ : Collections.singletonMap(fieldIdKey,
String.valueOf(fieldId));
+ }
+
+ private static class ArrowField {
+ private final String name;
+ private final boolean nullable;
+ private final ArrowTypeInfo type;
+ private final List<ArrowField> children;
+ private final Map<String, String> metadata;
+
+ private ArrowField(
+ String name,
+ boolean nullable,
+ ArrowTypeInfo type,
+ List<ArrowField> children,
+ Map<String, String> metadata) {
+ this.name = name;
+ this.nullable = nullable;
+ this.type = type;
+ this.children = children;
+ this.metadata = metadata;
+ }
+
+ private ArrowField withMetadata(Map<String, String> metadata) {
+ return new ArrowField(name, nullable, type, children, metadata);
+ }
+ }
+
+ private static class ArrowTypeInfo {
+ private final byte typeType;
+ private int bitWidth;
+ private boolean signed;
+ private short precision;
+ private int precisionValue;
+ private int scale;
+ private short unit;
+ private String timezone;
+ private int listSize;
+
+ private ArrowTypeInfo(byte typeType) {
+ this.typeType = typeType;
+ }
+
+ private static ArrowTypeInfo simple(byte typeType) {
+ return new ArrowTypeInfo(typeType);
+ }
+ }
+
+ private static class ArrowFieldTypeVisitor implements
DataTypeVisitor<ArrowTypeInfo> {
+
+ private static final ArrowFieldTypeVisitor INSTANCE = new
ArrowFieldTypeVisitor();
+
+ @Override
+ public ArrowTypeInfo visit(CharType charType) {
+ return ArrowTypeInfo.simple(TYPE_UTF8);
+ }
+
+ @Override
+ public ArrowTypeInfo visit(VarCharType varCharType) {
+ return ArrowTypeInfo.simple(TYPE_UTF8);
+ }
+
+ @Override
+ public ArrowTypeInfo visit(BooleanType booleanType) {
+ return ArrowTypeInfo.simple(TYPE_BOOL);
+ }
+
+ @Override
+ public ArrowTypeInfo visit(BinaryType binaryType) {
+ return ArrowTypeInfo.simple(TYPE_BINARY);
+ }
+
+ @Override
+ public ArrowTypeInfo visit(VarBinaryType varBinaryType) {
+ return ArrowTypeInfo.simple(TYPE_BINARY);
+ }
+
+ @Override
+ public ArrowTypeInfo visit(DecimalType decimalType) {
+ ArrowTypeInfo type = ArrowTypeInfo.simple(TYPE_DECIMAL);
+ type.precisionValue = decimalType.getPrecision();
+ type.scale = decimalType.getScale();
+ type.bitWidth = 128;
+ return type;
+ }
+
+ @Override
+ public ArrowTypeInfo visit(TinyIntType tinyIntType) {
+ return integer(8);
+ }
+
+ @Override
+ public ArrowTypeInfo visit(SmallIntType smallIntType) {
+ return integer(16);
+ }
+
+ @Override
+ public ArrowTypeInfo visit(IntType intType) {
+ return integer(32);
+ }
+
+ @Override
+ public ArrowTypeInfo visit(BigIntType bigIntType) {
+ return integer(64);
+ }
+
+ @Override
+ public ArrowTypeInfo visit(FloatType floatType) {
+ ArrowTypeInfo type = ArrowTypeInfo.simple(TYPE_FLOATING_POINT);
+ type.precision = PRECISION_SINGLE;
+ return type;
+ }
+
+ @Override
+ public ArrowTypeInfo visit(DoubleType doubleType) {
+ ArrowTypeInfo type = ArrowTypeInfo.simple(TYPE_FLOATING_POINT);
+ type.precision = PRECISION_DOUBLE;
+ return type;
+ }
+
+ @Override
+ public ArrowTypeInfo visit(DateType dateType) {
+ return ArrowTypeInfo.simple(TYPE_DATE);
+ }
+
+ @Override
+ public ArrowTypeInfo visit(TimeType timeType) {
+ ArrowTypeInfo type = ArrowTypeInfo.simple(TYPE_TIME);
+ type.unit = TIME_UNIT_MILLISECOND;
+ type.bitWidth = 32;
+ return type;
+ }
+
+ @Override
+ public ArrowTypeInfo visit(TimestampType timestampType) {
+ ArrowTypeInfo type = ArrowTypeInfo.simple(TYPE_TIMESTAMP);
+ type.unit = getTimeUnit(timestampType.getPrecision());
+ return type;
+ }
+
+ @Override
+ public ArrowTypeInfo visit(LocalZonedTimestampType
localZonedTimestampType) {
+ ArrowTypeInfo type = ArrowTypeInfo.simple(TYPE_TIMESTAMP);
+ type.unit = getTimeUnit(localZonedTimestampType.getPrecision());
+ type.timezone = "UTC";
+ return type;
+ }
+
+ @Override
+ public ArrowTypeInfo visit(VariantType variantType) {
+ return ArrowTypeInfo.simple(TYPE_STRUCT);
+ }
+
+ @Override
+ public ArrowTypeInfo visit(BlobType blobType) {
+ throw new UnsupportedOperationException("Doesn't support
BlobType.");
+ }
+
+ @Override
+ public ArrowTypeInfo visit(ArrayType arrayType) {
+ return ArrowTypeInfo.simple(TYPE_LIST);
+ }
+
+ @Override
+ public ArrowTypeInfo visit(VectorType vectorType) {
+ ArrowTypeInfo type = ArrowTypeInfo.simple(TYPE_FIXED_SIZE_LIST);
+ type.listSize = vectorType.getLength();
+ return type;
+ }
+
+ @Override
+ public ArrowTypeInfo visit(MultisetType multisetType) {
+ throw new UnsupportedOperationException("Doesn't support
MultisetType.");
+ }
+
+ @Override
+ public ArrowTypeInfo visit(MapType mapType) {
+ return ArrowTypeInfo.simple(TYPE_MAP);
+ }
+
+ @Override
+ public ArrowTypeInfo visit(RowType rowType) {
+ return ArrowTypeInfo.simple(TYPE_STRUCT);
+ }
+
+ private ArrowTypeInfo integer(int bitWidth) {
+ ArrowTypeInfo type = ArrowTypeInfo.simple(TYPE_INT);
+ type.bitWidth = bitWidth;
+ type.signed = true;
+ return type;
+ }
+
+ private short getTimeUnit(int precision) {
+ if (precision == 0) {
+ return TIME_UNIT_SECOND;
+ } else if (precision >= 1 && precision <= 3) {
+ return TIME_UNIT_MILLISECOND;
+ } else if (precision >= 4 && precision <= 6) {
+ return TIME_UNIT_MICROSECOND;
+ } else {
+ return TIME_UNIT_NANOSECOND;
+ }
+ }
+ }
+}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/FormatMetadataUtils.java
b/paimon-format/src/main/java/org/apache/paimon/format/FormatMetadataUtils.java
new file mode 100644
index 0000000000..342f1ad442
--- /dev/null
+++
b/paimon-format/src/main/java/org/apache/paimon/format/FormatMetadataUtils.java
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.format;
+
+import org.apache.paimon.types.RowType;
+
+import javax.annotation.Nullable;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/** Utilities for format metadata encoded at file boundaries. */
+public class FormatMetadataUtils {
+
+ public static final String ARROW_SCHEMA_METADATA_KEY = "ARROW:schema";
+ public static final String PARQUET_FIELD_ID_KEY = "PARQUET:field_id";
+
+ private FormatMetadataUtils() {}
+
+ /**
+ * Encodes raw metadata values as base64 strings so they can be stored in
format key-value
+ * metadata.
+ */
+ public static Map<String, String> encodeMetadata(Map<String, byte[]>
metadata) {
+ Map<String, String> encoded = new LinkedHashMap<>();
+ for (Map.Entry<String, byte[]> entry : metadata.entrySet()) {
+ encoded.put(entry.getKey(),
Base64.getEncoder().encodeToString(entry.getValue()));
+ }
+ return encoded;
+ }
+
+ /**
+ * Decodes base64-encoded metadata values. Values that are not valid
base64 are returned as
+ * UTF-8 bytes.
+ */
+ public static Map<String, byte[]> decodeMetadata(Map<String, String>
metadata) {
+ Map<String, byte[]> decoded = new LinkedHashMap<>();
+ for (Map.Entry<String, String> entry : metadata.entrySet()) {
+ try {
+ decoded.put(entry.getKey(),
Base64.getDecoder().decode(entry.getValue()));
+ } catch (IllegalArgumentException e) {
+ decoded.put(entry.getKey(),
entry.getValue().getBytes(StandardCharsets.UTF_8));
+ }
+ }
+ return decoded;
+ }
+
+ /**
+ * Builds serialized Arrow schema metadata from a Paimon row type and
injects metadata into
+ * top-level fields.
+ *
+ * <p>The keys of {@code fieldMetadata} are top-level field names. Nested
fields are converted
+ * from the {@link RowType} but do not receive metadata from this map. If
injected metadata
+ * conflicts with metadata produced during Arrow conversion, the Arrow
conversion metadata wins
+ * to preserve format-specific field information. Set {@code fieldIdKey}
to the field id
+ * metadata key used by the target format, or {@code null} if field id
metadata should not be
+ * written.
+ */
+ public static byte[] buildArrowSchemaMetadata(
+ RowType rowType,
+ Map<String, Map<String, String>> fieldMetadata,
+ @Nullable String fieldIdKey) {
+ return ArrowSchemaMetadata.serialize(rowType, fieldMetadata,
fieldIdKey);
+ }
+
+ /**
+ * Reads field metadata from serialized Arrow schema metadata.
+ *
+ * <p>The returned map contains top-level fields only, keyed by field
name. If the input is
+ * {@code null} or cannot be parsed as an Arrow schema message, this
method returns an empty
+ * map. Fields without any metadata are omitted from the returned map.
+ */
+ public static Map<String, Map<String, String>> readFieldMetadata(@Nullable
byte[] schemaBytes) {
+ if (schemaBytes == null) {
+ return Collections.emptyMap();
+ }
+ try {
+ return ArrowSchemaMetadata.readFieldMetadata(schemaBytes);
+ } catch (RuntimeException e) {
+ return Collections.emptyMap();
+ }
+ }
+}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/SupportsReaderFieldMetadata.java
b/paimon-format/src/main/java/org/apache/paimon/format/SupportsReaderFieldMetadata.java
new file mode 100644
index 0000000000..0dc9f191fb
--- /dev/null
+++
b/paimon-format/src/main/java/org/apache/paimon/format/SupportsReaderFieldMetadata.java
@@ -0,0 +1,35 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.util.Map;
+
+/** Reader capability for formats that can recover top-level field metadata. */
+public interface SupportsReaderFieldMetadata {
+
+ /**
+ * Reads metadata from top-level file fields.
+ *
+ * <p>The returned map is keyed by field name. Each value contains the
metadata key-value pairs
+ * attached to that field. Implementations return an empty map when the
file does not contain
+ * field metadata.
+ */
+ Map<String, Map<String, String>> readFieldMetadata() throws IOException;
+}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/SupportsWriterMetadata.java
b/paimon-format/src/main/java/org/apache/paimon/format/SupportsWriterMetadata.java
new file mode 100644
index 0000000000..06aa89029b
--- /dev/null
+++
b/paimon-format/src/main/java/org/apache/paimon/format/SupportsWriterMetadata.java
@@ -0,0 +1,33 @@
+/*
+ * 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;
+
+import java.util.Map;
+
+/** Writer capability for formats that can attach key-value metadata before
close. */
+public interface SupportsWriterMetadata {
+
+ /**
+ * Adds file-level metadata to the writer.
+ *
+ * <p>The metadata values are raw bytes. Format implementations are
responsible for converting
+ * them to the physical file metadata representation.
+ */
+ void addMetadata(Map<String, byte[]> metadata);
+}
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 b1de74242a..63f07ad152 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,8 +24,10 @@ 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.format.FormatMetadataUtils;
import org.apache.paimon.format.FormatReaderFactory;
import org.apache.paimon.format.OrcFormatReaderContext;
+import org.apache.paimon.format.SupportsReaderFieldMetadata;
import org.apache.paimon.format.fs.HadoopReadOnlyFileSystem;
import org.apache.paimon.format.orc.filter.OrcFilters;
import org.apache.paimon.fs.FileIO;
@@ -54,7 +56,10 @@ import org.apache.orc.impl.RecordReaderImpl;
import javax.annotation.Nullable;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
import java.util.List;
+import java.util.Map;
import static org.apache.paimon.format.orc.OrcTypeUtil.convertToOrcSchema;
import static
org.apache.paimon.format.orc.reader.AbstractOrcColumnVector.createPaimonVector;
@@ -104,7 +109,7 @@ public class OrcReaderFactory implements
FormatReaderFactory {
Pool<OrcReaderBatch> poolOfBatches =
createPoolOfBatches(context.filePath(), poolSize,
context.fileIO());
- RecordReader orcReader =
+ OrcRecordReader orcReader =
createRecordReader(
hadoopConfig,
schema,
@@ -224,12 +229,14 @@ public class OrcReaderFactory implements
FormatReaderFactory {
* batch is addressed by the starting row number of the batch, plus the
number of records to be
* skipped before.
*/
- private static final class OrcVectorizedReader implements
FileRecordReader<InternalRow> {
+ private static final class OrcVectorizedReader
+ implements FileRecordReader<InternalRow>,
SupportsReaderFieldMetadata {
- private final RecordReader orcReader;
+ private final OrcRecordReader orcReader;
private final Pool<OrcReaderBatch> pool;
- private OrcVectorizedReader(final RecordReader orcReader, final
Pool<OrcReaderBatch> pool) {
+ private OrcVectorizedReader(
+ final OrcRecordReader orcReader, final Pool<OrcReaderBatch>
pool) {
this.orcReader = checkNotNull(orcReader, "orcReader");
this.pool = checkNotNull(pool, "pool");
}
@@ -240,8 +247,8 @@ public class OrcReaderFactory implements
FormatReaderFactory {
final OrcReaderBatch batch = getCachedEntry();
final VectorizedRowBatch orcVectorBatch =
batch.orcVectorizedRowBatch();
- long rowNumber = orcReader.getRowNumber();
- if (!nextBatch(orcReader, orcVectorBatch)) {
+ long rowNumber = orcReader.recordReader.getRowNumber();
+ if (!nextBatch(orcReader.recordReader, orcVectorBatch)) {
batch.recycle();
return null;
}
@@ -249,9 +256,37 @@ public class OrcReaderFactory implements
FormatReaderFactory {
return batch.convertAndGetIterator(orcVectorBatch, rowNumber);
}
+ @Override
+ public Map<String, Map<String, String>> readFieldMetadata() {
+ org.apache.orc.Reader fileReader = orcReader.fileReader;
+ if (!fileReader
+ .getMetadataKeys()
+ .contains(FormatMetadataUtils.ARROW_SCHEMA_METADATA_KEY)) {
+ return Collections.emptyMap();
+ }
+ String encodedSchema =
+ StandardCharsets.UTF_8
+ .decode(
+ fileReader
+ .getMetadataValue(
+
FormatMetadataUtils.ARROW_SCHEMA_METADATA_KEY)
+ .duplicate())
+ .toString();
+ return FormatMetadataUtils.readFieldMetadata(
+ FormatMetadataUtils.decodeMetadata(
+ Collections.singletonMap(
+
FormatMetadataUtils.ARROW_SCHEMA_METADATA_KEY,
+ encodedSchema))
+
.get(FormatMetadataUtils.ARROW_SCHEMA_METADATA_KEY));
+ }
+
@Override
public void close() throws IOException {
- orcReader.close();
+ try {
+ orcReader.recordReader.close();
+ } finally {
+ orcReader.fileReader.close();
+ }
}
private OrcReaderBatch getCachedEntry() throws IOException {
@@ -264,7 +299,18 @@ public class OrcReaderFactory implements
FormatReaderFactory {
}
}
- private static RecordReader createRecordReader(
+ private static final class OrcRecordReader {
+
+ private final org.apache.orc.Reader fileReader;
+ private final RecordReader recordReader;
+
+ private OrcRecordReader(org.apache.orc.Reader fileReader, RecordReader
recordReader) {
+ this.fileReader = fileReader;
+ this.recordReader = recordReader;
+ }
+ }
+
+ private static OrcRecordReader createRecordReader(
org.apache.hadoop.conf.Configuration conf,
TypeDescription schema,
List<OrcFilters.Predicate> conjunctPredicates,
@@ -314,7 +360,7 @@ public class OrcReaderFactory implements
FormatReaderFactory {
// assign ids
schema.getId();
- return orcRowsReader;
+ return new OrcRecordReader(orcReader, orcRowsReader);
} catch (IOException e) {
// exception happened, we need to close the reader
IOUtils.closeQuietly(orcReader);
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/orc/writer/OrcBulkWriter.java
b/paimon-format/src/main/java/org/apache/paimon/format/orc/writer/OrcBulkWriter.java
index c44e3f26d6..9655e193d6 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/orc/writer/OrcBulkWriter.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/orc/writer/OrcBulkWriter.java
@@ -20,7 +20,9 @@ package org.apache.paimon.format.orc.writer;
import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.format.FormatMetadataUtils;
import org.apache.paimon.format.FormatWriter;
+import org.apache.paimon.format.SupportsWriterMetadata;
import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.options.MemorySize;
@@ -28,19 +30,27 @@ import
org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
import org.apache.orc.Writer;
import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
+import static org.apache.paimon.utils.Preconditions.checkState;
/** A {@link FormatWriter} implementation that writes data in ORC format. */
-public class OrcBulkWriter implements FormatWriter {
+public class OrcBulkWriter implements FormatWriter, SupportsWriterMetadata {
private final Writer writer;
private final Vectorizer<InternalRow> vectorizer;
private final VectorizedRowBatch rowBatch;
private final PositionOutputStream underlyingStream;
+ private final Map<String, byte[]> metadata;
private long currentBatchMemoryUsage = 0;
private final long memoryLimit;
+ private boolean closed = false;
public OrcBulkWriter(
Vectorizer<InternalRow> vectorizer,
@@ -54,6 +64,16 @@ public class OrcBulkWriter implements FormatWriter {
this.rowBatch = vectorizer.getSchema().createRowBatch(batchSize);
this.underlyingStream = underlyingStream;
this.memoryLimit = memoryLimit.getBytes();
+ this.metadata = new HashMap<>();
+ }
+
+ @Override
+ public void addMetadata(Map<String, byte[]> metadata) {
+ checkState(!closed, "Cannot add metadata after writer is closed.");
+ for (Map.Entry<String, byte[]> entry : metadata.entrySet()) {
+ this.metadata.put(
+ entry.getKey(), Arrays.copyOf(entry.getValue(),
entry.getValue().length));
+ }
}
@Override
@@ -74,7 +94,14 @@ public class OrcBulkWriter implements FormatWriter {
@Override
public void close() throws IOException {
+ this.closed = true;
flush();
+ for (Map.Entry<String, String> entry :
+ FormatMetadataUtils.encodeMetadata(metadata).entrySet()) {
+ writer.addUserMetadata(
+ entry.getKey(),
+
ByteBuffer.wrap(entry.getValue().getBytes(StandardCharsets.UTF_8)));
+ }
writer.close();
}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetWriterFactory.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetWriterFactory.java
index 282805897a..a26a0f3d1a 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetWriterFactory.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetWriterFactory.java
@@ -22,8 +22,10 @@ import org.apache.paimon.data.InternalRow;
import org.apache.paimon.format.FormatWriter;
import org.apache.paimon.format.FormatWriterFactory;
import org.apache.paimon.format.HadoopCompressionType;
+import org.apache.paimon.format.parquet.writer.MetadataParquetBuilder;
import org.apache.paimon.format.parquet.writer.ParquetBuilder;
import org.apache.paimon.format.parquet.writer.ParquetBulkWriter;
+import org.apache.paimon.format.parquet.writer.ParquetMetadataBulkWriter;
import org.apache.paimon.format.parquet.writer.RowDataParquetBuilder;
import org.apache.paimon.format.parquet.writer.StreamOutputFile;
import org.apache.paimon.format.variant.SupportsVariantInference;
@@ -34,6 +36,8 @@ import org.apache.parquet.hadoop.ParquetWriter;
import org.apache.parquet.io.OutputFile;
import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
/** A factory that creates a Parquet {@link FormatWriter}. */
public class ParquetWriterFactory implements FormatWriterFactory,
SupportsVariantInference {
@@ -57,6 +61,11 @@ public class ParquetWriterFactory implements
FormatWriterFactory, SupportsVarian
compression = null;
}
+ if (writerBuilder instanceof MetadataParquetBuilder) {
+ return createMetadataWriter(
+ (MetadataParquetBuilder<InternalRow>) writerBuilder, out,
compression);
+ }
+
final ParquetWriter<InternalRow> writer =
writerBuilder.createWriter(out, compression);
return new ParquetBulkWriter(writer);
}
@@ -70,10 +79,20 @@ public class ParquetWriterFactory implements
FormatWriterFactory, SupportsVarian
compression = null;
}
- ParquetBuilder<InternalRow> newBuilder =
+ RowDataParquetBuilder newBuilder =
((RowDataParquetBuilder) writerBuilder)
.withShreddingSchemas(inferredShreddingSchema);
- final ParquetWriter<InternalRow> writer = newBuilder.createWriter(out,
compression);
- return new ParquetBulkWriter(writer);
+ return createMetadataWriter(newBuilder, out, compression);
+ }
+
+ private FormatWriter createMetadataWriter(
+ MetadataParquetBuilder<InternalRow> builder, OutputFile out,
String compression)
+ throws IOException {
+ // Keep this exact map instance shared by ParquetBulkWriter and
WriteSupport. The writer
+ // collects metadata before close, and WriteSupport reads it when
finalizing the footer.
+ Map<String, byte[]> metadata = new HashMap<>();
+ final ParquetWriter<InternalRow> writer =
+ builder.createWriter(out, compression, () -> metadata);
+ return new ParquetMetadataBulkWriter(writer, metadata);
}
}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/VectorizedParquetRecordReader.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/VectorizedParquetRecordReader.java
index af63a7d9f9..ebc38e31a7 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/VectorizedParquetRecordReader.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/VectorizedParquetRecordReader.java
@@ -20,6 +20,8 @@ package org.apache.paimon.format.parquet.reader;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.columnar.writable.WritableColumnVector;
+import org.apache.paimon.format.FormatMetadataUtils;
+import org.apache.paimon.format.SupportsReaderFieldMetadata;
import org.apache.paimon.format.parquet.type.ParquetField;
import org.apache.paimon.format.parquet.type.ParquetPrimitiveField;
import org.apache.paimon.fs.FileIO;
@@ -39,8 +41,10 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
@@ -48,7 +52,8 @@ import static java.lang.String.format;
import static
org.apache.paimon.format.parquet.reader.ParquetReaderUtil.createReadableColumnVectors;
/** Record reader for parquet. */
-public class VectorizedParquetRecordReader implements
FileRecordReader<InternalRow> {
+public class VectorizedParquetRecordReader
+ implements FileRecordReader<InternalRow>, SupportsReaderFieldMetadata {
private ParquetFileReader reader;
@@ -269,6 +274,24 @@ public class VectorizedParquetRecordReader implements
FileRecordReader<InternalR
}
}
+ @Override
+ public Map<String, Map<String, String>> readFieldMetadata() throws
IOException {
+ String encodedSchema =
+ reader.getFooter()
+ .getFileMetaData()
+ .getKeyValueMetaData()
+ .get(FormatMetadataUtils.ARROW_SCHEMA_METADATA_KEY);
+ if (encodedSchema == null) {
+ return FormatMetadataUtils.readFieldMetadata(null);
+ }
+ return FormatMetadataUtils.readFieldMetadata(
+ FormatMetadataUtils.decodeMetadata(
+ Collections.singletonMap(
+
FormatMetadataUtils.ARROW_SCHEMA_METADATA_KEY,
+ encodedSchema))
+ .get(FormatMetadataUtils.ARROW_SCHEMA_METADATA_KEY));
+ }
+
@Override
public void close() throws IOException {
if (reader != null) {
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/MetadataParquetBuilder.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/MetadataParquetBuilder.java
new file mode 100644
index 0000000000..463414163d
--- /dev/null
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/MetadataParquetBuilder.java
@@ -0,0 +1,37 @@
+/*
+ * 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.parquet.writer;
+
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.io.OutputFile;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.function.Supplier;
+
+/** A {@link ParquetBuilder} that can provide writer metadata before the file
is finalized. */
+public interface MetadataParquetBuilder<T> extends ParquetBuilder<T> {
+
+ /**
+ * Creates and configures a parquet writer that can read metadata before
finalizing the file.
+ */
+ ParquetWriter<T> createWriter(
+ OutputFile out, String compression, Supplier<Map<String, byte[]>>
metadataSupplier)
+ throws IOException;
+}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/ParquetMetadataBulkWriter.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/ParquetMetadataBulkWriter.java
new file mode 100644
index 0000000000..a1d24cf2ed
--- /dev/null
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/ParquetMetadataBulkWriter.java
@@ -0,0 +1,60 @@
+/*
+ * 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.parquet.writer;
+
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.format.SupportsWriterMetadata;
+
+import org.apache.parquet.hadoop.ParquetWriter;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+import static org.apache.paimon.utils.Preconditions.checkState;
+
+/** A {@link ParquetBulkWriter} that supports adding file metadata before
close. */
+public class ParquetMetadataBulkWriter extends ParquetBulkWriter implements
SupportsWriterMetadata {
+
+ private final Map<String, byte[]> metadata;
+
+ private boolean closed = false;
+
+ public ParquetMetadataBulkWriter(
+ ParquetWriter<InternalRow> parquetWriter, Map<String, byte[]>
metadata) {
+ super(parquetWriter);
+ this.metadata = checkNotNull(metadata, "metadata");
+ }
+
+ @Override
+ public void addMetadata(Map<String, byte[]> metadata) {
+ checkState(!closed, "Cannot add metadata after writer is closed.");
+ for (Map.Entry<String, byte[]> entry : metadata.entrySet()) {
+ this.metadata.put(
+ entry.getKey(), Arrays.copyOf(entry.getValue(),
entry.getValue().length));
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ this.closed = true;
+ super.close();
+ }
+}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/ParquetRowDataBuilder.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/ParquetRowDataBuilder.java
index 14970e548e..60fd1097fe 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/ParquetRowDataBuilder.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/ParquetRowDataBuilder.java
@@ -19,12 +19,14 @@
package org.apache.paimon.format.parquet.writer;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.format.FormatMetadataUtils;
import org.apache.paimon.format.parquet.VariantUtils;
import org.apache.paimon.types.RowType;
import org.apache.hadoop.conf.Configuration;
import org.apache.parquet.hadoop.ParquetWriter;
import org.apache.parquet.hadoop.api.WriteSupport;
+import org.apache.parquet.hadoop.api.WriteSupport.FinalizedWriteContext;
import org.apache.parquet.io.OutputFile;
import org.apache.parquet.io.api.RecordConsumer;
import org.apache.parquet.schema.MessageType;
@@ -32,6 +34,8 @@ import org.apache.parquet.schema.MessageType;
import javax.annotation.Nullable;
import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Supplier;
import static
org.apache.paimon.format.parquet.ParquetSchemaConverter.convertToParquetMessageType;
@@ -41,12 +45,20 @@ public class ParquetRowDataBuilder
private final RowType rowType;
@Nullable private final RowType shreddingSchemas;
+ private Supplier<Map<String, byte[]>> metadataSupplier;
public ParquetRowDataBuilder(
OutputFile path, RowType rowType, @Nullable RowType
shreddingSchemas) {
super(path);
this.rowType = rowType;
this.shreddingSchemas = shreddingSchemas;
+ this.metadataSupplier = HashMap::new;
+ }
+
+ public ParquetRowDataBuilder withMetadataSupplier(
+ Supplier<Map<String, byte[]>> metadataSupplier) {
+ this.metadataSupplier = metadataSupplier;
+ return this;
}
@Override
@@ -89,5 +101,11 @@ public class ParquetRowDataBuilder
public void write(InternalRow record) {
this.writer.write(record);
}
+
+ @Override
+ public FinalizedWriteContext finalizeWrite() {
+ return new FinalizedWriteContext(
+
FormatMetadataUtils.encodeMetadata(metadataSupplier.get()));
+ }
}
}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/RowDataParquetBuilder.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/RowDataParquetBuilder.java
index 2e84df0932..6239fc546b 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/RowDataParquetBuilder.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/RowDataParquetBuilder.java
@@ -35,9 +35,12 @@ import org.apache.parquet.io.OutputFile;
import javax.annotation.Nullable;
import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Supplier;
/** A {@link ParquetBuilder} for {@link InternalRow}. */
-public class RowDataParquetBuilder implements ParquetBuilder<InternalRow> {
+public class RowDataParquetBuilder implements
MetadataParquetBuilder<InternalRow> {
private final RowType rowType;
private final Configuration conf;
@@ -58,8 +61,16 @@ public class RowDataParquetBuilder implements
ParquetBuilder<InternalRow> {
@Override
public ParquetWriter<InternalRow> createWriter(OutputFile out, String
compression)
throws IOException {
+ return createWriter(out, compression, HashMap::new);
+ }
+
+ @Override
+ public ParquetWriter<InternalRow> createWriter(
+ OutputFile out, String compression, Supplier<Map<String, byte[]>>
metadataSupplier)
+ throws IOException {
ParquetRowDataBuilder builder =
new ParquetRowDataBuilder(out, rowType, shreddingSchemas)
+ .withMetadataSupplier(metadataSupplier)
.withConf(conf)
.withCompressionCodec(getCompressionCodec(getCompression(compression)))
.withRowGroupSize(
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/FormatMetadataUtilsTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/FormatMetadataUtilsTest.java
new file mode 100644
index 0000000000..f01854f9e3
--- /dev/null
+++
b/paimon-format/src/test/java/org/apache/paimon/format/FormatMetadataUtilsTest.java
@@ -0,0 +1,148 @@
+/*
+ * 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;
+
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link FormatMetadataUtils}. */
+public class FormatMetadataUtilsTest {
+
+ @Test
+ public void testEncodeAndDecodeMetadata() {
+ Map<String, byte[]> metadata = new LinkedHashMap<>();
+ metadata.put("encoded",
"paimon-value".getBytes(StandardCharsets.UTF_8));
+
+ Map<String, String> encoded =
FormatMetadataUtils.encodeMetadata(metadata);
+ encoded.put("plain", "plain-value");
+ assertThat(encoded)
+ .containsEntry(
+ "encoded",
+ Base64.getEncoder()
+
.encodeToString("paimon-value".getBytes(StandardCharsets.UTF_8)));
+
+ Map<String, byte[]> decoded =
FormatMetadataUtils.decodeMetadata(encoded);
+
+ assertThat(new String(decoded.get("encoded"), StandardCharsets.UTF_8))
+ .isEqualTo("paimon-value");
+ assertThat(new String(decoded.get("plain"), StandardCharsets.UTF_8))
+ .isEqualTo("plain-value");
+ }
+
+ @Test
+ public void testReadFieldMetadataFromArrowSchemaMetadata() {
+ RowType rowType = DataTypes.ROW(DataTypes.FIELD(0, "field",
DataTypes.STRING()));
+ Map<String, String> fieldMetadata = new LinkedHashMap<>();
+ fieldMetadata.put("paimon.test.field-key", "field-value");
+ Map<String, Map<String, String>> expected = new LinkedHashMap<>();
+ expected.put("field", fieldMetadata);
+ byte[] schemaBytes =
FormatMetadataUtils.buildArrowSchemaMetadata(rowType, expected, null);
+
+
assertThat(FormatMetadataUtils.readFieldMetadata(schemaBytes).get("field"))
+ .containsAllEntriesOf(fieldMetadata);
+ assertThat(FormatMetadataUtils.readFieldMetadata(null)).isEmpty();
+ assertThat(
+ FormatMetadataUtils.readFieldMetadata(
+
"not-arrow-schema".getBytes(StandardCharsets.UTF_8)))
+ .isEmpty();
+ }
+
+ @Test
+ public void testReadFieldMetadata() {
+ RowType rowType =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "with_metadata", DataTypes.INT()),
+ DataTypes.FIELD(1, "without_metadata",
DataTypes.INT()));
+ Map<String, String> fieldMetadata = new LinkedHashMap<>();
+ fieldMetadata.put("paimon.test.field-key", "field-value");
+ Map<String, Map<String, String>> expected = new LinkedHashMap<>();
+ expected.put("with_metadata", fieldMetadata);
+ byte[] schemaBytes =
FormatMetadataUtils.buildArrowSchemaMetadata(rowType, expected, null);
+
+ Map<String, Map<String, String>> metadata =
+ FormatMetadataUtils.readFieldMetadata(schemaBytes);
+
assertThat(metadata.get("with_metadata")).containsAllEntriesOf(fieldMetadata);
+ assertThat(metadata).doesNotContainKey("without_metadata");
+ }
+
+ @Test
+ public void testBuildArrowSchemaWithFieldMetadata() {
+ RowType rowType =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "id", DataTypes.INT()),
+ DataTypes.FIELD(
+ 1, "tags", DataTypes.MAP(DataTypes.STRING(),
DataTypes.INT())),
+ DataTypes.FIELD(
+ 2,
+ "nested",
+ DataTypes.ROW(
+ DataTypes.FIELD(3, "name",
DataTypes.STRING()),
+ DataTypes.FIELD(
+ 4, "scores",
DataTypes.ARRAY(DataTypes.INT())))));
+ Map<String, String> tagsMetadata = new LinkedHashMap<>();
+ tagsMetadata.put("paimon.test.tags", "enabled");
+
+ Map<String, Map<String, String>> fieldMetadata = new LinkedHashMap<>();
+ fieldMetadata.put("tags", tagsMetadata);
+
+ byte[] schemaBytes =
+ FormatMetadataUtils.buildArrowSchemaMetadata(
+ rowType, fieldMetadata,
FormatMetadataUtils.PARQUET_FIELD_ID_KEY);
+
+ Map<String, Map<String, String>> metadata =
+ FormatMetadataUtils.readFieldMetadata(schemaBytes);
+ assertThat(metadata).containsOnlyKeys("id", "tags", "nested");
+
assertThat(metadata.get("id")).containsEntry(FormatMetadataUtils.PARQUET_FIELD_ID_KEY,
"0");
+ assertThat(metadata.get("tags")).containsAllEntriesOf(tagsMetadata);
+ assertThat(metadata.get("tags"))
+ .containsEntry(FormatMetadataUtils.PARQUET_FIELD_ID_KEY, "1");
+
assertThat(metadata.get("nested")).doesNotContainKey("paimon.test.tags");
+ }
+
+ @Test
+ public void testBuildArrowSchemaWithoutFieldIdMetadata() {
+ RowType rowType =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "id", DataTypes.INT()),
+ DataTypes.FIELD(1, "name", DataTypes.STRING()));
+ Map<String, String> nameMetadata = new LinkedHashMap<>();
+ nameMetadata.put("paimon.test.name", "enabled");
+ Map<String, Map<String, String>> fieldMetadata = new LinkedHashMap<>();
+ fieldMetadata.put("name", nameMetadata);
+
+ byte[] schemaBytes =
+ FormatMetadataUtils.buildArrowSchemaMetadata(rowType,
fieldMetadata, null);
+
+ Map<String, Map<String, String>> metadata =
+ FormatMetadataUtils.readFieldMetadata(schemaBytes);
+ assertThat(metadata).containsOnlyKeys("name");
+ assertThat(metadata.get("name")).containsAllEntriesOf(nameMetadata);
+ assertThat(metadata.get("name"))
+ .doesNotContainKey(FormatMetadataUtils.PARQUET_FIELD_ID_KEY);
+ }
+}
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/orc/OrcFormatReadWriteTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/orc/OrcFormatReadWriteTest.java
index fb625c68da..a03af8dc25 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/orc/OrcFormatReadWriteTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/orc/OrcFormatReadWriteTest.java
@@ -24,12 +24,16 @@ import org.apache.paimon.data.Timestamp;
import org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.format.FileFormat;
import org.apache.paimon.format.FileFormatFactory;
+import org.apache.paimon.format.FormatMetadataUtils;
import org.apache.paimon.format.FormatReadWriteTest;
import org.apache.paimon.format.FormatReaderContext;
import org.apache.paimon.format.FormatWriter;
import org.apache.paimon.format.OrcOptions;
+import org.apache.paimon.format.SupportsReaderFieldMetadata;
+import org.apache.paimon.format.SupportsWriterMetadata;
import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.options.Options;
+import org.apache.paimon.reader.FileRecordReader;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
@@ -37,14 +41,19 @@ import org.apache.paimon.types.RowType;
import org.junit.jupiter.api.Test;
import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.TimeZone;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** An orc {@link FormatReadWriteTest}. */
public class OrcFormatReadWriteTest extends FormatReadWriteTest {
@@ -81,6 +90,64 @@ public class OrcFormatReadWriteTest extends
FormatReadWriteTest {
super("orc");
}
+ @Test
+ public void testWriteMetadata() throws IOException {
+ RowType rowType =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "id", DataTypes.INT()),
+ DataTypes.FIELD(1, "name", DataTypes.STRING()));
+
+ PositionOutputStream out = fileIO.newOutputStream(file, false);
+ FormatWriter writer =
newFormat.createWriterFactory(rowType).create(out, "zstd");
+ Map<String, String> fieldMetadata = new HashMap<>();
+ fieldMetadata.put("paimon.test.field-key", "field-value");
+ fieldMetadata.put("paimon.test.field-version", "1");
+ Map<String, Map<String, String>> fieldMetadataByName = new HashMap<>();
+ fieldMetadataByName.put("name", fieldMetadata);
+ byte[] arrowSchemaBytes =
+ FormatMetadataUtils.buildArrowSchemaMetadata(
+ rowType, fieldMetadataByName,
OrcTypeUtil.PAIMON_ORC_FIELD_ID_KEY);
+ Map<String, byte[]> metadata = new HashMap<>();
+ metadata.put("paimon.test.key",
"paimon-test-value".getBytes(StandardCharsets.UTF_8));
+ metadata.put(FormatMetadataUtils.ARROW_SCHEMA_METADATA_KEY,
arrowSchemaBytes);
+ ((SupportsWriterMetadata) writer).addMetadata(metadata);
+ writer.addElement(GenericRow.of(1,
org.apache.paimon.data.BinaryString.fromString("one")));
+ writer.close();
+ assertThatThrownBy(() -> ((SupportsWriterMetadata)
writer).addMetadata(metadata))
+ .isInstanceOf(IllegalStateException.class);
+ out.close();
+
+ try (org.apache.orc.Reader reader =
+ OrcReaderFactory.createReader(
+ new org.apache.hadoop.conf.Configuration(false),
fileIO, file, null)) {
+ ByteBuffer value = reader.getMetadataValue("paimon.test.key");
+ Map<String, byte[]> decodedMetadata =
+ FormatMetadataUtils.decodeMetadata(
+ Collections.singletonMap(
+ "paimon.test.key",
+
StandardCharsets.UTF_8.decode(value.duplicate()).toString()));
+ assertThat(new String(decodedMetadata.get("paimon.test.key"),
StandardCharsets.UTF_8))
+ .isEqualTo("paimon-test-value");
+ }
+
+ FormatReaderContext context =
+ new FormatReaderContext(fileIO, file,
fileIO.getFileSize(file));
+ RowType emptyRowType = new RowType(Collections.emptyList());
+ try (FileRecordReader<InternalRow> reader =
+ newFormat
+ .createReaderFactory(emptyRowType, emptyRowType,
Collections.emptyList())
+ .createReader(context)) {
+ Map<String, Map<String, String>> readFieldMetadata =
+ ((SupportsReaderFieldMetadata) reader).readFieldMetadata();
+ assertThat(readFieldMetadata).containsOnlyKeys("id", "name");
+ assertThat(readFieldMetadata.get("id"))
+ .containsEntry(OrcTypeUtil.PAIMON_ORC_FIELD_ID_KEY, "0");
+
assertThat(readFieldMetadata.get("name")).containsAllEntriesOf(fieldMetadata);
+ assertThat(readFieldMetadata.get("name"))
+ .containsEntry(OrcTypeUtil.PAIMON_ORC_FIELD_ID_KEY, "1");
+ }
+ }
+
@Test
public void testTimestampLTZWithLegacyWriteAndRead() throws IOException {
RowType rowType =
DataTypes.ROW(DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE());
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFormatReadWriteTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFormatReadWriteTest.java
index 137da29aa6..e4d03b8afc 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFormatReadWriteTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFormatReadWriteTest.java
@@ -20,12 +20,18 @@ package org.apache.paimon.format.parquet;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
import org.apache.paimon.format.FileFormat;
import org.apache.paimon.format.FileFormatFactory;
+import org.apache.paimon.format.FormatMetadataUtils;
import org.apache.paimon.format.FormatReadWriteTest;
+import org.apache.paimon.format.FormatReaderContext;
import org.apache.paimon.format.FormatWriter;
+import org.apache.paimon.format.SupportsReaderFieldMetadata;
+import org.apache.paimon.format.SupportsWriterMetadata;
import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.options.Options;
+import org.apache.paimon.reader.FileRecordReader;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
@@ -40,6 +46,8 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -58,6 +66,66 @@ public class ParquetFormatReadWriteTest extends
FormatReadWriteTest {
new FileFormatFactory.FormatContext(new Options(), 1024,
1024));
}
+ @Test
+ public void testWriteMetadata() throws Exception {
+ ParquetFileFormat format =
+ new ParquetFileFormat(
+ new FileFormatFactory.FormatContext(new Options(),
1024, 1024));
+ RowType rowType =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "id", DataTypes.INT()),
+ DataTypes.FIELD(1, "name", DataTypes.STRING()));
+
+ PositionOutputStream out = fileIO.newOutputStream(file, false);
+ FormatWriter writer = format.createWriterFactory(rowType).create(out,
"zstd");
+ Map<String, String> fieldMetadata = new HashMap<>();
+ fieldMetadata.put("paimon.test.field-key", "field-value");
+ fieldMetadata.put("paimon.test.field-version", "1");
+ Map<String, Map<String, String>> fieldMetadataByName = new HashMap<>();
+ fieldMetadataByName.put("name", fieldMetadata);
+ byte[] arrowSchemaBytes =
+ FormatMetadataUtils.buildArrowSchemaMetadata(
+ rowType, fieldMetadataByName,
FormatMetadataUtils.PARQUET_FIELD_ID_KEY);
+ Map<String, byte[]> metadata = new HashMap<>();
+ metadata.put("paimon.test.key",
"paimon-test-value".getBytes(StandardCharsets.UTF_8));
+ metadata.put(FormatMetadataUtils.ARROW_SCHEMA_METADATA_KEY,
arrowSchemaBytes);
+ ((SupportsWriterMetadata) writer).addMetadata(metadata);
+ writer.addElement(GenericRow.of(1, BinaryString.fromString("one")));
+ writer.close();
+ Assertions.assertThatThrownBy(() -> ((SupportsWriterMetadata)
writer).addMetadata(metadata))
+ .isInstanceOf(IllegalStateException.class);
+ out.close();
+
+ try (ParquetFileReader reader =
+ ParquetUtil.getParquetReader(
+ fileIO, file, fileIO.getFileSize(file), new
Options())) {
+ Map<String, String> fileMetadata =
+ reader.getFooter().getFileMetaData().getKeyValueMetaData();
+ Map<String, byte[]> decodedMetadata =
FormatMetadataUtils.decodeMetadata(fileMetadata);
+ Assertions.assertThat(
+ new String(
+ decodedMetadata.get("paimon.test.key"),
StandardCharsets.UTF_8))
+ .isEqualTo("paimon-test-value");
+ }
+
+ FormatReaderContext context =
+ new FormatReaderContext(fileIO, file,
fileIO.getFileSize(file));
+ RowType emptyRowType = new RowType(Collections.emptyList());
+ try (FileRecordReader<InternalRow> reader =
+ format.createReaderFactory(emptyRowType, emptyRowType,
Collections.emptyList())
+ .createReader(context)) {
+ Map<String, Map<String, String>> readFieldMetadata =
+ ((SupportsReaderFieldMetadata) reader).readFieldMetadata();
+
Assertions.assertThat(readFieldMetadata).containsKey("id").containsKey("name");
+ Assertions.assertThat(readFieldMetadata.get("id"))
+ .containsEntry(FormatMetadataUtils.PARQUET_FIELD_ID_KEY,
"0");
+ Assertions.assertThat(readFieldMetadata.get("name"))
+ .containsAllEntriesOf(fieldMetadata);
+ Assertions.assertThat(readFieldMetadata.get("name"))
+ .containsEntry(FormatMetadataUtils.PARQUET_FIELD_ID_KEY,
"1");
+ }
+ }
+
@ParameterizedTest
@ValueSource(booleans = {true, false})
public void testEnableBloomFilter(boolean enabled) throws Exception {
