Skip to content

[format] Support narrowing Parquet reads #8505

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
leaves12138 wants to merge 1 commit into
base: master
Choose a base branch
from
+161 −2
Open

[format] Support narrowing Parquet reads #8505

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
85 changes: 83 additions & 2 deletions ...at/src/main/java/org/apache/paimon/format/parquet/reader/ParquetVectorUpdaterFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -340,6 +352,40 @@ public void decodeSingleDictionaryId(
}
}

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(
Expand Down Expand Up @@ -682,6 +728,41 @@ public void decodeSingleDictionaryId(
}
}

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(
Expand Down
78 changes: 78 additions & 0 deletions ...t/src/test/java/org/apache/paimon/format/parquet/reader/FileTypeNotMatchReadTypeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<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";
Expand Down
Loading