Skip to content

chore(): Added benchmark tests for parquet and hdf5 #2509

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
Ichbinkiana wants to merge 9 commits into
base: main
Choose a base branch
from
Open

chore(): Added benchmark tests for parquet and hdf5 #2509

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
359c381
chore(): Added benchmark tests for parquet and hdf5
Jun 21, 2026
0353cfd
chore(): Add performance related tests for parquet and hdf5
Jun 21, 2026
e9cc1af
chore(): Added Optimization for HDF5 read and write
Jul 28, 2026
cc9198c
Merge branch 'systemds-3929-add-tests' into feature/add-benchmark-tests
Jul 28, 2026
7e8e3ff
cleanup(): Moved benchmark tests away from JUNIT + cleanup
Jul 28, 2026
d935c7a
cleanup(): Deleted old commits irrelevant files as cleanup
Jul 28, 2026
2d200ff
chore(): Optimization for parquet reader and writer.
Jul 29, 2026
b1e6633
chore(): cleanup and add missing commits
Aug 3, 2026
11b9c95
cleanup(): Remove unnecessary tests
Aug 3, 2026
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
92 changes: 53 additions & 39 deletions src/main/java/org/apache/sysds/runtime/io/FrameReaderParquet.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/

package org.apache.sysds.runtime.io;

import java.io.IOException;
Expand All @@ -27,7 +28,6 @@
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;
Expand All @@ -43,6 +43,44 @@
*/
public class FrameReaderParquet extends FrameReader {

protected PrimitiveType.PrimitiveTypeName[] getParquetColumnTypes(MessageType parquetSchema, int[] columnIndices) {
PrimitiveType.PrimitiveTypeName[] columnTypes = new PrimitiveType.PrimitiveTypeName[columnIndices.length];
for(int i = 0; i < columnIndices.length; i++)
columnTypes[i] = parquetSchema.getType(columnIndices[i]).asPrimitiveType().getPrimitiveTypeName();
return columnTypes;
}

protected int[] getParquetColumnIndices(MessageType parquetSchema, String[] columnNames) {
int[] columnIndices = new int[columnNames.length];
for (int i = 0; i < columnNames.length; i++) {
columnIndices[i] = parquetSchema.getFieldIndex(columnNames[i]);
}
return columnIndices;
}

protected Object readTypedParquetValue(Group group, PrimitiveType.PrimitiveTypeName type, int columnIndex) throws IOException {
if (group.getFieldRepetitionCount(columnIndex) == 0) {
return null;
}

switch (type) {
case INT32:
return group.getInteger(columnIndex, 0);
case INT64:
return group.getLong(columnIndex, 0);
case FLOAT:
return group.getFloat(columnIndex, 0);
case DOUBLE:
return group.getDouble(columnIndex, 0);
case BOOLEAN:
return group.getBoolean(columnIndex, 0);
case BINARY:
return group.getBinary(columnIndex, 0).toStringUsingUTF8();
default:
throw new IOException("Unsupported data type: " + type);
}
}

/**
* Reads a Parquet file from HDFS and converts it into a FrameBlock.
*
Expand All @@ -59,7 +97,7 @@ public FrameBlock readFrameFromHDFS(String fname, ValueType[] schema, String[] n
Configuration conf = ConfigurationManager.getCachedJobConf();
Path path = new Path(fname);

// Check existence and non-empty file
// Check existence
if (!HDFSTool.existsFileOnHDFS(path.toString())) {
throw new IOException("File does not exist on HDFS: " + fname);
}
Expand All @@ -70,7 +108,7 @@ public FrameBlock readFrameFromHDFS(String fname, ValueType[] schema, String[] n
FrameBlock ret = createOutputFrameBlock(lschema, lnames, rlen);

// Read Parquet file
readParquetFrameFromHDFS(path, conf, ret, lschema, rlen, clen);
readParquetFrameFromHDFS(path, conf, ret, rlen, clen);

return ret;
}
Expand All @@ -84,59 +122,35 @@ public FrameBlock readFrameFromHDFS(String fname, ValueType[] schema, String[] n
* @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 {
protected void readParquetFrameFromHDFS(Path path, Configuration conf, FrameBlock dest, 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();
MessageType parquetSchema;
try (ParquetFileReader fileReader = ParquetFileReader.open(HadoopInputFile.fromPath(path, conf))) {
parquetSchema = fileReader.getFooter().getFileMetaData().getSchema();
}

// 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]);
}
int[] columnIndices = getParquetColumnIndices(parquetSchema, columnNames);
PrimitiveType.PrimitiveTypeName[] columnTypes = getParquetColumnTypes(parquetSchema, columnIndices);

// Read data usind ParquetReader
// Read data using ParquetReader
try (ParquetReader<Group> rowReader = ParquetReader.builder(new GroupReadSupport(), path)
.withConf(conf)
.build()) {

Group group;
int row = 0;
while ((group = rowReader.read()) != null) {
if(row >= rlen)
throw new IOException("Mismatch in row count: expected " + rlen + ", but got more rows.");

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);
}
dest.set(row, col, readTypedParquetValue(group, columnTypes[col], colIndex));
}
row++;
}
Expand Down
112 changes: 91 additions & 21 deletions src/main/java/org/apache/sysds/runtime/io/FrameReaderParquetParallel.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,20 +16,28 @@
* specific language governing permissions and limitations
* under the License.
*/

package org.apache.sysds.runtime.io;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.parquet.example.data.Group;
import org.apache.parquet.hadoop.metadata.BlockMetaData;
import org.apache.parquet.hadoop.ParquetFileReader;
import org.apache.parquet.hadoop.ParquetReader;
import org.apache.parquet.hadoop.example.GroupReadSupport;
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.hops.OptimizerUtils;
import org.apache.sysds.runtime.DMLRuntimeException;
Expand All @@ -41,32 +49,82 @@
*
*/
public class FrameReaderParquetParallel extends FrameReaderParquet {

private Path[] getParquetDataFilePaths(FileSystem fs, Path path) throws IOException {
FileStatus status = fs.getFileStatus(path);

if (status.isFile())
return new Path[] {path};

List<Path> files = new ArrayList<>();
for (FileStatus child : fs.listStatus(path)) {
if(child.isFile() && isParquetDataFile(child.getPath()))
files.add(child.getPath());
}

return files.toArray(new Path[0]);
}

private boolean isParquetDataFile(Path path) {
String name = path.getName();

return !name.startsWith("_")
&& !name.startsWith(".")
&& !name.endsWith(".crc");
}

private long getParquetRowCount(Path path, Configuration conf) throws IOException {
long rowCount = 0;
try (ParquetFileReader fileReader = ParquetFileReader.open(HadoopInputFile.fromPath(path, conf))) {
for (BlockMetaData block : fileReader.getFooter().getBlocks()) {
rowCount += block.getRowCount();
}
}
return rowCount;
}

/**
* Reads a Parquet frame in parallel and populates the provided FrameBlock with the data.
* The method retrieves all file paths from the sequence files at that location, it then determines
* The method retrieves all Parquet data file paths at the given location, it then determines
* the number of threads to use based on the available files and a configured parallelism setting.
* A thread pool is created to run a reading task for each file concurrently.
*
* @param path The HDFS path to the Parquet file or the directory containing sequence files.
* @param path The HDFS path to the Parquet file or the directory containing part files.
* @param conf The Hadoop configuration.
* @param dest The FrameBlock to be updated with the data read from the files.
* @param schema The expected value types for the frame columns.
* @param rlen The expected number of rows.
* @param clen The expected number of columns.
*/
@Override
protected void readParquetFrameFromHDFS(Path path, Configuration conf, FrameBlock dest, ValueType[] schema, long rlen, long clen) throws IOException, DMLRuntimeException {
FileSystem fs = IOUtilFunctions.getFileSystem(path);
Path[] files = IOUtilFunctions.getSequenceFilePaths(fs, path);
int numThreads = Math.min(OptimizerUtils.getParallelBinaryReadParallelism(), files.length);
protected void readParquetFrameFromHDFS(Path path, Configuration conf, FrameBlock dest, long rlen, long clen) throws IOException, DMLRuntimeException {
FileSystem fs = IOUtilFunctions.getFileSystem(path, conf);
Path[] files = getParquetDataFilePaths(fs, path);

if (files.length == 0)
throw new IOException("No Parquet data files found at path: " + path);

Arrays.sort(files);
long[] rowCounts = new long[files.length];
long totalRows = 0;

for (int i = 0; i < files.length; i++) {
rowCounts[i] = getParquetRowCount(files[i], conf);
totalRows += rowCounts[i];
}

if (rlen >= 0 && totalRows != rlen)
throw new IOException("Mismatch in row count: expected " + rlen + ", but got " + totalRows);

int numThreads = Math.min(OptimizerUtils.getParallelBinaryReadParallelism(), files.length);
// Create and execute read tasks
ExecutorService pool = CommonThreadPool.get(numThreads);
try {
List<ReadFileTask> tasks = new ArrayList<>();
for (Path file : files) {
tasks.add(new ReadFileTask(file, conf, dest, schema, clen));
long rowOffset = 0;

for (int i = 0; i < files.length; i++) {
tasks.add(new ReadFileTask(files[i], conf, dest, clen, rowOffset, rowCounts[i]));
rowOffset += rowCounts[i];
}

for (Future<Object> task : pool.invokeAll(tasks)) {
Expand All @@ -83,36 +141,48 @@ private class ReadFileTask implements Callable<Object> {
private Path path;
private Configuration conf;
private FrameBlock dest;
@SuppressWarnings("unused")
private ValueType[] schema;
private long clen;
private long rowOffset;
private long expectedRows;

public ReadFileTask(Path path, Configuration conf, FrameBlock dest, ValueType[] schema, long clen) {
public ReadFileTask(Path path, Configuration conf, FrameBlock dest, long clen, long rowOffset, long expectedRows) {
this.path = path;
this.conf = conf;
this.dest = dest;
this.schema = schema;
this.clen = clen;
this.rowOffset = rowOffset;
this.expectedRows = expectedRows;
}

// When executed, a ParquetReader for the assigned file opens and iterates over each row processing every column.
@Override
public Object call() throws Exception {
MessageType parquetSchema;
try (ParquetFileReader fileReader = ParquetFileReader.open(HadoopInputFile.fromPath(path, conf))) {
parquetSchema = fileReader.getFooter().getFileMetaData().getSchema();
}
String[] columnNames = dest.getColumnNames();
int[] columnIndices = getParquetColumnIndices(parquetSchema, columnNames);
PrimitiveType.PrimitiveTypeName[] columnTypes = getParquetColumnTypes(parquetSchema, columnIndices);
try (ParquetReader<Group> reader = ParquetReader.builder(new GroupReadSupport(), path).withConf(conf).build()) {
Group group;
int row = 0;
long localRow = 0;

while ((group = reader.read()) != null) {
if(localRow >= expectedRows)
throw new IOException("Mismatch in row count for file " + path + ": expected " + expectedRows + ", but got more rows.");
int outRow = Math.toIntExact(rowOffset + localRow);
for (int col = 0; col < clen; col++) {
if (group.getFieldRepetitionCount(col) > 0) {
dest.set(row, col, group.getValueToString(col, 0));
} else {
dest.set(row, col, null);
}
int colIndex = columnIndices[col];
dest.set(outRow, col, readTypedParquetValue(group, columnTypes[col], colIndex));
}
row++;
localRow++;
}

if (localRow != expectedRows)
throw new IOException("Mismatch in row count for file " + path + ": expected " + expectedRows + ", but got " + localRow);
}
return null;
}
}
}
}
Loading