From 3a48366e01c382616e7fa4d6a4e03d8d43e3b31e Mon Sep 17 00:00:00 2001 From: YeJunHao <41894543+leaves12138@users.noreply.github.com> Date: Wed, 8 Jul 2026 14:27:57 +0800 Subject: [PATCH] [format] Support narrowing Parquet reads --- .../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 fdf3ecdc3066..afa60ff070f8 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 UpdaterFactory visit(SmallIntType smallIntType) { @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 UpdaterFactory visit(BigIntType bigIntType) { @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 void decodeSingleDictionaryId( } } + private static class IntegerFromLongUpdater implements ParquetVectorUpdater { + @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 { @Override public void readValues( @@ -682,6 +728,41 @@ public void decodeSingleDictionaryId( } } + private static class FloatFromDoubleUpdater + implements ParquetVectorUpdater { + @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 { @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 7afcbacc9d7a..964f4a21eb89 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 void testDecimal() throws Exception { } } + @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 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 fileRecordReader = + parquetReaderFactory.createReader( + new FormatReaderContext( + LocalFileIO.create(), + new org.apache.paimon.fs.Path(tempDir.toString(), fileName), + file.length())); + + FileRecordIterator 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 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 fileRecordReader = + parquetReaderFactory.createReader( + new FormatReaderContext( + LocalFileIO.create(), + new org.apache.paimon.fs.Path(tempDir.toString(), fileName), + file.length())); + + FileRecordIterator 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";