diff --git a/.github/resources/compare-scale-benchmark.properties b/.github/resources/compare-scale-benchmark.properties index c9a6dd4f..f9662768 100644 --- a/.github/resources/compare-scale-benchmark.properties +++ b/.github/resources/compare-scale-benchmark.properties @@ -24,7 +24,7 @@ default.data.distribution=random generator.pause.per.row=0 millis # Compression used for generating and storing records (SNAPPY, ZSTD, LZ4, LZO, GZIP, NONE) -record.compression=LZO +record.compression=SNAPPY # Row count to scale tests (Tests can override but typically do not) scale.row.count=70000000 diff --git a/src/it/java/io/deephaven/benchmark/tests/compare/iterate/RowIteratorTest.java b/src/it/java/io/deephaven/benchmark/tests/compare/iterate/RowIteratorTest.java new file mode 100644 index 00000000..bb6eba2c --- /dev/null +++ b/src/it/java/io/deephaven/benchmark/tests/compare/iterate/RowIteratorTest.java @@ -0,0 +1,110 @@ +/* Copyright (c) 2022-2024 Deephaven Data Labs and Patent Pending */ +package io.deephaven.benchmark.tests.compare.iterate; + +import org.junit.jupiter.api.*; +import org.junit.jupiter.api.MethodOrderer.OrderAnnotation; +import io.deephaven.benchmark.tests.compare.CompareTestRunner; + +/** + * Product comparison tests for iterating and summing table columns. Tests read the same parquet data. To avoid an + * unfair advantage where some products may partition or group data during the read, parquet read time is included in + * the benchmark results. + *

+ * Each test produces a table result containing one row with one column that is the total of the result of the sum of + * two columns for each row. ex. sum((r1c1 + r1c2)..(rNc1 + rNc2)). This is achieved without creating an extra column to + * hold the column sums. + *

+ * Data generation only happens in the first test, the Deephaven test. Tests can be run individually, but only after the + * desired data has been generated. + */ +@TestMethodOrder(OrderAnnotation.class) +public class RowIteratorTest { + final CompareTestRunner runner = new CompareTestRunner(this); + + @Test + @Order(1) + public void deephavenRowIterator() { + runner.initDeephaven(2, "source", null, "int250", "int640"); + var setup = "from deephaven.parquet import read"; + var op = """ + source = read('/data/source.parquet').select() + result = new_table([ + long_col('total', [sum(row.int250 + row.int640 for row in source.iter_tuple())]) + ]) + """; + var msize = "source.size"; + var rsize = "result.size"; + runner.test("Deephaven Row Iterator", setup, op, msize, rsize); + } + + @Test + @Order(2) + public void pyarrowRowIterator() { + runner.initPython("pyarrow"); + var setup = """ + import pyarrow as pa + import pyarrow.dataset as ds + import pyarrow.compute as pc + + def iterdicts(table, cols=[]): + for batch in table.to_batches(1024): + d = batch.to_pydict() + int250 = d['int250'] + int640 = d['int640'] + for i in range(len(int250)): + row = {'int250':int250[i],'int640':int640[i]} + yield row + """; + var op = """ + source = ds.dataset('/data/source.parquet', format="parquet").to_table() + rsum = sum(row['int250'] + row['int640'] for row in iterdicts(source)) + result = pa.Table.from_pydict({'total':[rsum]}) + """; + var msize = "source.num_rows"; + var rsize = "result.num_rows"; + runner.test("PyArrow Row Iterator", setup, op, msize, rsize); + } + + @Test + @Order(3) + public void pandasRowIteratior() { + runner.initPython("fastparquet", "pandas"); + var setup = "import pandas as pd"; + var op = """ + source = pd.read_parquet('/data/source.parquet') + rsum = sum(row.int250 + row.int640 for row in source.itertuples()) + result = pd.DataFrame([[rsum]], columns=['total']) + """; + var msize = "len(source)"; + var rsize = "len(result)"; + runner.test("Pandas Row Iterator", setup, op, msize, rsize); + } + + @Test + @Order(4) + public void duckdbRowIterator() { + runner.initPython("duckdb"); + var setup = """ + import duckdb as db + + def iterdicts(table): + while batch := table.fetchmany(1024): + for row in batch: + r = {'int250':row[0],'int640':row[1]} + yield r + """; + var op = """ + source = db.sql("SELECT * FROM '/data/source.parquet'") + + db.sql("CREATE TABLE results(total INT)") + rsum = sum(row['int250'] + row['int640'] for row in iterdicts(source)) + db.sql("INSERT INTO results VALUES(" + str(rsum) + ")") + sourceLen = db.sql("SELECT count(*) FROM source").fetchone()[0] + resultLen = db.sql("SELECT count(*) FROM results").fetchone()[0] + """; + var msize = "sourceLen"; + var rsize = "resultLen"; + runner.test("DuckDb Row Iterator", setup, op, msize, rsize); + } + +} diff --git a/src/it/java/io/deephaven/benchmark/tests/standard/formula/RowIteratorTest.java b/src/it/java/io/deephaven/benchmark/tests/standard/formula/RowIteratorTest.java new file mode 100644 index 00000000..ff9b54ea --- /dev/null +++ b/src/it/java/io/deephaven/benchmark/tests/standard/formula/RowIteratorTest.java @@ -0,0 +1,42 @@ +/* Copyright (c) 2022-2024 Deephaven Data Labs and Patent Pending */ +package io.deephaven.benchmark.tests.standard.formula; + +import org.junit.jupiter.api.*; +import io.deephaven.benchmark.tests.standard.StandardTestRunner; + +/** + * Standard tests for iterating through tables to access column values directly. These benchmarks iterate through the + * same columns and do the same sums. + */ +public class RowIteratorTest { + final StandardTestRunner runner = new StandardTestRunner(this); + + void setup(int rowFactor) { + runner.setRowFactor(rowFactor); + runner.tables("source"); + runner.setScaleFactors(1, 0); + } + + @Test + void iterDict2Cols() { + setup(2); + var q = """ + new_table([ + double_col('total', [sum(row['num1'] + row['num2'] for row in source.iter_dict())]) + ]) + """; + runner.test("Row-IterDict- Sum 2 Double Cols", 1, q, "num1", "num2"); + } + + @Test + void iterTuple2Cols() { + setup(4); + var q = """ + new_table([ + double_col('total', [sum(row.num1 + row.num2 for row in source.iter_tuple())]) + ]) + """; + runner.test("Row-IterTuple- Sum 2 Double Cols", 1, q, "num1", "num2"); + } + +} diff --git a/src/main/resources/io/deephaven/benchmark/run/profile/default.properties b/src/main/resources/io/deephaven/benchmark/run/profile/default.properties index 9f1d382e..bc4cab82 100644 --- a/src/main/resources/io/deephaven/benchmark/run/profile/default.properties +++ b/src/main/resources/io/deephaven/benchmark/run/profile/default.properties @@ -24,7 +24,7 @@ default.data.distribution=random generator.pause.per.row=0 millis # Compression used for generating and storing records (ZSTD, LZ4, LZO, GZIP, SNAPPY, NONE) -record.compression=LZO +record.compression=SNAPPY # Row count to scale tests (Tests can override but typically do not) scale.row.count=100000 diff --git a/src/main/resources/io/deephaven/benchmark/run/profile/queries/compare.py b/src/main/resources/io/deephaven/benchmark/run/profile/queries/compare.py new file mode 100644 index 00000000..8c5d32c1 --- /dev/null +++ b/src/main/resources/io/deephaven/benchmark/run/profile/queries/compare.py @@ -0,0 +1,33 @@ +# Copyright (c) 2022-2024 Deephaven Data Labs and Patent Pending +# +# Supporting Deephaven queries to use the benchmark_snippet to investigate product comparisons. +# - Generate a table that shows the comparisons between products for each benchmark +# Requirements: Deephaven 0.23.0 or greater + +from urllib.request import urlopen; import os + +benchmark_set_arg = 'stanbrub/full-set-140M' + +root = 'file:///data' if os.path.exists('/data/deephaven-benchmark') else 'https://storage.googleapis.com' +with urlopen(root + '/deephaven-benchmark/benchmark_tables.dh.py') as r: + benchmark_storage_uri_arg = root + '/deephaven-benchmark' + benchmark_category_arg ='adhoc' + benchmark_actor_filter_arg = os.path.dirname(benchmark_set_arg) + benchmark_set_filter_arg = os.path.basename(benchmark_set_arg) + exec(r.read().decode(), globals(), locals()) + + +product_compare = bench_results_sets.view([ + 'Product=benchmark_name.replaceAll(`[ ].*$`,``)', 'Benchmark=benchmark_name.replaceAll(`^[^ ]+[ ]`,``)', + 'Rate=op_rate' +]).sort(['Benchmark','Product']) + +from deephaven import numpy as dhnp +products = dhnp.to_numpy(product_compare.select_distinct(['Product'])) +products = [str(prod[0]) for prod in products] + +product_compare = product_result.group_by(['Benchmark']).view(['Benchmark'] + [ + products[i] + '=Rate[' + str(i) + ']' for i in range(len(products)) +]) + +bench_results = bench_metrics = bench_platforms = bench_results_sets = bench_results_change = None