diff --git a/src/it/java/io/deephaven/benchmark/tests/standard/file/FileTestRunner.java b/src/it/java/io/deephaven/benchmark/tests/standard/file/FileTestRunner.java index 7159d7c0..ed75f4f9 100644 --- a/src/it/java/io/deephaven/benchmark/tests/standard/file/FileTestRunner.java +++ b/src/it/java/io/deephaven/benchmark/tests/standard/file/FileTestRunner.java @@ -92,6 +92,52 @@ void runParquetReadTest(String testName) { runReadTest(testName, q); } + /** + * Run a benchmark that measures a filtered parquet read. The filter reads {@code long10K} in full and keeps 1 in + * {@code modulus} rows, spread evenly through the file, then the given columns are read for only those rows. The + * rate is file rows per second, like the full read tests. This test always runs after a corresponding write test. + * + * @param testName name that will appear in the results as the benchmark name + * @param modulus keep the rows where {@code long10K % modulus == 1}; should divide 10000 to keep exactly 1 in + * {@code modulus} + * @param columnNames the columns to read for the remaining rows + */ + void runParquetFilteredReadTest(String testName, int modulus, String... columnNames) { + runFilteredReadTest(testName, modulus, "filtered.select([${columns}])", columnNames); + } + + /** + * Run a benchmark like {@link #runParquetFilteredReadTest}, but read the remaining rows with + * {@code ForceReadUtility} instead of {@code select}. That reads the column data in chunks on one thread through + * {@code getChunk}, and keeps none of it, rather than copying it in parallel into a new table through + * {@code fillChunk}. + * + * @param testName name that will appear in the results as the benchmark name + * @param modulus keep the rows where {@code long10K % modulus == 1}; should divide 10000 to keep exactly 1 in + * {@code modulus} + * @param columnNames the columns to read for the remaining rows + */ + void runParquetFilteredForceReadTest(String testName, int modulus, String... columnNames) { + var forceRead = "jpy.get_type('io.deephaven.engine.table.impl.ForceReadUtility').of(" + + "filtered.j_table, jpy.array('java.lang.String', [${columns}]))"; + runFilteredReadTest(testName, modulus, forceRead, columnNames); + } + + private void runFilteredReadTest(String testName, int modulus, String readExpr, String... columnNames) { + var q = """ + read('/data/source.ptr.parquet') + filtered = source.where('long10K % ${modulus} == 1') + ${readExpr} + metric_file_size = os.path.getsize('/data/source.ptr.parquet') + bench_api_metrics_add('data', 'file.size', str(metric_file_size), 'parquet') + """; + q = q.replace("${readExpr}", readExpr); + q = q.replace("${modulus}", "" + modulus); + var columns = Arrays.stream(columnNames).map(c -> "'" + c + "'").collect(Collectors.joining(",")); + q = q.replace("${columns}", columns); + runReadTest(testName, q); + } + /** * Run a benchmark that measures parquet write performance. * @@ -269,7 +315,7 @@ private String getType(String columnName) { private Bench initialize(Object testInst) { var query = """ - import time, os, shutil + import time, os, shutil, jpy from deephaven import empty_table, garbage_collect, new_table, merge from deephaven.column import long_col, double_col from deephaven.parquet import read, write, write_partitioned diff --git a/src/it/java/io/deephaven/benchmark/tests/standard/file/ParquetMultiColTest.java b/src/it/java/io/deephaven/benchmark/tests/standard/file/ParquetMultiColTest.java index 0d1dd750..162bcb1f 100644 --- a/src/it/java/io/deephaven/benchmark/tests/standard/file/ParquetMultiColTest.java +++ b/src/it/java/io/deephaven/benchmark/tests/standard/file/ParquetMultiColTest.java @@ -11,88 +11,102 @@ @Tag("Iterate") class ParquetMultiColTest { final String[] usedColumns = {"str10K", "long10K", "int10K", "short10K", "bigDec10K", "intArr5", "intVec5"}; + // Array columns can't be read sparsely (DH-23641), so the filtered reads leave them out + final String[] filteredColumns = {"str10K", "int10K", "short10K", "bigDec10K"}; final FileTestRunner runner = new FileTestRunner(this); - @BeforeEach - void setup() { - runner.setScaleFactors(3, 3); + void setup(int scaleFactor) { + runner.setScaleFactors(3, scaleFactor); } @Test @Order(1) void writeMultiColSnappy() { + setup(3); runner.runParquetWriteTest("ParquetWrite- Snappy Multi Col -Static", "SNAPPY", usedColumns); } @Test @Order(2) void readMultiColSnappy() { + setup(3); runner.runParquetReadTest("ParquetRead- Snappy Multi Col -Static"); } @Test @Order(3) void writeMultiColZstd() { + setup(3); runner.runParquetWriteTest("ParquetWrite- Zstd Multi Col -Static", "ZSTD", usedColumns); } @Test @Order(4) void readMultiColZstd() { + setup(3); runner.runParquetReadTest("ParquetRead- Zstd Multi Col -Static"); } @Test @Order(5) void writeMultiColLzo() { + setup(3); runner.runParquetWriteTest("ParquetWrite- Lzo Multi Col -Static", "LZO", usedColumns); } @Test @Order(6) void readMultiColLzo() { + setup(3); runner.runParquetReadTest("ParquetRead- Lzo Multi Col -Static"); } @Test @Order(7) void writeMultiColLz4Raw() { + setup(3); runner.runParquetWriteTest("ParquetWrite- Lz4Raw Multi Col -Static", "LZ4_RAW", usedColumns); } @Test @Order(8) void readMultiColLz4Raw() { + setup(3); runner.runParquetReadTest("ParquetRead- Lz4Raw Multi Col -Static"); } @Test @Order(9) void writeMultiColGzip() { + setup(1); runner.runParquetWriteTest("ParquetWrite- Gzip Multi Col -Static", "GZIP", usedColumns); } @Test @Order(10) void readMultiColGzip() { + setup(1); runner.runParquetReadTest("ParquetRead- Gzip Multi Col -Static"); } @Test @Order(11) void writeMultiColNone() { + setup(3); runner.runParquetWriteTest("ParquetWrite- No Codec Multi Col -Static", "NONE", usedColumns); } @Test @Order(12) void readMultiColNone() { + setup(3); runner.runParquetReadTest("ParquetRead- No Codec Multi Col -Static"); } @Test @Order(13) void writeMultiColDefaultSnappy() { + setup(3); runner.useParquetDefaultSettings(); runner.runParquetWriteTest("ParquetWrite- Snappy Multi Col Defaults -Static", "SNAPPY", usedColumns); } @@ -100,8 +114,30 @@ void writeMultiColDefaultSnappy() { @Test @Order(14) void readMultiColDefaultSnappy() { - runner.useParquetDefaultSettings(); + setup(3); runner.runParquetReadTest("ParquetRead- Snappy Multi Col Defaults -Static"); } + @Test + @Order(15) + void readMultiColDefaultSnappySparse() { + setup(3); + runner.runParquetFilteredReadTest("ParquetRead- Sparse Snappy Multi Col Defaults -Static", 100, filteredColumns); + } + + @Test + @Order(16) + void readMultiColDefaultSnappyDense() { + setup(3); + runner.runParquetFilteredReadTest("ParquetRead- Dense Snappy Multi Col Defaults -Static", 2, filteredColumns); + } + + @Test + @Order(17) + void forceReadMultiColDefaultSnappySparse() { + setup(3); + runner.runParquetFilteredForceReadTest("ParquetRead- Sparse ForceRead Snappy Multi Col Defaults -Static", 100, + filteredColumns); + } + }