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 1379eefb21 [format] Support narrowing Parquet reads (#8505)
1379eefb21 is described below
commit 1379eefb21f7f7647720e57d318d85fdb015dbc4
Author: YeJunHao <[email protected]>
AuthorDate: Mon Jul 13 11:29:32 2026 +0800
[format] Support narrowing Parquet reads (#8505)
Some externally produced Parquet files can store values with a wider
physical type than the Paimon table schema expects. For example, an
`INT` field may be stored as Parquet `INT64` and a `FLOAT` field may be
stored as Parquet `DOUBLE`. When such columns are dictionary encoded,
the vectorized reader currently selects the updater only from the Paimon
logical type and calls methods such as `decodeToInt`, which fails for
dictionaries such as `PlainLongDictionary`.
---
.../reader/ParquetVectorUpdaterFactory.java | 85 +++++++++++++++++++++-
.../reader/FileTypeNotMatchReadTypeTest.java | 78 ++++++++++++++++++++
2 files changed, 161 insertions(+), 2 deletions(-)
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/ParquetVectorUpdaterFactory.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/ParquetVectorUpdaterFactory.java
index fdf3ecdc30..afa60ff070 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/ParquetVectorUpdaterFactory.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/ParquetVectorUpdaterFactory.java
@@ -157,7 +157,13 @@ public class ParquetVectorUpdaterFactory {
@Override
public UpdaterFactory visit(IntType intType) {
- return c -> new IntegerUpdater();
+ return c -> {
+ if (c.getPrimitiveType().getPrimitiveTypeName()
+ == PrimitiveType.PrimitiveTypeName.INT64) {
+ return new IntegerFromLongUpdater();
+ }
+ return new IntegerUpdater();
+ };
}
@Override
@@ -167,7 +173,13 @@ public class ParquetVectorUpdaterFactory {
@Override
public UpdaterFactory visit(FloatType floatType) {
- return c -> new FloatUpdater();
+ return c -> {
+ if (c.getPrimitiveType().getPrimitiveTypeName()
+ == PrimitiveType.PrimitiveTypeName.DOUBLE) {
+ return new FloatFromDoubleUpdater();
+ }
+ return new FloatUpdater();
+ };
}
@Override
@@ -340,6 +352,40 @@ public class ParquetVectorUpdaterFactory {
}
}
+ private static class IntegerFromLongUpdater implements
ParquetVectorUpdater<WritableIntVector> {
+ @Override
+ public void readValues(
+ int total,
+ int offset,
+ WritableIntVector values,
+ VectorizedValuesReader valuesReader) {
+ for (int i = 0; i < total; i++) {
+ values.setInt(offset + i,
Math.toIntExact(valuesReader.readLong()));
+ }
+ }
+
+ @Override
+ public void skipValues(int total, VectorizedValuesReader valuesReader)
{
+ valuesReader.skipLongs(total);
+ }
+
+ @Override
+ public void readValue(
+ int offset, WritableIntVector values, VectorizedValuesReader
valuesReader) {
+ values.setInt(offset, Math.toIntExact(valuesReader.readLong()));
+ }
+
+ @Override
+ public void decodeSingleDictionaryId(
+ int offset,
+ WritableIntVector values,
+ WritableIntVector dictionaryIds,
+ Dictionary dictionary) {
+ values.setInt(
+ offset,
Math.toIntExact(dictionary.decodeToLong(dictionaryIds.getInt(offset))));
+ }
+ }
+
private static class ByteUpdater implements
ParquetVectorUpdater<WritableByteVector> {
@Override
public void readValues(
@@ -682,6 +728,41 @@ public class ParquetVectorUpdaterFactory {
}
}
+ private static class FloatFromDoubleUpdater
+ implements ParquetVectorUpdater<WritableFloatVector> {
+ @Override
+ public void readValues(
+ int total,
+ int offset,
+ WritableFloatVector values,
+ VectorizedValuesReader valuesReader) {
+ for (int i = 0; i < total; i++) {
+ values.setFloat(offset + i, (float) valuesReader.readDouble());
+ }
+ }
+
+ @Override
+ public void skipValues(int total, VectorizedValuesReader valuesReader)
{
+ valuesReader.skipDoubles(total);
+ }
+
+ @Override
+ public void readValue(
+ int offset, WritableFloatVector values, VectorizedValuesReader
valuesReader) {
+ values.setFloat(offset, (float) valuesReader.readDouble());
+ }
+
+ @Override
+ public void decodeSingleDictionaryId(
+ int offset,
+ WritableFloatVector values,
+ WritableIntVector dictionaryIds,
+ Dictionary dictionary) {
+ values.setFloat(
+ offset, (float)
dictionary.decodeToDouble(dictionaryIds.getInt(offset)));
+ }
+ }
+
private static class DoubleUpdater implements
ParquetVectorUpdater<WritableDoubleVector> {
@Override
public void readValues(
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/FileTypeNotMatchReadTypeTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/FileTypeNotMatchReadTypeTest.java
index 7afcbacc9d..964f4a21eb 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/FileTypeNotMatchReadTypeTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/FileTypeNotMatchReadTypeTest.java
@@ -159,6 +159,84 @@ public class FileTypeNotMatchReadTypeTest {
}
}
+ @Test
+ public void testReadIntFromInt64() throws Exception {
+ String fileName = "test.parquet";
+ String fileWholePath = tempDir + "/" + fileName;
+
+ RowType rowTypeWrite = RowType.of(new DataField(0, "int_col",
DataTypes.BIGINT()));
+ RowType rowTypeRead = RowType.of(new DataField(0, "int_col",
DataTypes.INT()));
+ MessageType messageType =
Util.convertToParquetMessageType(rowTypeWrite);
+ ParquetRowDataBuilderForTest parquetRowDataBuilder =
+ new ParquetRowDataBuilderForTest(
+ new LocalOutputFile(new
File(fileWholePath).toPath()),
+ rowTypeWrite,
+ messageType)
+ .enableDictionaryEncoding();
+ ParquetWriter<InternalRow> parquetWriter =
parquetRowDataBuilder.build();
+
+ for (int i = 0; i < 100; i++) {
+ parquetWriter.write(GenericRow.of((long) i));
+ }
+ parquetWriter.close();
+
+ ParquetReaderFactory parquetReaderFactory =
+ new ParquetReaderFactory(new Options(), rowTypeRead, 100,
null);
+
+ File file = new File(fileWholePath);
+ FileRecordReader<InternalRow> fileRecordReader =
+ parquetReaderFactory.createReader(
+ new FormatReaderContext(
+ LocalFileIO.create(),
+ new
org.apache.paimon.fs.Path(tempDir.toString(), fileName),
+ file.length()));
+
+ FileRecordIterator<InternalRow> batch = fileRecordReader.readBatch();
+ for (int i = 0; i < 100; i++) {
+ assertThat(batch.next().getInt(0)).isEqualTo(i);
+ }
+ file.delete();
+ }
+
+ @Test
+ public void testReadFloatFromDouble() throws Exception {
+ String fileName = "test.parquet";
+ String fileWholePath = tempDir + "/" + fileName;
+
+ RowType rowTypeWrite = RowType.of(new DataField(0, "float_col",
DataTypes.DOUBLE()));
+ RowType rowTypeRead = RowType.of(new DataField(0, "float_col",
DataTypes.FLOAT()));
+ MessageType messageType =
Util.convertToParquetMessageType(rowTypeWrite);
+ ParquetRowDataBuilderForTest parquetRowDataBuilder =
+ new ParquetRowDataBuilderForTest(
+ new LocalOutputFile(new
File(fileWholePath).toPath()),
+ rowTypeWrite,
+ messageType)
+ .enableDictionaryEncoding();
+ ParquetWriter<InternalRow> parquetWriter =
parquetRowDataBuilder.build();
+
+ for (int i = 0; i < 100; i++) {
+ parquetWriter.write(GenericRow.of(i + 0.5D));
+ }
+ parquetWriter.close();
+
+ ParquetReaderFactory parquetReaderFactory =
+ new ParquetReaderFactory(new Options(), rowTypeRead, 100,
null);
+
+ File file = new File(fileWholePath);
+ FileRecordReader<InternalRow> fileRecordReader =
+ parquetReaderFactory.createReader(
+ new FormatReaderContext(
+ LocalFileIO.create(),
+ new
org.apache.paimon.fs.Path(tempDir.toString(), fileName),
+ file.length()));
+
+ FileRecordIterator<InternalRow> batch = fileRecordReader.readBatch();
+ for (int i = 0; i < 100; i++) {
+ assertThat(batch.next().getFloat(0)).isEqualTo(i + 0.5F);
+ }
+ file.delete();
+ }
+
@Test
public void testArray() throws Exception {
String fileName = "test.parquet";
