diff --git a/src/main/java/org/apache/sysds/runtime/io/FrameReaderParquet.java b/src/main/java/org/apache/sysds/runtime/io/FrameReaderParquet.java index 79ca2cff38e..384ffa852b2 100644 --- a/src/main/java/org/apache/sysds/runtime/io/FrameReaderParquet.java +++ b/src/main/java/org/apache/sysds/runtime/io/FrameReaderParquet.java @@ -16,6 +16,7 @@ * specific language governing permissions and limitations * under the License. */ + package org.apache.sysds.runtime.io; import java.io.IOException; @@ -31,7 +32,8 @@ import org.apache.parquet.column.impl.ColumnReadStoreImpl; import org.apache.parquet.column.page.PageReadStore; import org.apache.parquet.hadoop.ParquetFileReader; -import org.apache.parquet.hadoop.metadata.FileMetaData; +import org.apache.parquet.hadoop.ParquetReader; +import org.apache.parquet.hadoop.example.GroupReadSupport; import org.apache.parquet.hadoop.util.HadoopInputFile; import org.apache.parquet.io.api.Converter; import org.apache.parquet.io.api.GroupConverter; @@ -58,6 +60,45 @@ */ 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. * @@ -73,7 +114,9 @@ public FrameBlock readFrameFromHDFS(String fname, ValueType[] schema, String[] n throws IOException, DMLRuntimeException { Configuration conf = ConfigurationManager.getCachedJobConf(); Path path = new Path(fname); - if(!HDFSTool.existsFileOnHDFS(path.toString())) + + // Check existence + if (!HDFSTool.existsFileOnHDFS(path.toString())) { throw new IOException("File does not exist on HDFS: " + fname); ValueType[] lschema = createOutputSchema(schema, clen); @@ -83,294 +126,61 @@ public FrameBlock readFrameFromHDFS(String fname, ValueType[] schema, String[] n for(int c = 0; c < clen; c++) dest[c] = ArrayFactory.allocateBacking(lschema[c], (int) rlen); - readParquetFrameFromHDFS(path, conf, dest, lschema, lnames, rlen); - - // zero-row output stays a schema-only frame (no zero-length column arrays) - if(rlen == 0) - return new FrameBlock(lschema, lnames, 0); + // Read Parquet file + readParquetFrameFromHDFS(path, conf, ret, rlen, clen); - Array[] columns = new Array[(int) clen]; - for(int c = 0; c < clen; c++) - columns[c] = ArrayFactory.create(lschema[c], dest[c]); - return new FrameBlock(columns, lnames); + return ret; } /** - * Reads an entire Parquet file (or directory of part files) into the pre-allocated column backing arrays. Part - * files, if any, are read sequentially in name order. + * 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 or directory. - * @param conf The Hadoop configuration. - * @param dest The per-column backing arrays to populate. - * @param schema The value types of the output columns. - * @param names The names of the output columns. - * @param rlen The expected number of rows. + * @param path The HDFS path to the Parquet file or directory. + * @param conf The Hadoop configuration. + * @param dest The FrameBlock to populate with data. + * @param rlen The expected number of rows. + * @param clen The expected number of columns. */ - protected void readParquetFrameFromHDFS(Path path, Configuration conf, Object[] dest, ValueType[] schema, - String[] names, long rlen) throws IOException { - FileSystem fs = IOUtilFunctions.getFileSystem(path); - Path[] files = IOUtilFunctions.getSequenceFilePaths(fs, path); - Arrays.sort(files, Comparator.comparing(Path::getName)); + protected void readParquetFrameFromHDFS(Path path, Configuration conf, FrameBlock dest, long rlen, long clen) + throws IOException { + // Retrieve schema from Parquet footer + MessageType parquetSchema; + try(ParquetFileReader fileReader = ParquetFileReader.open(HadoopInputFile.fromPath(path, conf))) { + parquetSchema = fileReader.getFooter().getFileMetaData().getSchema(); + } - long off = 0; - for(Path file : files) - off += readSingleParquetFile(file, conf, dest, schema, names, rlen, (int) off); - if(off != rlen) - throw new IOException("Mismatch in row count: expected " + rlen + ", but got " + off); - } + // Map column names to Parquet schema indices + String[] columnNames = dest.getColumnNames(); + int[] columnIndices = getParquetColumnIndices(parquetSchema, columnNames); + PrimitiveType.PrimitiveTypeName[] columnTypes = getParquetColumnTypes(parquetSchema, columnIndices); - /** - * Decodes a single Parquet file into the column backing arrays at the given row offset, one row group at a time and - * column-at-a-time within each row group. Thread-safe for distinct row ranges, so the parallel reader assigns each - * file its own offset and shares the output arrays. - * - * @param path The HDFS path to the Parquet file. - * @param conf The Hadoop configuration. - * @param dest The per-column backing arrays to populate. - * @param schema The value types of the output columns. - * @param names The names of the output columns. - * @param rlen The total number of output rows (exclusive upper bound of this file's rows). - * @param rowOffset The row offset of this file's first row. - * @return The number of rows read. - */ - protected int readSingleParquetFile(Path path, Configuration conf, Object[] dest, ValueType[] schema, - String[] names, long rlen, int rowOffset) throws IOException { - final int ncol = schema.length; - try(ParquetFileReader reader = ParquetFileReader.open(HadoopInputFile.fromPath(path, conf))) { - FileMetaData meta = reader.getFooter().getFileMetaData(); - MessageType parquetSchema = meta.getSchema(); - String createdBy = meta.getCreatedBy(); + // Read data using ParquetReader + try(ParquetReader rowReader = ParquetReader.builder(new GroupReadSupport(), path).withConf(conf) + .build()) { - // map each requested frame column (by name) to its parquet column descriptor - final ColumnDescriptor[] descs = new ColumnDescriptor[ncol]; - for(int c = 0; c < ncol; c++) - descs[c] = validateDecodable(parquetSchema, names[c]); + 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."); - GroupConverter root = dummyConverter(parquetSchema.getFieldCount()); - int off = rowOffset; - PageReadStore pages; - while((pages = reader.readNextRowGroup()) != null) { - int nrow = (int) pages.getRowCount(); - if(off + nrow > rlen) - throw new IOException( - "Mismatch in row count: expected " + rlen + ", but got at least " + (off + nrow)); - ColumnReadStoreImpl store = new ColumnReadStoreImpl(pages, root, parquetSchema, createdBy); - for(int c = 0; c < ncol; c++) { - ColumnReader creader = store.getColumnReader(descs[c]); - int maxDef = descs[c].getMaxDefinitionLevel(); - PrimitiveTypeName ptype = descs[c].getPrimitiveType().getPrimitiveTypeName(); - if(directDecodable(ptype, schema[c])) - decodeColumnInto(creader, maxDef, nrow, ptype, dest[c], off); - else - decodeColumnConvert(creader, maxDef, nrow, ptype, schema[c], dest[c], off); + for(int col = 0; col < clen; col++) { + int colIndex = columnIndices[col]; + dest.set(row, col, readTypedParquetValue(group, columnTypes[col], colIndex)); } - off += nrow; + row++; } - return off - rowOffset; - } - } - - /** - * Resolve the descriptor of one parquet column and verify it is decodable: a non-nested primitive of a physical - * type the reader supports (INT96 timestamps and nested/repeated groups are not). - */ - private static ColumnDescriptor validateDecodable(MessageType parquetSchema, String name) throws IOException { - if(!parquetSchema.containsField(name)) - throw new IOException("Column not found in Parquet schema: " + name); - Type t = parquetSchema.getType(name); - if(!t.isPrimitive() || t.isRepetition(Repetition.REPEATED)) - throw new IOException("Nested Parquet columns are not supported: " + name); - PrimitiveTypeName ptype = t.asPrimitiveType().getPrimitiveTypeName(); - switch(ptype) { - case INT32: - case INT64: - case FLOAT: - case DOUBLE: - case BOOLEAN: - case BINARY: - return parquetSchema.getColumnDescription(new String[] {name}); - default: - throw new IOException("Unsupported Parquet type " + ptype + " for column: " + name - + (ptype == PrimitiveTypeName.INT96 ? " (deprecated INT96 timestamps; re-encode as INT64)" : "")); - } - } - /** - * Whether the parquet physical type is the one the frame value type's backing array stores, i.e. the column can be - * decoded through the typed direct path without per-value conversion. - */ - private static boolean directDecodable(PrimitiveTypeName ptype, ValueType vt) { - switch(ptype) { - case INT32: - return vt == ValueType.INT32; - case INT64: - return vt == ValueType.INT64; - case FLOAT: - return vt == ValueType.FP32; - case DOUBLE: - return vt == ValueType.FP64; - case BOOLEAN: - return vt == ValueType.BOOLEAN; - default: - return vt == ValueType.STRING; // BINARY - } - } - - /** - * Decode one parquet column of the current row group into a pre-allocated typed array at the given offset. Null - * cells (definition level below max) keep the array default (0 for numerics, null for strings). - */ - private static void decodeColumnInto(ColumnReader creader, int maxDef, int nrow, PrimitiveTypeName ptype, - Object dest, int off) { - final int end = off + nrow; - switch(ptype) { - case INT32: { - int[] a = (int[]) dest; - for(int r = off; r < end; r++) { - if(creader.getCurrentDefinitionLevel() == maxDef) - a[r] = creader.getInteger(); - creader.consume(); - } - break; - } - case INT64: { - long[] a = (long[]) dest; - for(int r = off; r < end; r++) { - if(creader.getCurrentDefinitionLevel() == maxDef) - a[r] = creader.getLong(); - creader.consume(); - } - break; - } - case FLOAT: { - float[] a = (float[]) dest; - for(int r = off; r < end; r++) { - if(creader.getCurrentDefinitionLevel() == maxDef) - a[r] = creader.getFloat(); - creader.consume(); - } - break; - } - case DOUBLE: { - double[] a = (double[]) dest; - for(int r = off; r < end; r++) { - if(creader.getCurrentDefinitionLevel() == maxDef) - a[r] = creader.getDouble(); - creader.consume(); - } - break; - } - case BOOLEAN: { - boolean[] a = (boolean[]) dest; - for(int r = off; r < end; r++) { - if(creader.getCurrentDefinitionLevel() == maxDef) - a[r] = creader.getBoolean(); - creader.consume(); - } - break; + // Check frame dimensions + if(row != rlen) { + throw new IOException("Mismatch in row count: expected " + rlen + ", but got " + row); } - default: { // BINARY - String[] a = (String[]) dest; - for(int r = off; r < end; r++) { - if(creader.getCurrentDefinitionLevel() == maxDef) - a[r] = creader.getBinary().toStringUsingUTF8(); - creader.consume(); - } - break; - } - } - } - - /** - * Decode one parquet column whose physical type does not match the frame value type, converting each value per cell - * (same semantics as {@link FrameBlock#set(int, int, Object)}, e.g. a DOUBLE file column read into a STRING frame - * column). - */ - private static void decodeColumnConvert(ColumnReader creader, int maxDef, int nrow, PrimitiveTypeName ptype, - ValueType vt, Object dest, int off) { - final int end = off + nrow; - for(int r = off; r < end; r++) { - if(creader.getCurrentDefinitionLevel() == maxDef) - setConverted(dest, vt, r, readValue(creader, ptype)); - creader.consume(); } } - private static Object readValue(ColumnReader creader, PrimitiveTypeName ptype) { - switch(ptype) { - case INT32: - return creader.getInteger(); - case INT64: - return creader.getLong(); - case FLOAT: - return creader.getFloat(); - case DOUBLE: - return creader.getDouble(); - case BOOLEAN: - return creader.getBoolean(); - default: - return creader.getBinary().toStringUsingUTF8(); // BINARY - } - } - - /** - * Store one converted value into the backing array of the given value type. - */ - private static void setConverted(Object dest, ValueType vt, int r, Object val) { - Object converted = UtilFunctions.objectToObject(vt, val); - if(converted == null) - return; - switch(vt) { - case FP64: - ((double[]) dest)[r] = (Double) converted; - break; - case FP32: - ((float[]) dest)[r] = (Float) converted; - break; - case INT64: - case HASH64: - ((long[]) dest)[r] = (Long) converted; - break; - case UINT4: - case UINT8: - case INT32: - case HASH32: - ((int[]) dest)[r] = (Integer) converted; - break; - case BOOLEAN: - ((boolean[]) dest)[r] = (Boolean) converted; - break; - case CHARACTER: - ((char[]) dest)[r] = (Character) converted; - break; - default: - ((String[]) dest)[r] = (String) converted; - break; - } - } - - /** No-op converter tree; the column API requires one, but values are pulled via the typed getters. */ - private static GroupConverter dummyConverter(int nFields) { - final PrimitiveConverter[] leaves = new PrimitiveConverter[nFields]; - for(int i = 0; i < nFields; i++) - leaves[i] = new PrimitiveConverter() { - }; - return new GroupConverter() { - @Override - public Converter getConverter(int fieldIndex) { - return leaves[fieldIndex]; - } - - @Override - public void start() { - } - - @Override - public void end() { - } - }; - } - //not implemented @Override public FrameBlock readFrameFromInputStream(InputStream is, ValueType[] schema, String[] names, long rlen, long clen) diff --git a/src/main/java/org/apache/sysds/runtime/io/FrameReaderParquetParallel.java b/src/main/java/org/apache/sysds/runtime/io/FrameReaderParquetParallel.java index a28a366a317..76ba06d213a 100644 --- a/src/main/java/org/apache/sysds/runtime/io/FrameReaderParquetParallel.java +++ b/src/main/java/org/apache/sysds/runtime/io/FrameReaderParquetParallel.java @@ -16,23 +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.Comparator; 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.hadoop.ParquetFileReader; +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.sysds.common.Types.ValueType; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.PrimitiveType; import org.apache.sysds.hops.OptimizerUtils; import org.apache.sysds.runtime.util.CommonThreadPool; @@ -42,31 +47,81 @@ */ 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 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 + * 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 part files. + * @param conf The Hadoop configuration. + * @param dest The FrameBlock to be updated with the data read from the files. + * @param rlen The expected number of rows. + * @param clen The expected number of columns. + */ @Override - protected void readParquetFrameFromHDFS(Path path, Configuration conf, Object[] dest, ValueType[] schema, - String[] names, long rlen) throws IOException { - FileSystem fs = IOUtilFunctions.getFileSystem(path); - Path[] files = IOUtilFunctions.getSequenceFilePaths(fs, path); - Arrays.sort(files, Comparator.comparing(Path::getName)); - 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; - long[] offsets = new long[files.length]; - long cumulative = 0; for(int i = 0; i < files.length; i++) { - offsets[i] = cumulative; - try(ParquetFileReader reader = ParquetFileReader.open(HadoopInputFile.fromPath(files[i], conf))) { - for(BlockMetaData block : reader.getFooter().getBlocks()) - cumulative += block.getRowCount(); - } + rowCounts[i] = getParquetRowCount(files[i], conf); + totalRows += rowCounts[i]; } - if(cumulative != rlen) - throw new IOException("Mismatch in row count: expected " + rlen + ", but got " + cumulative); + 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 tasks = new ArrayList<>(); - for(int i = 0; i < files.length; i++) - tasks.add(new ReadFileTask(files[i], conf, dest, schema, names, rlen, (int) offsets[i])); + 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 task : pool.invokeAll(tasks)) task.get(); @@ -80,28 +135,53 @@ protected void readParquetFrameFromHDFS(Path path, Configuration conf, Object[] } private class ReadFileTask implements Callable { - private final Path path; - private final Configuration conf; - private final Object[] dest; - private final ValueType[] schema; - private final String[] names; - private final long rlen; - private final int rowOffset; - - public ReadFileTask(Path path, Configuration conf, Object[] dest, ValueType[] schema, String[] names, long rlen, - int rowOffset) { + private Path path; + private Configuration conf; + private FrameBlock dest; + private long clen; + private long rowOffset; + private long expectedRows; + + 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.names = names; - this.rlen = rlen; + this.clen = clen; this.rowOffset = rowOffset; + this.expectedRows = expectedRows; } @Override public Object call() throws Exception { - readSingleParquetFile(path, conf, dest, schema, names, rlen, rowOffset); + 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 reader = ParquetReader.builder(new GroupReadSupport(), path).withConf(conf) + .build()) { + Group group; + 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++) { + int colIndex = columnIndices[col]; + dest.set(outRow, col, readTypedParquetValue(group, columnTypes[col], colIndex)); + } + localRow++; + } + + if(localRow != expectedRows) + throw new IOException("Mismatch in row count for file " + path + ": expected " + expectedRows + + ", but got " + localRow); + } return null; } } diff --git a/src/main/java/org/apache/sysds/runtime/io/FrameWriterParquet.java b/src/main/java/org/apache/sysds/runtime/io/FrameWriterParquet.java index 775bfde0ce7..d88d9e7bd39 100644 --- a/src/main/java/org/apache/sysds/runtime/io/FrameWriterParquet.java +++ b/src/main/java/org/apache/sysds/runtime/io/FrameWriterParquet.java @@ -16,11 +16,10 @@ * specific language governing permissions and limitations * under the License. */ + package org.apache.sysds.runtime.io; import java.io.IOException; -import java.util.HashMap; -import java.util.Map; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; @@ -48,26 +47,8 @@ */ public class FrameWriterParquet extends FrameWriter { - public enum DictEncoding { - ALL_ON, ALL_OFF, STRING_ONLY - } - - private final CompressionCodecName codec; - private final DictEncoding dictEncoding; - private final long rowGroupSize; - - public FrameWriterParquet() { - this(CompressionCodecName.ZSTD, DictEncoding.STRING_ONLY, ParquetWriter.DEFAULT_BLOCK_SIZE); - } - - public FrameWriterParquet(CompressionCodecName codec, DictEncoding dictEncoding) { - this(codec, dictEncoding, ParquetWriter.DEFAULT_BLOCK_SIZE); - } - - public FrameWriterParquet(CompressionCodecName codec, DictEncoding dictEncoding, long rowGroupSize) { - this.codec = codec; - this.dictEncoding = dictEncoding; - this.rowGroupSize = rowGroupSize; + protected void writeParquetFrameToHDFS(Path path, Configuration conf, FrameBlock src) throws IOException { + writeParquetFrameToHDFS(path, conf, src, 0, src.getNumRows()); } /** @@ -104,36 +85,69 @@ public final void writeFrameToHDFS(FrameBlock src, String fname, long rlen, long * @param path The HDFS path where the Parquet file will be written. * @param conf The Hadoop configuration. * @param src The FrameBlock containing the data to write. + * @param startRow The starting row index for the write operation. + * @param endRow The ending row index for the write operation. */ - protected void writeParquetFrameToHDFS(Path path, Configuration conf, FrameBlock src) - throws IOException - { + protected void writeParquetFrameToHDFS(Path path, Configuration conf, FrameBlock src, int startRow, int endRow) + throws IOException { + if(startRow < 0 || endRow < startRow || endRow > src.getNumRows()) + throw new IOException("Invalid row range for Parquet write: " + startRow + " to " + endRow); + FileSystem fs = IOUtilFunctions.getFileSystem(path, conf); // Create schema based on frame block metadata MessageType schema = createParquetSchema(src); - String[] columnNames = src.getColumnNames(); - ValueType[] columnTypes = src.getSchema(); - - FrameParquetWriterBuilder writerBuilder = new FrameParquetWriterBuilder(path, schema, src).withConf(conf) - .withCompressionCodec( - CompressionCodecName.fromConf(conf.get(ParquetOutputFormat.COMPRESSION, codec.name()))) - .withRowGroupSize(conf.getLong(ParquetOutputFormat.BLOCK_SIZE, rowGroupSize)) - .withPageSize(conf.getInt(ParquetOutputFormat.PAGE_SIZE, ParquetWriter.DEFAULT_PAGE_SIZE)) - .withDictionaryPageSize( - conf.getInt(ParquetOutputFormat.DICTIONARY_PAGE_SIZE, ParquetWriter.DEFAULT_PAGE_SIZE)) - .withDictionaryEncoding( - conf.getBoolean(ParquetOutputFormat.ENABLE_DICTIONARY, dictEncoding == DictEncoding.ALL_ON)); - - if(dictEncoding == DictEncoding.STRING_ONLY) - for(int j = 0; j < src.getNumColumns(); j++) - if(columnTypes[j] == ValueType.STRING) - writerBuilder = writerBuilder.withDictionaryEncoding(columnNames[j], true); + // TODO:Experiment with different batch sizes? + // int batchSize = 1000; + // int rowCount = 0; + + // Write data using ParquetWriter //FIXME replace example writer? + try(ParquetWriter writer = ExampleParquetWriter.builder(path).withConf(conf).withType(schema) + .withCompressionCodec(ParquetWriter.DEFAULT_COMPRESSION_CODEC_NAME) + .withRowGroupSize((long) ParquetWriter.DEFAULT_BLOCK_SIZE).withPageSize(ParquetWriter.DEFAULT_PAGE_SIZE) + .withDictionaryEncoding(true).build()) { + final int numCols = src.getNumColumns(); + final String[] columnNames = src.getColumnNames(); + final ValueType[] schemaTypes = src.getSchema(); + + SimpleGroupFactory groupFactory = new SimpleGroupFactory(schema); + + // List rowBuffer = new ArrayList<>(batchSize); + + for(int i = startRow; i < endRow; i++) { + Group group = groupFactory.newGroup(); + for(int j = 0; j < numCols; j++) { + Object value = src.get(i, j); + if(value != null) { + ValueType type = schemaTypes[j]; + switch(type) { + case STRING: + group.add(columnNames[j], value.toString()); + break; + case INT32: + group.add(columnNames[j], (int) value); + break; + case INT64: + group.add(columnNames[j], (long) value); + break; + case FP32: + group.add(columnNames[j], (float) value); + break; + case FP64: + group.add(columnNames[j], (double) value); + break; + case BOOLEAN: + group.add(columnNames[j], (boolean) value); + break; + default: + throw new IOException("Unsupported value type: " + type); + } + } + } - try(ParquetWriter writer = writerBuilder.build()) { - for(int i = 0; i < src.getNumRows(); i++) - writer.write(i); + writer.write(group); + } } // Delete CRC files created by Hadoop if necessary @@ -176,97 +190,7 @@ protected MessageType createParquetSchema(FrameBlock src) { throw new IllegalArgumentException("Unsupported data type: " + columnTypes[i]); } } - return builder.named("FrameSchema"); - } - - /** - * WriteSupport implementation that writes rows from a FrameBlock directly to the Parquet RecordConsumer. - */ - private static class FrameWriteSupport extends WriteSupport { - private final MessageType schema; - private final FrameBlock src; - private RecordConsumer recordConsumer; - // constant across all rows - private String[] colNames; - private ValueType[] colTypes; - private int numCols; - - FrameWriteSupport(MessageType schema, FrameBlock src) { - this.schema = schema; - this.src = src; - } - - @Override - public WriteContext init(Configuration configuration) { - Map metadata = new HashMap<>(); - return new WriteContext(schema, metadata); - } - - @Override - public void prepareForWrite(RecordConsumer consumer) { - this.recordConsumer = consumer; - this.colNames = src.getColumnNames(); - this.colTypes = src.getSchema(); - this.numCols = src.getNumColumns(); - } - - @Override - public void write(Integer rowIndex) { - recordConsumer.startMessage(); - for(int j = 0; j < numCols; j++) { - Object value = src.get(rowIndex, j); - if(value != null) { - recordConsumer.startField(colNames[j], j); - switch(colTypes[j]) { - case STRING: - recordConsumer.addBinary(Binary.fromString(value.toString())); - break; - case INT32: - recordConsumer.addInteger((int) value); - break; - case INT64: - recordConsumer.addLong((long) value); - break; - case FP32: - recordConsumer.addFloat((float) value); - break; - case FP64: - recordConsumer.addDouble((double) value); - break; - case BOOLEAN: - recordConsumer.addBoolean((boolean) value); - break; - default: - throw new IllegalArgumentException("Unsupported value type: " + colTypes[j]); - } - recordConsumer.endField(colNames[j], j); - } - } - recordConsumer.endMessage(); - } - } - - /** - * ParquetWriter builder wired to FrameWriteSupport. - */ - private static class FrameParquetWriterBuilder extends ParquetWriter.Builder { - private final MessageType schema; - private final FrameBlock src; - - FrameParquetWriterBuilder(Path path, MessageType schema, FrameBlock src) { - super(path); - this.schema = schema; - this.src = src; - } - - @Override - protected FrameParquetWriterBuilder self() { - return this; - } - - @Override - protected WriteSupport getWriteSupport(Configuration conf) { - return new FrameWriteSupport(schema, src); - } + schemaBuilder.append("}"); + return MessageTypeParser.parseMessageType(schemaBuilder.toString()); } } diff --git a/src/main/java/org/apache/sysds/runtime/io/FrameWriterParquetParallel.java b/src/main/java/org/apache/sysds/runtime/io/FrameWriterParquetParallel.java index 3efbbbefef0..b979089c12c 100644 --- a/src/main/java/org/apache/sysds/runtime/io/FrameWriterParquetParallel.java +++ b/src/main/java/org/apache/sysds/runtime/io/FrameWriterParquetParallel.java @@ -16,6 +16,7 @@ * specific language governing permissions and limitations * under the License. */ + package org.apache.sysds.runtime.io; import java.io.IOException; @@ -26,41 +27,115 @@ import java.util.concurrent.Future; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; +import org.apache.sysds.common.Types.ValueType; import org.apache.sysds.conf.DMLConfig; import org.apache.sysds.hops.OptimizerUtils; -import org.apache.sysds.runtime.DMLRuntimeException; import org.apache.sysds.runtime.frame.data.FrameBlock; import org.apache.sysds.runtime.util.CommonThreadPool; import org.apache.sysds.runtime.util.HDFSTool; import org.apache.sysds.utils.stats.InfrastructureAnalyzer; /** - * Multi-threaded frame parquet writer. - * + * Multi-threaded frame parquet writer. Multi-threaded frame parquet writer. + * */ public class FrameWriterParquetParallel extends FrameWriterParquet { + private static final String PARQUET_WRITER_TARGET_PART_SIZE_MB = "sysds.io.parquet.writer.target.part.size.mb"; + + private static final String PARQUET_WRITER_THREADS = "sysds.io.parquet.writer.threads"; + + private static final long DEFAULT_TARGET_PART_SIZE_BYTES = 128L * 1024 * 1024; + + private long getTargetPartSizeBytes() throws IOException { + String value = System.getProperty(PARQUET_WRITER_TARGET_PART_SIZE_MB); + + if(value != null && !value.trim().isEmpty()) { + try { + long mb = Long.parseLong(value.trim()); + if(mb <= 0) + throw new IOException("Invalid Parquet writer target part size: " + value); + if(mb > Long.MAX_VALUE / (1024L * 1024L)) + throw new IOException("Parquet writer target part size is too large: " + value); + return mb * 1024L * 1024L; + } + catch(NumberFormatException e) { + throw new IOException("Invalid value for " + PARQUET_WRITER_TARGET_PART_SIZE_MB + ": " + value, e); + } + } + + long hdfsBlockSize = InfrastructureAnalyzer.getHDFSBlockSize(); + return hdfsBlockSize > 0 ? hdfsBlockSize : DEFAULT_TARGET_PART_SIZE_BYTES; + } + + private int getMaxWriterThreads() throws IOException { + int configuredParallelism = OptimizerUtils.getParallelBinaryWriteParallelism(); + String value = System.getProperty(PARQUET_WRITER_THREADS); + + if(value != null && !value.trim().isEmpty()) { + try { + int requestedThreads = Integer.parseInt(value.trim()); + if(requestedThreads <= 0) + throw new IOException("Invalid Parquet writer thread count: " + value); + + return Math.min(requestedThreads, configuredParallelism); + } + catch(NumberFormatException e) { + throw new IOException("Invalid value for " + PARQUET_WRITER_THREADS + ": " + value, e); + } + } + return configuredParallelism; + } + + private long estimateFrameSizeBytes(FrameBlock src) { + long rows = src.getNumRows(); + long estimatedRowSize = 0; + + for(ValueType type : src.getSchema()) { + switch(type) { + case FP64: + case INT64: + estimatedRowSize += 8; + break; + case FP32: + case INT32: + estimatedRowSize += 4; + break; + case BOOLEAN: + estimatedRowSize += 1; + break; + case STRING: + estimatedRowSize += 16; + break; + default: + estimatedRowSize += 8; + } + } + + return Math.max(rows * estimatedRowSize, 1); + } + /** - * Writes the FrameBlock data to HDFS in parallel. The method estimates the number of output partitions by comparing - * the estimated output size of the FrameBlock with the HDFS block size. It then determines the number of threads to - * use based on the parallelism configuration and the number of partitions. In case of parallelism, it divides the - * FrameBlock into chunks and a thread pool is created to execute a write task for each partition concurrently. + * Writes the FrameBlock data to HDFS in parallel. The number of output part files is derived from an estimated byte + * size and a target part size. The number of active writer threads is limited independently from the number of part + * files. * * @param path The HDFS path where the Parquet files will be written. * @param conf The Hadoop configuration. * @param src The FrameBlock containing the data to write. */ @Override - protected void writeParquetFrameToHDFS(Path path, Configuration conf, FrameBlock src) - throws IOException, DMLRuntimeException - { - // Estimate output partitions from output size in bytes - int numPartFiles = Math - .max((int) (OptimizerUtils.estimateSizeExactFrame(src.getNumRows(), src.getNumColumns()) / - InfrastructureAnalyzer.getHDFSBlockSize()), 1); + protected void writeParquetFrameToHDFS(Path path, Configuration conf, FrameBlock src) throws IOException { + // Estimate number of output partitions + long estimatedSizeBytes = estimateFrameSizeBytes(src); + long targetPartSizeBytes = getTargetPartSizeBytes(); + + int numPartFiles = (int) Math.max(1, (long) Math.ceil((double) estimatedSizeBytes / targetPartSizeBytes)); + + numPartFiles = Math.min(numPartFiles, src.getNumRows()); - // Determine parallelism - int numThreads = Math.min(OptimizerUtils.getParallelBinaryWriteParallelism(), numPartFiles); + int maxWriterThreads = getMaxWriterThreads(); + int numThreads = Math.min(maxWriterThreads, numPartFiles); if(!_forcedParallel && numThreads <= 1) { super.writeParquetFrameToHDFS(path, conf, src); @@ -70,18 +145,20 @@ protected void writeParquetFrameToHDFS(Path path, Configuration conf, FrameBlock // Create directory for concurrent tasks HDFSTool.createDirIfNotExistOnHDFS(path, DMLConfig.DEFAULT_SHARED_DIR_PERMISSION); + // Materialize default column names before parallel tasks to avoid lazy initialization in workers. + src.getColumnNames(); // Create and execute write tasks ExecutorService pool = CommonThreadPool.get(numThreads); try { List tasks = new ArrayList<>(); - int chunkSize = (int) Math.ceil((double) src.getNumRows() / numThreads); + int chunkSize = (int) Math.ceil((double) src.getNumRows() / numPartFiles); - for (int i = 0; i < numThreads; i++) { + for(int i = 0; i < numPartFiles; i++) { int startRow = i * chunkSize; int endRow = Math.min((i + 1) * chunkSize, (int) src.getNumRows()); if (startRow < endRow) { Path newPath = new Path(path, IOUtilFunctions.getPartFileName(i)); - tasks.add(new WriteFileTask(newPath, conf, src.slice(startRow, endRow - 1))); + tasks.add(new WriteFileTask(newPath, conf, src, startRow, endRow)); } } @@ -94,26 +171,29 @@ protected void writeParquetFrameToHDFS(Path path, Configuration conf, FrameBlock } } - protected void writeSingleParquetFile(Path path, Configuration conf, FrameBlock src) - throws IOException, DMLRuntimeException - { - super.writeParquetFrameToHDFS(path, conf, src); + protected void writeSingleParquetFile(Path path, Configuration conf, FrameBlock src, int startRow, int endRow) + throws IOException { + super.writeParquetFrameToHDFS(path, conf, src, startRow, endRow); } private class WriteFileTask implements Callable { private Path path; private Configuration conf; private FrameBlock src; + private final int startRow; + private final int endRow; - public WriteFileTask(Path path, Configuration conf, FrameBlock src) { + public WriteFileTask(Path path, Configuration conf, FrameBlock src, int startRow, int endRow) { this.path = path; this.conf = conf; this.src = src; + this.startRow = startRow; + this.endRow = endRow; } @Override public Object call() throws Exception { - writeSingleParquetFile(path, conf, src); + writeSingleParquetFile(path, conf, src, startRow, endRow); return null; } } diff --git a/src/main/java/org/apache/sysds/runtime/io/ReaderHDF5.java b/src/main/java/org/apache/sysds/runtime/io/ReaderHDF5.java index 71d710d3f15..277f8dcfcc7 100644 --- a/src/main/java/org/apache/sysds/runtime/io/ReaderHDF5.java +++ b/src/main/java/org/apache/sysds/runtime/io/ReaderHDF5.java @@ -81,10 +81,16 @@ public class ReaderHDF5 extends MatrixReader { protected static final int HDF5_READ_PARALLEL_MIN_BYTES = getHdf5ReadInt("sysds.hdf5.read.parallel.min.bytes", DEFAULT_HDF5_READ_PARALLEL_MIN_BYTES); + private static final String HDF5_READ_SPARSE_LAYOUT = System.getProperty("sysds.hdf5.read.sparse.layout", "dense"); + public ReaderHDF5(FileFormatPropertiesHDF5 props) { _props = props; } + protected static boolean useSparseCOORead() { + return "coo".equalsIgnoreCase(HDF5_READ_SPARSE_LAYOUT); + } + @Override public MatrixBlock readMatrixFromHDFS(String fname, long rlen, long clen, int blen, long estnnz) throws IOException, DMLRuntimeException { @@ -219,14 +225,18 @@ public static long readMatrixFromHDF5(H5ByteReader byteReader, String datasetNam LOG.trace("[HDF5] Forcing dense output for dataset=" + datasetName); } H5RootObject rootObject = H5.H5Fopen(byteReader); - H5ContiguousDataset contiguousDataset = H5.H5Dopen(rootObject, datasetName); - - int ncol = (int) rootObject.getCol(); - LOG.trace("[HDF5] readMatrix dataset=" + datasetName + " dims=" + rootObject.getRow() + "x" - + rootObject.getCol() + " loop=[" + rl + "," + ru + ") dest=" + dest.getNumRows() + "x" - + dest.getNumColumns()); try { + H5ContiguousDataset contiguousDataset = H5.H5Dopen(rootObject, datasetName); + + if(useSparseCOORead()) + return readSparseCOOFromHDF5(contiguousDataset, dest, clen); + + int ncol = (int) rootObject.getCol(); + LOG.trace( + "[HDF5] readMatrix dataset=" + datasetName + " dims=" + rootObject.getRow() + "x" + rootObject.getCol() + + " loop=[" + rl + "," + ru + ") dest=" + dest.getNumRows() + "x" + dest.getNumColumns()); + double[] row = null; double[] blockBuffer = null; int[] ixBuffer = null; @@ -360,6 +370,46 @@ public static long readMatrixFromHDF5(H5ByteReader byteReader, String datasetNam return lnnz; } + private static long readSparseCOOFromHDF5(H5ContiguousDataset dataset, MatrixBlock dest, long expectedClen) { + double[] meta = new double[3]; + dataset.readRowDoubles(0, meta, 0); + + long originalRows = (long) meta[0]; + long originalCols = (long) meta[1]; + long nnz = (long) meta[2]; + + if(originalRows != dest.getNumRows() || originalCols != dest.getNumColumns()) + throw new DMLRuntimeException("Sparse COO HDF5 metadata mismatch: " + originalRows + "x" + originalCols + + " vs " + dest.getNumRows() + "x" + dest.getNumColumns() + "."); + + if(expectedClen >= 0 && originalCols != expectedClen) + throw new DMLRuntimeException( + "Sparse COO HDF5 column metadata mismatch: " + originalCols + " vs " + expectedClen + "."); + + if(nnz > Integer.MAX_VALUE) + throw new DMLRuntimeException("Sparse COO HDF5 nnz exceeds supported row index range: " + nnz); + + if(!dest.isInSparseFormat()) + throw new DMLRuntimeException("Sparse COO HDF5 read requires sparse output."); + + SparseBlock sb = dest.getSparseBlock(); + double[] triple = new double[3]; + + for(int i = 1; i <= (int) nnz; i++) { + dataset.readRowDoubles(i, triple, 0); + + int row = (int) triple[0]; + int col = (int) triple[1]; + double val = triple[2]; + + // TODO Generalize COO reconstruction by pre-counting nonzeros per row and preallocating sparse rows. + sb.allocate(row, 1); + sb.append(row, col, val); + } + + return nnz; + } + public static MatrixBlock computeHDF5Size(List files, FileSystem fs, String datasetName, long estnnz) throws IOException, DMLRuntimeException { diff --git a/src/main/java/org/apache/sysds/runtime/io/ReaderHDF5Parallel.java b/src/main/java/org/apache/sysds/runtime/io/ReaderHDF5Parallel.java index d1651f94206..f28176e4182 100644 --- a/src/main/java/org/apache/sysds/runtime/io/ReaderHDF5Parallel.java +++ b/src/main/java/org/apache/sysds/runtime/io/ReaderHDF5Parallel.java @@ -54,7 +54,7 @@ public class ReaderHDF5Parallel extends ReaderHDF5 { - final private int _numThreads; + private final int _numThreads; protected JobConf _job; public ReaderHDF5Parallel(FileFormatPropertiesHDF5 props) { @@ -65,6 +65,9 @@ public ReaderHDF5Parallel(FileFormatPropertiesHDF5 props) { @Override public MatrixBlock readMatrixFromHDFS(String fname, long rlen, long clen, int blen, long estnnz) throws IOException, DMLRuntimeException { + // COO stores sparse triples, not dense matrix rows; reuse the COO-aware sequential reader. + if(useSparseCOORead()) + return new ReaderHDF5(_props).readMatrixFromHDFS(fname, rlen, clen, blen, estnnz); // prepare file access _job = new JobConf(ConfigurationManager.getCachedJobConf()); @@ -99,7 +102,7 @@ public MatrixBlock readMatrixFromHDFS(String fname, long rlen, long clen, int bl ArrayList tasks = new ArrayList<>(); rlen = src.getNumRows(); int blklen = (int) Math.ceil((double) rlen / numParts); - for(int i = 0; i < _numThreads & i * blklen < rlen; i++) { + for(int i = 0; i < _numThreads && i * blklen < rlen; i++) { int rl = i * blklen; int ru = (int) Math.min((i + 1) * blklen, rlen); Path newPath = HDFSTool.isDirectory(fs, path) ? diff --git a/src/main/java/org/apache/sysds/runtime/io/WriterHDF5.java b/src/main/java/org/apache/sysds/runtime/io/WriterHDF5.java index e8581c9c842..87671703abc 100644 --- a/src/main/java/org/apache/sysds/runtime/io/WriterHDF5.java +++ b/src/main/java/org/apache/sysds/runtime/io/WriterHDF5.java @@ -39,12 +39,42 @@ public class WriterHDF5 extends MatrixWriter { + private static final int DEFAULT_HDF5_WRITE_BATCH_ROWS = 1024; + private static final int DEFAULT_HDF5_WRITE_BATCH_BYTES = 1024 * 1024; + + private static final int HDF5_WRITE_BATCH_ROWS = getHdf5WriteInt("sysds.hdf5.write.batch.rows", + DEFAULT_HDF5_WRITE_BATCH_ROWS); + + private static final int HDF5_WRITE_BATCH_BYTES = getHdf5WriteInt("sysds.hdf5.write.batch.bytes", + DEFAULT_HDF5_WRITE_BATCH_BYTES); + + private static final String HDF5_WRITE_SPARSE_LAYOUT = System.getProperty("sysds.hdf5.write.sparse.layout", + "dense"); + + private static int getHdf5WriteInt(String key, int defaultValue) { + String value = System.getProperty(key); + if(value == null) + return defaultValue; + + try { + int parsed = Integer.parseInt(value.trim()); + return parsed > 0 ? parsed : defaultValue; + } + catch(NumberFormatException ex) { + return defaultValue; + } + } + protected static FileFormatPropertiesHDF5 _props = null; public WriterHDF5(FileFormatPropertiesHDF5 _props) { WriterHDF5._props = _props; } + private static boolean useSparseCOO(MatrixBlock src) { + return src.isInSparseFormat() && "coo".equalsIgnoreCase(HDF5_WRITE_SPARSE_LAYOUT); + } + @Override public void writeMatrixToHDFS(MatrixBlock src, String fname, long rlen, long clen, int blen, long nnz, boolean diag) throws IOException, DMLRuntimeException @@ -65,8 +95,10 @@ public void writeMatrixToHDFS(MatrixBlock src, String fname, long rlen, long cle //if the file already exists on HDFS, remove it. HDFSTool.deleteFileIfExistOnHDFS(fname); - //core write (sequential/parallel) - writeHDF5MatrixToHDFS(path, job, fs, src); + if(useSparseCOO(src)) + writeSparseCOOMatrixToFile(path, fs, src, rlen, clen, nnz); + else + writeHDF5MatrixToHDFS(path, job, fs, src); IOUtilFunctions.deleteCrcFilesFromLocalFileSystem(fs, path); } @@ -88,48 +120,148 @@ protected static void writeHDF5MatrixToFile(Path path, JobConf job, FileSystem f throws IOException { int clen = src.getNumColumns(); - BufferedOutputStream bos = new BufferedOutputStream(fs.create(path, true)); String datasetName = _props.getDatasetName(); - H5RootObject rootObject = H5.H5Screate(bos, src.getNumRows(), src.getNumColumns()); - H5.H5Dcreate(rootObject, src.getNumRows(), src.getNumColumns(), datasetName); - //write headers - if(rl == 0) { - H5.H5WriteHeaders(rootObject); + try(BufferedOutputStream bos = new BufferedOutputStream(fs.create(path, true))) { + H5RootObject rootObject = H5.H5Screate(bos, src.getNumRows(), src.getNumColumns()); + H5.H5Dcreate(rootObject, src.getNumRows(), src.getNumColumns(), datasetName); + + if(rl == 0) + H5.H5WriteHeaders(rootObject); + + int batchRows = getWriteBatchRows(clen); + if(src.isInSparseFormat()) + writeSparseBatched(rootObject, src, rl, rlen, clen, batchRows); + else + writeDenseBatched(rootObject, src, rl, rlen, clen, batchRows); } + } - try { - // Write the data to the datasets. - double[] row = new double[clen]; - if( src.isInSparseFormat() ) { - SparseBlock sb = src.getSparseBlock(); - for(int i = rl; i < rlen; i++) { - Arrays.fill(row, 0); - if( !sb.isEmpty(i) ) { - int apos = sb.pos(i); - int alen = sb.size(i); - double[] avals = sb.values(i); - int[] aix = sb.indexes(i); - for(int j = apos; j < apos+alen; j++) - row[aix[j]] = avals[j]; - } - H5.H5Dwrite(rootObject, row); - } + private static int getWriteBatchRows(int clen) { + long rowBytes = (long) clen * Double.BYTES; + + int rowsByBytes = rowBytes > 0 ? (int) Math.max(1, HDF5_WRITE_BATCH_BYTES / rowBytes) : 1; + + int rows = Math.max(1, Math.min(HDF5_WRITE_BATCH_ROWS, rowsByBytes)); + rows = roundDownPowerOfTwo(rows); + long cells = (long) rows * clen; + + if(cells > Integer.MAX_VALUE) + throw new DMLRuntimeException("HDF5 write batch too large: " + rows + " x " + clen); + + return rows; + } + + private static int roundDownPowerOfTwo(int value) { + int ret = 1; + while(ret <= value / 2) + ret *= 2; + return ret; + } + + private static void writeDenseBatched(H5RootObject rootObject, MatrixBlock src, int rl, int ru, int clen, + int batchRows) { + + DenseBlock db = src.getDenseBlock(); + double[] batch = new double[batchRows * clen]; + + for(int rowStart = rl; rowStart < ru; rowStart += batchRows) { + int rows = Math.min(batchRows, ru - rowStart); + + for(int r = 0; r < rows; r++) { + int srcRow = rowStart + r; + int off = r * clen; + + for(int c = 0; c < clen; c++) + batch[off + c] = db.get(srcRow, c); } - else { - DenseBlock db = src.getDenseBlock(); - for(int i = rl; i < rlen; i++) { - for(int j = 0; j < clen;j++) { - double lvalue = db!=null ? db.get(i, j) : 0; - row[j] = lvalue; - } - H5.H5Dwrite(rootObject, row); - } + + if(rows == batchRows) + H5.H5Dwrite(rootObject, batch); + else + H5.H5Dwrite(rootObject, Arrays.copyOf(batch, rows * clen)); + } + } + + private static void writeSparseBatched(H5RootObject rootObject, MatrixBlock src, int rl, int ru, int clen, + int batchRows) { + SparseBlock sb = src.getSparseBlock(); + double[] batch = new double[batchRows * clen]; + + for(int rowStart = rl; rowStart < ru; rowStart += batchRows) { + int rows = Math.min(batchRows, ru - rowStart); + Arrays.fill(batch, 0, rows * clen, 0.0); + + for(int r = 0; r < rows; r++) { + int srcRow = rowStart + r; + + if(sb == null || sb.isEmpty(srcRow)) + continue; + + int apos = sb.pos(srcRow); + int alen = sb.size(srcRow); + int[] aix = sb.indexes(srcRow); + double[] avals = sb.values(srcRow); + + int off = r * clen; + for(int k = apos; k < apos + alen; k++) + batch[off + aix[k]] = avals[k]; } + + if(rows == batchRows) + H5.H5Dwrite(rootObject, batch); + else + H5.H5Dwrite(rootObject, Arrays.copyOf(batch, rows * clen)); } - finally { - IOUtilFunctions.closeSilently(bos); + } + + private static void writeSparseCOOMatrixToFile(Path path, FileSystem fs, MatrixBlock src, long rlen, long clen, + long nnz) throws IOException { + String datasetName = _props.getDatasetName(); + + long cooRows = nnz + 1; + long cooCols = 3; + + try(BufferedOutputStream bos = new BufferedOutputStream(fs.create(path, true))) { + H5RootObject rootObject = H5.H5Screate(bos, cooRows, cooCols); + H5.H5Dcreate(rootObject, cooRows, cooCols, datasetName); + H5.H5WriteHeaders(rootObject); + + H5.H5Dwrite(rootObject, new double[] {(double) rlen, (double) clen, (double) nnz}); + + writeSparseCOOEntries(rootObject, src); + } + } + + private static void writeSparseCOOEntries(H5RootObject rootObject, MatrixBlock src) { + SparseBlock sb = src.getSparseBlock(); + int batchRows = getWriteBatchRows(3); + double[] batch = new double[batchRows * 3]; + + int pos = 0; + for(int i = 0; i < src.getNumRows(); i++) { + if(sb == null || sb.isEmpty(i)) + continue; + + int apos = sb.pos(i); + int alen = sb.size(i); + int[] aix = sb.indexes(i); + double[] avals = sb.values(i); + + for(int k = apos; k < apos + alen; k++) { + batch[pos++] = i; + batch[pos++] = aix[k]; + batch[pos++] = avals[k]; + + if(pos == batch.length) { + H5.H5Dwrite(rootObject, batch); + pos = 0; + } + } } + + if(pos > 0) + H5.H5Dwrite(rootObject, Arrays.copyOf(batch, pos)); } @Override diff --git a/src/main/java/org/apache/sysds/runtime/io/hdf5/H5.java b/src/main/java/org/apache/sysds/runtime/io/hdf5/H5.java index 0f640490ed6..576b6e49830 100644 --- a/src/main/java/org/apache/sysds/runtime/io/hdf5/H5.java +++ b/src/main/java/org/apache/sysds/runtime/io/hdf5/H5.java @@ -219,8 +219,8 @@ public static void H5WriteHeaders(H5RootObject rootObject) { public static void H5Dwrite(H5RootObject rootObject, double[] data) { try { H5BufferBuilder bb = new H5BufferBuilder(); - for(Double d : data) { - bb.writeDouble(d); + for(int i = 0; i < data.length; i++) { + bb.writeDouble(data[i]); } rootObject.getBufferedOutputStream().write(bb.noOrderBuild().array()); } diff --git a/src/test/java/org/apache/sysds/performance/Main.java b/src/test/java/org/apache/sysds/performance/Main.java index 0622e789baa..b3a4439e287 100644 --- a/src/test/java/org/apache/sysds/performance/Main.java +++ b/src/test/java/org/apache/sysds/performance/Main.java @@ -32,6 +32,8 @@ import org.apache.sysds.performance.generators.GenMatrices; import org.apache.sysds.performance.generators.IGenerate; import org.apache.sysds.performance.generators.MatrixFile; +import org.apache.sysds.performance.io.HDF5IOBenchmark; +import org.apache.sysds.performance.io.ParquetIOBenchmark; import org.apache.sysds.performance.matrix.MatrixAppend; import org.apache.sysds.performance.matrix.MatrixBinaryCellPerf; import org.apache.sysds.performance.matrix.MatrixMultiplicationPerf; @@ -147,6 +149,13 @@ private static void exec(int prog, String[] args) throws Exception { case 1009: MatrixMultiplicationPerf.main(args); break; + case 1010: + System.out.println("Starting ParquetIOBenchmark!"); + ParquetIOBenchmark.main(args); + break; + case 1011: + HDF5IOBenchmark.main(args); + break; default: break; } diff --git a/src/test/java/org/apache/sysds/performance/io/HDF5IOBenchmark.java b/src/test/java/org/apache/sysds/performance/io/HDF5IOBenchmark.java new file mode 100644 index 00000000000..352f33afe29 --- /dev/null +++ b/src/test/java/org/apache/sysds/performance/io/HDF5IOBenchmark.java @@ -0,0 +1,512 @@ +/* + * 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.performance.io; + +import java.io.IOException; +import java.lang.management.GarbageCollectorMXBean; +import java.lang.management.ManagementFactory; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.Locale; +import java.util.stream.Stream; + +import org.apache.sysds.hops.OptimizerUtils; +import org.apache.sysds.runtime.data.DenseBlock; +import org.apache.sysds.runtime.data.SparseBlock; +import org.apache.sysds.runtime.io.FileFormatPropertiesHDF5; +import org.apache.sysds.runtime.io.ReaderHDF5; +import org.apache.sysds.runtime.io.ReaderHDF5Parallel; +import org.apache.sysds.runtime.io.WriterHDF5; +import org.apache.sysds.runtime.io.WriterHDF5Parallel; +import org.apache.sysds.runtime.matrix.data.MatrixBlock; + +public class HDF5IOBenchmark { + private static final String PROFILE = "sysds.test.hdf5.profile"; + private static final String KEEP = "sysds.test.hdf5.keep.files"; + private static final String LABEL = "sysds.test.hdf5.benchmark.label"; + private static final String DATASET = "DATASET_1"; + + private static final int ROWS = Integer.getInteger("sysds.test.hdf5.rows", 250_000); + private static final int COLS = Integer.getInteger("sysds.test.hdf5.cols", 1_000); + private static final int BLOCK = Integer.getInteger("sysds.test.hdf5.block.size", 1024); + private static final int WARMUP = Integer.getInteger("sysds.test.hdf5.warmup.reps", 2); + private static final int REPS = Integer.getInteger("sysds.test.hdf5.measure.reps", 3); + private static final int SPARSE_NNZ_PER_ROW = Integer.getInteger("sysds.test.hdf5.sparse.nnz.per.row", 1); + + public static void main(String[] args) throws Exception { + checkProperties(); + + String profile = System.getProperty(PROFILE, "sparse").trim().toLowerCase(Locale.ROOT); + switch(profile) { + case "dense": + benchmarkDense(); + break; + case "sparse": + benchmarkSparse(); + break; + case "all": + benchmarkDense(); + benchmarkSparse(); + break; + default: + throw new IllegalArgumentException( + "Unsupported HDF5 benchmark profile: " + profile + ". Use dense, sparse, or all."); + } + } + + private static void benchmarkDense() throws Exception { + long nnz = Math.multiplyExact((long) ROWS, (long) COLS); + MatrixBlock mb = denseMatrix(); + run(new Profile("dense_double", nnz, false), mb); + } + + private static void benchmarkSparse() throws Exception { + long nnz = Math.multiplyExact((long) ROWS, (long) SPARSE_NNZ_PER_ROW); + MatrixBlock mb = sparseMatrix(nnz); + run(new Profile("sparse_double", nnz, true), mb); + } + + private static void run(Profile p, MatrixBlock input) throws Exception { + long cells = Math.multiplyExact((long) ROWS, (long) COLS); + long logicalDenseBytes = Math.multiplyExact(cells, (long) Double.BYTES); + + Path target = Paths.get("target").toAbsolutePath().normalize(); + Path work = target.resolve("hdf5-bench-" + p.name + "-" + System.currentTimeMillis()); + Files.createDirectories(work); + + Path csv = target.resolve("hdf5-bench-" + p.name + ".csv"); + Path json = target.resolve("hdf5-bench-" + p.name + ".json"); + + FileFormatPropertiesHDF5 props = new FileFormatPropertiesHDF5(DATASET); + WriterHDF5 seqWriter = new WriterHDF5(props); + WriterHDF5Parallel parWriter = new WriterHDF5Parallel(props); + ReaderHDF5 seqReader = new ReaderHDF5(props); + ReaderHDF5Parallel parReader = new ReaderHDF5Parallel(props); + + List results = new ArrayList<>(); + int total = WARMUP + REPS; + + try { + for(int rep = 0; rep < total; rep++) { + boolean warmup = rep < WARMUP; + int outRep = warmup ? rep : rep - WARMUP; + + Path seqPath = work.resolve(p.name + "_seq_" + rep + ".h5").toAbsolutePath().normalize(); + Path parPath = work.resolve(p.name + "_par_" + rep + ".h5").toAbsolutePath().normalize(); + + String seqFile = seqPath.toUri().toString(); + String parFile = parPath.toUri().toString(); + + Result seqWrite = measure(p, "seq", "", "hdf5_write", warmup, outRep, seqPath, logicalDenseBytes, + () -> seqWriter.writeMatrixToHDFS(input, seqFile, ROWS, COLS, BLOCK, p.nnz, false)); + results.add(seqWrite); + + if(seqWrite.ok()) { + results.add(measure(p, "seq", "seq", "hdf5_read", warmup, outRep, seqPath, logicalDenseBytes, + () -> validate(p, seqReader.readMatrixFromHDFS(seqFile, ROWS, COLS, BLOCK, p.nnz)))); + results.add(measure(p, "seq", "par", "hdf5_read", warmup, outRep, seqPath, logicalDenseBytes, + () -> validate(p, parReader.readMatrixFromHDFS(seqFile, ROWS, COLS, BLOCK, p.nnz)))); + } + else { + results.add(skip(p, "seq", "seq", "hdf5_read", warmup, outRep, seqPath, logicalDenseBytes)); + results.add(skip(p, "seq", "par", "hdf5_read", warmup, outRep, seqPath, logicalDenseBytes)); + } + + Result parWrite = measure(p, "par", "", "hdf5_write", warmup, outRep, parPath, logicalDenseBytes, + () -> parWriter.writeMatrixToHDFS(input, parFile, ROWS, COLS, BLOCK, p.nnz, false)); + results.add(parWrite); + + if(parWrite.ok()) { + results.add(measure(p, "par", "seq", "hdf5_read", warmup, outRep, parPath, logicalDenseBytes, + () -> validate(p, seqReader.readMatrixFromHDFS(parFile, ROWS, COLS, BLOCK, p.nnz)))); + results.add(measure(p, "par", "par", "hdf5_read", warmup, outRep, parPath, logicalDenseBytes, + () -> validate(p, parReader.readMatrixFromHDFS(parFile, ROWS, COLS, BLOCK, p.nnz)))); + } + else { + results.add(skip(p, "par", "seq", "hdf5_read", warmup, outRep, parPath, logicalDenseBytes)); + results.add(skip(p, "par", "par", "hdf5_read", warmup, outRep, parPath, logicalDenseBytes)); + } + } + } + finally { + writeCsv(results, csv); + writeJson(results, json); + + System.out.println("HDF5 benchmark CSV: " + csv); + System.out.println("HDF5 benchmark JSON: " + json); + System.out.println("HDF5 benchmark work: " + work); + + if(!Boolean.parseBoolean(System.getProperty(KEEP, "false"))) + deleteQuietly(work); + } + } + + private static Result measure(Profile p, String writer, String reader, String op, boolean warmup, int rep, + Path path, long logicalDenseBytes, CheckedRunnable action) throws Exception { + gc(); + + long heap0 = usedHeap(); + Gc gc0 = Gc.now(); + long t0 = System.nanoTime(); + + Result res = baseResult(p, writer, reader, op, warmup, rep, path, logicalDenseBytes); + try { + action.run(); + res.status = "PASS"; + } + catch(Exception | AssertionError ex) { + res.status = "FAIL"; + res.errorClass = ex.getClass().getName(); + res.errorMessage = ex.getMessage(); + } + + long t1 = System.nanoTime(); + Gc gc1 = Gc.now(); + long heap1 = usedHeap(); + + res.wallMs = (t1 - t0) / 1_000_000.0; + res.heapBefore = heap0; + res.heapAfter = heap1; + res.heapDelta = heap1 - heap0; + res.gcCount = gc1.count - gc0.count; + res.gcMs = gc1.ms - gc0.ms; + res.fileSize = Files.exists(path) ? fileSize(path) : 0; + res.numFiles = Files.exists(path) ? numFiles(path) : 0; + return res; + } + + private static Result skip(Profile p, String writer, String reader, String op, boolean warmup, int rep, Path path, + long logicalDenseBytes) throws IOException { + Result res = baseResult(p, writer, reader, op, warmup, rep, path, logicalDenseBytes); + res.status = "SKIP"; + res.errorMessage = "writer failed"; + res.fileSize = Files.exists(path) ? fileSize(path) : 0; + res.numFiles = Files.exists(path) ? numFiles(path) : 0; + return res; + } + + private static Result baseResult(Profile p, String writer, String reader, String op, boolean warmup, int rep, + Path path, long logicalDenseBytes) { + Result r = new Result(); + r.label = System.getProperty(LABEL, "base"); + r.sparseLayout = System.getProperty("sysds.hdf5.write.sparse.layout", "dense"); + r.profile = p.name; + r.writer = writer; + r.reader = reader; + r.operation = op; + r.status = "RUNNING"; + r.rows = ROWS; + r.cols = COLS; + r.cells = Math.multiplyExact((long) ROWS, (long) COLS); + r.nnz = p.nnz; + r.sparsity = r.cells == 0 ? 0 : (double) p.nnz / r.cells; + r.rep = rep; + r.warmup = warmup; + r.logicalDenseBytes = logicalDenseBytes; + r.readParallelism = OptimizerUtils.getParallelBinaryReadParallelism(); + r.writeParallelism = OptimizerUtils.getParallelTextWriteParallelism(); + r.path = path.toString(); + return r; + } + + private static MatrixBlock denseMatrix() { + MatrixBlock mb = new MatrixBlock(ROWS, COLS, false); + mb.allocateDenseBlockUnsafe(ROWS, COLS); + + DenseBlock db = mb.getDenseBlock(); + for(int i = 0; i < ROWS; i++) + for(int j = 0; j < COLS; j++) + db.set(i, j, denseValue(i, j)); + + mb.setNonZeros(Math.multiplyExact((long) ROWS, (long) COLS)); + mb.examSparsity(); + return mb; + } + + private static MatrixBlock sparseMatrix(long nnz) { + MatrixBlock mb = new MatrixBlock(ROWS, COLS, true, nnz); + mb.allocateSparseRowsBlock(); + + SparseBlock sb = mb.getSparseBlock(); + for(int i = 0; i < ROWS; i++) { + sb.allocate(i, SPARSE_NNZ_PER_ROW); + for(int j = 0; j < SPARSE_NNZ_PER_ROW; j++) + sb.append(i, j, sparseValue(i, j)); + } + + mb.setNonZeros(nnz); + mb.examSparsity(); + check(mb.isInSparseFormat(), "Sparse input converted to dense."); + return mb; + } + + private static void validate(Profile p, MatrixBlock mb) { + checkEquals(ROWS, mb.getNumRows(), "Unexpected number of rows."); + checkEquals(COLS, mb.getNumColumns(), "Unexpected number of columns."); + checkEquals(p.nnz, mb.getNonZeros(), "Unexpected number of nonzeros."); + + if(p.sparse) { + int lastNnzCol = SPARSE_NNZ_PER_ROW - 1; + check(mb.isInSparseFormat(), "Expected sparse output."); + checkEquals(sparseValue(0, 0), value(mb, 0, 0), "Unexpected sparse value at first row."); + checkEquals(sparseValue(ROWS / 2, 0), value(mb, ROWS / 2, 0), "Unexpected sparse value at middle row."); + checkEquals(sparseValue(ROWS - 1, 0), value(mb, ROWS - 1, 0), "Unexpected sparse value at last row."); + checkEquals(sparseValue(ROWS / 2, lastNnzCol), value(mb, ROWS / 2, lastNnzCol), + "Unexpected sparse value at last nonzero column."); + + if(SPARSE_NNZ_PER_ROW < COLS) + checkEquals(0, value(mb, ROWS - 1, SPARSE_NNZ_PER_ROW), "Expected zero after sparse nonzero range."); + } + else { + check(!mb.isInSparseFormat(), "Expected dense output."); + checkEquals(denseValue(0, 0), value(mb, 0, 0), "Unexpected dense value at first cell."); + checkEquals(denseValue(ROWS / 2, COLS / 2), value(mb, ROWS / 2, COLS / 2), + "Unexpected dense value at middle cell."); + checkEquals(denseValue(ROWS - 1, COLS - 1), value(mb, ROWS - 1, COLS - 1), + "Unexpected dense value at last cell."); + } + } + + private static double denseValue(int row, int col) { + return 1.0 + row * 1000.0 + col; + } + + private static double sparseValue(int row, int col) { + return col < SPARSE_NNZ_PER_ROW ? denseValue(row, col) : 0.0; + } + + private static double value(MatrixBlock mb, int row, int col) { + if(!mb.isInSparseFormat()) + return mb.getDenseBlock().get(row, col); + + SparseBlock sb = mb.getSparseBlock(); + if(sb == null || sb.isEmpty(row)) + return 0; + + int pos = sb.pos(row); + int end = pos + sb.size(row); + int[] ix = sb.indexes(row); + double[] vals = sb.values(row); + + for(int p = pos; p < end; p++) + if(ix[p] == col) + return vals[p]; + return 0; + } + + private static void checkProperties() { + if(ROWS <= 0 || COLS <= 0 || BLOCK <= 0) + throw new IllegalArgumentException("rows, cols, and block size must be positive."); + if(WARMUP < 0 || REPS <= 0) + throw new IllegalArgumentException("warmup must be >= 0 and reps must be > 0."); + if(SPARSE_NNZ_PER_ROW <= 0 || SPARSE_NNZ_PER_ROW > COLS) + throw new IllegalArgumentException("sparse.nnz.per.row must be in [1, cols]."); + } + + private static void check(boolean condition, String message) { + if(!condition) + throw new IllegalStateException(message); + } + + private static void checkEquals(long expected, long actual, String message) { + if(expected != actual) + throw new IllegalStateException(message + " Expected=" + expected + ", actual=" + actual + "."); + } + + private static void checkEquals(double expected, double actual, String message) { + if(Double.compare(expected, actual) != 0) + throw new IllegalStateException(message + " Expected=" + expected + ", actual=" + actual + "."); + } + + private static long usedHeap() { + Runtime rt = Runtime.getRuntime(); + return rt.totalMemory() - rt.freeMemory(); + } + + private static void gc() { + System.gc(); + try { + Thread.sleep(100); + } + catch(InterruptedException ex) { + Thread.currentThread().interrupt(); + } + } + + private static long fileSize(Path p) throws IOException { + if(Files.isRegularFile(p)) + return Files.size(p); + if(!Files.isDirectory(p)) + return 0; + + final long[] ret = new long[] {0}; + try(Stream s = Files.walk(p)) { + s.filter(Files::isRegularFile).forEach(x -> { + try { + ret[0] += Files.size(x); + } + catch(IOException ex) { + throw new RuntimeException(ex); + } + }); + } + return ret[0]; + } + + private static int numFiles(Path p) throws IOException { + if(Files.isRegularFile(p)) + return 1; + if(!Files.isDirectory(p)) + return 0; + + final int[] ret = new int[] {0}; + try(Stream s = Files.walk(p)) { + s.filter(Files::isRegularFile).forEach(x -> ret[0]++); + } + return ret[0]; + } + + private static void deleteQuietly(Path p) { + try { + if(!Files.exists(p)) + return; + try(Stream s = Files.walk(p)) { + s.sorted(Comparator.reverseOrder()).forEach(x -> { + try { + Files.deleteIfExists(x); + } + catch(IOException ex) { + throw new RuntimeException(ex); + } + }); + } + } + catch(Exception ex) { + System.err.println("Could not delete benchmark directory: " + p); + } + } + + private static void writeCsv(List results, Path out) throws IOException { + StringBuilder sb = new StringBuilder(); + sb.append("label,sparse_layout,profile,writer,reader,operation,status,rep,warmup,rows,cols,cells,nnz,sparsity,") + .append("wall_ms,file_size,num_files,logical_dense_bytes,heap_before,heap_after,heap_delta,") + .append("gc_count,gc_ms,read_parallelism,write_parallelism,path,error_class,error_message\n"); + + for(Result r : results) + sb.append(csv(r.label)).append(',').append(csv(r.sparseLayout)).append(',').append(csv(r.profile)) + .append(',').append(csv(r.writer)).append(',').append(csv(r.reader)).append(',') + .append(csv(r.operation)).append(',').append(csv(r.status)).append(',').append(r.rep).append(',') + .append(r.warmup).append(',').append(r.rows).append(',').append(r.cols).append(',').append(r.cells) + .append(',').append(r.nnz).append(',').append(String.format(Locale.US, "%.8f", r.sparsity)).append(',') + .append(String.format(Locale.US, "%.3f", r.wallMs)).append(',').append(r.fileSize).append(',') + .append(r.numFiles).append(',').append(r.logicalDenseBytes).append(',').append(r.heapBefore).append(',') + .append(r.heapAfter).append(',').append(r.heapDelta).append(',').append(r.gcCount).append(',') + .append(r.gcMs).append(',').append(r.readParallelism).append(',').append(r.writeParallelism).append(',') + .append(csv(r.path)).append(',').append(csv(r.errorClass)).append(',').append(csv(r.errorMessage)) + .append('\n'); + + Files.write(out, sb.toString().getBytes(StandardCharsets.UTF_8)); + } + + private static void writeJson(List results, Path out) throws IOException { + StringBuilder sb = new StringBuilder("[\n"); + for(int i = 0; i < results.size(); i++) { + if(i > 0) + sb.append(",\n"); + sb.append(results.get(i).json()); + } + sb.append("\n]\n"); + Files.write(out, sb.toString().getBytes(StandardCharsets.UTF_8)); + } + + private static String csv(String s) { + return s == null ? "" : "\"" + s.replace("\"", "\"\"") + "\""; + } + + private static String json(String s) { + return s == null ? "null" : "\"" + s.replace("\\", "\\\\").replace("\"", "\\\"") + "\""; + } + + private interface CheckedRunnable { + void run() throws Exception; + } + + private static class Profile { + final String name; + final long nnz; + final boolean sparse; + + Profile(String name, long nnz, boolean sparse) { + this.name = name; + this.nnz = nnz; + this.sparse = sparse; + } + } + + private static class Gc { + long count; + long ms; + + static Gc now() { + Gc g = new Gc(); + for(GarbageCollectorMXBean b : ManagementFactory.getGarbageCollectorMXBeans()) { + long c = b.getCollectionCount(); + long t = b.getCollectionTime(); + if(c > 0) + g.count += c; + if(t > 0) + g.ms += t; + } + return g; + } + } + + private static class Result { + String label, sparseLayout, profile, writer, reader, operation, status, path, errorClass, errorMessage; + int rows, cols, rep, numFiles, readParallelism, writeParallelism; + long cells, nnz, fileSize, logicalDenseBytes, heapBefore, heapAfter, heapDelta, gcCount, gcMs; + boolean warmup; + double sparsity, wallMs; + + boolean ok() { + return "PASS".equals(status); + } + + String json() { + return "{" + "\"label\":" + HDF5IOBenchmark.json(label) + ",\"sparse_layout\":" + + HDF5IOBenchmark.json(sparseLayout) + ",\"profile\":" + HDF5IOBenchmark.json(profile) + ",\"writer\":" + + HDF5IOBenchmark.json(writer) + ",\"reader\":" + HDF5IOBenchmark.json(reader) + ",\"operation\":" + + HDF5IOBenchmark.json(operation) + ",\"status\":" + HDF5IOBenchmark.json(status) + ",\"rep\":" + rep + + ",\"warmup\":" + warmup + ",\"rows\":" + rows + ",\"cols\":" + cols + ",\"cells\":" + cells + + ",\"nnz\":" + nnz + ",\"sparsity\":" + String.format(Locale.US, "%.8f", sparsity) + ",\"wall_ms\":" + + String.format(Locale.US, "%.3f", wallMs) + ",\"file_size\":" + fileSize + ",\"num_files\":" + numFiles + + ",\"logical_dense_bytes\":" + logicalDenseBytes + ",\"heap_before\":" + heapBefore + + ",\"heap_after\":" + heapAfter + ",\"heap_delta\":" + heapDelta + ",\"gc_count\":" + gcCount + + ",\"gc_ms\":" + gcMs + ",\"read_parallelism\":" + readParallelism + ",\"write_parallelism\":" + + writeParallelism + ",\"path\":" + HDF5IOBenchmark.json(path) + ",\"error_class\":" + + HDF5IOBenchmark.json(errorClass) + ",\"error_message\":" + HDF5IOBenchmark.json(errorMessage) + "}"; + } + } +} diff --git a/src/test/java/org/apache/sysds/performance/io/ParquetIOBenchmark.java b/src/test/java/org/apache/sysds/performance/io/ParquetIOBenchmark.java new file mode 100644 index 00000000000..e796080dd2d --- /dev/null +++ b/src/test/java/org/apache/sysds/performance/io/ParquetIOBenchmark.java @@ -0,0 +1,833 @@ +/* + * 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.performance.io; + +import java.io.File; +import java.io.FileWriter; +import java.io.IOException; +import java.io.PrintWriter; +import java.lang.management.GarbageCollectorMXBean; +import java.lang.management.ManagementFactory; +import java.util.ArrayList; +import java.util.List; +import java.util.Locale; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FSDataInputStream; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.hadoop.ParquetFileReader; +import org.apache.parquet.hadoop.metadata.ParquetMetadata; +import org.apache.parquet.hadoop.util.HadoopInputFile; +import org.apache.sysds.common.Types.ValueType; +import org.apache.sysds.conf.ConfigurationManager; +import org.apache.sysds.hops.OptimizerUtils; +import org.apache.sysds.runtime.frame.data.FrameBlock; +import org.apache.sysds.runtime.io.FrameReader; +import org.apache.sysds.runtime.io.FrameReaderParquet; +import org.apache.sysds.runtime.io.FrameReaderParquetParallel; +import org.apache.sysds.runtime.io.FrameWriter; +import org.apache.sysds.runtime.io.FrameWriterParquet; +import org.apache.sysds.runtime.io.FrameWriterParquetParallel; +import org.apache.sysds.utils.stats.InfrastructureAnalyzer; + +public class ParquetIOBenchmark { + private static final String PROFILE_DENSE_FP64 = "dense_fp64_only"; + private static final String PROFILE_MIXED_SCHEMA = "mixed_schema"; + private static final String PROFILE_SPARSE_LIKE_FP64 = "sparse_like_fp64"; + + private static volatile long _blackhole = 0; + + private final File resultDir; + private final File workDir; + + public ParquetIOBenchmark() { + resultDir = new File("target/performance/parquet"); + workDir = new File(resultDir, "work"); + } + + public static void main(String[] args) throws Exception { + new ParquetIOBenchmark().run(); + } + + public void run() throws Exception { + final int rows = getIntProperty("sysds.performance.parquet.rows", "sysds.test.parquet.rows", 100_000); + final int cols = getIntProperty("sysds.performance.parquet.cols", "sysds.test.parquet.cols", 50); + final int warmup = getIntProperty("sysds.performance.parquet.warmup", "sysds.test.parquet.warmup", 1); + final int reps = getIntProperty("sysds.performance.parquet.reps", "sysds.test.parquet.reps", 3); + + final String[] profiles = parseProfileList( + getStringProperty("sysds.performance.parquet.profiles", "sysds.test.parquet.profiles", + PROFILE_DENSE_FP64 + "," + PROFILE_MIXED_SCHEMA + "," + PROFILE_SPARSE_LIKE_FP64)); + + final int[] manualParts = parsePositiveIntList( + getStringProperty("sysds.performance.parquet.manual.parts", "sysds.test.parquet.manual.parts", "")); + + if(rows <= 0 || cols <= 0) + throw new IllegalArgumentException("Rows and columns must be positive."); + if(warmup < 0 || reps <= 0) + throw new IllegalArgumentException("Warmup must be >= 0 and reps must be > 0."); + + if(!resultDir.exists() && !resultDir.mkdirs()) + throw new IOException("Failed to create result directory: " + resultDir.getAbsolutePath()); + if(!workDir.exists() && !workDir.mkdirs()) + throw new IOException("Failed to create work directory: " + workDir.getAbsolutePath()); + + File csvFile = new File(resultDir, "parquet-benchmark.csv"); + File jsonFile = new File(resultDir, "parquet-benchmark.json"); + + System.out.println("Running ParquetIOBenchmark"); + System.out.println("rows=" + rows + ", cols=" + cols + ", warmup=" + warmup + ", reps=" + reps); + System.out.println("results=" + resultDir.getAbsolutePath()); + + try(PrintWriter csv = new PrintWriter(new FileWriter(csvFile)); + PrintWriter json = new PrintWriter(new FileWriter(jsonFile))) { + writeCsvHeader(csv); + + json.println("["); + JsonState jsonState = new JsonState(); + + for(String dataProfile : profiles) { + System.out.println("Benchmarking profile: " + dataProfile); + FrameBlock frameBlock = generateFrame(dataProfile, rows, cols); + benchmarkProfile(csv, json, jsonState, dataProfile, frameBlock, warmup, reps); + + if(manualParts.length > 0) + benchmarkManualMultipart(csv, json, jsonState, dataProfile, frameBlock, manualParts, warmup, reps); + } + + json.println(); + json.println("]"); + } + + System.out.println("Parquet CSV benchmark results saved under: " + csvFile.getAbsolutePath()); + System.out.println("Parquet JSON benchmark results saved under: " + jsonFile.getAbsolutePath()); + } + + private void benchmarkProfile(PrintWriter csv, PrintWriter json, JsonState jsonState, String dataProfile, + FrameBlock frameBlock, int warmup, int reps) throws Exception { + int rows = frameBlock.getNumRows(); + int cols = frameBlock.getNumColumns(); + String profileName = safeName(dataProfile); + + String seqInput = output("parquet_" + profileName + "_seq_input"); + new FrameWriterParquet().writeFrameToHDFS(frameBlock, seqInput, rows, cols); + + String parallelInput = output("parquet_" + profileName + "_parallel_input"); + new FrameWriterParquetParallel().writeFrameToHDFS(frameBlock, parallelInput, rows, cols); + + benchmarkWrite(csv, json, jsonState, dataProfile, "seq", new FrameWriterParquet(), frameBlock, warmup, reps, + ""); + benchmarkRawIORead(csv, json, jsonState, dataProfile, "seq", seqInput, frameBlock, warmup, reps, + "raw_bytes_only"); + benchmarkFooterRead(csv, json, jsonState, dataProfile, "seq", seqInput, frameBlock, warmup, reps, + "parquet_footer_only"); + benchmarkRead(csv, json, jsonState, dataProfile, "seq", new FrameReaderParquet(), seqInput, frameBlock, warmup, + reps, ""); + + benchmarkWrite(csv, json, jsonState, dataProfile, "parallel", new FrameWriterParquetParallel(), frameBlock, + warmup, reps, "current_parallel_write_path"); + benchmarkRawIORead(csv, json, jsonState, dataProfile, "parallel", parallelInput, frameBlock, warmup, reps, + "raw_bytes_only"); + benchmarkFooterRead(csv, json, jsonState, dataProfile, "parallel", parallelInput, frameBlock, warmup, reps, + "parquet_footer_only"); + benchmarkRead(csv, json, jsonState, dataProfile, "parallel", new FrameReaderParquetParallel(), parallelInput, + frameBlock, warmup, reps, "current_parallel_read_path_not_value_validated"); + } + + private void benchmarkManualMultipart(PrintWriter csv, PrintWriter json, JsonState jsonState, String dataProfile, + FrameBlock frameBlock, int[] manualParts, int warmup, int reps) throws Exception { + for(int numParts : manualParts) { + if(numParts <= 1 || numParts > frameBlock.getNumRows()) + continue; + + String manualInput = output("parquet_" + safeName(dataProfile) + "_manual_multipart_" + numParts); + writeManualMultipartInput(frameBlock, manualInput, numParts); + + String impl = "parallel_manual_parts_" + numParts; + benchmarkRawIORead(csv, json, jsonState, dataProfile, impl, manualInput, frameBlock, warmup, reps, + "raw_bytes_only"); + benchmarkFooterRead(csv, json, jsonState, dataProfile, impl, manualInput, frameBlock, warmup, reps, + "parquet_footer_only"); + benchmarkRead(csv, json, jsonState, dataProfile, impl, new FrameReaderParquetParallel(), manualInput, + frameBlock, warmup, reps, "manual_multipart_input_not_value_validated"); + } + } + + private void writeCsvHeader(PrintWriter csv) { + csv.println("data_profile,impl,operation,rows,cols,cells,rep,is_warmup,wall_ms," + + "file_size_bytes,num_part_files,avg_part_file_size_bytes," + + "hdfs_block_size_bytes,legacy_cell_based_estimated_num_part_files," + + "legacy_cell_based_estimated_num_threads," + + "configured_parallel_read_parallelism,configured_parallel_write_parallelism," + + "estimated_parallel_read_tasks," + + "dense_fp64_reference_size_bytes,actual_file_to_dense_fp64_reference_ratio," + + "expected_nonzero_fraction," + "reader_value_extraction_mode,frame_materialization," + + "heap_before_bytes,heap_after_bytes,heap_delta_bytes," + "gc_count_delta,gc_time_delta_ms,notes"); + } + + private void benchmarkWrite(PrintWriter csv, PrintWriter json, JsonState jsonState, String dataProfile, String impl, + FrameWriter writer, FrameBlock frameBlock, int warmup, int reps, String notes) throws Exception { + int rows = frameBlock.getNumRows(); + int cols = frameBlock.getNumColumns(); + + for(int i = 0; i < warmup + reps; i++) { + boolean isWarmup = i < warmup; + String path = output("parquet_" + safeName(dataProfile) + "_" + impl + "_write_" + i); + + long heapBefore = usedHeap(); + GcStats gcBefore = getGcStats(); + + long t0 = System.nanoTime(); + writer.writeFrameToHDFS(frameBlock, path, rows, cols); + long t1 = System.nanoTime(); + + GcStats gcAfter = getGcStats(); + long heapAfter = usedHeap(); + PathStats pathStats = getPathStats(path); + + writeResult(csv, json, jsonState, dataProfile, impl, "write", rows, cols, i, isWarmup, t0, t1, + pathStats.fileSizeBytes, pathStats.numPartFiles, heapBefore, heapAfter, gcAfter.count - gcBefore.count, + gcAfter.timeMs - gcBefore.timeMs, notes); + } + } + + private void benchmarkRead(PrintWriter csv, PrintWriter json, JsonState jsonState, String dataProfile, String impl, + FrameReader reader, String path, FrameBlock reference, int warmup, int reps, String notes) throws Exception { + int rows = reference.getNumRows(); + int cols = reference.getNumColumns(); + ValueType[] schema = reference.getSchema(); + String[] names = reference.getColumnNames(); + + for(int i = 0; i < warmup + reps; i++) { + boolean isWarmup = i < warmup; + PathStats pathStats = getPathStats(path); + + long heapBefore = usedHeap(); + GcStats gcBefore = getGcStats(); + + long t0 = System.nanoTime(); + FrameBlock ret = reader.readFrameFromHDFS(path, schema, names, rows, cols); + long t1 = System.nanoTime(); + + GcStats gcAfter = getGcStats(); + long heapAfter = usedHeap(); + blackhole(ret); + + writeResult(csv, json, jsonState, dataProfile, impl, "read", rows, cols, i, isWarmup, t0, t1, + pathStats.fileSizeBytes, pathStats.numPartFiles, heapBefore, heapAfter, gcAfter.count - gcBefore.count, + gcAfter.timeMs - gcBefore.timeMs, notes); + } + } + + private void benchmarkRawIORead(PrintWriter csv, PrintWriter json, JsonState jsonState, String dataProfile, + String impl, String path, FrameBlock reference, int warmup, int reps, String notes) throws Exception { + int rows = reference.getNumRows(); + int cols = reference.getNumColumns(); + + for(int i = 0; i < warmup + reps; i++) { + boolean isWarmup = i < warmup; + PathStats pathStats = getPathStats(path); + + long heapBefore = usedHeap(); + GcStats gcBefore = getGcStats(); + + long t0 = System.nanoTime(); + long bytesRead = readRawBytes(path); + long t1 = System.nanoTime(); + + GcStats gcAfter = getGcStats(); + long heapAfter = usedHeap(); + blackhole(bytesRead); + + writeResult(csv, json, jsonState, dataProfile, impl, "raw_io_read", rows, cols, i, isWarmup, t0, t1, + pathStats.fileSizeBytes, pathStats.numPartFiles, heapBefore, heapAfter, gcAfter.count - gcBefore.count, + gcAfter.timeMs - gcBefore.timeMs, notes); + } + } + + private void benchmarkFooterRead(PrintWriter csv, PrintWriter json, JsonState jsonState, String dataProfile, + String impl, String path, FrameBlock reference, int warmup, int reps, String notes) throws Exception { + int rows = reference.getNumRows(); + int cols = reference.getNumColumns(); + + for(int i = 0; i < warmup + reps; i++) { + boolean isWarmup = i < warmup; + PathStats pathStats = getPathStats(path); + + long heapBefore = usedHeap(); + GcStats gcBefore = getGcStats(); + + long t0 = System.nanoTime(); + long footerInfo = readParquetFooters(path); + long t1 = System.nanoTime(); + + GcStats gcAfter = getGcStats(); + long heapAfter = usedHeap(); + blackhole(footerInfo); + + writeResult(csv, json, jsonState, dataProfile, impl, "footer_read", rows, cols, i, isWarmup, t0, t1, + pathStats.fileSizeBytes, pathStats.numPartFiles, heapBefore, heapAfter, gcAfter.count - gcBefore.count, + gcAfter.timeMs - gcBefore.timeMs, notes); + } + } + + private void writeManualMultipartInput(FrameBlock frameBlock, String dirName, int numParts) throws Exception { + Configuration conf = ConfigurationManager.getCachedJobConf(); + Path dir = new Path(dirName); + FileSystem fs = dir.getFileSystem(conf); + + if(fs.exists(dir)) + fs.delete(dir, true); + + fs.mkdirs(dir); + + FrameWriter writer = new FrameWriterParquet(); + int rows = frameBlock.getNumRows(); + int cols = frameBlock.getNumColumns(); + int chunkSize = (int) Math.ceil((double) rows / numParts); + + for(int part = 0; part < numParts; part++) { + int startRow = part * chunkSize; + int endRow = Math.min((part + 1) * chunkSize, rows); + + if(startRow >= endRow) + continue; + + FrameBlock slice = frameBlock.slice(startRow, endRow - 1); + Path partPath = new Path(dir, getManualPartFileName(part)); + writer.writeFrameToHDFS(slice, partPath.toString(), slice.getNumRows(), cols); + } + } + + private String getManualPartFileName(int part) { + return String.format(Locale.US, "part-%05d", part); + } + + private String output(String name) { + return new File(workDir, name).toURI().toString(); + } + + private void writeResult(PrintWriter csv, PrintWriter json, JsonState jsonState, String dataProfile, String impl, + String operation, int rows, int cols, int rep, boolean isWarmup, long t0, long t1, long fileSizeBytes, + int numPartFiles, long heapBefore, long heapAfter, long gcCountDelta, long gcTimeDeltaMs, String notes) { + double wallMs = (t1 - t0) / 1e6; + BenchmarkDiagnostics diag = getBenchmarkDiagnostics(dataProfile, impl, operation, rows, cols, fileSizeBytes, + numPartFiles); + + csv.printf(Locale.US, + "%s,%s,%s,%d,%d,%d,%d,%s,%.3f,%d,%d,%.3f,%d,%d,%d,%d,%d,%d,%d,%.6f,%.6f,%s,%s,%d,%d,%d,%d,%d,%s%n", + escapeCsv(dataProfile), escapeCsv(impl), escapeCsv(operation), rows, cols, diag.cells, rep, isWarmup, + wallMs, fileSizeBytes, numPartFiles, diag.avgPartFileSizeBytes, diag.hdfsBlockSizeBytes, + diag.currentWriterEstimatedNumPartFiles, diag.currentWriterEstimatedNumThreads, + diag.configuredParallelReadParallelism, diag.configuredParallelWriteParallelism, + diag.estimatedParallelReadTasks, diag.denseFp64ReferenceSizeBytes, diag.actualFileToDenseFp64ReferenceRatio, + diag.expectedNonZeroFraction, escapeCsv(diag.readerValueExtractionMode), + escapeCsv(diag.frameMaterialization), heapBefore, heapAfter, heapAfter - heapBefore, gcCountDelta, + gcTimeDeltaMs, escapeCsv(notes)); + csv.flush(); + + if(jsonState.hasEntries) + json.println(","); + else + jsonState.hasEntries = true; + + json.printf(Locale.US, + " {\"data_profile\":\"%s\",\"impl\":\"%s\",\"operation\":\"%s\"," + "\"rows\":%d,\"cols\":%d,\"cells\":%d," + + "\"rep\":%d,\"is_warmup\":%s,\"wall_ms\":%.3f," + "\"file_size_bytes\":%d,\"num_part_files\":%d," + + "\"avg_part_file_size_bytes\":%.3f," + "\"hdfs_block_size_bytes\":%d," + + "\"legacy_cell_based_estimated_num_part_files\":%d," + + "\"legacy_cell_based_estimated_num_threads\":%d," + "\"configured_parallel_read_parallelism\":%d," + + "\"configured_parallel_write_parallelism\":%d," + "\"estimated_parallel_read_tasks\":%d," + + "\"dense_fp64_reference_size_bytes\":%d," + "\"actual_file_to_dense_fp64_reference_ratio\":%.6f," + + "\"expected_nonzero_fraction\":%.6f," + "\"reader_value_extraction_mode\":\"%s\"," + + "\"frame_materialization\":\"%s\"," + "\"heap_before_bytes\":%d,\"heap_after_bytes\":%d," + + "\"heap_delta_bytes\":%d," + "\"gc_count_delta\":%d,\"gc_time_delta_ms\":%d," + "\"notes\":\"%s\"}", + escapeJson(dataProfile), escapeJson(impl), escapeJson(operation), rows, cols, diag.cells, rep, isWarmup, + wallMs, fileSizeBytes, numPartFiles, diag.avgPartFileSizeBytes, diag.hdfsBlockSizeBytes, + diag.currentWriterEstimatedNumPartFiles, diag.currentWriterEstimatedNumThreads, + diag.configuredParallelReadParallelism, diag.configuredParallelWriteParallelism, + diag.estimatedParallelReadTasks, diag.denseFp64ReferenceSizeBytes, diag.actualFileToDenseFp64ReferenceRatio, + diag.expectedNonZeroFraction, escapeJson(diag.readerValueExtractionMode), + escapeJson(diag.frameMaterialization), heapBefore, heapAfter, heapAfter - heapBefore, gcCountDelta, + gcTimeDeltaMs, escapeJson(notes)); + json.flush(); + } + + private BenchmarkDiagnostics getBenchmarkDiagnostics(String dataProfile, String impl, String operation, int rows, + int cols, long fileSizeBytes, int numPartFiles) { + long cells = (long) rows * cols; + long hdfsBlockSize = InfrastructureAnalyzer.getHDFSBlockSize(); + + int configuredParallelReadParallelism = OptimizerUtils.getParallelBinaryReadParallelism(); + int configuredParallelWriteParallelism = OptimizerUtils.getParallelBinaryWriteParallelism(); + + long currentWriterEstimatedNumPartFiles = hdfsBlockSize > 0 ? Math.max(cells / hdfsBlockSize, 1) : 1; + int currentWriterEstimatedNumThreads = (int) Math.min(configuredParallelWriteParallelism, + currentWriterEstimatedNumPartFiles); + int estimatedParallelReadTasks = estimateParallelReadTasks(impl, operation, numPartFiles, + configuredParallelReadParallelism); + + long denseFp64ReferenceSizeBytes = cells * 8; + double avgPartFileSizeBytes = numPartFiles > 0 ? ((double) fileSizeBytes / numPartFiles) : -1.0; + double actualFileToDenseFp64ReferenceRatio = denseFp64ReferenceSizeBytes > 0 ? ((double) fileSizeBytes / + denseFp64ReferenceSizeBytes) : -1.0; + String readerValueExtractionMode = getReaderValueExtractionMode(impl, operation); + String frameMaterialization = "FrameBlock"; + double expectedNonZeroFraction = getExpectedNonZeroFraction(dataProfile); + + return new BenchmarkDiagnostics(cells, hdfsBlockSize, currentWriterEstimatedNumPartFiles, + currentWriterEstimatedNumThreads, configuredParallelReadParallelism, configuredParallelWriteParallelism, + estimatedParallelReadTasks, denseFp64ReferenceSizeBytes, avgPartFileSizeBytes, + actualFileToDenseFp64ReferenceRatio, expectedNonZeroFraction, readerValueExtractionMode, + frameMaterialization); + } + + private int estimateParallelReadTasks(String impl, String operation, int numPartFiles, + int configuredParallelReadParallelism) { + if(!"read".equals(operation)) + return 0; + if(!impl.startsWith("parallel")) + return 1; + if(numPartFiles <= 0) + return 0; + return Math.min(configuredParallelReadParallelism, numPartFiles); + } + + private String getReaderValueExtractionMode(String impl, String operation) { + if(!"read".equals(operation)) + return "not_applicable"; + if("seq".equals(impl)) + return "typed_getters"; + if(impl.startsWith("parallel")) + return "typed_getters"; + return "unknown"; + } + + private double getExpectedNonZeroFraction(String dataProfile) { + String profile = normalizeProfileName(dataProfile); + if(PROFILE_DENSE_FP64.equals(profile)) + return 1.0; + else if(PROFILE_SPARSE_LIKE_FP64.equals(profile)) + return 0.05; + else + return -1.0; + } + + private FrameBlock generateFrame(String dataProfile, int rows, int cols) { + String profile = normalizeProfileName(dataProfile); + if(PROFILE_DENSE_FP64.equals(profile)) + return generateDenseFp64Frame(rows, cols); + else if(PROFILE_MIXED_SCHEMA.equals(profile)) + return generateMixedSchemaFrame(rows, cols); + else if(PROFILE_SPARSE_LIKE_FP64.equals(profile)) + return generateSparseLikeFp64Frame(rows, cols); + else + throw new IllegalArgumentException("Unknown Parquet benchmark data profile: " + dataProfile); + } + + private FrameBlock generateDenseFp64Frame(int rows, int cols) { + ValueType[] schema = new ValueType[cols]; + for(int j = 0; j < cols; j++) + schema[j] = ValueType.FP64; + + FrameBlock frameBlock = new FrameBlock(schema); + for(int i = 0; i < rows; i++) { + Object[] row = new Object[cols]; + for(int j = 0; j < cols; j++) + row[j] = (double) ((long) i * cols + j + 1); + frameBlock.appendRow(row); + } + return frameBlock; + } + + private FrameBlock generateSparseLikeFp64Frame(int rows, int cols) { + ValueType[] schema = new ValueType[cols]; + for(int j = 0; j < cols; j++) + schema[j] = ValueType.FP64; + + FrameBlock frameBlock = new FrameBlock(schema); + for(int i = 0; i < rows; i++) { + Object[] row = new Object[cols]; + for(int j = 0; j < cols; j++) { + long linearIndex = (long) i * cols + j; + row[j] = linearIndex % 20 == 0 ? (double) (linearIndex + 1) : 0.0; + } + frameBlock.appendRow(row); + } + return frameBlock; + } + + private FrameBlock generateMixedSchemaFrame(int rows, int cols) { + ValueType[] schema = new ValueType[cols]; + for(int j = 0; j < cols; j++) { + switch(j % 5) { + case 0: + schema[j] = ValueType.FP64; + break; + case 1: + schema[j] = ValueType.INT64; + break; + case 2: + schema[j] = ValueType.BOOLEAN; + break; + case 3: + schema[j] = ValueType.STRING; + break; + default: + schema[j] = ValueType.INT32; + } + } + + FrameBlock frameBlock = new FrameBlock(schema); + for(int i = 0; i < rows; i++) { + Object[] row = new Object[cols]; + for(int j = 0; j < cols; j++) { + switch(schema[j]) { + case FP64: + row[j] = (double) ((long) i * cols + j + 1); + break; + case INT64: + row[j] = (long) ((long) i * cols + j + 1); + break; + case BOOLEAN: + row[j] = ((i + j) % 2 == 0); + break; + case STRING: + row[j] = "s_" + (i % 1000) + "_" + j; + break; + case INT32: + row[j] = (int) ((long) i * cols + j + 1); + break; + default: + throw new IllegalArgumentException("Unsupported generated type: " + schema[j]); + } + } + frameBlock.appendRow(row); + } + return frameBlock; + } + + private long readRawBytes(String fname) throws IOException { + Configuration conf = ConfigurationManager.getCachedJobConf(); + List files = listDataFiles(fname, conf); + byte[] buffer = new byte[1024 * 1024]; + long total = 0; + + for(Path file : files) { + FileSystem fs = file.getFileSystem(conf); + try(FSDataInputStream in = fs.open(file)) { + int n; + while((n = in.read(buffer)) != -1) + total += n; + } + } + return total; + } + + private long readParquetFooters(String fname) throws IOException { + Configuration conf = ConfigurationManager.getCachedJobConf(); + List files = listDataFiles(fname, conf); + long checksum = 0; + + for(Path file : files) { + try(ParquetFileReader reader = ParquetFileReader.open(HadoopInputFile.fromPath(file, conf))) { + ParquetMetadata metadata = reader.getFooter(); + checksum += metadata.getBlocks().size(); + checksum += metadata.getFileMetaData().getSchema().getFieldCount(); + for(int i = 0; i < metadata.getBlocks().size(); i++) + checksum += metadata.getBlocks().get(i).getRowCount(); + } + } + return checksum; + } + + private List listDataFiles(String fname, Configuration conf) throws IOException { + Path path = new Path(fname); + FileSystem fileSystem = path.getFileSystem(conf); + List files = new ArrayList<>(); + + if(!fileSystem.exists(path)) + return files; + + FileStatus status = fileSystem.getFileStatus(path); + if(status.isFile()) { + files.add(path); + return files; + } + + for(FileStatus child : fileSystem.listStatus(path)) + if(child.isFile() && isDataFile(child.getPath())) + files.add(child.getPath()); + + return files; + } + + private boolean isDataFile(Path path) { + String name = path.getName(); + return !name.startsWith("_") && !name.startsWith(".") && !name.endsWith(".crc"); + } + + private PathStats getPathStats(String fname) { + try { + Configuration conf = ConfigurationManager.getCachedJobConf(); + List files = listDataFiles(fname, conf); + long totalSize = 0; + int numFiles = 0; + + for(Path file : files) { + FileSystem fileSystem = file.getFileSystem(conf); + FileStatus status = fileSystem.getFileStatus(file); + if(status.isFile()) { + totalSize += status.getLen(); + numFiles++; + } + } + return new PathStats(totalSize, numFiles); + } + catch(Exception ex) { + System.err.println("Failed to collect path stats for " + fname + ": " + ex.getMessage()); + return new PathStats(-1, -1); + } + } + + private String[] parseProfileList(String value) { + if(value == null || value.trim().isEmpty()) + return new String[] {PROFILE_DENSE_FP64, PROFILE_MIXED_SCHEMA, PROFILE_SPARSE_LIKE_FP64}; + + String[] tokens = value.split(","); + List profiles = new ArrayList<>(); + + for(String token : tokens) { + String profile = normalizeProfileName(token); + if(profile.isEmpty()) + continue; + if(!isSupportedProfile(profile)) + throw new IllegalArgumentException("Unknown Parquet benchmark data profile: " + token); + profiles.add(profile); + } + + if(profiles.isEmpty()) + throw new IllegalArgumentException("No valid Parquet benchmark data profiles specified."); + + return profiles.toArray(new String[0]); + } + + private boolean isSupportedProfile(String profile) { + return PROFILE_DENSE_FP64.equals(profile) || PROFILE_MIXED_SCHEMA.equals(profile) || + PROFILE_SPARSE_LIKE_FP64.equals(profile); + } + + private String normalizeProfileName(String value) { + if(value == null) + return ""; + + String profile = value.trim().toLowerCase(Locale.US).replace('-', '_').replace(" ", "_"); + if("dense".equals(profile) || "dense_double".equals(profile) || "dense_double_only".equals(profile) || + "dense_fp64".equals(profile) || PROFILE_DENSE_FP64.equals(profile)) + return PROFILE_DENSE_FP64; + if("mixed".equals(profile) || "mixed_types".equals(profile) || PROFILE_MIXED_SCHEMA.equals(profile)) + return PROFILE_MIXED_SCHEMA; + if("sparse".equals(profile) || "sparse_like".equals(profile) || "sparse_double".equals(profile) || + "sparse_like_double".equals(profile) || "sparse_fp64".equals(profile) || + PROFILE_SPARSE_LIKE_FP64.equals(profile)) + return PROFILE_SPARSE_LIKE_FP64; + return profile; + } + + private int[] parsePositiveIntList(String value) { + if(value == null || value.trim().isEmpty()) + return new int[0]; + + String[] tokens = value.split(","); + int[] tmp = new int[tokens.length]; + int count = 0; + for(String token : tokens) { + String trimmed = token.trim(); + if(trimmed.isEmpty()) + continue; + int v = Integer.parseInt(trimmed); + if(v > 0) + tmp[count++] = v; + } + + int[] ret = new int[count]; + System.arraycopy(tmp, 0, ret, 0, count); + return ret; + } + + private int getIntProperty(String primaryKey, String legacyKey, int defaultValue) { + String value = getStringProperty(primaryKey, legacyKey, Integer.toString(defaultValue)); + return Integer.parseInt(value); + } + + private String getStringProperty(String primaryKey, String legacyKey, String defaultValue) { + String value = System.getProperty(primaryKey); + if(value != null) + return value; + value = System.getProperty(legacyKey); + return value != null ? value : defaultValue; + } + + private String safeName(String s) { + return s.replaceAll("[^A-Za-z0-9_\\-]", "_"); + } + + private long usedHeap() { + Runtime rt = Runtime.getRuntime(); + return rt.totalMemory() - rt.freeMemory(); + } + + private GcStats getGcStats() { + long count = 0; + long timeMs = 0; + + for(GarbageCollectorMXBean bean : ManagementFactory.getGarbageCollectorMXBeans()) { + long c = bean.getCollectionCount(); + long t = bean.getCollectionTime(); + if(c >= 0) + count += c; + if(t >= 0) + timeMs += t; + } + return new GcStats(count, timeMs); + } + + private void blackhole(FrameBlock frameBlock) { + if(frameBlock == null) + throw new IllegalArgumentException("Unexpected null FrameBlock."); + _blackhole ^= frameBlock.getNumRows(); + _blackhole ^= frameBlock.getNumColumns(); + } + + private void blackhole(long value) { + _blackhole ^= value; + } + + private String escapeCsv(String s) { + if(s == null) + return ""; + boolean needsQuotes = s.indexOf(',') >= 0 || s.indexOf('"') >= 0 || s.indexOf('\n') >= 0 || + s.indexOf('\r') >= 0; + if(!needsQuotes) + return s; + return "\"" + s.replace("\"", "\"\"") + "\""; + } + + private String escapeJson(String s) { + if(s == null) + return ""; + + StringBuilder stringBuilder = new StringBuilder(); + for(int i = 0; i < s.length(); i++) { + char c = s.charAt(i); + switch(c) { + case '"': + stringBuilder.append("\\\""); + break; + case '\\': + stringBuilder.append("\\\\"); + break; + case '\b': + stringBuilder.append("\\b"); + break; + case '\f': + stringBuilder.append("\\f"); + break; + case '\n': + stringBuilder.append("\\n"); + break; + case '\r': + stringBuilder.append("\\r"); + break; + case '\t': + stringBuilder.append("\\t"); + break; + default: + if(c < 0x20) + stringBuilder.append(String.format("\\u%04x", (int) c)); + else + stringBuilder.append(c); + } + } + return stringBuilder.toString(); + } + + private static class JsonState { + private boolean hasEntries = false; + } + + private static class GcStats { + private final long count; + private final long timeMs; + + private GcStats(long count, long timeMs) { + this.count = count; + this.timeMs = timeMs; + } + } + + private static class PathStats { + private final long fileSizeBytes; + private final int numPartFiles; + + private PathStats(long fileSizeBytes, int numPartFiles) { + this.fileSizeBytes = fileSizeBytes; + this.numPartFiles = numPartFiles; + } + } + + private static class BenchmarkDiagnostics { + private final long cells; + private final long hdfsBlockSizeBytes; + private final long currentWriterEstimatedNumPartFiles; + private final int currentWriterEstimatedNumThreads; + private final int configuredParallelReadParallelism; + private final int configuredParallelWriteParallelism; + private final int estimatedParallelReadTasks; + private final long denseFp64ReferenceSizeBytes; + private final double avgPartFileSizeBytes; + private final double actualFileToDenseFp64ReferenceRatio; + private final double expectedNonZeroFraction; + private final String readerValueExtractionMode; + private final String frameMaterialization; + + private BenchmarkDiagnostics(long cells, long hdfsBlockSizeBytes, long currentWriterEstimatedNumPartFiles, + int currentWriterEstimatedNumThreads, int configuredParallelReadParallelism, + int configuredParallelWriteParallelism, int estimatedParallelReadTasks, long denseFp64ReferenceSizeBytes, + double avgPartFileSizeBytes, double actualFileToDenseFp64ReferenceRatio, double expectedNonZeroFraction, + String readerValueExtractionMode, String frameMaterialization) { + this.cells = cells; + this.hdfsBlockSizeBytes = hdfsBlockSizeBytes; + this.currentWriterEstimatedNumPartFiles = currentWriterEstimatedNumPartFiles; + this.currentWriterEstimatedNumThreads = currentWriterEstimatedNumThreads; + this.configuredParallelReadParallelism = configuredParallelReadParallelism; + this.configuredParallelWriteParallelism = configuredParallelWriteParallelism; + this.estimatedParallelReadTasks = estimatedParallelReadTasks; + this.denseFp64ReferenceSizeBytes = denseFp64ReferenceSizeBytes; + this.avgPartFileSizeBytes = avgPartFileSizeBytes; + this.actualFileToDenseFp64ReferenceRatio = actualFileToDenseFp64ReferenceRatio; + this.expectedNonZeroFraction = expectedNonZeroFraction; + this.readerValueExtractionMode = readerValueExtractionMode; + this.frameMaterialization = frameMaterialization; + } + } +}