Baunsgaard commented on code in PR #2528:
URL: https://github.com/apache/systemds/pull/2528#discussion_r3567032745
##########
src/main/java/org/apache/sysds/runtime/io/FrameReaderParquet.java:
##########
@@ -89,65 +97,147 @@ public FrameBlock readFrameFromHDFS(String fname,
ValueType[] schema, String[] n
* @param clen The expected number of columns.
*/
protected void readParquetFrameFromHDFS(Path path, Configuration conf,
FrameBlock dest, ValueType[] schema, long rlen, long clen) throws IOException {
- // Retrieve schema from Parquet footer
- ParquetMetadata metadata =
ParquetFileReader.open(HadoopInputFile.fromPath(path, conf)).getFooter();
- MessageType parquetSchema =
metadata.getFileMetaData().getSchema();
+ int row = readSingleParquetFile(path, conf, dest, clen, 0);
- // Map column names to Parquet schema indices
- String[] columnNames = dest.getColumnNames();
- int[] columnIndices = new int[columnNames.length];
- for (int i = 0; i < columnNames.length; i++) {
- columnIndices[i] =
parquetSchema.getFieldIndex(columnNames[i]);
+ // Check frame dimensions
+ if (row != rlen) {
+ throw new IOException("Mismatch in row count: expected
" + rlen + ", but got " + row);
}
+ }
+
+ // Constants for decoding legacy INT96 timestamps
+ private static final int JULIAN_EPOCH_OFFSET_DAYS = 2_440_588;
+ private static final long MILLIS_IN_DAY = TimeUnit.DAYS.toMillis(1);
+ private static final long NANOS_PER_MILLISECOND =
TimeUnit.MILLISECONDS.toNanos(1);
+
+ /**
+ * Reads a single Parquet file into the destination FrameBlock using
the column API.
+ * Iterates row groups; within each row group reads each requested
column and writes the
+ * value into the FrameBlock.
+ *
+ * @param path The HDFS path to the Parquet file.
+ * @param conf The Hadoop configuration.
+ * @param dest The FrameBlock to populate.
+ * @param clen The number of columns.
+ * @param rowOffset The starting row offset in the destination
FrameBlock.
+ * @return The number of rows read.
+ */
+ protected int readSingleParquetFile(Path path, Configuration conf,
FrameBlock dest,
+ long clen, long rowOffset) throws IOException
+ {
+ String[] columnNames = dest.getColumnNames();
+
+ try (ParquetFileReader reader =
ParquetFileReader.open(HadoopInputFile.fromPath(path, conf))) {
+ ParquetMetadata metadata = reader.getFooter();
+ MessageType parquetSchema =
metadata.getFileMetaData().getSchema();
+ String createdBy =
metadata.getFileMetaData().getCreatedBy();
+
+ // Map each requested frame column (by name) to its
Parquet column descriptor.
+ ColumnDescriptor[] descriptors = new
ColumnDescriptor[(int) clen];
+ for (int col = 0; col < clen; col++) {
+ ColumnDescriptor desc =
parquetSchema.getColumnDescription(new String[]{ columnNames[col] });
+ // Nested columns cannot be represented by a
flat FrameBlock column
+ if (desc.getMaxRepetitionLevel() > 0)
+ throw new IOException("Nested Parquet
columns are not supported: " + columnNames[col]);
+ descriptors[col] = desc;
+ }
+
+ // no-op converter tree, only used by
ColumnReadStoreImpl to resolve a converter per column
+ GroupConverter rootConverter =
newNoOpRootConverter(parquetSchema);
+
+ int row = (int) rowOffset;
+ PageReadStore pages;
+
+ while (true) {
+ pages = reader.readNextRowGroup();
- // Read data usind ParquetReader
- try (ParquetReader<Group> rowReader = ParquetReader.builder(new
GroupReadSupport(), path)
- .withConf(conf)
- .build()) {
+ if (pages == null)
+ break;
+
+ int rowsInGroup = (int) pages.getRowCount();
+ ColumnReadStoreImpl colStore = new
ColumnReadStoreImpl(pages, rootConverter, parquetSchema, createdBy);
- Group group;
- int row = 0;
- while ((group = rowReader.read()) != null) {
for (int col = 0; col < clen; col++) {
- int colIndex = columnIndices[col];
- if
(group.getFieldRepetitionCount(colIndex) > 0) {
- PrimitiveType.PrimitiveTypeName
type =
parquetSchema.getType(columnNames[col]).asPrimitiveType().getPrimitiveTypeName();
- switch (type) {
- case INT32:
- dest.set(row,
col, group.getInteger(colIndex, 0));
- break;
- case INT64:
- dest.set(row,
col, group.getLong(colIndex, 0));
- break;
- case FLOAT:
- dest.set(row,
col, group.getFloat(colIndex, 0));
- break;
- case DOUBLE:
- dest.set(row,
col, group.getDouble(colIndex, 0));
- break;
- case BOOLEAN:
- dest.set(row,
col, group.getBoolean(colIndex, 0));
- break;
- case BINARY:
- dest.set(row,
col, group.getBinary(colIndex, 0).toStringUsingUTF8());
- break;
- default:
- throw new
IOException("Unsupported data type: " + type);
- }
- } else {
- dest.set(row, col, null);
- }
+ ColumnDescriptor desc =
descriptors[col];
+ ColumnReader creader =
colStore.getColumnReader(desc);
+ int maxDef =
desc.getMaxDefinitionLevel();
+ PrimitiveType.PrimitiveTypeName ptype =
desc.getPrimitiveType().getPrimitiveTypeName();
+ readColumn(dest, creader, col, row,
rowsInGroup, maxDef, ptype);
}
- row++;
+ row += rowsInGroup;
}
+ return row - (int) rowOffset;
+ }
+ }
- // Check frame dimensions
- if (row != rlen) {
- throw new IOException("Mismatch in row count:
expected " + rlen + ", but got " + row);
+ /**
+ * Reads one column of a row group, writing each value (or null) into
the destination FrameBlock.
+ */
+ private void readColumn(FrameBlock dest, ColumnReader creader, int col,
int rowStart,
+ int rowsInGroup, int maxDef, PrimitiveType.PrimitiveTypeName
ptype) throws IOException
+ {
+ for (int i = 0; i < rowsInGroup; i++) {
+ int row = rowStart + i;
+ if (creader.getCurrentDefinitionLevel() == maxDef) {
+ switch (ptype) {
+ case INT32:
+ dest.set(row, col,
creader.getInteger());
Review Comment:
Instead of going though the FrameBlock interface, we should directly
allocate and use underlying Arrays. Just like the Delta Pr you just made.
##########
src/main/java/org/apache/sysds/runtime/io/FrameReaderParquet.java:
##########
@@ -89,65 +97,147 @@ public FrameBlock readFrameFromHDFS(String fname,
ValueType[] schema, String[] n
* @param clen The expected number of columns.
*/
protected void readParquetFrameFromHDFS(Path path, Configuration conf,
FrameBlock dest, ValueType[] schema, long rlen, long clen) throws IOException {
- // Retrieve schema from Parquet footer
- ParquetMetadata metadata =
ParquetFileReader.open(HadoopInputFile.fromPath(path, conf)).getFooter();
- MessageType parquetSchema =
metadata.getFileMetaData().getSchema();
+ int row = readSingleParquetFile(path, conf, dest, clen, 0);
- // Map column names to Parquet schema indices
- String[] columnNames = dest.getColumnNames();
- int[] columnIndices = new int[columnNames.length];
- for (int i = 0; i < columnNames.length; i++) {
- columnIndices[i] =
parquetSchema.getFieldIndex(columnNames[i]);
+ // Check frame dimensions
+ if (row != rlen) {
+ throw new IOException("Mismatch in row count: expected
" + rlen + ", but got " + row);
}
+ }
+
+ // Constants for decoding legacy INT96 timestamps
+ private static final int JULIAN_EPOCH_OFFSET_DAYS = 2_440_588;
+ private static final long MILLIS_IN_DAY = TimeUnit.DAYS.toMillis(1);
+ private static final long NANOS_PER_MILLISECOND =
TimeUnit.MILLISECONDS.toNanos(1);
+
+ /**
+ * Reads a single Parquet file into the destination FrameBlock using
the column API.
+ * Iterates row groups; within each row group reads each requested
column and writes the
+ * value into the FrameBlock.
+ *
+ * @param path The HDFS path to the Parquet file.
+ * @param conf The Hadoop configuration.
+ * @param dest The FrameBlock to populate.
+ * @param clen The number of columns.
+ * @param rowOffset The starting row offset in the destination
FrameBlock.
+ * @return The number of rows read.
+ */
+ protected int readSingleParquetFile(Path path, Configuration conf,
FrameBlock dest,
+ long clen, long rowOffset) throws IOException
+ {
+ String[] columnNames = dest.getColumnNames();
+
+ try (ParquetFileReader reader =
ParquetFileReader.open(HadoopInputFile.fromPath(path, conf))) {
+ ParquetMetadata metadata = reader.getFooter();
+ MessageType parquetSchema =
metadata.getFileMetaData().getSchema();
+ String createdBy =
metadata.getFileMetaData().getCreatedBy();
+
+ // Map each requested frame column (by name) to its
Parquet column descriptor.
+ ColumnDescriptor[] descriptors = new
ColumnDescriptor[(int) clen];
+ for (int col = 0; col < clen; col++) {
+ ColumnDescriptor desc =
parquetSchema.getColumnDescription(new String[]{ columnNames[col] });
+ // Nested columns cannot be represented by a
flat FrameBlock column
+ if (desc.getMaxRepetitionLevel() > 0)
+ throw new IOException("Nested Parquet
columns are not supported: " + columnNames[col]);
+ descriptors[col] = desc;
+ }
+
+ // no-op converter tree, only used by
ColumnReadStoreImpl to resolve a converter per column
+ GroupConverter rootConverter =
newNoOpRootConverter(parquetSchema);
+
+ int row = (int) rowOffset;
+ PageReadStore pages;
+
+ while (true) {
+ pages = reader.readNextRowGroup();
- // Read data usind ParquetReader
- try (ParquetReader<Group> rowReader = ParquetReader.builder(new
GroupReadSupport(), path)
- .withConf(conf)
- .build()) {
+ if (pages == null)
+ break;
+
+ int rowsInGroup = (int) pages.getRowCount();
+ ColumnReadStoreImpl colStore = new
ColumnReadStoreImpl(pages, rootConverter, parquetSchema, createdBy);
- Group group;
- int row = 0;
- while ((group = rowReader.read()) != null) {
for (int col = 0; col < clen; col++) {
- int colIndex = columnIndices[col];
- if
(group.getFieldRepetitionCount(colIndex) > 0) {
- PrimitiveType.PrimitiveTypeName
type =
parquetSchema.getType(columnNames[col]).asPrimitiveType().getPrimitiveTypeName();
- switch (type) {
- case INT32:
- dest.set(row,
col, group.getInteger(colIndex, 0));
- break;
- case INT64:
- dest.set(row,
col, group.getLong(colIndex, 0));
- break;
- case FLOAT:
- dest.set(row,
col, group.getFloat(colIndex, 0));
- break;
- case DOUBLE:
- dest.set(row,
col, group.getDouble(colIndex, 0));
- break;
- case BOOLEAN:
- dest.set(row,
col, group.getBoolean(colIndex, 0));
- break;
- case BINARY:
- dest.set(row,
col, group.getBinary(colIndex, 0).toStringUsingUTF8());
- break;
- default:
- throw new
IOException("Unsupported data type: " + type);
- }
- } else {
- dest.set(row, col, null);
- }
+ ColumnDescriptor desc =
descriptors[col];
+ ColumnReader creader =
colStore.getColumnReader(desc);
+ int maxDef =
desc.getMaxDefinitionLevel();
+ PrimitiveType.PrimitiveTypeName ptype =
desc.getPrimitiveType().getPrimitiveTypeName();
+ readColumn(dest, creader, col, row,
rowsInGroup, maxDef, ptype);
}
- row++;
+ row += rowsInGroup;
}
+ return row - (int) rowOffset;
+ }
+ }
- // Check frame dimensions
- if (row != rlen) {
- throw new IOException("Mismatch in row count:
expected " + rlen + ", but got " + row);
+ /**
+ * Reads one column of a row group, writing each value (or null) into
the destination FrameBlock.
+ */
+ private void readColumn(FrameBlock dest, ColumnReader creader, int col,
int rowStart,
+ int rowsInGroup, int maxDef, PrimitiveType.PrimitiveTypeName
ptype) throws IOException
+ {
+ for (int i = 0; i < rowsInGroup; i++) {
+ int row = rowStart + i;
+ if (creader.getCurrentDefinitionLevel() == maxDef) {
+ switch (ptype) {
+ case INT32:
+ dest.set(row, col,
creader.getInteger());
+ break;
+ case INT64:
+ dest.set(row, col,
creader.getLong());
+ break;
+ case FLOAT:
+ dest.set(row, col,
creader.getFloat());
+ break;
+ case DOUBLE:
+ dest.set(row, col,
creader.getDouble());
+ break;
+ case BOOLEAN:
+ dest.set(row, col,
creader.getBoolean());
+ break;
+ case INT96: {
+ // Legacy INT96 timestamp,
narrowed to epoch millis.
+ // See
https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#timestamp
+ Binary binary =
creader.getBinary();
+ ByteBuffer buf =
ByteBuffer.wrap(binary.getBytes()).order(ByteOrder.LITTLE_ENDIAN);
+ long nanosOfDay = buf.getLong();
+ int julianDay = buf.getInt();
+ long millis = (julianDay -
JULIAN_EPOCH_OFFSET_DAYS) * MILLIS_IN_DAY
+ + nanosOfDay /
NANOS_PER_MILLISECOND;
+ dest.set(row, col, millis);
+ break;
+ }
Review Comment:
I think, we probably should not support this Int96, seems like a feature
going out of the format.
##########
src/test/java/org/apache/sysds/test/functions/io/parquet/ParquetWriterBenchmark.java:
##########
Review Comment:
remove the writer benchmark, move it similarly to the reader.
##########
src/test/java/org/apache/sysds/test/functions/io/parquet/FrameReaderParquetLegacy.java:
##########
Review Comment:
Remove these legacy files.
##########
src/test/java/org/apache/sysds/test/functions/io/parquet/FrameReaderParquetLegacy.java:
##########
@@ -0,0 +1,186 @@
+/*
+ * 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.sysds.test.functions.io.parquet;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.ParquetReader;
+import org.apache.parquet.hadoop.example.GroupReadSupport;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.parquet.hadoop.util.HadoopInputFile;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.sysds.common.Types.ValueType;
+import org.apache.sysds.conf.ConfigurationManager;
+import org.apache.sysds.runtime.DMLRuntimeException;
+import org.apache.sysds.runtime.frame.data.FrameBlock;
+import org.apache.sysds.runtime.io.FrameReader;
+import org.apache.sysds.runtime.util.HDFSTool;
+import org.apache.parquet.io.api.Binary;
+
+/**
+ * Single-threaded frame parquet reader using the {@code Group} record API
(legacy baseline).
+ *
+ */
+public class FrameReaderParquetLegacy extends FrameReader {
+
+ /**
+ * Reads a Parquet file from HDFS and converts it into a FrameBlock.
+ *
+ * @param fname The HDFS file path to the Parquet file.
+ * @param schema The expected data types of the columns.
+ * @param names The names of the columns.
+ * @param rlen The expected number of rows.
+ * @param clen The expected number of columns.
+ * @return A FrameBlock containing the data read from the Parquet file.
+ */
+ @Override
+ public FrameBlock readFrameFromHDFS(String fname, ValueType[] schema,
String[] names, long rlen, long clen) throws IOException, DMLRuntimeException {
+ // Prepare file access
+ Configuration conf = ConfigurationManager.getCachedJobConf();
+ Path path = new Path(fname);
+
+ // Check existence and non-empty file
+ if (!HDFSTool.existsFileOnHDFS(path.toString())) {
+ throw new IOException("File does not exist on HDFS: " +
fname);
+ }
+
+ // Allocate output frame block
+ ValueType[] lschema = createOutputSchema(schema, clen);
+ String[] lnames = createOutputNames(names, clen);
+ FrameBlock ret = createOutputFrameBlock(lschema, lnames, rlen);
+
+ // Read Parquet file
+ readParquetFrameFromHDFS(path, conf, ret, lschema, rlen, clen);
+
+ return ret;
+ }
+
+ /**
+ * Reads data from a Parquet file on HDFS and fills the provided
FrameBlock.
+ * The method retrieves the Parquet schema from the file footer, maps
the required column names
+ * to their corresponding indices, and then uses a ParquetReader to
iterate over each row.
+ * Data is extracted based on the column type and set into the output
FrameBlock.
+ *
+ * @param path The HDFS path to the Parquet file.
+ * @param conf The Hadoop configuration.
+ * @param dest The FrameBlock to populate with data.
+ * @param schema The expected value types for the output columns.
+ * @param rlen The expected number of rows.
+ * @param clen The expected number of columns.
+ */
+ protected void readParquetFrameFromHDFS(Path path, Configuration conf,
FrameBlock dest, ValueType[] schema, long rlen, long clen) throws IOException {
+ int row = readSingleParquetFile(path, conf, dest, clen, 0);
+
+ // Check frame dimensions
+ if (row != rlen) {
+ throw new IOException("Mismatch in row count: expected
" + rlen + ", but got " + row);
+ }
+ }
+
+ // Constants for decoding legacy INT96 timestamps
+ private static final int JULIAN_EPOCH_OFFSET_DAYS = 2_440_588;
+ private static final long MILLIS_IN_DAY = TimeUnit.DAYS.toMillis(1);
+ private static final long NANOS_PER_MILLISECOND =
TimeUnit.MILLISECONDS.toNanos(1);
+
+ protected int readSingleParquetFile(Path path, Configuration conf,
FrameBlock dest,
Review Comment:
This seems odd, why do we have a custom reader in the test ?
I suggest, instead of this, we compare to the Spark Parquet reader (that is
builtin to the spark dependency)
##########
src/test/java/org/apache/sysds/test/functions/io/parquet/FrameParquetSchemaTest.java:
##########
@@ -76,11 +77,11 @@ public void testParquetWriteReadAllSchemaTypes() {
// Populate frame block
Object[][] rows = new Object[][] {
- { 1.0, 1.1f, 10, 100L, true, "A" },
- { 2.0, 2.1f, 20, 200L, false, "B" },
- { 3.0, 3.1f, 30, 300L, true, "C" },
- { 4.0, 4.1f, 40, 400L, false, "D" },
- { 5.0, 5.1f, 50, 500L, true, "E" }
+ { 1.0, 1.1f, 10, 100L, true, "A" },
+ { 2.0, 2.1f, 20, 200L, false, "B" },
+ { 3.0, 3.1f, 30, 300L, true, "C" },
+ { 4.0, 4.1f, 40, 400L, false, "D" },
+ { 5.0, 5.1f, 50, 500L, true, "E" }
Review Comment:
formatting only change, please refrain from things like this.
##########
src/test/java/org/apache/sysds/test/functions/io/parquet/ParquetReaderBenchmark.java:
##########
Review Comment:
Do not add a benchmark file here, Instead put it in
test/java/org/aparche/sysds/performance/
##########
src/test/java/org/apache/sysds/test/functions/io/parquet/FrameWriterParquetLegacy.java:
##########
Review Comment:
remove
##########
src/main/java/org/apache/sysds/runtime/io/FrameReaderParquet.java:
##########
@@ -89,65 +97,147 @@ public FrameBlock readFrameFromHDFS(String fname,
ValueType[] schema, String[] n
* @param clen The expected number of columns.
*/
protected void readParquetFrameFromHDFS(Path path, Configuration conf,
FrameBlock dest, ValueType[] schema, long rlen, long clen) throws IOException {
- // Retrieve schema from Parquet footer
- ParquetMetadata metadata =
ParquetFileReader.open(HadoopInputFile.fromPath(path, conf)).getFooter();
- MessageType parquetSchema =
metadata.getFileMetaData().getSchema();
+ int row = readSingleParquetFile(path, conf, dest, clen, 0);
- // Map column names to Parquet schema indices
- String[] columnNames = dest.getColumnNames();
- int[] columnIndices = new int[columnNames.length];
- for (int i = 0; i < columnNames.length; i++) {
- columnIndices[i] =
parquetSchema.getFieldIndex(columnNames[i]);
+ // Check frame dimensions
+ if (row != rlen) {
+ throw new IOException("Mismatch in row count: expected
" + rlen + ", but got " + row);
}
+ }
+
+ // Constants for decoding legacy INT96 timestamps
+ private static final int JULIAN_EPOCH_OFFSET_DAYS = 2_440_588;
+ private static final long MILLIS_IN_DAY = TimeUnit.DAYS.toMillis(1);
+ private static final long NANOS_PER_MILLISECOND =
TimeUnit.MILLISECONDS.toNanos(1);
+
+ /**
+ * Reads a single Parquet file into the destination FrameBlock using
the column API.
+ * Iterates row groups; within each row group reads each requested
column and writes the
+ * value into the FrameBlock.
+ *
+ * @param path The HDFS path to the Parquet file.
+ * @param conf The Hadoop configuration.
+ * @param dest The FrameBlock to populate.
+ * @param clen The number of columns.
+ * @param rowOffset The starting row offset in the destination
FrameBlock.
+ * @return The number of rows read.
+ */
+ protected int readSingleParquetFile(Path path, Configuration conf,
FrameBlock dest,
+ long clen, long rowOffset) throws IOException
+ {
+ String[] columnNames = dest.getColumnNames();
+
+ try (ParquetFileReader reader =
ParquetFileReader.open(HadoopInputFile.fromPath(path, conf))) {
+ ParquetMetadata metadata = reader.getFooter();
+ MessageType parquetSchema =
metadata.getFileMetaData().getSchema();
+ String createdBy =
metadata.getFileMetaData().getCreatedBy();
+
+ // Map each requested frame column (by name) to its
Parquet column descriptor.
+ ColumnDescriptor[] descriptors = new
ColumnDescriptor[(int) clen];
+ for (int col = 0; col < clen; col++) {
+ ColumnDescriptor desc =
parquetSchema.getColumnDescription(new String[]{ columnNames[col] });
+ // Nested columns cannot be represented by a
flat FrameBlock column
+ if (desc.getMaxRepetitionLevel() > 0)
+ throw new IOException("Nested Parquet
columns are not supported: " + columnNames[col]);
+ descriptors[col] = desc;
+ }
+
+ // no-op converter tree, only used by
ColumnReadStoreImpl to resolve a converter per column
+ GroupConverter rootConverter =
newNoOpRootConverter(parquetSchema);
+
+ int row = (int) rowOffset;
+ PageReadStore pages;
+
+ while (true) {
+ pages = reader.readNextRowGroup();
- // Read data usind ParquetReader
- try (ParquetReader<Group> rowReader = ParquetReader.builder(new
GroupReadSupport(), path)
- .withConf(conf)
- .build()) {
+ if (pages == null)
+ break;
+
+ int rowsInGroup = (int) pages.getRowCount();
+ ColumnReadStoreImpl colStore = new
ColumnReadStoreImpl(pages, rootConverter, parquetSchema, createdBy);
- Group group;
- int row = 0;
- while ((group = rowReader.read()) != null) {
for (int col = 0; col < clen; col++) {
- int colIndex = columnIndices[col];
- if
(group.getFieldRepetitionCount(colIndex) > 0) {
- PrimitiveType.PrimitiveTypeName
type =
parquetSchema.getType(columnNames[col]).asPrimitiveType().getPrimitiveTypeName();
- switch (type) {
- case INT32:
- dest.set(row,
col, group.getInteger(colIndex, 0));
- break;
- case INT64:
- dest.set(row,
col, group.getLong(colIndex, 0));
- break;
- case FLOAT:
- dest.set(row,
col, group.getFloat(colIndex, 0));
- break;
- case DOUBLE:
- dest.set(row,
col, group.getDouble(colIndex, 0));
- break;
- case BOOLEAN:
- dest.set(row,
col, group.getBoolean(colIndex, 0));
- break;
- case BINARY:
- dest.set(row,
col, group.getBinary(colIndex, 0).toStringUsingUTF8());
- break;
- default:
- throw new
IOException("Unsupported data type: " + type);
- }
- } else {
- dest.set(row, col, null);
- }
+ ColumnDescriptor desc =
descriptors[col];
+ ColumnReader creader =
colStore.getColumnReader(desc);
+ int maxDef =
desc.getMaxDefinitionLevel();
+ PrimitiveType.PrimitiveTypeName ptype =
desc.getPrimitiveType().getPrimitiveTypeName();
+ readColumn(dest, creader, col, row,
rowsInGroup, maxDef, ptype);
}
- row++;
+ row += rowsInGroup;
}
+ return row - (int) rowOffset;
+ }
+ }
- // Check frame dimensions
- if (row != rlen) {
- throw new IOException("Mismatch in row count:
expected " + rlen + ", but got " + row);
+ /**
+ * Reads one column of a row group, writing each value (or null) into
the destination FrameBlock.
+ */
+ private void readColumn(FrameBlock dest, ColumnReader creader, int col,
int rowStart,
+ int rowsInGroup, int maxDef, PrimitiveType.PrimitiveTypeName
ptype) throws IOException
+ {
+ for (int i = 0; i < rowsInGroup; i++) {
+ int row = rowStart + i;
Review Comment:
Here similarly to the other Pr, you can move the row calculation out of the
for loop and iterate
int row = rowStart ; i < rowsInGroup + rowstarte; i++.
Please use the same variable names as your other PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
