diff --git a/.github/resources/adhoc-benchmark-docker-compose.yml b/.github/resources/adhoc-benchmark-docker-compose.yml index c02c30e3..e83c6db9 100644 --- a/.github/resources/adhoc-benchmark-docker-compose.yml +++ b/.github/resources/adhoc-benchmark-docker-compose.yml @@ -1,13 +1,15 @@ services: deephaven: - image: ${DOCKER_IMG} + image: ghcr.io/stanbrub/server:jvm25 ports: - "${DEEPHAVEN_PORT:-10000}:10000" volumes: - ./data:/data - ./minio:/minio environment: + - "JAVA_OPTS=-XX:+UseG1GC" - "START_OPTS=-DAuthHandlers=io.deephaven.auth.AnonymousAuthenticationHandler ${CONFIG_OPTS}" + - "DEEPHAVEN_HOST_OS_DIR=${ENV_DEEPHAVEN_HOST_OS_DIR}" redpanda: command: diff --git a/.github/resources/adhoc-scale-benchmark.properties b/.github/resources/adhoc-scale-benchmark.properties index adab86ad..31a2fd15 100644 --- a/.github/resources/adhoc-scale-benchmark.properties +++ b/.github/resources/adhoc-scale-benchmark.properties @@ -15,7 +15,7 @@ schema.registry.addr=redpanda:8081 kafka.consumer.addr=redpanda:29092 # Default timeout to complete processes (Executing queries, generating records) -default.completion.timeout=10 minutes +default.completion.timeout=20 minutes # Default data distribution for column data (random, ascending, descending, runlength) default.data.distribution=${baseDistrib} diff --git a/.github/scripts/build-benchmark-artifact-remote.sh b/.github/scripts/build-benchmark-artifact-remote.sh index 8f679ef0..edce0c0c 100755 --- a/.github/scripts/build-benchmark-artifact-remote.sh +++ b/.github/scripts/build-benchmark-artifact-remote.sh @@ -25,6 +25,7 @@ title "- Building Remote Benchmark Artifact on ${HOST} -" title "-- Building and Verifying --" cd ${GIT_DIR} +sleep 10 # Some JVM command lines start DH slower than others mvn verify title "-- Cleanup After Build --" diff --git a/.github/scripts/manage-deephaven-remote.sh b/.github/scripts/manage-deephaven-remote.sh index e1e49cf8..57af19cb 100755 --- a/.github/scripts/manage-deephaven-remote.sh +++ b/.github/scripts/manage-deephaven-remote.sh @@ -35,6 +35,7 @@ if [[ ${CONFIG_OPTS} == "" ]]; then CONFIG_OPTS="-Xmx24g" fi echo "CONFIG_OPTS=${CONFIG_OPTS}" > .env +echo "ENV_DEEPHAVEN_HOST_OS_DIR=${DEEPHAVEN_DIR}" >> .env IS_BRANCH="false" if [[ ${DOCKER_IMG} == *"@sha"*":"* ]]; then diff --git a/.github/scripts/run-benchmarks-remote.sh b/.github/scripts/run-benchmarks-remote.sh index b28a25c1..259915e6 100755 --- a/.github/scripts/run-benchmarks-remote.sh +++ b/.github/scripts/run-benchmarks-remote.sh @@ -43,7 +43,7 @@ title "-- Running Benchmarks --" set +f cd ${RUN_DIR} cat ${RUN_TYPE}-scale-benchmark.properties | sed 's|${baseRowCount}|'"${ROW_COUNT}|g" | sed 's|${baseDistrib}|'"${DISTRIB}|g" | sed 's|${userHome}|'"${HOME}|g" > scale-benchmark.properties -JAVA_OPTS=$(echo -Dbenchmark.profile=scale-benchmark.properties -jar deephaven-benchmark-*-standalone.jar -cp standard-tests.jar) +JAVA_OPTS=$(echo -Xmx4g -Dbenchmark.profile=scale-benchmark.properties -jar deephaven-benchmark-*-standalone.jar -cp standard-tests.jar) set -f if [ "${TAG_NAME}" = "Any" ]; then diff --git a/.github/scripts/setup-test-server-remote.sh b/.github/scripts/setup-test-server-remote.sh index 43033300..920e6c4d 100755 --- a/.github/scripts/setup-test-server-remote.sh +++ b/.github/scripts/setup-test-server-remote.sh @@ -135,7 +135,8 @@ sudo docker system prune --volumes --force sudo rm -rf ${DEEPHAVEN_DIR} title "-- Staging Docker Resources --" -mkdir -p ${DEEPHAVEN_DIR} +mkdir -p ${DEEPHAVEN_DIR}/data +chmod 777 ${DEEPHAVEN_DIR}/data cd ${DEEPHAVEN_DIR} cp ${GIT_DIR}/benchmark/.github/resources/${RUN_TYPE}-benchmark-docker-compose.yml docker-compose.yml diff --git a/docs/gc/all-gc-rankings.png b/docs/gc/all-gc-rankings.png new file mode 100644 index 00000000..ab219c72 Binary files /dev/null and b/docs/gc/all-gc-rankings.png differ diff --git a/docs/gc/gc-report.md b/docs/gc/gc-report.md new file mode 100644 index 00000000..8ee369fd --- /dev/null +++ b/docs/gc/gc-report.md @@ -0,0 +1,138 @@ +# Overview + +This report provides some insight into how various Deephaven CE operations behave with different Garbage Collectors and settings. It is not possible to test every combination of JVM version, GC option, heap size, data size/cardinality, DHC operation, OS, and hardware platform. What follows are results based on a small subset of the possibilities. Each user will have a different setup, but hopefully we can glean some useful tips from the data that has been collected. + +We are covering only JVM 17 and JVM 25. This presents an interesting challenge, because GC options have changed dramatically in the four years in between. For example, ZGC generational didn't exist in JVM 17, while ZGC non-generational doesn't exist in JVM 25. The approach taken here was to use the options that will likely be the defaults in JVM 25+ and then try to use the nearest approximation in JVM 17. + +One other thing to note is that, judging by recent JDK release notes, the strategy for GCs going forward is to make GCs more self-tuning. Since the GC's top priority is keeping the application running, the GC can override user-supplied settings if it needs to. The JVM options used for this effort are intentionally minimal and tend to rely more on defaults. + +## TL;DR + +Here are some highlights from the data: +- For best Static throughput, use G1 or G1 THP. (Others degrade by 4% to 22%) +- For best Ticking throughput, use G1, but expect higher jitter. (5% to 12% more) +- For lowest jitter, use Shenandoah, though it's lower throughput than G1. (9% to 15% less) +- For best 100ms cycle throughput-jitter combo, use Shenandoah THP (3rd-5th best throughput, best jitter) +- For best 1sec cycle combo of throughput and jitter, use G1 THP (best throughput, 3rd best jitter) + +## The Benchmarks + +The benchmarks used for this effort are not the nightly benchmarks, which run single operations typically for 8-10 seconds. Instead a longer running and smaller benchmark set is used that covers categories of operations. This set typically runs for 1.5 mins to 2 mins, lazily reading from a large parquet file. The idea is to use few benchmarks to cover much of the operational code base. These are referred to as "Training Benchmarks". + +- AggBy: Runs multiple mathematical operations like avg, std, var, min, max, etc +- Filter: Runs `where_in` and `where` filters +- Formula: Runs multiple UDFs and Inline Formulas +- NaturalJoin: Runs a natural join operation +- Ordered: Runs multiple aggBy ordered operations like median, unique, sorted_first, etc +- UpdateBy: Runs multiple windowed operations + +## Naming Conventions + +The set names for the various runs use a naming convention like `___[_]` +- gc-type: `g1gc`, `shen`, `zgc`, `para` +- cycle-time: `1000` (1sec), `100` (100ms) +- autotune-load: `p80`, `p90`, `p100`, `p110` (Percentage throughput targets) +- extra: `huge` (Transparent Huge Pages) + +## Supporting Evidence for TL;DR + +What follows are charts showing the results of many benchmarks runs using various options. The naming convention for the gc sets displayed have been described previously. First, there are three kinds of summary tables; throughput, jitter, on-budget. There is also a table that combines all three. +- Throughput: Starting with rows/sec, a normalized fraction starting from 1. So 1.20 would be 20% faster than 1.0. +- Jitter: Starting with Coefficient of Variation, a normalized fraction starting from 1. So 1.04 would be 4% more jitter than 1.0. +- On Budget: A percentage a cycles that finished on or below the cycle time (e.g. 1000ms or 100ms) + +## Overall Rankings + +Overall ranking can show the distance between the overall throughput, jitter, or on-budget metrics for each GC for the same set of configuration. For example, on JVM 25 for Static data, ZGC is 16% slower than G1. + +![All GC Rankings](./all-gc-rankings.png) + +### Static Data Throughput + +The following chart shows Throughput for Static data for JVM 17. G1 comes in first with ZGC a surprising second. +![Static Throughput for JVM 17](./static-j17.throughput.png) + +The following chart shows Throughput for Static data for JVM 25. The story is similar to JVM 17 with G1 in front but followed by parallel GC. +![Static Throughput for JVM 25](./static-j25.throughput.png) + +### Ticking Data Throughput + +There are three considerations captured here; throughput, jitter, and on-budget percentage. These can be weighted based on the needs of the workflow. Typically, but not always, higher throughput means higher jitter or lower on-budget percentage. + +The following chart shows throughput for JVM 17 with 1s UGP cycles. +![Ticking Throughput for JVM 17 1s](./inc-j17-1s.throughput.png) + +The following chart shows throughput for JVM 25 with 1s UGP cycles. +![Ticking Throughput for JVM 25 1s](./inc-j25-1s.throughput.png) + +The following charts show summaries of JVM 17 and JVM 25 with three factors organized according to their ranks for the benchmarks overall. For example, the values in the "jitter" column are the geometric means for each "gc_type" set when compare to every other set in the table. This gives a ranking of each column that can provide a quicker comparison than a dozen individual tables. + + + + +
GC Summary for JVM 17 1sGC Summary for JVM 25 1s
+ +The following charts show summaries of JVM 25 for 100ms running at an autotune target of 80%, 90%, and 100% + + + + + +
GC Summary 100ms p80GC Summary 100ms p90GC Summary 100ms p100
+ +## Test Setup + +- OS: Ubuntu 24.04 +- Hardware: Intel(R) Xeon(R) E-2388G (8C/16T), 64G DDR4-3200, NVME drive +- OS Settings: THPMode=madvise, CPUPower=schedutil, ASLR=disabled +- Java: 48G Heap, Temurin 17 and 25 +- Container: Docker Images built from a recent edge build +- Test Iterations: 7 +- Static Data Provider: Parquet +- Ticking Data Provider: Autotuning Incremental Release Filter from Parquet + +## JVM Options + +The strategy for the JVM options selected for each JVM (e.g. 17, 25) was to make them equivalent. Of course, this was impossible, because different options are supported in different versions of the same GC. For the JVM 25 options, an attempt was made to use options that will be the defaults in future versions of the JVM. + +The JVM options used for these benchmarks are as follows: +- All GCs: `-Xms48g -Xmx48g -XX:+UseStringDeduplication -DPeriodicUpdateGraph.targetCycleDurationMillis=<100 or 1000>` +- Huge: `-XX:+AlwaysPreTouch -XX:+UseTransparentHugePages` +- JDK 17 + - G1GC: `-XX:+UseG1GC -XX:G1HeapRegionSize=32m -XX:+G1UseAdaptiveIHOP` + - ParallelGC: `-XX:+UseParallelGC -XX:ParallelGCThreads=13 ` + - Shenandoah: `-XX:+UseShenandoahGC -XX:+UnlockExperimentalVMOptions -XX:+ShenandoahImplicitGCInvokesConcurrent` + - ZGC: `-XX:+UseZGC -XX:ConcGCThreads=6` +- JDK 25 + - G1GC: `-XX:+UseG1GC -XX:+UseCompactObjectHeaders` + - ParallelGC: `-XX:+UseParallelGC -XX:+UseCompactObjectHeaders` + - Shenandoah: `-XX:+UseShenandoahGC -XX:ShenandoahGCMode=generational -XX:+UseCompactObjectHeaders` + - ZGC: `-XX:+UseZGC -XX:+UseCompactObjectHeaders` + +### What are Huge Memory Pages? + +All of the benchmarks sets that end with "_huge" have the options `-XX:+AlwaysPreTouch -XX:+UseTransparentHugePages`. In a nutshell, these options pair well together to initialize huge memory pages (e.g. 2MB) at JVM init. This provides better memory performance for larger heaps (> 8G) but can suffer from larger startup costs. In these benchmarks THP provided sizable throughput increases for some benchmarks like natural join in exchange for a small increase in jitter. + +What about `-XX:+UseLargePages`? These have not been tested but could provide similar benefits as THP but with a faster JVM startup time, since they are preallocated during OS start. This requires strict rules, however, and is best for systems that are locked down in terms of memory configuration. It is tricky to size as well, since it is used for more than just JVM heap, and that depends on JVM version. + +### What are Compact Headers? + +The `-XX:+UseCompactObjectHeaders` option cuts the Java Object header in half (16 to 8 bytes) to reduce object footprint and improve cache utilization. Systems that use lots of small objects can benefit with better throughput. Minimal testing for Deephaven showed no significant change in performance, but the option will be a default in JVM 26+. + +## Needed for Further Investigation + +In order to improve investigations like this one, we need much more information on how customers are using Deephaven. Without violating any NDA, we should be able to produce the following: + +- Query structure and complexity like a flow chart for the table chains +- General order of operations. Do they filter first, then agg, then sort? +- Static and ticking schema width, column types, and value cardinality +- Complexity of inline formulas and UDFs +- Type of hardware and containerization +- Java heap and cpu thread count typically used +- Where the data typically comes from. DHE live tables? Kafka? Parquet? +- What JVM options have already been tried in the field + + + + + diff --git a/docs/gc/inc-j17-1s.summary.png b/docs/gc/inc-j17-1s.summary.png new file mode 100644 index 00000000..f30f773b Binary files /dev/null and b/docs/gc/inc-j17-1s.summary.png differ diff --git a/docs/gc/inc-j17-1s.throughput.png b/docs/gc/inc-j17-1s.throughput.png new file mode 100644 index 00000000..08fadd8b Binary files /dev/null and b/docs/gc/inc-j17-1s.throughput.png differ diff --git a/docs/gc/inc-j25-100ms-p100.summary.png b/docs/gc/inc-j25-100ms-p100.summary.png new file mode 100644 index 00000000..eaed673c Binary files /dev/null and b/docs/gc/inc-j25-100ms-p100.summary.png differ diff --git a/docs/gc/inc-j25-100ms-p80.summary.png b/docs/gc/inc-j25-100ms-p80.summary.png new file mode 100644 index 00000000..1ec8ce05 Binary files /dev/null and b/docs/gc/inc-j25-100ms-p80.summary.png differ diff --git a/docs/gc/inc-j25-100ms-p90.summary.png b/docs/gc/inc-j25-100ms-p90.summary.png new file mode 100644 index 00000000..49e97342 Binary files /dev/null and b/docs/gc/inc-j25-100ms-p90.summary.png differ diff --git a/docs/gc/inc-j25-1s.summary.png b/docs/gc/inc-j25-1s.summary.png new file mode 100644 index 00000000..9b8a9558 Binary files /dev/null and b/docs/gc/inc-j25-1s.summary.png differ diff --git a/docs/gc/inc-j25-1s.throughput.png b/docs/gc/inc-j25-1s.throughput.png new file mode 100644 index 00000000..0351ea5c Binary files /dev/null and b/docs/gc/inc-j25-1s.throughput.png differ diff --git a/docs/gc/static-j17.throughput.png b/docs/gc/static-j17.throughput.png new file mode 100644 index 00000000..353edf63 Binary files /dev/null and b/docs/gc/static-j17.throughput.png differ diff --git a/docs/gc/static-j25.throughput.png b/docs/gc/static-j25.throughput.png new file mode 100644 index 00000000..67d01577 Binary files /dev/null and b/docs/gc/static-j25.throughput.png differ diff --git a/pom.xml b/pom.xml index 304b6632..70a4bcfa 100644 --- a/pom.xml +++ b/pom.xml @@ -178,7 +178,7 @@ ${project.basedir}/eclipse-java-google-style.xml - /* Copyright (c) 2022-$YEAR Deephaven Data Labs and Patent Pending */ + /* Copyright (c) $YEAR Deephaven Data Labs and Patent Pending */ @@ -271,6 +271,11 @@ kafka-protobuf-serializer 8.1.1 + + blue.strategic.parquet + parquet-floor + 1.64 + io.deephaven deephaven-java-client-barrage-dagger diff --git a/src/it/java/io/deephaven/benchmark/tests/compare/CompareTestRunner.java b/src/it/java/io/deephaven/benchmark/tests/compare/CompareTestRunner.java index 755633f7..b7b31fe4 100644 --- a/src/it/java/io/deephaven/benchmark/tests/compare/CompareTestRunner.java +++ b/src/it/java/io/deephaven/benchmark/tests/compare/CompareTestRunner.java @@ -28,6 +28,9 @@ * practical purposes, though it is not ideal. */ public class CompareTestRunner { + static { + System.setProperty("root.test.package", "io.deephaven.benchmark.tests"); + } final Object testInst; final Set requiredPackages = new LinkedHashSet<>(); final Map downloadFiles = new LinkedHashMap<>(); diff --git a/src/it/java/io/deephaven/benchmark/tests/standard/StandardTestRunner.java b/src/it/java/io/deephaven/benchmark/tests/standard/StandardTestRunner.java index 3ca12176..5f3da6e5 100644 --- a/src/it/java/io/deephaven/benchmark/tests/standard/StandardTestRunner.java +++ b/src/it/java/io/deephaven/benchmark/tests/standard/StandardTestRunner.java @@ -1,4 +1,4 @@ -/* Copyright (c) 2022-2024 Deephaven Data Labs and Patent Pending */ +/* Copyright (c) 2022-2026 Deephaven Data Labs and Patent Pending */ package io.deephaven.benchmark.tests.standard; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -10,6 +10,7 @@ import io.deephaven.benchmark.controller.Controller; import io.deephaven.benchmark.controller.DeephavenDockerController; import io.deephaven.benchmark.metric.Metrics; +import io.deephaven.benchmark.util.Log; import io.deephaven.benchmark.util.Timer; /** @@ -21,17 +22,25 @@ * conventions are followed (ex. main file is "source") */ final public class StandardTestRunner { + static { + System.setProperty("root.test.package", "io.deephaven.benchmark.tests"); + } final Object testInst; final List supportTables = new ArrayList<>(); final List setupQueries = new ArrayList<>(); final List preOpQueries = new ArrayList<>(); + final List teardownQueries = new ArrayList<>(); final Set requiredServices = new TreeSet<>(List.of("deephaven")); private String mainTable = "source"; private Bench api; private Controller controller; + private int rowCountFactor = 1; private int staticFactor = 1; private int incFactor = 1; - private int rowCountFactor = 1; + private boolean useCachedSource = true; + private boolean useLocalParquet = false; + private float incCycleFactor = 1.0f; + private long incReleaseRowCount = 1000000; public StandardTestRunner(Object testInst) { this.testInst = testInst; @@ -96,6 +105,25 @@ public void setServices(String... services) { requiredServices.addAll(Arrays.asList(services)); } + /** + * Set if the generated tables are loaded into memory before running the test queries. + * + * @return true if in memory source, otherwise false + */ + public void useCachedSource(boolean useMemorySource) { + this.useCachedSource = useMemorySource; + } + + /** + * Set if the generated tables are created through Deephaven (i.e. real client-server) or through the local file + * system (i.e. a local copy). The default of "false" is preferred. + * + * @param useLocalParquet false to generate tables through Deephaven, otherwise false + */ + public void useLocalParquet(boolean useLocalParquet) { + this.useLocalParquet = useLocalParquet; + } + /** * Add a query to be run directly after the main table is loaded. It is not measured. This query can transform the * main table or supporting table, set up aggregations or updateby operations, etc. @@ -117,6 +145,16 @@ public void addPreOpQuery(String query) { preOpQueries.add(query); } + /** + * Add a query to be run after everything else is done. This is useful for teardown of any resources after the test + * is run like logging, temporary files, perf table retrieval, etc. + * + * @param query the query to run after the measured operation + */ + public void addTeardownQuery(String query) { + teardownQueries.add(query); + } + /** * The {@code scale.row.count} property supplies a default for the number of rows generated for benchmark tests. * Given that some operations use less memory than others, scaling up the generated rows per operation is more @@ -143,6 +181,18 @@ public void setScaleFactors(int staticFactor, int incFactor) { this.incFactor = incFactor; } + /** + * Set the incremental release filter to use for the incremental test. By default, this is an auto-tuning release + * filter with a target cycle factor of 1.0f. + * + * @param cycleFactor if isAutoTune is true, the target cycle factor, otherwise ignored + * @param releaseRowsCount if isAutoTune is true, ignored, otherwise the number of rows to release per cycle + */ + public void setIncReleaseFilter(float cycleFactor, long releaseRowCount) { + this.incCycleFactor = cycleFactor; + this.incReleaseRowCount = releaseRowCount; + } + /** * Run a single operation test through the Bench API with no upper bound expected on the resulting row count * @@ -193,12 +243,12 @@ public void test(String name, long maxExpectedRowCount, String operation, String } } - long getWarmupRowCount() { - return (long) (api.propertyAsIntegral("warmup.row.count", "0") * rowCountFactor); + public long getGeneratedRowCount() { + return (long) (api.propertyAsIntegral("scale.row.count", "100000") * rowCountFactor); } - long getGeneratedRowCount() { - return (long) (api.propertyAsIntegral("scale.row.count", "100000") * rowCountFactor); + long getWarmupRowCount() { + return (long) (api.propertyAsIntegral("warmup.row.count", "0") * rowCountFactor); } long getMaxExpectedRowCount(long expectedRowCount, long scaleFactor) { @@ -206,27 +256,29 @@ long getMaxExpectedRowCount(long expectedRowCount, long scaleFactor) { } String getReadOperation(int scaleFactor, long rowCount, String... loadColumns) { - var headRows = (rowCount >= getGeneratedRowCount())?"":".head(${rows})"; + var headRows = (rowCount >= getGeneratedRowCount()) ? "" : ".head(${rows})"; + var selectStr = useCachedSource ? "select" : "view"; if (scaleFactor > 1 && mainTable.equals("timed") && Arrays.asList(loadColumns).contains("timestamp")) { var read = """ merge([ - read('/data/timed.parquet').view(formulas=[${loadColumns}])${headRows} + bench_api_read('/data/timed.parquet').view(formulas=[${loadColumns}])${headRows} ] * ${scaleFactor}).update_view([ 'timestamp=timestamp.plusMillis((long)(ii / ${rows}) * ${rows})' - ]).select() + ]).${selectStr}() """; - read = read.replace("${headRows}",headRows); + read = read.replace("${headRows}", headRows).replace("${selectStr}", selectStr); return read.replace("${scaleFactor}", "" + scaleFactor).replace("${rows}", "" + rowCount); } - var read = "read('/data/${mainTable}.parquet')${headRows}.select(formulas=[${loadColumns}])"; + var read = "bench_api_read('/data/${mainTable}.parquet')${headRows}.${selectStr}(formulas=[${loadColumns}])"; read = (loadColumns.length == 0) ? ("empty_table(${rows})") : read; if (scaleFactor > 1) { read = "merge([${readTable}] * ${scaleFactor})".replace("${readTable}", read); read = read.replace("${scaleFactor}", "" + scaleFactor); } - return read.replace("${headRows}",headRows).replace("${rows}", "" + rowCount); + read = read.replace("${headRows}", headRows).replace("${rows}", "" + rowCount); + return read.replace("${selectStr}", selectStr); } String getStaticQuery(String name, String operation, long rowCount, String... loadColumns) { @@ -241,9 +293,11 @@ String getStaticQuery(String name, String operation, long rowCount, String... lo bench_api_metrics_start() print('${logOperationBegin}') + begin_clock = time.time_ns() begin_time = time.perf_counter_ns() result = ${operation} end_time = time.perf_counter_ns() + end_clock = time.time_ns() print('${logOperationEnd}') bench_api_metrics_end() @@ -253,7 +307,10 @@ String getStaticQuery(String name, String operation, long rowCount, String... lo double_col("elapsed_nanos", [end_time - begin_time]), long_col("processed_row_count", [loaded_tbl_size]), long_col("result_row_count", [result.size]), + long_col("begin_clock_nanos", [begin_clock]), + long_col("end_clock_nanos", [end_clock]), ]) + ${teardownQueries} """; var read = getReadOperation(staticFactor, rowCount, loadColumns); return populateQuery(name, staticQuery, operation, read, loadColumns); @@ -268,16 +325,21 @@ String getIncQuery(String name, String operation, long rowCount, String... loadC loaded_tbl_size = ${mainTable}.size ${setupQueries} - autotune = jpy.get_type('io.deephaven.engine.table.impl.select.AutoTuningIncrementalReleaseFilter') - source_filter = autotune(0, 1000000, 1.0, True) + isat = System.getProperty('train.autotune', 'true').lower() == 'true' or ${incReleaseRowCount} <= 0 + filter_name = 'AutoTuningIncrementalReleaseFilter' if isat else 'IncrementalReleaseFilter' + autotune = jpy.get_type(f'io.deephaven.engine.table.impl.select.{filter_name}') + print("******* ISAT:",isat, "FILTER:", filter_name) + + source_filter = autotune(0,1000000,${incCycleFactor},True) if isat else autotune(0,${incReleaseRowCount}) ${mainTable} = ${mainTable}.where(source_filter) if right: - right_filter = autotune(0, 1010000, 1.0, True) + right_filter = autotune(0,1010000,${incCycleFactor},True) if isat else autotune(0,${incReleaseRowCount}) right = right.where(right_filter) ${preOpQueries} bench_api_metrics_start() print('${logOperationBegin}') + begin_clock = time.time_ns() begin_time = time.perf_counter_ns() result = ${operation} @@ -291,6 +353,7 @@ String getIncQuery(String name, String operation, long rowCount, String... loadC source_filter.waitForCompletion() end_time = time.perf_counter_ns() + end_clock = time.time_ns() print('${logOperationEnd}') bench_api_metrics_end() standard_metrics = bench_api_metrics_collect() @@ -298,8 +361,13 @@ String getIncQuery(String name, String operation, long rowCount, String... loadC stats = new_table([ double_col("elapsed_nanos", [end_time - begin_time]), long_col("processed_row_count", [loaded_tbl_size]), - long_col("result_row_count", [result.size]) + long_col("result_row_count", [result.size]), + long_col("begin_clock_nanos", [begin_clock]), + long_col("end_clock_nanos", [end_clock]), ]) + ${teardownQueries} + print("STANDARD EVENTS: ", f'start_ns > {begin_clock}L', f'start_ns < {end_clock}L') + standard_events = standard_events.where([f'start_ns > {begin_clock}L', f'start_ns < {end_clock}L']) """; var read = getReadOperation(incFactor, rowCount, loadColumns); return populateQuery(name, incQuery, operation, read, loadColumns); @@ -307,14 +375,17 @@ String getIncQuery(String name, String operation, long rowCount, String... loadC String populateQuery(String name, String query, String operation, String read, String... loadColumns) { query = query.replace("${readTable}", read); - query = query.replace("${mainTable}", mainTable); query = query.replace("${loadSupportTables}", loadSupportTables()); query = query.replace("${loadColumns}", listStr(loadColumns)); query = query.replace("${setupQueries}", String.join("\n", setupQueries)); query = query.replace("${preOpQueries}", String.join("\n", preOpQueries)); query = query.replace("${operation}", operation); + query = query.replace("${teardownQueries}", String.join("\n", teardownQueries)); query = query.replace("${logOperationBegin}", getLogSnippet("Begin", name)); query = query.replace("${logOperationEnd}", getLogSnippet("End", name)); + query = query.replace("${incCycleFactor}", "" + incCycleFactor); + query = query.replace("${incReleaseRowCount}", "" + incReleaseRowCount); + query = query.replace("${mainTable}", mainTable); return query; } @@ -326,6 +397,7 @@ Result runTest(String name, String warmupQuery, String mainQuery) { stopUnusedServices(requiredServices); try { + Log.info("Running Test: %s", name); if (getWarmupRowCount() > 0) api.query(warmupQuery).execute(); var result = new AtomicReference(); @@ -342,6 +414,8 @@ Result runTest(String name, String warmupQuery, String mainQuery) { metrics.set("inc.factor", incFactor); metrics.set("row.factor", rowCountFactor); api.metrics().add(metrics); + }).fetchAfter("standard_events", table -> { + api.events().add(table); }).execute(); api.result().test("deephaven-engine", result.get().elapsedTime(), result.get().loadedRowCount()); return result.get(); @@ -356,7 +430,7 @@ String listStr(String... values) { } String loadSupportTables() { - return supportTables.stream().map(t -> t + " = read('/data/" + t + ".parquet').select()\n") + return supportTables.stream().map(t -> t + " = bench_api_read('/data/" + t + ".parquet').select()\n") .collect(Collectors.joining("")); } @@ -435,7 +509,7 @@ boolean generateNamedTable(String name, String distribution, String[] groups) { } boolean generateSourceTable(String distribution, String[] groups) { - return api.table("source") + var t = api.table("source") .add("num1", "double", "[0-4]", distribution) .add("num2", "double", "[1-10]", distribution) .add("key1", "string", "[1-100]", distribution) @@ -444,8 +518,8 @@ boolean generateSourceTable(String distribution, String[] groups) { .add("key4", "int", "[0-98]", distribution) .add("key5", "string", "[1-1000000]", distribution) .withRowCount(getGeneratedRowCount()) - .withColumnGrouping(groups) - .generateParquet(); + .withColumnGrouping(groups); + return useLocalParquet ? t.generateLocalParquet() : t.generateParquet(); } boolean generateRightTable(String distribution, String[] groups) { @@ -469,7 +543,7 @@ boolean generateRightTable(String distribution, String[] groups) { boolean generateTimedTable(String distribution, String[] groups) { long minTime = 1676557157537L; long maxTime = minTime + getGeneratedRowCount() - 1; - return api.table("timed") + var t = api.table("timed") .add("timestamp", "timestamp-millis", "[" + minTime + "-" + maxTime + "]", "ascending") .add("num1", "double", "[0-4]", distribution) .add("num2", "double", "[1-10]", distribution) @@ -478,8 +552,8 @@ boolean generateTimedTable(String distribution, String[] groups) { .add("key3", "int", "[0-8]", distribution) .add("key4", "int", "[0-98]", distribution) .withFixedRowCount(true) - .withColumnGrouping(groups) - .generateParquet(); + .withColumnGrouping(groups); + return useLocalParquet ? t.generateLocalParquet() : t.generateParquet(); } record Result(long loadedRowCount, Duration elapsedTime, long resultRowCount) { 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 5fadc8ba..4126feac 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 @@ -14,6 +14,9 @@ * Test reading and writing parquet files with various data types and compression codecs. */ class FileTestRunner { + static { + System.setProperty("root.test.package", "io.deephaven.benchmark.tests"); + } final String parquetCfg = "max_dictionary_keys=1048576, max_dictionary_size=1048576, target_page_size=65536"; final Object testInst; final Set requiredServices = new TreeSet<>(List.of("deephaven")); diff --git a/src/it/java/io/deephaven/benchmark/tests/standard/kafka/KafkaTestRunner.java b/src/it/java/io/deephaven/benchmark/tests/standard/kafka/KafkaTestRunner.java index d65014d8..c919f852 100644 --- a/src/it/java/io/deephaven/benchmark/tests/standard/kafka/KafkaTestRunner.java +++ b/src/it/java/io/deephaven/benchmark/tests/standard/kafka/KafkaTestRunner.java @@ -19,6 +19,9 @@ * append/blink table types. Results are checked to ensure the correct number of rows has been processed. */ class KafkaTestRunner { + static { + System.setProperty("root.test.package", "io.deephaven.benchmark.tests"); + } final Object testInst; final Bench api; final Controller controller; diff --git a/src/it/java/io/deephaven/benchmark/tests/train/AggByTrainTest.java b/src/it/java/io/deephaven/benchmark/tests/train/AggByTrainTest.java new file mode 100644 index 00000000..5d9bd4f5 --- /dev/null +++ b/src/it/java/io/deephaven/benchmark/tests/train/AggByTrainTest.java @@ -0,0 +1,42 @@ +/* Copyright (c) 2026-2026 Deephaven Data Labs and Patent Pending */ +package io.deephaven.benchmark.tests.train; + +import org.junit.jupiter.api.*; + +/** + * Training tests for the aggBy table operations that do aggregations (e.g. sum, std, min/max. var, avg). See + * TrainTestRunner for more information. + */ +public class AggByTrainTest { + final TrainTestRunner runner = new TrainTestRunner(this); + + void setup(double staticRowFactor, double incRowFactor) { + runner.tables(staticRowFactor, incRowFactor, "timed"); + + var setupStr = """ + from deephaven import agg + + aggs = [ + agg.sum_('Sum=num1'), agg.std('Std=num2'), agg.min_('Min=num1'), agg.max_('Max=num2'), + agg.avg('Avg=num1'), agg.var('Var=num2'), agg.count_('num1') + ] + """; + runner.addSetupQuery(setupStr); + } + + @Test + @Disabled + void aggBy0Groups() { + setup(572, 286); + var q = "timed.agg_by(aggs)"; + runner.test("AggBy- No Groups", 1, q, "num1", "num2"); + } + + @Test + void aggBy2Groups() { + setup(66, 38); + var q = "timed.agg_by(aggs, by=['key1', 'key2'])"; + runner.test("AggBy- 2 Groups 10K Unique Combos", 10100, q, "key1", "key2", "num1", "num2"); + } + +} diff --git a/src/it/java/io/deephaven/benchmark/tests/train/FilterTrainTest.java b/src/it/java/io/deephaven/benchmark/tests/train/FilterTrainTest.java new file mode 100644 index 00000000..a3721852 --- /dev/null +++ b/src/it/java/io/deephaven/benchmark/tests/train/FilterTrainTest.java @@ -0,0 +1,46 @@ +/* Copyright (c) 2026-2026 Deephaven Data Labs and Patent Pending */ +package io.deephaven.benchmark.tests.train; + +import org.junit.jupiter.api.*; + +/** + * Standard tests for the whereIn table operation. Filters rows of data from the source table where the rows match + * column values in the filter table. + */ +public class FilterTrainTest { + final TrainTestRunner runner = new TrainTestRunner(this); + + void setup(double staticRowFactor, double incRowFactor) { + runner.tables(staticRowFactor, incRowFactor, "timed"); + var setup = """ + from deephaven.column import string_col, int_col + where_filter = new_table([ + string_col("set1", ['1', '2', '3', '4', '5', '6', '7', '8']), + string_col("set2", ['10', '20', '30', '40', '50', '60', '70', '80']), + int_col("set3", [-1, -2, -3, -4, 1, 2, 3, 4]) + ]) + """; + runner.addSetupQuery(setup); + } + + @Test + @Disabled + void filter2Cols() { + runner.setIncReleaseRowCount(3382734); + setup(815, 815); + var q = "timed.where_in(where_filter, cols=['key1 = set1']).where(['inRange(num1, 0, 100)'])"; + runner.test("Filter- 2 Cols", 0, q, "key1", "key2", "num1"); + } + + @Test + void filter3Cols() { + runner.setIncReleaseRowCount(3468549); + setup(620, 620); + var q = """ + timed.where_in(where_filter, cols=['key1 = set1', 'key2 = set2', 'key3 = set3']) \ + .where(filters=["key1 = '1'", 'inRange(num1, 0, 100)', 'key3 in -2, -1, 0, 1, 2']) + """; + runner.test("Filter- 3 Cols", 0, q, "key1", "key2", "key3", "num1"); + } + +} diff --git a/src/it/java/io/deephaven/benchmark/tests/train/FormulaTrainTest.java b/src/it/java/io/deephaven/benchmark/tests/train/FormulaTrainTest.java new file mode 100644 index 00000000..c136c639 --- /dev/null +++ b/src/it/java/io/deephaven/benchmark/tests/train/FormulaTrainTest.java @@ -0,0 +1,51 @@ +/* Copyright (c) 2026-2026 Deephaven Data Labs and Patent Pending */ +package io.deephaven.benchmark.tests.train; + +import org.junit.jupiter.api.*; + +/** + * Training tests for the formula table operations (e.g. udf, inline). See TrainTestRunner for more + * information. + */ +public class FormulaTrainTest { + final TrainTestRunner runner = new TrainTestRunner(this); + + void setup(double staticRowFactor, double incRowFactor) { + runner.tables(staticRowFactor, incRowFactor, "timed"); + } + + @Test + void formulaUdf() { + setup(9, 9); + var setup = """ + def f_py(num1: float, num2: float) -> float: + return (num2 + num1) / 2 + def f_np(num1: np.float64, num2: np.float64) -> np.float64: + return num1 + num2 + """; + runner.addSetupQuery(setup); + var q = "timed.view(['New1 = f_py(num1, num2)','New2 = f_np(num1, num2)']).sum_by()"; + runner.test("Formula- UDF 2 Calcs", 1, q, "num1", "num2"); + } + + @Test + void formulaInline() { + setup(467, 467); + var q = "timed.view(['New1 = (float)((num2 + num1) / 2)', 'New2 = (float)(num1 + num2)']).sum_by()"; + runner.test("Formula- Inline 2 Calcs", 1, q, "num1", "num2"); + } + + @Test + @Disabled + void formulaDate() { + setup(3, 3); + var q = """ + timed.view([ + 'New1 = parseDuration(`PT4H52M14S`).toHours()', + 'New1 = parseInstant(`2023-05-31T04:52:14.001 ET`).getEpochSecond()' + ]).sum_by() + """; + runner.test("Formula- Inline 2 Dates", 1, q, "num1", "num2"); + } + +} diff --git a/src/it/java/io/deephaven/benchmark/tests/train/NaturalJoinTrainTest.java b/src/it/java/io/deephaven/benchmark/tests/train/NaturalJoinTrainTest.java new file mode 100644 index 00000000..eb2559f4 --- /dev/null +++ b/src/it/java/io/deephaven/benchmark/tests/train/NaturalJoinTrainTest.java @@ -0,0 +1,36 @@ +/* Copyright (c) 2026-2026 Deephaven Data Labs and Patent Pending */ +package io.deephaven.benchmark.tests.train; + +import org.junit.jupiter.api.*; + +/** + * Training tests for the aggBy table operations that do joins (e.g. natural join). See TrainTestRunner for + * more information. + */ +public class NaturalJoinTrainTest { + final TrainTestRunner runner = new TrainTestRunner(this); + + void setup(double staticRowFactor, double incRowFactor) { + runner.tables(staticRowFactor, incRowFactor, "timed", "right"); + } + + @Test + @Disabled + void naturalJoinOn1Col() { + runner.setIncReleaseRowCount(448939); + setup(230, 120); + var r = "right = right.select_distinct(['r_wild'])"; + runner.addSetupQuery(r); + var q = "timed.natural_join(right, on=['key1 = r_wild'])"; + runner.test("NaturalJoin- Join On 1 Col", 0, q, "key1", "num1"); + } + + @Test + void naturalJoinOn3Cols() { + runner.setIncReleaseRowCount(52043); + setup(100, 20); + var q = "timed.natural_join(right, on=['key1 = r_wild', 'key2 = r_key2', 'key1 = r_key1'])"; + runner.test("NaturalJoin- Join On 3 Cols", 0, q, "key1", "key2", "num1"); + } + +} diff --git a/src/it/java/io/deephaven/benchmark/tests/train/OrderedTrainTest.java b/src/it/java/io/deephaven/benchmark/tests/train/OrderedTrainTest.java new file mode 100644 index 00000000..961f1c5d --- /dev/null +++ b/src/it/java/io/deephaven/benchmark/tests/train/OrderedTrainTest.java @@ -0,0 +1,43 @@ +/* Copyright (c) 2026-2026 Deephaven Data Labs and Patent Pending */ +package io.deephaven.benchmark.tests.train; + +import org.junit.jupiter.api.*; + +/** + * Training tests for the aggBy table operations that do ordering (e.g.. median, percentile, sorted_first/last). See + * TrainTestRunner for more information. + * + */ +public class OrderedTrainTest { + final TrainTestRunner runner = new TrainTestRunner(this); + + void setup(double staticRowFactor, double incRowFactor) { + runner.tables(staticRowFactor, incRowFactor, "timed"); + + var setupStr = """ + from deephaven import agg + aggs = [ + agg.median('Median=num1'), agg.pct(0.50, ['Percentile=num1']), + agg.unique('Unique=num2'), agg.sorted_first('key4', ['num2']), + agg.sorted_last('key3', ['num1']) + ] + """; + runner.addSetupQuery(setupStr); + } + + @Test + @Disabled + void ordered0Groups() { + setup(145, 18); + var q = "timed.agg_by(aggs)"; + runner.test("Ordered- No Groups", 100, q, "key3", "key4", "num1", "num2"); + } + + @Test + void ordered2Groups() { + setup(30, 5.7); + var q = "timed.agg_by(aggs, by=['key1', 'key2'])"; + runner.test("Ordered- 2 Groups 10K Unique Combos", 10100, q, "key1", "key2", "key3", "key4", "num1", "num2"); + } + +} diff --git a/src/it/java/io/deephaven/benchmark/tests/train/TrainTestRunner.java b/src/it/java/io/deephaven/benchmark/tests/train/TrainTestRunner.java new file mode 100644 index 00000000..5e9d5b24 --- /dev/null +++ b/src/it/java/io/deephaven/benchmark/tests/train/TrainTestRunner.java @@ -0,0 +1,248 @@ +/* Copyright (c) 2026-2026 Deephaven Data Labs and Patent Pending */ +package io.deephaven.benchmark.tests.train; + +import java.util.*; +import io.deephaven.benchmark.tests.standard.StandardTestRunner; + +/** + * A wrapper for the Bench api that allows the running of operational training (think AOT) tests without requiring the + * boilerplate logic like imports, parquet reads, time measurement logic, etc. Each test runs two + * benchmarks; one reading from a static parquet, and the other exercising ticking tables through the + * AutotuningIncrementalReleaseFilter. This is different from the StandardTestRunner in that + * it runs more than one operation per benchmark and attempts to cover the majority of the query code base with fewer + * benchmarks. It is meant for training AOT and for "representative" benchmarks used to compare things like JDK/Python + * versions and GC types. + */ +final public class TrainTestRunner { + static final int maxRowFactor = 620; + static final float incCycleFactor = 0.90f; + static final double incReleaseFactor = 1.0f; + final Object testInst; + final List setupQueries = new ArrayList<>(); + final List teardownQueries = new ArrayList<>(); + private double staticRowFactor = 1; + private double incRowFactor = 1; + private long incReleaseRowCount = 0; + private String[] tableNames = null; + + TrainTestRunner(Object testInst) { + this.testInst = testInst; + } + + public void tables(double staticRowFactor, double incRowFactor, String... names) { + if (Math.max(staticRowFactor, incRowFactor) > maxRowFactor) + throw new IllegalArgumentException("Row factors cannot be greater than " + maxRowFactor); + this.staticRowFactor = 0; // staticRowFactor; + this.incRowFactor = incRowFactor; + tableNames = names; + } + + public void addSetupQuery(String query) { + setupQueries.add(query); + } + + public void setIncReleaseRowCount(long incReleaseRowCount) { + this.incReleaseRowCount = incReleaseRowCount; + } + + public void test(String name, long maxExpectedRowCount, String operation, String... loadColumns) { + if (staticRowFactor <= 0 && incRowFactor <= 0) + throw new IllegalStateException("At least one of staticRowFactor or incRowFactor must be > 0"); + + setupQueries.add(startJfrQuery); + teardownQueries.add(stopJfrQuery); + + if (staticRowFactor > 0) + test(name, maxExpectedRowCount, operation, staticRowFactor, true, loadColumns); + + if (incRowFactor > 0) { + setupQueries.add(startUgpQuery); + teardownQueries.add(stopUgpQuery); + operation += "\ntrain_ugp_listener = listen(result, train_ugp_update)"; + test(name, maxExpectedRowCount, operation, incRowFactor, false, loadColumns); + } + } + + void test(String name, long maxExpectedRowCount, String operation, double rowFactor, boolean isStatic, + String... loadColumns) { + var delegate = new StandardTestRunner(testInst); + var baseRowCount = delegate.getGeneratedRowCount(); + delegate.useCachedSource(false); + delegate.useLocalParquet(true); + delegate.setRowFactor(maxRowFactor); + delegate.tables(tableNames); + delegate.setScaleFactors(isStatic ? 1 : 0, isStatic ? 0 : 1); + delegate.setIncReleaseFilter(incCycleFactor, (long) (incReleaseRowCount * incReleaseFactor)); + + var headQuery = """ + ${mainTable} = ${mainTable}.head(${trainRowCount}) + loaded_tbl_size = ${mainTable}.size + """.replace("${trainRowCount}", String.valueOf((long) (baseRowCount * rowFactor * incReleaseFactor))); + + delegate.addSetupQuery(headQuery); + setupQueries.forEach(delegate::addSetupQuery); + teardownQueries.forEach(delegate::addTeardownQuery); + delegate.test(name, maxExpectedRowCount, operation, loadColumns); + } + + static final String startJfrQuery = """ + import jpy + Recording = jpy.get_type("jdk.jfr.Recording") + rec = Recording() + rec.setName("benchmark") + + enabled_events=['jdk.ExecutionSample','jdk.NativeMethodSample','jdk.ThreadCPULoad','jdk.GarbageCollection', + 'jdk.GCPhasePause','jdk.SafepointBegin','jdk.SafepointEnd','jdk.SafepointState', + 'jdk.ObjectAllocationInNewTLAB','jdk.ObjectAllocationOutsideTLAB'] + for n in enabled_events: + try: + rec.enable(n) + except Exception: + print(f"Event Not Enabled: {n}") + + disabled_events=['jdk.GCPhaseConcurrent','jdk.GCPhaseConcurrentMark','jdk.GCPhaseConcurrentEvacuation', + 'jdk.G1GarbageCollection','jdk.ShenandoahGarbageCollection','jdk.ZGarbageCollection','jdk.GCHeapSummary', + 'jdk.GCReferenceStatistics','jdk.GCWorkerData','jdk.GCCPUTime','jdk.GCPhasePause'] + for n in disabled_events: + try: + rec.disable(n) + except Exception: + print(f"Event Not Disabled: {n}") + + rec.start() + """; + + static final String stopJfrQuery = """ + Paths = jpy.get_type("java.nio.file.Paths") + RecordingFile = jpy.get_type("jdk.jfr.consumer.RecordingFile") + + rec.dump(Paths.get("/data/benchmark.jfr")) + rec.stop() + rec.close() + + events = RecordingFile.readAllEvents(Paths.get("/data/benchmark.jfr")) + jfr_rows = [] + + def getEventValue(ev, field): + try: + return ev.getValue(field) + except Exception: + return None + + def getNanoValue(ev, duration_field): + val = ev.getValue(duration_field) + if val is None or str(val) == "null": return 0 + if isinstance(val, int): return val + if hasattr(val, "size") and hasattr(val, "get"): + total = 0 + for i in range(val.size()): + d = val.get(i) + if d is not None and str(d) != "null": total += d.toNanos() + return total + if hasattr(val, "toNanos"): return val.toNanos() + raise TypeError(f"Unsupported JFR value type: {type(val)}") + + for i in range(events.size()): + e = events.get(i) + etype = e.getEventType().getName() + start = e.getStartTime().getEpochSecond() * 1000000000 + e.getStartTime().getNano() + + if etype == 'jdk.GarbageCollection': + duration = getNanoValue(e, 'duration') + name = 'sumOfPauses' + value = getNanoValue(e, 'sumOfPauses') + elif etype == 'jdk.GCPhasePause': + duration = getNanoValue(e, 'duration') + name = getEventValue(e, 'name') + value = duration + else: + continue + + jfr_rows.append([etype, start, duration, name, value]) + + if len(jfr_rows) > 0: + jfr_gc = new_table([ + string_col("origin", ["deephaven-engine" for r in jfr_rows]), + string_col("type", [r[0] for r in jfr_rows]), + long_col("start_ns", [r[1] for r in jfr_rows]), + long_col("duration_ns", [r[2] for r in jfr_rows]), + string_col("name", [r[3] for r in jfr_rows]), + double_col("value", [r[4] for r in jfr_rows]), + ]) + standard_events = merge([standard_events, jfr_gc]) + """; + + static final String startUgpQuery = """ + from deephaven import time_table, perfmon + from deephaven.table_listener import listen + import time + + ss_log = perfmon.server_state_log() + if 'train_ugp_listener' in globals(): train_ugp_listener.stop() + train_wall_epoch_ns = time.time_ns() + train_ugp_times = [(time.perf_counter_ns(), 0, 0)] + + def train_ugp_update(update, is_replay): + ugp_cycle_cost = 0 + if autotune: + ug = update.table.update_graph.j_update_graph + ugp_cycle_cost = System.nanoTime() - ug.cycleStartNanoTime() + train_ugp_times.append((time.perf_counter_ns(), ${mainTable}.size, ugp_cycle_cost)) + """; + + static final String stopUgpQuery = """ + if 'train_ugp_listener' in globals(): train_ugp_listener.stop() + if len(train_ugp_times) > 1: + mono_start = train_ugp_times[0][0] + ugp_rows = [] + for i in range(1, len(train_ugp_times)): + mono_prev = train_ugp_times[i - 1][0] + mono_curr = train_ugp_times[i][0] + size_prev = train_ugp_times[i - 1][1] + size_curr = train_ugp_times[i][1] + ugp_cycle_cost = train_ugp_times[i][2] + delta_ns = mono_curr - mono_prev + wall_clock_ns = train_wall_epoch_ns + (mono_curr - mono_start) + delta_rows = max(0, size_curr - size_prev) + ugp_cycle_cost = max(0, ugp_cycle_cost) + ugp_rows.append([wall_clock_ns, delta_ns, delta_rows, ugp_cycle_cost]) + + ugp_events = new_table([ + string_col("origin", ["deephaven-engine"] * len(ugp_rows)), + string_col("type", ["ugp.delta"] * len(ugp_rows)), + long_col("start_ns", [r[0] for r in ugp_rows]), + long_col("duration_ns", [r[1] for r in ugp_rows]), + string_col("name", ["duration_rows"] * len(ugp_rows)), + double_col("value", [float(r[2]) for r in ugp_rows]), + ]) + + ugp_cycle_events = new_table([ + string_col("origin", ["deephaven-engine"] * len(ugp_rows)), + string_col("type", ["ugp.cycle.cost"] * len(ugp_rows)), + long_col("start_ns", [r[0] for r in ugp_rows]), + long_col("duration_ns", [r[3] for r in ugp_rows]), + string_col("name", ["duration_rows"] * len(ugp_rows)), + double_col("value", [float(r[2]) for r in ugp_rows]), + ]) + + standard_events = merge([standard_events, ugp_events, ugp_cycle_events]) + + ss_log = perfmon.server_state_log().ungroup(["IntervalUGPCyclesTimeMicros"]).snapshot() + if ss_log.size > 0: + ss_rows = [] + for row in ss_log.iter_dict(): + start = row['IntervalStartTime'].getEpochSecond() * 1000000000 + row['IntervalStartTime'].getNano() + ss_rows.append((start, row['IntervalCollectionTimeMicros'] * 1000, + row['IntervalUGPCyclesTimeMicros'] * 1000)) + + ss_events = new_table([ + string_col("origin", ["deephaven-engine"] * len(ss_rows)), + string_col("type", ["server_state_log"] * len(ss_rows)), + long_col("start_ns", [r[0] for r in ss_rows]), + long_col("duration_ns", [r[1] for r in ss_rows]), + string_col("name", ["ugp.cycle.time"] * len(ss_rows)), + double_col("value", [r[2] for r in ss_rows]), + ]) + standard_events = merge([standard_events, ss_events]) + """; +} diff --git a/src/it/java/io/deephaven/benchmark/tests/train/UpdateByTrainTest.java b/src/it/java/io/deephaven/benchmark/tests/train/UpdateByTrainTest.java new file mode 100644 index 00000000..cdfd8844 --- /dev/null +++ b/src/it/java/io/deephaven/benchmark/tests/train/UpdateByTrainTest.java @@ -0,0 +1,54 @@ +/* Copyright (c) 2022-2026 Deephaven Data Labs and Patent Pending */ +package io.deephaven.benchmark.tests.train; + +import org.junit.jupiter.api.*; + +/** + * Standard tests for the updateBy table operation. Combines a mixture of rolling operations and cumulative operations + */ +public class UpdateByTrainTest { + final TrainTestRunner runner = new TrainTestRunner(this); + final String noGroups = """ + avg_contains = rolling_avg_time(ts_col='timestamp',cols=['A=num1','B=num2'],rev_time='PT5S',fwd_time='PT5S') + max_before = rolling_max_tick(cols=['C=num1','D=num2'], rev_ticks=3000,fwd_ticks=-1000) + prod_after = rolling_prod_time(ts_col='timestamp',cols=['E=num1','F=num2'],rev_time='-PT1S',fwd_time='PT4S') + """; + final String group10K = """ + avg_contains = rolling_avg_time(ts_col='timestamp',cols=['A=num1','B=num2'],rev_time='PT4M',fwd_time='PT5M') + max_before = rolling_max_tick(cols=['C=num1','D=num2'], rev_ticks=30,fwd_ticks=-10) + prod_after = rolling_prod_time(ts_col='timestamp',cols=['E=num1','F=num2'],rev_time='-PT1M',fwd_time='PT4M') + """; + + void setup(double staticRowFactor, double incRowFactor) { + runner.tables(staticRowFactor, incRowFactor, "timed"); + var setup = """ + from deephaven.updateby import rolling_avg_time, rolling_max_tick, rolling_prod_time + from deephaven.updateby import ema_tick, cum_min, cum_sum + + ema_tick_op = ema_tick(decay_ticks=10000,cols=['G=num1','H=num2']) + min_op = cum_min(cols=['I=num1','J=num2']) + sum_op = cum_sum(cols=['K=num1','L=num2']) + """; + runner.addSetupQuery(setup); + } + + @Test + @Disabled + void mixedComboNoGroups() { + setup(21.8, 17); + runner.addSetupQuery(noGroups); + var q = "timed.update_by(ops=[avg_contains, max_before, prod_after, ema_tick_op, min_op, sum_op])"; + runner.test("UpdateBy- No Groups 12 Cols", 0, q, "num1", "num2", "timestamp"); + } + + @Test + void rollingCombo2Groups() { + setup(9, 0.2); // 0.2 for 100ms, 4.5 for 1s + runner.addSetupQuery(group10K); + var q = """ + timed.update_by(ops=[avg_contains,max_before,prod_after,ema_tick_op,min_op,sum_op], by=['key1','key2']) + """; + runner.test("UpdateBy- 2 Groups 10K Unique Combos", 0, q, "key1", "key2", "num1", "num2", "timestamp"); + } + +} diff --git a/src/main/java/io/deephaven/benchmark/api/Bench.java b/src/main/java/io/deephaven/benchmark/api/Bench.java index c4330dd9..1d3e5ab6 100644 --- a/src/main/java/io/deephaven/benchmark/api/Bench.java +++ b/src/main/java/io/deephaven/benchmark/api/Bench.java @@ -1,8 +1,7 @@ -/* Copyright (c) 2022-2025 Deephaven Data Labs and Patent Pending */ +/* Copyright (c) 2022-2026 Deephaven Data Labs and Patent Pending */ package io.deephaven.benchmark.api; import java.io.Closeable; -import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; import java.time.Duration; @@ -11,9 +10,7 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import io.deephaven.benchmark.metric.Metrics; -import io.deephaven.benchmark.util.Filer; -import io.deephaven.benchmark.util.Ids; -import io.deephaven.benchmark.util.Timer; +import io.deephaven.benchmark.util.*; /** * The root accessor class for the API. Use Bench.create(this) in a typical JUnit test to start things off @@ -27,6 +24,7 @@ final public class Bench { * The root benchmark result directory */ static final public Path rootOutputDir = Paths.get("results"); + /** * The name of the benchmark results csv file */ @@ -35,6 +33,10 @@ final public class Bench { * The name of the benchmark metrics csv file */ static final public String metricsFileName = "benchmark-metrics.csv"; + /** + * The name of the benchmark events csv file + */ + static final public String eventsFileName = "benchmark-events.csv"; /** * The name of the benchmark platform csv file */ @@ -61,9 +63,9 @@ static public Bench create(Object testInst) { return v; } - final Object testInst; final BenchResult result; final BenchMetrics metrics; + final BenchEvents events; final BenchPlatform platform; final QueryLog queryLog; final BenchLog runLog; @@ -72,13 +74,13 @@ static public Bench create(Object testInst) { final Session session = new Session(); private boolean isClosed = false; - Bench(Class testInst) { - this.testInst = testInst; + Bench(Class testClass) { this.result = new BenchResult(outputDir); this.metrics = new BenchMetrics(outputDir); + this.events = new BenchEvents(outputDir); this.platform = new BenchPlatform(this, outputDir); - this.queryLog = new QueryLog(outputDir, testInst); - this.runLog = new BenchLog(outputDir, testInst); + this.queryLog = new QueryLog(getLogDir(testClass), testClass); + this.runLog = new BenchLog(getLogDir(testClass), testClass); } /** @@ -91,6 +93,7 @@ public void setName(String name) { throw new RuntimeException("No blank Benchmark names allowed"); this.result.setName(name); this.metrics.setName(name); + this.events.setName(name); this.queryLog.setName(name); this.runLog.setName(name); } @@ -199,6 +202,15 @@ public BenchMetrics metrics() { return metrics; } + /** + * Get the events for this Benchmark instance (e.g. test) used for collecting event values + * + * @return the events instance + */ + public BenchEvents events() { + return events; + } + /** * Get the platform for this Benchmark instance (e.g. test) used for collecting platform properties * @@ -209,9 +221,9 @@ public BenchPlatform platform() { } /** - * Get the metrics for this Benchmark instance (e.g. test) used for collecting metric values + * Get the query log for this Benchmark instance (e.g. test) used for recording queries * - * @return the metrics instance + * @return the query log instance */ public BenchLog log() { return runLog; @@ -243,6 +255,7 @@ public void close() { closeables.clear(); result.commit(); metrics.commit(); + events.commit(); platform.commit(); runLog.close(); queryLog.close(); @@ -270,6 +283,12 @@ > T addFuture(T future) { return future; } + static private Path getLogDir(Class testClass) { + var pkgRoot = profile.property("root.test.package", Bench.class.getPackageName().replaceAll("[.][^.]+$", "")); + var name = testClass.getPackageName().replaceAll(pkgRoot + '.', "") + '.' + testClass.getSimpleName(); + return Filer.createDirectory(outputDir.resolve("test-logs").resolve(name).toString()); + } + static private Path initializeOutputDirectory() { setSystemProperties(); boolean isTimestamped = profile.propertyAsBoolean("timestamp.test.results", "false"); @@ -277,11 +296,7 @@ static private Path initializeOutputDirectory() { if (isTimestamped) dir = dir.resolve(Ids.runId()); Filer.delete(dir); - try { - return Files.createDirectories(dir); - } catch (Exception ex) { - throw new RuntimeException("Failed initialize benchmark result directory: " + dir, ex); - } + return Filer.createDirectory(dir.toString()); } static private void setSystemProperties() { diff --git a/src/main/java/io/deephaven/benchmark/api/BenchEvents.java b/src/main/java/io/deephaven/benchmark/api/BenchEvents.java new file mode 100644 index 00000000..1ecef26d --- /dev/null +++ b/src/main/java/io/deephaven/benchmark/api/BenchEvents.java @@ -0,0 +1,85 @@ +/* Copyright (c) 2026-2026 Deephaven Data Labs and Patent Pending */ +package io.deephaven.benchmark.api; + +import java.io.BufferedWriter; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.util.*; +import io.deephaven.benchmark.connect.ResultTable; + +/** + * Represents the events gathered during usage of the Bench API. These can include events gather by the API or the user. + */ +final public class BenchEvents { + static final String header = "benchmark_name,origin,type,start,duration,name,value"; + final List events = new ArrayList<>(); + final Path file; + private String name = null; + + BenchEvents(Path parent) { + this(parent, Bench.eventsFileName); + } + + BenchEvents(Path parent, String resultFileName) { + this.file = parent.resolve(resultFileName); + } + + /** + * Add the results from a table as events to persist to the file system. This table must have columns defined as + * origin, type, start_ns, duration_ns, detail + * + * @param table a table containing events + * @return this instance + */ + public BenchEvents add(ResultTable table) { + for (int r = 0, rn = table.getRowCount(); r < rn; r++) { + var origin = table.getValue(r, "origin").toString(); + var type = table.getValue(r, "type").toString(); + var startNanos = table.getNumber(r, "start_ns").longValue(); + var durationNanos = table.getNumber(r, "duration_ns").longValue(); + var name = String.valueOf(table.getValue(r, "name")); + var value = table.getNumber(r, "value").doubleValue(); + var event = new Event(origin, type, startNanos, durationNanos, name, value); + events.add(event); + } + return this; + } + + /** + * Save the collected events to a csv file. + */ + public void commit() { + if (!hasHeader()) + writeLine(header, file); + + for (Event event : events) { + var line = name + ',' + event.toCsv(); + writeLine(line, file); + } + } + + void setName(String name) { + this.name = name; + } + + private boolean hasHeader() { + return Files.exists(file); + } + + static void writeLine(String line, Path file) { + try (BufferedWriter out = Files.newBufferedWriter(file, StandardOpenOption.CREATE, StandardOpenOption.APPEND)) { + out.write(line); + out.newLine(); + } catch (Exception ex) { + throw new RuntimeException("Failed to write result to file: " + file, ex); + } + } + + record Event(String origin, String type, long startNanos, long durationNanos, String name, double value) { + String toCsv() { + return origin + "," + type + "," + startNanos + "," + durationNanos + "," + name + "," + value; + } + } + +} diff --git a/src/main/java/io/deephaven/benchmark/api/BenchLog.java b/src/main/java/io/deephaven/benchmark/api/BenchLog.java index 3f4de79b..288c5484 100644 --- a/src/main/java/io/deephaven/benchmark/api/BenchLog.java +++ b/src/main/java/io/deephaven/benchmark/api/BenchLog.java @@ -1,4 +1,4 @@ -/* Copyright (c) 2022-2023 Deephaven Data Labs and Patent Pending */ +/* Copyright (c) 2022-2026 Deephaven Data Labs and Patent Pending */ package io.deephaven.benchmark.api; import static java.nio.file.StandardOpenOption.*; @@ -26,7 +26,7 @@ final public class BenchLog { BenchLog(Path parent, Class testClass) { this.testClass = testClass; this.parent = parent; - this.logFile = getLogFile(parent, testClass); + this.logFile = parent.resolve("engine.log"); } /** diff --git a/src/main/java/io/deephaven/benchmark/api/BenchPlatform.java b/src/main/java/io/deephaven/benchmark/api/BenchPlatform.java index 1266eac5..85d1dece 100644 --- a/src/main/java/io/deephaven/benchmark/api/BenchPlatform.java +++ b/src/main/java/io/deephaven/benchmark/api/BenchPlatform.java @@ -1,4 +1,4 @@ -/* Copyright (c) 2022-2025 Deephaven Data Labs and Patent Pending */ +/* Copyright (c) 2022-2026 Deephaven Data Labs and Patent Pending */ package io.deephaven.benchmark.api; import java.io.BufferedWriter; @@ -180,7 +180,7 @@ def bench_api_add_proc_info(name, proctype, prockeys, delim=' '): def bench_api_add_proc_args(name, proctype, len_name, includes=(), excludes=()): prockeys = [] for i in range(int(bench_api_get_proc_info(proctype,len_name))): - value = bench_api_get_proc_info('runtime-mx.jvm-args',str(i)) + value = bench_api_get_proc_info(proctype,str(i)) if(value.startswith(includes) and not value.startswith(excludes)): prockeys.append(str(i)) bench_api_add_proc_info(name, proctype, prockeys) @@ -195,11 +195,15 @@ def bench_api_add_proc_args(name, proctype, len_name, includes=(), excludes=()): bench_api_add_proc_info('java.version','runtime-mx.sys-props',['java.vendor.version']) bench_api_add_proc_info('java.max.heap','memory-mx.heap',['max']) bench_api_add_platform('java.available.processors',Runtime.getRuntime().availableProcessors()) - bench_api_add_proc_args('java.xx.args','runtime-mx.jvm-args','len',('-XX'),('-XX:CompilerDirectivesFile=','-XX:+UnlockD')) + bench_api_add_proc_args('java.xx.args','runtime-mx.jvm-args','len',('-XX'),('-XX:CompilerDirectivesFile=', + '-XX:+UnlockD')) + bench_api_add_proc_args('java.d.args','runtime-mx.jvm-args','len',('-D'),('-Dauthentication','-DAuthHandlers', + '-Ddeephaven.console')) dhInst = jpy.get_type('io.deephaven.engine.exceptions.ArgumentException')() deephaven_version = dhInst.getClass().getPackage().getImplementationVersion() bench_api_add_platform('deephaven.version', deephaven_version) - python_version = '.'.join([str(sys.version_info.major), str(sys.version_info.minor), str(sys.version_info.micro)]) + python_version = '.'.join([str(sys.version_info.major), str(sys.version_info.minor), + str(sys.version_info.micro)]) bench_api_add_platform('python.version', python_version) # Java Dependency Versions diff --git a/src/main/java/io/deephaven/benchmark/api/BenchQuery.java b/src/main/java/io/deephaven/benchmark/api/BenchQuery.java index 358c4a3e..22d90bdd 100644 --- a/src/main/java/io/deephaven/benchmark/api/BenchQuery.java +++ b/src/main/java/io/deephaven/benchmark/api/BenchQuery.java @@ -1,4 +1,4 @@ -/* Copyright (c) 2022-2025 Deephaven Data Labs and Patent Pending */ +/* Copyright (c) 2022-2026 Deephaven Data Labs and Patent Pending */ package io.deephaven.benchmark.api; import java.io.Closeable; @@ -107,8 +107,13 @@ public void close() { return; if (!conn.getUsedVariableNames().isEmpty()) { - String logic = String.join("=None; ", conn.getUsedVariableNames()) + "=None\n"; - executeBarrageQuery(logic); + StringBuilder sb = new StringBuilder(); + sb.append("from deephaven.execution_context import get_exec_ctx\n"); + sb.append("_qs = get_exec_ctx().j_exec_ctx.getQueryScope()\n"); + for (String name : conn.getUsedVariableNames()) { + sb.append("_qs.putParam('").append(name).append("', None)\n"); + } + executeBarrageQuery(sb.toString()); } } diff --git a/src/main/java/io/deephaven/benchmark/api/BenchTable.java b/src/main/java/io/deephaven/benchmark/api/BenchTable.java index b00c00e3..50198c42 100644 --- a/src/main/java/io/deephaven/benchmark/api/BenchTable.java +++ b/src/main/java/io/deephaven/benchmark/api/BenchTable.java @@ -1,17 +1,17 @@ -/* Copyright (c) 2022-2024 Deephaven Data Labs and Patent Pending */ +/* Copyright (c) 2022-2026 Deephaven Data Labs and Patent Pending */ package io.deephaven.benchmark.api; import java.io.Closeable; import java.time.Duration; import java.time.temporal.ChronoUnit; +import java.util.ArrayList; +import java.util.List; import java.util.concurrent.Future; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import io.deephaven.benchmark.generator.*; import io.deephaven.benchmark.metric.Metrics; -import io.deephaven.benchmark.util.Ids; -import io.deephaven.benchmark.util.Log; -import io.deephaven.benchmark.util.Numbers; -import io.deephaven.benchmark.util.Timer; +import io.deephaven.benchmark.util.*; /** * Represents the configuration of table name and columns. @@ -24,7 +24,7 @@ final public class BenchTable implements Closeable { private int durationSecs = -1; private int rowPauseMillis = -1; private String compression = null; - private Generator generator = null; + private List generators = new ArrayList<>(); private boolean isFixed = false; private String defaultDistro = null; private String[] columnGrouping = null; @@ -188,7 +188,7 @@ public boolean generateParquet() { }).execute(); if (usedExistingParquet.get()) { - Log.info("Using existing table '%s' with %s rows", tableName, getRowCount()); + Log.info("\nUsing existing table '%s' with %s rows", tableName, getRowCount()); return false; } Log.info("Generating table '%s' with %s rows", tableName, getRowCount()); @@ -197,14 +197,71 @@ public boolean generateParquet() { if (rowPauseMillis < 0) withRowPause(0, ChronoUnit.MILLIS); - bench.awaitCompletion(generateWithAvro()); - Log.info("Produce Data Duration: " + timer.duration().toMillis()); + var m = bench.awaitCompletion(generateWithAvro()); + Log.info("Produce Send Rate: %.2f recs/sec", m.getValue("send.rate")); + Log.info("Produce Data Duration: %d secs", timer.duration().toSeconds()); timer = Timer.start(); q = replaceTableAndGeneratorFields(kafkaToParquetQuery); bench.query(q).execute(); - Log.info("DH Write Table Duration: " + timer.duration().toMillis()); + Log.info("DH Write Table Duration: %d secs", timer.duration().toSeconds()); + return true; + } + + /** + * Generate the table synchronously to a parquet file in the engine's data directory. If a parquet file already + * exists in the Deephaven data directory that matches this table definition, use it and skip generation. + *

+ * Note: This is the same as generateParquet() except it generates the parquet file directly to the + * engine's data directory without going through kafka. As such, it will not work when the test runner and the + * engine are not co-located. + * + * @return true if file was generated, otherwise false + */ + public boolean generateLocalParquet() { + columns.setDefaultDistribution(getDefaultDistro()); + var q = replaceTableAndGeneratorFields(useExistingParquetQuery); + + var usedExistingParquet = new AtomicBoolean(false); + var tableGenParquet = new AtomicReference(""); + var dhHostOsDir = new AtomicReference(""); + bench.query(q).fetchAfter("used_existing_parquet_" + tableName, table -> { + usedExistingParquet.set(table.getValue(0, "UsedExistingParquet").toString().equalsIgnoreCase("true")); + tableGenParquet.set(table.getValue(0, "TableGenParquet").toString()); + dhHostOsDir.set(table.getValue(0, "DhHostOsDir").toString()); + }).execute(); + + if (usedExistingParquet.get()) { + Log.info("Using existing table '%s' with %s rows", tableName, getRowCount()); + return false; + } + Log.info("Generating table '%s' with %s rows", tableName, getRowCount()); + var timer = Timer.start(); + + if (rowPauseMillis < 0) + withRowPause(0, ChronoUnit.MILLIS); + + if (dhHostOsDir.get().isEmpty()) + throw new RuntimeException("DEEPHAVEN_HOST_OS_DIR env must be set to use local parquet generation"); + + var parquetPath = (dhHostOsDir.get() + "/" + tableGenParquet.get()).replace(".parquet", ".dataset"); + var threadCount = 8; + var rowsPerThread = getRowCount() / threadCount; + var futures = new ArrayList>(threadCount); + for (int i = 0; i < threadCount; i++) { + long rows = (i < threadCount - 1) ? rowsPerThread : (getRowCount() - (rowsPerThread * i)); + var future = generateWithLocalParquet(parquetPath, String.format("%04d.parquet", i), i, rows); + futures.add(future); + } + futures.stream().forEach(future -> bench.awaitCompletion(future)); + close(); // Needed for the final parquet flushes + + bench.query(localToParquetQuery).execute(); + var durMillis = timer.duration().toMillis(); + Log.info("Produce Send Rate: %.2f recs/sec", getRowCount() / (durMillis / 1000.0)); + Log.info("Produce Data Duration: %.2f secs", durMillis / 1000.0); + Log.info("Produce Write Rate: %.2f MB/sec", Filer.getByteSize(parquetPath) * 1000.0 / durMillis / 1024 / 1024); return true; } @@ -212,29 +269,41 @@ public boolean generateParquet() { * Shutdown and cleanup any running generator */ public void close() { - if (generator != null) + for (Generator generator : generators) generator.close(); + generators.clear(); } private Future generateWithAvro() { String bootstrapServer = bench.property("client.redpanda.addr", "localhost:9092"); String schemaRegistry = "http://" + bench.property("client.schema.registry.addr", "localhost:8081"); - generator = new AvroKafkaGenerator(bootstrapServer, schemaRegistry, tableName, columns, getCompression()); - return generator.produce(getRowPause(), getRowCount(), getRunDuration()); + var gen = new AvroKafkaGenerator(bootstrapServer, schemaRegistry, tableName, columns, getCompression()); + generators.add(gen); + return gen.produce(getRowPause(), getRowCount(), getRunDuration()); } private Future generateWithJson() { String bootstrapServer = bench.property("client.redpanda.addr", "localhost:9092"); String schemaRegistry = "http://" + bench.property("client.schema.registry.addr", "localhost:8081"); - generator = new JsonKafkaGenerator(bootstrapServer, schemaRegistry, tableName, columns, getCompression()); - return generator.produce(getRowPause(), getRowCount(), getRunDuration()); + var gen = new JsonKafkaGenerator(bootstrapServer, schemaRegistry, tableName, columns, getCompression()); + generators.add(gen); + return gen.produce(getRowPause(), getRowCount(), getRunDuration()); } private Future generateWithProtobuf() { String bootstrapServer = bench.property("client.redpanda.addr", "localhost:9092"); String schemaRegistry = "http://" + bench.property("client.schema.registry.addr", "localhost:8081"); - generator = new ProtobufKafkaGenerator(bootstrapServer, schemaRegistry, tableName, columns, getCompression()); - return generator.produce(getRowPause(), getRowCount(), getRunDuration()); + var gen = new ProtobufKafkaGenerator(bootstrapServer, schemaRegistry, tableName, columns, getCompression()); + generators.add(gen); + return gen.produce(getRowPause(), getRowCount(), getRunDuration()); + } + + private Future generateWithLocalParquet(String parquetPath, String parquetPart, long startRow, + long rowCount) { + var parquetFile = Filer.createFile(parquetPath, parquetPart).toString(); + var gen = new LocalParquetGenerator(parquetFile, tableName, columns.copy(), startRow); + generators.add(gen); + return gen.produce(getRowPause(), rowCount, getRunDuration()); } private int getRowPause() { @@ -342,11 +411,18 @@ with open(path) as f: usedExisting = False matching_gen_parquet = findMatchingGenParquet(table_gen_def_text) - if matching_gen_parquet is not None and os.path.exists(str(matching_gen_parquet) + '.gen.parquet'): - os.link(str(matching_gen_parquet) + '.gen.parquet', table_parquet) + if matching_gen_parquet and os.path.exists(f"{matching_gen_parquet}.gen.parquet"): + bench_api_link(str(matching_gen_parquet) + '.gen.parquet', table_parquet) + usedExisting = True + if matching_gen_parquet and os.path.exists(f"{matching_gen_parquet}.gen.dataset"): + bench_api_link(str(matching_gen_parquet) + '.gen.dataset', table_parquet) usedExisting = True - used_existing_parquet_${table.name} = new_table([string_col("UsedExistingParquet", [str(usedExisting)])]) + used_existing_parquet_${table.name} = new_table([ + string_col("UsedExistingParquet", [str(usedExisting)]), + string_col("TableGenParquet", [table_gen_parquet]), + string_col("DhHostOsDir", [os.getenv("DEEPHAVEN_HOST_OS_DIR","")]) + ]) """; static final String kafkaToParquetQuery = """ @@ -379,7 +455,7 @@ with open(table_gen_def_file, 'w') as f: column_grouping=${column.grouping} if column_grouping: ${table.name} = ${table.name}.sort([${column.grouping}]) write(${table.name}, table_gen_parquet ${compression.codec} ${max.dict.keys} ${max.dict.bytes} ${target.page.bytes}) - os.link(table_gen_parquet, table_parquet) + bench_api_link(table_gen_parquet, table_parquet) del ${table.name} @@ -387,4 +463,20 @@ with open(table_gen_def_file, 'w') as f: garbage_collect() """; + static final String localToParquetQuery = """ + # Link an already created parquet dataset directory + import jpy, os + + if os.path.exists(table_parquet): + os.remove(table_parquet) + + with open(table_gen_def_file, 'w') as f: + f.write(table_gen_def_text) + + bench_api_link(table_gen_parquet.replace(".parquet", ".dataset"), table_parquet) + + from deephaven import garbage_collect + garbage_collect() + """; + } diff --git a/src/main/java/io/deephaven/benchmark/api/QueryLog.java b/src/main/java/io/deephaven/benchmark/api/QueryLog.java index b14b615e..bf307893 100644 --- a/src/main/java/io/deephaven/benchmark/api/QueryLog.java +++ b/src/main/java/io/deephaven/benchmark/api/QueryLog.java @@ -1,4 +1,4 @@ -/* Copyright (c) 2022-2023 Deephaven Data Labs and Patent Pending */ +/* Copyright (c) 2022-2026 Deephaven Data Labs and Patent Pending */ package io.deephaven.benchmark.api; import static java.nio.file.StandardOpenOption.*; @@ -30,7 +30,7 @@ class QueryLog implements Closeable { QueryLog(Path parent, Class testClass) { this.testClass = testClass; this.parent = parent; - this.logFile = getLogFile(parent, testClass); + this.logFile = parent.resolve("query.md"); } /** @@ -51,9 +51,9 @@ public void close() { write("## " + label + " - " + name, 2); for (int i = 0, n = queries.size(); i < n; i++) { write("### Query " + (i + 1), 1); - write("````", 1); + write("```", 1); write(queries.get(i), 0); - write("````", 2); + write("```", 2); } } @@ -93,14 +93,4 @@ private void write(String text, int newLineCount) { } } - static Path getLogFile(Path parent, Class testClass) { - Path logFile = parent.resolve("test-logs/" + testClass.getName() + ".query.md"); - try { - Files.createDirectories(logFile.getParent()); - return logFile; - } catch (Exception ex) { - throw new RuntimeException("Failed to create query log directory" + logFile.getParent(), ex); - } - } - } diff --git a/src/main/java/io/deephaven/benchmark/api/Snippets.java b/src/main/java/io/deephaven/benchmark/api/Snippets.java index 07ad6f5e..0008f0d3 100644 --- a/src/main/java/io/deephaven/benchmark/api/Snippets.java +++ b/src/main/java/io/deephaven/benchmark/api/Snippets.java @@ -1,4 +1,4 @@ -/* Copyright (c) 2022-2024 Deephaven Data Labs and Patent Pending */ +/* Copyright (c) 2022-2026 Deephaven Data Labs and Patent Pending */ package io.deephaven.benchmark.api; /** @@ -75,14 +75,26 @@ with exclusive_lock(table): /** * Initialize the container for storing benchmark metrics. Define functions for getting some MX Bean data for gc, - * jit and heap + * jit and heap. *

* ex. bench_api_metrics_init() */ static String bench_api_metrics_init = """ + import jpy + from deephaven import new_table + from deephaven.column import string_col, long_col, double_col + System = jpy.get_type('java.lang.System') def bench_api_metrics_init(): - global bench_api_metrics + global bench_api_metrics, standard_events bench_api_metrics = [] + standard_events = new_table([ + string_col("origin", []), + string_col("type", []), + long_col("start_ns", []), + long_col("duration_ns", []), + string_col("name", []), + double_col("value", []), + ]) """; /** @@ -175,7 +187,8 @@ def bench_api_metrics_add(category, name, value, note=''): * ex. bench_api_metrics_table = bench_api_metrics_collect() */ static String bench_api_metrics_collect = """ - from deephaven import input_table, empty_table, dtypes as dht + from deephaven import input_table, empty_table, new_table, dtypes as dht + from deephaven.column import string_col, long_col, double_col def bench_api_metrics_collect(): s = dht.string t = input_table({'timestamp':s,'origin':s,'category':s,'name':s,'value':s,'note':s}) @@ -186,6 +199,47 @@ def bench_api_metrics_collect(): return t """; + /** + * Make a file containing a one line reference to another file. Note: This is to get around the fact that + * Deephaven's parquet can't read from symbolic links that are directories. + *

+ * ex. bench_api_link('my_parquet_dir_or_file', 'my_link_name') + * + * @param target the table to link + * @param link_name the name to link the table to for retrieval + */ + static String bench_api_link = """ + import os, glob + def bench_api_link(target, link_name): + for f in glob.glob(link_name + '*'): + os.remove(f) + if target.endswith('.dataset'): + with open(link_name + '.link', 'w') as f: + f.write(target) + else: + os.link(target, link_name) + """; + + /** + * Read a parquet file or dataset into a Deephaven table. If the filename is a link (e.g. ".link") grab the file + * reference within it. + *

+ * ex. source = bench_api_read('/data/timed.parquet') + * + * @param file_name the name of the file containing the link reference + * @return a table containing the contents of the linked parquet file or dataset + */ + static String bench_api_read = """ + import os + from deephaven.parquet import read + def bench_api_read(file_name): + link_path = file_name + '.link' + if os.path.exists(link_path): + with open(link_path, 'r') as f: + file_name = f.read().strip() + return read(file_name) + """; + /** * Returns a query containing the api functions called by the query * @@ -206,6 +260,8 @@ static String getFunctions(String query) { defs += getFunc("bench_api_metrics_add", bench_api_metrics_add, query, defs); defs += getFunc("bench_api_metrics_collect", bench_api_metrics_collect, query, defs); defs += getFunc("bench_api_await_column_value_limit", bench_api_await_column_value_limit, query, defs); + defs += getFunc("bench_api_link", bench_api_link, query, defs); + defs += getFunc("bench_api_read", bench_api_read, query, defs); return defs; } diff --git a/src/main/java/io/deephaven/benchmark/connect/BarrageConnector.java b/src/main/java/io/deephaven/benchmark/connect/BarrageConnector.java index 632811ca..b7910511 100644 --- a/src/main/java/io/deephaven/benchmark/connect/BarrageConnector.java +++ b/src/main/java/io/deephaven/benchmark/connect/BarrageConnector.java @@ -1,4 +1,4 @@ -/* Copyright (c) 2022-2025 Deephaven Data Labs and Patent Pending */ +/* Copyright (c) 2022-2026 Deephaven Data Labs and Patent Pending */ package io.deephaven.benchmark.connect; import java.util.*; @@ -35,7 +35,8 @@ class BarrageConnector implements Connector { static { System.setProperty("thread.initialization", ""); // Remove server side initializers (e.g. DebuggingInitializer) } - static final int maxFetchCount = 1000; + static final int maxFetchCount = 100000; + static final int inboundMessageMB = 64; final private BarrageSession session; final private ConsoleSession console; final private ManagedChannel channel; @@ -181,12 +182,8 @@ public void close() { if (isClosed.get()) return; isClosed.set(true); - subscriptions.keySet().forEach(t -> { - closeSubscription(t); - }); - snapshots.keySet().forEach(t -> { - closeSubscription(t); - }); + new ArrayList<>(subscriptions.keySet()).forEach(this::closeSubscription); + new ArrayList<>(snapshots.keySet()).forEach(this::closeSnapshot); variableNames.clear(); try { @@ -226,6 +223,16 @@ private void closeSubscription(String tableName) { } } + private void closeSnapshot(String tableName) { + try { + var snapshot = snapshots.remove(tableName); + if (snapshot != null) + snapshot.handle.close(); + } catch (Exception ex) { + Log.info("Failed to close handle for snapshot: %s", tableName); + } + } + private FieldInfo findTable(String table) { Optional found = changes.changes().created().stream().filter(f -> f.name().equals(table)).findFirst(); @@ -243,6 +250,9 @@ private ManagedChannel getManagedChannel(String host, int port) { final ManagedChannelBuilder channelBuilder = ManagedChannelBuilder.forAddress(host, port); channelBuilder.usePlaintext(); // channelBuilder.useTransportSecurity(); If eventually security is needed + // Increase the maximum inbound message size so large Barrage snapshots (e.g. standard_events) + // do not trip the default 4 MiB gRPC limit while prototyping benchmarks. + channelBuilder.maxInboundMessageSize(inboundMessageMB * 1024 * 1024); // 32 MiB return channelBuilder.build(); } diff --git a/src/main/java/io/deephaven/benchmark/controller/DeephavenDockerController.java b/src/main/java/io/deephaven/benchmark/controller/DeephavenDockerController.java index 7df8ee76..15956804 100644 --- a/src/main/java/io/deephaven/benchmark/controller/DeephavenDockerController.java +++ b/src/main/java/io/deephaven/benchmark/controller/DeephavenDockerController.java @@ -1,4 +1,4 @@ -/* Copyright (c) 2022-2024 Deephaven Data Labs and Patent Pending */ +/* Copyright (c) 2022-2026 Deephaven Data Labs and Patent Pending */ package io.deephaven.benchmark.controller; import java.net.HttpURLConnection; @@ -111,7 +111,7 @@ public String getLog() { void waitForEngineReady() { long beginTime = System.currentTimeMillis(); - while (System.currentTimeMillis() - beginTime < 20000) { + while (System.currentTimeMillis() - beginTime < 30000) { if (getUrlStatus("http://" + httpHostPort + "/ide/")) return; Threads.sleep(100); diff --git a/src/main/java/io/deephaven/benchmark/generator/AvroKafkaGenerator.java b/src/main/java/io/deephaven/benchmark/generator/AvroKafkaGenerator.java index db662a88..42bfbe08 100644 --- a/src/main/java/io/deephaven/benchmark/generator/AvroKafkaGenerator.java +++ b/src/main/java/io/deephaven/benchmark/generator/AvroKafkaGenerator.java @@ -1,4 +1,4 @@ -/* Copyright (c) 2022-2023 Deephaven Data Labs and Patent Pending */ +/* Copyright (c) 2022-2026 Deephaven Data Labs and Patent Pending */ package io.deephaven.benchmark.generator; import static org.apache.kafka.clients.producer.ProducerConfig.*; @@ -128,11 +128,13 @@ private Producer createProducer(String bootstrapServer, S props.put(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); props.put(VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class); props.put("schema.registry.url", schemaRegistryUrl); - props.put(ACKS_CONFIG, "0"); + props.put(ENABLE_IDEMPOTENCE_CONFIG, "true"); + props.put(MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "1"); + props.put(ACKS_CONFIG, "all"); props.put(COMPRESSION_TYPE_CONFIG, getCompression(compression)); - props.put(BATCH_SIZE_CONFIG, 16384); - props.put(BUFFER_MEMORY_CONFIG, 16384 * 4); - props.put(LINGER_MS_CONFIG, 50); + props.put(BATCH_SIZE_CONFIG, "512000"); + props.put(BUFFER_MEMORY_CONFIG, "67108864"); + props.put(LINGER_MS_CONFIG, 10); return new KafkaProducer<>(props); } diff --git a/src/main/java/io/deephaven/benchmark/generator/ColumnDefs.java b/src/main/java/io/deephaven/benchmark/generator/ColumnDefs.java index a2211eb4..8d57292f 100644 --- a/src/main/java/io/deephaven/benchmark/generator/ColumnDefs.java +++ b/src/main/java/io/deephaven/benchmark/generator/ColumnDefs.java @@ -1,4 +1,4 @@ -/* Copyright (c) 2022-2023 Deephaven Data Labs and Patent Pending */ +/* Copyright (c) 2022-2026 Deephaven Data Labs and Patent Pending */ package io.deephaven.benchmark.generator; import java.util.*; @@ -12,16 +12,16 @@ * Note: All possible data values are loaded up front to prevent object-creation during production. This can take a * considerable amount of memory for larger scales, especially for generated strings. */ -public class ColumnDefs { +final public class ColumnDefs { final int valueCacheSize; final List columns = new ArrayList<>(); private String defaultDistribution = "random"; /** - * Initialize the instance with a default cache size of 1024 + * Initialize the instance with a default cache size large enough to cover typical column value ranges */ public ColumnDefs() { - this(1024); + this(2_000_000); } ColumnDefs(int valueCacheSize) { @@ -105,6 +105,21 @@ public ColumnDefs add(String name, String type, String valueDef) { return add(name, type, valueDef, null); } + /** + * Create an independent copy of this column definitions instance. Each copy has its own Maker objects, value + * caches, and distribution functions, making it safe to use from a separate thread without contention. + * + * @return a new independent ColumnDefs with the same column definitions + */ + public ColumnDefs copy() { + var c = new ColumnDefs(valueCacheSize); + c.defaultDistribution = defaultDistribution; + for (ColumnDef col : columns) { + c.add(col.name(), col.type(), col.valueDef(), col.maker().distributionName); + } + return c; + } + /** * Get the next value for the column in the given index according to the columns defined distribution. * diff --git a/src/main/java/io/deephaven/benchmark/generator/LocalParquetGenerator.java b/src/main/java/io/deephaven/benchmark/generator/LocalParquetGenerator.java new file mode 100644 index 00000000..692e6fd8 --- /dev/null +++ b/src/main/java/io/deephaven/benchmark/generator/LocalParquetGenerator.java @@ -0,0 +1,162 @@ +/* Copyright (c) 2026-2026 Deephaven Data Labs and Patent Pending */ +package io.deephaven.benchmark.generator; + +import java.io.File; +import java.io.IOException; +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicBoolean; +import blue.strategic.parquet.*; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.MessageTypeParser; +import io.deephaven.benchmark.metric.Metrics; +import io.deephaven.benchmark.util.Log; +import io.deephaven.benchmark.util.Threads; + +/** + * Generator that produces rows to a local Parquet file according to the provided column definitions. Note: This + * generator MUST generate the same row and column data in the same order and types as the non-local + * AvroKafkaGenerator when the two generators have the same column definitions. (The "same data" is defined + * by how it looks in Deephaven tables, not byte-for-byte in the files.) + */ +public class LocalParquetGenerator implements Generator { + final private ExecutorService queue = Threads.single("LocalParquetGenerator"); + final private ColumnDefs columnDefs; + final private String topic; + final private long startSeed; + final private MessageType schema; + final private File parquetFile; + final private AtomicBoolean isClosed = new AtomicBoolean(false); + private ParquetWriter writer; + + /** + * Create a local Parquet generator with the provided column definitions and output file. The column definitions + * determine the schema of the Parquet file and the data generated for each column. + * + * @param parquetFile output Parquet file path + * @param topic topic name (used for logging and schema generation) + * @param columnDefs column definitions that determine the schema and generated data + * @param startSeed starting seed for data generation + */ + public LocalParquetGenerator(String parquetFile, String topic, ColumnDefs columnDefs, long startSeed) { + this.topic = topic; + this.columnDefs = columnDefs; + this.startSeed = startSeed; + this.parquetFile = new File(parquetFile); + this.schema = MessageTypeParser.parseMessageType(getSchemaMessage(topic, columnDefs)); + try { + this.writer = ParquetWriter.writeFile(schema, this.parquetFile, createDehydrator()); + } catch (IOException ex) { + throw new RuntimeException("Failed to create Parquet writer for topic: " + topic, ex); + } + } + + /** + * Produce a maximum number of records asynchronously. + * + * @param perRecordPauseMillis wait time between each record sent + * @param maxRecordCount maximum records to produce + * @param maxDurationSecs maximum duration to produce + */ + public Future produce(int perRecordPauseMillis, long maxRecordCount, int maxDurationSecs) { + checkClosed(); + return queue.submit(() -> { + final long maxDuration = maxDurationSecs * 1000L; + final long beginTime = System.currentTimeMillis(); + final int colCount = columnDefs.getCount(); + + long recCount = startSeed; + long totalWritten = 0; + long duration = 0; + Object[] row = new Object[colCount]; + + while (!isClosed.get() && recCount < maxRecordCount) { + for (int i = 0; i < colCount; i++) { + row[i] = columnDefs.nextValue(i, recCount, maxRecordCount); + } + writer.write(row); + recCount++; + + if (++totalWritten % 10_000_000 == 0) + Log.info("Produced %s records to topic: %s", totalWritten, topic); + + duration = System.currentTimeMillis() - beginTime; + if (duration > maxDuration) + break; + } + + Log.info("Produced %s records to topic: %s", totalWritten, topic); + duration = System.currentTimeMillis() - beginTime; + return new Metrics("test-runner", "generate." + topic) + .set("duration.secs", duration / 1000.0) + .set("record.count", totalWritten) + .set("send.rate", totalWritten / (duration / 1000.0)); + }); + } + + /** + * Close the producer and shutdown any async threads created during production + */ + public void close() { + if (isClosed.get()) + return; + isClosed.set(true); + queue.shutdown(); + try { + writer.close(); + } catch (Exception ex) { + throw new RuntimeException("Failed to close Parquet writer for topic: " + topic, ex); + } + } + + private void checkClosed() { + if (isClosed.get()) + throw new RuntimeException("Generator is closed"); + } + + private Dehydrator createDehydrator() { + final String[] colNames = columnDefs.toTypeMap().keySet().toArray(new String[0]); + return (row, valueWriter) -> { + for (int i = 0; i < colNames.length; i++) { + valueWriter.write(colNames[i], row[i]); + } + }; + } + + private String getSchemaMessage(String topic, ColumnDefs fieldDefs) { + var schema = """ + message ${topic} { + ${fields} + } + """; + var fields = ""; + for (Map.Entry e : fieldDefs.toTypeMap().entrySet()) { + var name = e.getKey(); + var type = e.getValue(); + fields += String.format("required %s %s %s;\n", getFieldType(type), name, getCharEncoding(type)); + } + schema = schema.replace("${topic}", topic); + return schema.replace("${fields}", fields); + } + + private String getFieldType(String type) { + return switch (type) { + case "long" -> "int64"; + case "int" -> "int32"; + case "double" -> "double"; + case "float" -> "float"; + case "string" -> "binary"; + case "timestamp-millis" -> "int64"; + default -> throw new RuntimeException("Unsupported generator data type: " + type); + }; + } + + private String getCharEncoding(String type) { + return switch (type) { + case "string" -> "(UTF8)"; + case "timestamp-millis" -> "(TIMESTAMP(MILLIS,true))"; + default -> ""; + }; + } + +} diff --git a/src/main/java/io/deephaven/benchmark/util/Filer.java b/src/main/java/io/deephaven/benchmark/util/Filer.java index f14a85a9..30010640 100644 --- a/src/main/java/io/deephaven/benchmark/util/Filer.java +++ b/src/main/java/io/deephaven/benchmark/util/Filer.java @@ -1,4 +1,4 @@ -/* Copyright (c) 2022-2023 Deephaven Data Labs and Patent Pending */ +/* Copyright (c) 2022-2026 Deephaven Data Labs and Patent Pending */ package io.deephaven.benchmark.util; import static java.nio.file.StandardOpenOption.*; @@ -6,6 +6,8 @@ import java.net.URL; import java.nio.file.Files; import java.nio.file.Path; +import java.nio.file.Paths; +import java.nio.file.attribute.PosixFilePermissions; import java.util.Comparator; import java.util.stream.Collectors; @@ -34,6 +36,53 @@ static public void delete(Path path) { } } + /** + * Create a file with the given name in the given parent directory. Create the parent directory if it does not + * exist. Permissions are 755 for directories and 644 for files. + * + * @param parentDir the parent directory to contain the file + * @param fileName the name of the file to create + * @return the path of the created file + */ + static public Path createFile(String parentDir, String fileName) { + try { + return Files.createFile(createDirectory(parentDir).resolve(fileName), + PosixFilePermissions.asFileAttribute(PosixFilePermissions.fromString("rw-r--r--"))); + } catch (Exception ex) { + throw new RuntimeException("Failed to create file: " + fileName, ex); + } + } + + /** + * Create a directory with the given name. Create parent directories if they do not exist. Permissions are 755. + * + * @param dir the directory to create + * @return the path of the created directories + */ + static public Path createDirectory(String dir) { + try { + return Files.createDirectories(Paths.get(dir), PosixFilePermissions.asFileAttribute( + PosixFilePermissions.fromString("rwxr-xr-x"))); + } catch (Exception ex) { + throw new RuntimeException("Failed to create temp directory: " + dir, ex); + } + } + + /** + * Get the size of a file or directory in bytes. Directory sizes are calculated recursively by summing the sizes of + * all regular files contained within. + * + * @param file the file or directory to get the size of + * @return the size of the file or directory in bytes + */ + static public long getByteSize(String path) { + try { + return Files.walk(Paths.get(path)).filter(Files::isRegularFile).mapToLong(f -> f.toFile().length()).sum(); + } catch (Exception ex) { + throw new RuntimeException("Failed to get size of file: " + path, ex); + } + } + /** * Read the text of a file while preserving newlines and getting rid of carriage returns * diff --git a/src/main/resources/io/deephaven/benchmark/run/profile/queries/dashboards/benchmark_functions.dh.py b/src/main/resources/io/deephaven/benchmark/run/profile/queries/dashboards/benchmark_functions.dh.py index be362158..c62173fd 100644 --- a/src/main/resources/io/deephaven/benchmark/run/profile/queries/dashboards/benchmark_functions.dh.py +++ b/src/main/resources/io/deephaven/benchmark/run/profile/queries/dashboards/benchmark_functions.dh.py @@ -1,16 +1,14 @@ -# Copyright (c) 2022-2024 Deephaven Data Labs and Patent Pending +# Copyright (c) 2022-2026 Deephaven Data Labs and Patent Pending # # Deephaven python functions to support Benchmark Dashboards. These functions produce basic tables, # format strings, and do calculations. The data for creating tables is downloaded and cached from # either the Deephaven Benchmark GCloud bucket or from NFS on one of Deephaven's demos servers. # -# Requirements: Deephaven 0.36.1 or greater +# Requirements: Deephaven 41.3 or greater -import os, re, glob, jpy -import deephaven.dtypes as dht +import os, re, jpy from deephaven import read_csv, merge, agg, empty_table, input_table, dtypes as dht from urllib.request import urlopen, urlretrieve -from numpy import typing as npt # Convert the given name to a name suitable for a DH column name def normalize_name(name): @@ -123,9 +121,12 @@ def convert_result(table): # Do any conversions of type or column name needed from benchmark-metrics.csv def convert_metric(table): - return table.view(['benchmark_name','origin','timestamp=(long)timestamp','name', - 'value=(double)value','note']) + return table.view(['benchmark_name','origin','timestamp=(long)timestamp','name','value=(double)value','note']) +# Do any conversions of type or column name needed from benchmark-events.csv +def convert_event(table): + return table.view(['benchmark_name','origin','type','start','duration','name','value=(double)value']) + # Do any conversions of type or column name needed from benchmark-platform.csv def convert_platform(table): return table.view(['origin','name','value']) @@ -171,37 +172,92 @@ def load_bench_results(storage_uri, category='adhoc', actor_filter=None, set_fil def load_bench_metrics(storage_uri, category='adhoc', actor_filter=None, set_filter=None): run_ids = get_run_paths(storage_uri, category, actor_filter, set_filter, 100) return merge_run_tables(storage_uri, run_ids, category, 'benchmark-metrics.csv', convert_metric) + +# Load all benchmark-events.csv data collected from the given storage, category, and filters +def load_bench_events(storage_uri, category='adhoc', actor_filter=None, set_filter=None): + run_ids = get_run_paths(storage_uri, category, actor_filter, set_filter, 100) + return merge_run_tables(storage_uri, run_ids, category, 'benchmark-events.csv', convert_event) # Load all benchmark-platform.csv data collected from the given storage, category, and filters def load_bench_platform(storage_uri, category='adhoc', actor_filter=None, set_filter=None): run_ids = get_run_paths(storage_uri, category, actor_filter, set_filter, 100) return merge_run_tables(storage_uri, run_ids, category, 'benchmark-platform.csv', convert_platform) -# Load all benchmark-results.csv data collected from the given storage, category, and filters by set -# Sets contain one or more runs for each benchmark. This function loads the median run by rate for each benchmark -def load_bench_result_sets(storage_uri, category='adhoc', actor_filter=None, set_filter=None): - bench_results = load_bench_results(storage_uri,category,actor_filter,set_filter) +# Compute result sets from raw bench results +def compute_result_sets(bench_results): bench_results_sets = bench_results.sort(['benchmark_name','origin','set_id','op_rate']) \ .group_by(['benchmark_name','origin','set_id']) \ .view(['benchmark_name','origin','timestamp=(long)mid_item(timestamp)','test_duration=(double)mid_item(test_duration)', 'set_op_rates=op_rate','op_duration=(double)mid_item(op_duration)','op_rate=(long)mid_item(op_rate)', 'row_count=(long)mid_item(row_count)','variability=(float)rstd(set_op_rates)','set_id', 'run_id=(String)mid_item(run_id)','set_count=count(set_op_rates)']) - # Attach columns for specified metrics and platform properties - #local_platform_props = [] - #local_metric_props = [] - #bench_results_sets = add_platform_values(bench_results_sets, ['deephaven.version'] + platform_props, local_platform_props) - #bench_results_sets = add_metric_values(bench_results_sets, metric_props, local_metric_props) return bench_results_sets, bench_results + +# Load all benchmark-results.csv data collected from the given storage, category, and filters by set +# Sets contain one or more runs for each benchmark. This function loads the median run by rate for each benchmark +def load_bench_result_sets(storage_uri, category='adhoc', actor_filter=None, set_filter=None): + bench_results = load_bench_results(storage_uri,category,actor_filter,set_filter) + return compute_result_sets(bench_results) +# Check for a pulled-up consolidated file (produced by pullup.py) +def find_pullup_file(storage_uri, category, actor, table_name): + csv_name = f'benchmark-{table_name}' + # Always check local cache first + local_root = re.sub('^file:/+', '/', storage_uri) + if local_root.startswith('http'): + local_root = re.sub('^http.*/deephaven-benchmark', '/data/deephaven-benchmark', local_root) + pullup_dir = os.path.join(local_root, category, actor) + for ext in ['.parquet', '.csv.gz']: + path = os.path.join(pullup_dir, csv_name + ext) + if os.path.exists(path): + return path + # Try to download from remote if storage_uri is HTTP + if storage_uri.startswith('http'): + pullup_base = f'{storage_uri}/{category}/{actor}/{csv_name}' + for ext in ['.parquet', '.csv.gz']: + uri = pullup_base + ext + try: + out_path = re.sub('^http.*/deephaven-benchmark/', '/data/deephaven-benchmark/', uri) + os.makedirs(os.path.dirname(out_path), mode=0o777, exist_ok=True) + if not os.path.exists(out_path): + urlretrieve(uri, out_path) + print('Cache', uri) + return out_path + except Exception: + continue + return None + +# Read a pulled-up file (parquet or csv.gz) +def read_pullup_file(path): + if path.endswith('.parquet'): + from deephaven.parquet import read as pq_read + return pq_read(path) + return read_csv(path) + def load_table_or_empty(table_name, storage_uri, category='adhoc', actor_filter='', set_filter=''): actor = actor_filter.strip(); prefix = set_filter.strip() if actor and prefix: + pullup_path = find_pullup_file(storage_uri, category, actor, table_name) + if pullup_path: + print(f'Pullup {pullup_path}') + tbl = read_pullup_file(pullup_path) + if table_name == 'result_sets': + results_path = find_pullup_file(storage_uri, category, actor, 'results') + results_tbl = read_pullup_file(results_path) if results_path else tbl + return tbl, results_tbl + return tbl + if table_name == 'result_sets': + results_path = find_pullup_file(storage_uri, category, actor, 'results') + if results_path: + print(f'Pullup {results_path} (computing result_sets)') + return compute_result_sets(read_pullup_file(results_path)) return globals()[f'load_bench_{table_name}'](storage_uri, category, actor, prefix) return globals()[f'empty_bench_{table_name}']() # Add columns for the specified platform properties -def add_platform_values(table, pnames=[], cnames = []): +def add_platform_values(table, pnames=None, cnames=None): + pnames = pnames if pnames is not None else [] + cnames = cnames if cnames is not None else [] pnames = list(dict.fromkeys(pnames)) for pname in pnames: new_pname = normalize_name(pname) @@ -213,14 +269,16 @@ def add_platform_values(table, pnames=[], cnames = []): return table # Add columns for the specified metric properties -def add_metric_values(table, pnames=[], cnames=[]): +def add_metric_values(table, pnames=None, cnames=None): + pnames = pnames if pnames is not None else [] + cnames = cnames if cnames is not None else [] pnames = list(dict.fromkeys(pnames)) for pname in pnames: new_pname = normalize_name(pname) cnames.append(new_pname) - single_metrtics = bench_metrics.where(['name=pname']).first_by(['benchmark_name','set_id','run_id','origin']) + single_metrics = bench_metrics.where(['name=pname']).first_by(['benchmark_name','set_id','run_id','origin']) table = table.natural_join( - single_metrtics, on=['benchmark_name','set_id','run_id','origin'], joins=[new_pname+'=value'] + single_metrics, on=['benchmark_name','set_id','run_id','origin'], joins=[new_pname+'=value'] ) return table @@ -239,12 +297,14 @@ def format_columns(table,pct_cols=(),int_cols=()): # Get a percentage standard deviation for the given list of rates def rstd(rates) -> float: rates = [i for i in rates if i >= 0] + if not rates: return 0.0 mean = statistics.mean(rates) return (statistics.pstdev(rates) / mean) if mean != 0 else 0.0 # Get the zscore of one rate against a list of rates def zscore(rate, rates) -> float: rates = [i for i in rates if i >= 0] + if not rates: return 0.0 std = statistics.pstdev(rates) return ((rate - statistics.mean(rates)) / std) if std != 0 else 0.0 @@ -260,11 +320,11 @@ def rchange(rates) -> float: rates = array('l', rates) if(len(rates) < 2): return 0.0 m = statistics.mean(rates[:-1]) - return (rates[-1] - m) / m + return ((rates[-1] - m) / m) if m != 0 else 0.0 # Get the percentage gain between two values def gain(start, end) -> float: - return (end - start) / start + return ((end - start) / start) if start != 0 else 0.0 # Format a list of rates to make them easier to read in a DHC table def format_rates(rates): diff --git a/src/main/resources/io/deephaven/benchmark/run/profile/queries/dashboards/pullup.py b/src/main/resources/io/deephaven/benchmark/run/profile/queries/dashboards/pullup.py new file mode 100644 index 00000000..30f4817c --- /dev/null +++ b/src/main/resources/io/deephaven/benchmark/run/profile/queries/dashboards/pullup.py @@ -0,0 +1,48 @@ +# Copyright (c) 2026-2026 Deephaven Data Labs and Patent Pending +# +# Takes the adhoc-style directory structure and generates csv with all sets and runs. The +# resulting files are 'benchmark-events.csv', 'benchmark-metrics.csv', 'benchmark-platform.csv', +# 'benchmark-results.csv', and 'benchmark-result_sets.csv'. These files look, for example, like +# the tables loaded from "load_table_or_empty('metrics', storage_uri, 'adhoc', actor, prefix)" +# function seen in other benchmark queries. The csv files are written to the "/data" directory. +# +# Requirements: Deephaven 41.7 or greater + +from urllib.request import urlopen; import os, gzip, shutil, re +from deephaven import csv as dhcsv +from deephaven.parquet import write as pq_write + +def glob_to_regex(pat): + return '^' + re.escape(pat).replace(r'\*', '.*').replace(r'\?', '.') + '$' + +actor = "Final-gc-report" +dest_actor = "gc-fun" +prefix = "gc_" +exclude_sets = ['*_p110_*'] +output_dir = f'/data/deephaven-benchmark/demo/{dest_actor}' + +root = 'file:///data' if os.path.exists('/data/deephaven-benchmark') else 'https://storage.googleapis.com' +with urlopen(f'{root}/deephaven-benchmark/benchmark_functions.dh.py') as r: + exec(r.read().decode(), globals(), locals()) + storage_uri = f'{root}/deephaven-benchmark' + +os.makedirs(output_dir, exist_ok=True) + +for name in ['metrics', 'events', 'platform', 'results', 'result_sets']: + data = load_table_or_empty(name, storage_uri, 'adhoc', actor, prefix) + table = data[0] if name == 'result_sets' else data + if exclude_sets: + table = table.where(' && '.join(f'!set_id.matches(`{glob_to_regex(pat)}`)' for pat in exclude_sets)) + if dest_actor != actor: + table = table.update_view([f'set_id = set_id.replaceFirst(`{actor}`, `{dest_actor}`)']) + output_csv = f'{output_dir}/benchmark-{name}.csv' + dhcsv.write(table, output_csv) + with open(output_csv, 'rb') as f_in, gzip.open(output_csv + '.gz', 'wb') as f_out: + shutil.copyfileobj(f_in, f_out) + os.remove(output_csv) + print(f'Wrote csv: {output_csv}.gz') + output_pq = f'{output_dir}/benchmark-{name}.parquet' + pq_write(table, output_pq, compression_codec_name='ZSTD') + print(f'Wrote parquet: {output_pq}') + +print("Done!") diff --git a/src/main/resources/io/deephaven/benchmark/run/profile/queries/dashboards/train_dashboard.dh.py b/src/main/resources/io/deephaven/benchmark/run/profile/queries/dashboards/train_dashboard.dh.py new file mode 100644 index 00000000..cb819e06 --- /dev/null +++ b/src/main/resources/io/deephaven/benchmark/run/profile/queries/dashboards/train_dashboard.dh.py @@ -0,0 +1,966 @@ +# Copyright (c) 2026-2026 Deephaven Data Labs and Patent Pending +# +# Deephaven Train dashboard for visualizing benchmark data captured with the training benchmarks. +# The dashboard shows the throughput, jitter, and cycle budget metrics for different sets of JVM options. +# +# Requirements: Deephaven 41.3 or greater +# +# ruff: noqa: F821 +from urllib.request import urlopen; import os +from deephaven import ui +from deephaven.ui import use_memo, use_state, use_ref +from deephaven.execution_context import get_exec_ctx +from deephaven import pandas as dhpd +import deephaven.plot.express as dx +import numpy as np + +prefix = "gc_" + +COLORS = [ + "#636EFA", "#EF553B", "#00CC96", "#AB63FA", + "#FFA15A", "#19D3F3", "#FF6692", "#B6E880", + "#FF97FF", "#FECB52", +] + +root = 'file:///nfs' if os.path.exists('/nfs/deephaven-benchmark') else 'https://storage.googleapis.com' +with urlopen(f'{root}/deephaven-benchmark/benchmark_functions2.dh.py') as r: + exec(r.read().decode(), globals(), locals()) + storage_uri = f'{root}/deephaven-benchmark' + +# These globals are set by the dashboard component on each render +actor = "gc-fun" +static_inc_filter = 'Inc' +benchmark_filter = "*" +cycle_budget_ms = 100 +autotune_pct = '100' +jvm_version = '25' +setid_filter = f'*_{cycle_budget_ms}_p{autotune_pct}_j{jvm_version}*' +bench_result_sets = None +bench_results = None +bench_metrics = None +bench_events = None +gc_metrics = None + +# Load raw tables once at module level (just table handles; data stays on server) +_raw_rs, _raw_br = load_table_or_empty('result_sets', storage_uri, 'demo', actor, prefix) +_raw_rs = _raw_rs.where("origin = `deephaven-engine`") +_raw_br = _raw_br.where("origin = `deephaven-engine`") +_raw_bm = load_table_or_empty('metrics', storage_uri, 'demo', actor, prefix) +_raw_be = load_table_or_empty('events', storage_uri, 'demo', actor, prefix) + +def filter_benchmarks(table): + if not benchmark_filter: return table + benchmark_regex = "|".join(p.strip().replace("*", ".*") for p in benchmark_filter.split(",")) + setid_regex = "|".join(p.strip().replace("*", ".*") for p in setid_filter.split(",")) + filters = [ + f"benchmark_name.matches(`{benchmark_regex}`)", + f"set_id.matches(`{setid_regex}`)", + ] + if static_inc_filter: + type_regex = "|".join(f".*{t.strip()}.*" for t in static_inc_filter.split(",")) + filters.append(f"benchmark_name.matches(`{type_regex}`)") + return table.where(filters) + +def get_metric_table(metric_name, col_type, metric_table=gc_metrics): + setids = metric_table.select_distinct(['set_id']) + benchmarks = metric_table.select_distinct(['benchmark_name']) + names = metric_table.select_distinct(['name']) + all_combos = benchmarks.join(setids).join(names) + base = all_combos.natural_join( + metric_table.view(['benchmark_name','set_id','name','value']), + ['benchmark_name','set_id','name'] + ) + tbl = base.group_by(['benchmark_name','name']) + for idx, (setid,) in enumerate(setids.iter_tuple()): + setid = setid.replace(actor + '/' + prefix,'') + tbl = tbl.update(f"{setid}=({col_type})value[{idx}]") + return tbl.where([f"name=`{metric_name}`"]).drop_columns(['name','set_id','value']) \ + .sort(['benchmark_name']) + +def get_event_time_table(event_type, metric_name, col_type, agg, value_col='value'): + gc_events = bench_result_sets.join(bench_events, ['benchmark_name','origin','set_id'], + ['type','start','duration','name','value']).sort(['set_id']) + tbl = gc_events.where([f"type=`{event_type}`"]).group_by(['benchmark_name','set_id','name']) \ + .update([f"value={agg}({value_col})/1_000_000_000"]) + return get_metric_table(metric_name, col_type, tbl) + +def get_throughput_table(): + setids = bench_result_sets.select_distinct(['set_id']) + benchmarks = bench_result_sets.select_distinct(['benchmark_name']) + all_combos = benchmarks.join(setids) + base = all_combos.natural_join( + bench_result_sets.view(['benchmark_name','set_id','op_rate','op_duration']), + ['benchmark_name','set_id'] + ) + tbl = base.group_by(['benchmark_name']) \ + .update(['max=max(op_rate)','min=min(op_rate)']) \ + .ungroup(['set_id','op_rate','op_duration']) \ + .update(['op_factor=op_rate/min']) \ + .group_by(['benchmark_name']) \ + .update(['max=max(op_factor)','min=min(op_factor)']) + for idx, (setid,) in enumerate(setids.iter_tuple()): + setid_clean = setid.replace(actor + '/' + prefix,'') + tbl = tbl.update(f"{setid_clean}=op_factor[{idx}]") \ + .format_columns(f"{setid_clean} = {setid_clean}=max ? DEEP_GREEN : {setid_clean}=min ? DEEP_RED : NO_FORMATTING") + return tbl.drop_columns(['set_id','op_rate','max','min','op_duration','op_factor']).sort(['benchmark_name']) + +def get_jitter_table(): + gc_events = bench_result_sets.join(bench_events, ['benchmark_name','origin','set_id'], + ['type','start','duration','name','value']).sort(['set_id']) + jitter = gc_events.where(["type=`server_state_log`", "name=`ugp.cycle.time`"]) \ + .group_by(['benchmark_name','set_id']) \ + .update(['jitter=sqrt(avg(value * value - avg(value) * avg(value)))/avg(value)']) + setids = bench_result_sets.select_distinct(['set_id']) + benchmarks = jitter.select_distinct(['benchmark_name']) + all_combos = benchmarks.join(setids) + base = all_combos.natural_join( + jitter.view(['benchmark_name','set_id','jitter']), + ['benchmark_name','set_id'] + ) + tbl = base.group_by(['benchmark_name']) \ + .update(['max=max(jitter)','min=min(jitter)']) .ungroup(['set_id','jitter']) \ + .update(['jitter_factor=jitter/min']).group_by(['benchmark_name']) \ + .update(['max=max(jitter_factor)','min=min(jitter_factor)']) + for idx, (setid,) in enumerate(setids.iter_tuple()): + setid = setid.replace(actor + '/' + prefix, '') + tbl = tbl.update(f"{setid}=(double)jitter_factor[{idx}]") \ + .format_columns(f"{setid} = {setid}=max ? DEEP_RED : {setid}=min ? DEEP_GREEN : NO_FORMATTING") + return tbl.drop_columns(['set_id','jitter','jitter_factor','max','min']).sort(['benchmark_name']) + +def get_cycle_onbudget_table(): + budget_nanos = cycle_budget_ms * 1_000_000 + gc_events = bench_result_sets.join(bench_events, ['benchmark_name','origin','set_id'], + ['type','start','duration','name','value']).sort(['set_id']) + cycle_events = gc_events.where(["type=`server_state_log`", "name=`ugp.cycle.time`"]) + total = cycle_events.count_by('total', ['benchmark_name','set_id']) + under = cycle_events.where(f"value <= {budget_nanos}L").count_by('under', ['benchmark_name','set_id']) + cycles = total.join(under, ['benchmark_name','set_id'], ['under']) \ + .update(['pct_under_budget = 100.0 * under / total']) + setids = bench_result_sets.select_distinct(['set_id']) + benchmarks = cycles.select_distinct(['benchmark_name']) + all_combos = benchmarks.join(setids) + base = all_combos.natural_join( + cycles.view(['benchmark_name','set_id','pct_under_budget']), + ['benchmark_name','set_id'] + ) + tbl = base.group_by(['benchmark_name']) \ + .update(['max=max(pct_under_budget)','min=min(pct_under_budget)']) \ + .ungroup(['set_id','pct_under_budget']) \ + .group_by(['benchmark_name']) \ + .update(['max=max(pct_under_budget)','min=min(pct_under_budget)']) + for idx, (setid,) in enumerate(setids.iter_tuple()): + setid = setid.replace(actor + '/' + prefix, '') + tbl = tbl.update(f"{setid}=(double)pct_under_budget[{idx}]") \ + .format_columns(f"{setid} = {setid}==max ? DEEP_GREEN : {setid}==min ? DEEP_RED : NO_FORMATTING") + return tbl.drop_columns(['set_id','pct_under_budget','max','min']).sort(['benchmark_name']) + +def get_throughput_hilow_table(): + return bench_results.view(['benchmark_name','set_id','op_rate']).group_by(['benchmark_name','set_id']) \ + .update(['E=1.96*(std(op_rate)/sqrt(count(op_rate)))', 'M=avg(op_rate)', 'L=max(0,M-E)', 'H=M+E', + 'set_id=set_id.replace(actor + "/" + prefix,"")']) + +def add_summary_rows(tbl, higher_is_better=True, use_geo_mean=True): + from deephaven import pandas as dhpd + import pandas as pd + + visible_cols = [col.name for col in tbl.columns if '__TABLE_STYLE_FORMAT' not in col.name] + df = dhpd.to_pandas(tbl.view(visible_cols)) + gc_cols = [c for c in visible_cols if c != 'benchmark_name'] + + row = {'benchmark_name': 'GEO MEAN' if use_geo_mean else 'MEAN'} + for gc in gc_cols: + vals = df[gc].dropna().values.astype(float) + if use_geo_mean: + vals = vals[vals > 0] + row[gc] = float(np.exp(np.log(vals).mean())) if len(vals) > 0 else float('nan') + else: + row[gc] = float(vals.mean()) if len(vals) > 0 else float('nan') + + full_df = pd.concat([df, pd.DataFrame([row])], ignore_index=True) + result = dhpd.to_table(full_df) + + max_expr = 'max(' + ','.join(gc_cols) + ')' + min_expr = 'min(' + ','.join(gc_cols) + ')' + result = result.update([f'max_val={max_expr}', f'min_val={min_expr}']) + for gc in gc_cols: + if higher_is_better: + result = result.format_columns(f"{gc} = {gc}==max_val ? DEEP_GREEN : {gc}==min_val ? DEEP_RED : NO_FORMATTING") + else: + result = result.format_columns(f"{gc} = {gc}==min_val ? DEEP_GREEN : {gc}==max_val ? DEEP_RED : NO_FORMATTING") + return result.drop_columns(['max_val', 'min_val']) + +def get_summary_table(): + from deephaven.column import string_col, double_col + from deephaven import new_table, pandas as dhpd + + tput_tbl = get_throughput_table() + + def visible_cols(tbl): + return [c.name for c in tbl.columns if c.name != 'benchmark_name' and '__TABLE_STYLE_FORMAT' not in c.name] + + def geo_mean(df, col): + vals = df[col].dropna().values.astype(float) + vals = vals[vals > 0] + return float(np.exp(np.log(vals).mean())) if len(vals) > 0 else float('nan') + + tput_cols = visible_cols(tput_tbl) + tput_df = dhpd.to_pandas(tput_tbl.view(['benchmark_name'] + tput_cols)) + + columns = [ + string_col("gc_type", tput_cols), + double_col("throughput_factor", [geo_mean(tput_df, gc) for gc in tput_cols]), + ] + + import math + try: + jit_tbl = get_jitter_table() + jit_cols = visible_cols(jit_tbl) + jit_df = dhpd.to_pandas(jit_tbl.view(['benchmark_name'] + jit_cols)) + jit_vals = [geo_mean(jit_df, gc) for gc in jit_cols] + if not all(math.isnan(v) for v in jit_vals): + columns.append(double_col("jitter_factor", jit_vals)) + except Exception: + pass + + try: + cc_tbl = get_cycle_onbudget_table() + cc_cols = visible_cols(cc_tbl) + cc_df = dhpd.to_pandas(cc_tbl.view(['benchmark_name'] + cc_cols)) + budget_col = f"pct_under_{cycle_budget_ms}ms" + cc_vals = [float(cc_df[gc].dropna().mean()) for gc in cc_cols] + if not all(math.isnan(v) for v in cc_vals): + columns.append(double_col(budget_col, cc_vals)) + except Exception: + pass + + return new_table(columns) + +_ctx = get_exec_ctx() + +def _compute_rank(t): + with _ctx: + return t.update_view(["Rank = ii + 1"]) + + +@ui.component +def gc_bench_summary_dashboard(): + # --- Filter state --- + static_inc_val, set_static_inc = use_state("Inc") + cycle_budget_str, set_cycle_budget = use_state("100") + autotune_val, set_autotune = use_state("100") + jvm_val, set_jvm = use_state("25") + + # Applied state — tables only recompute when Apply is clicked + applied, set_applied = use_state({ + "static_inc": "Inc", + "cycle_budget": "100", "autotune": "100", "jvm": "25", + }) + + def handle_apply(): + set_applied({ + "static_inc": static_inc_val, + "cycle_budget": cycle_budget_str, "autotune": autotune_val, "jvm": jvm_val, + }) + + # Derived from applied (committed) values + a_cycle = applied["cycle_budget"] + a_autotune = applied["autotune"] + a_jvm = applied["jvm"] + cycle_budget_ms_local = int(a_cycle) if a_cycle.isdigit() else 100 + setid_filter_local = f'*_{a_cycle}_p{a_autotune}_j{a_jvm}*' + + def compute_tables(): + global actor, static_inc_filter, benchmark_filter, cycle_budget_ms, autotune_pct, jvm_version + global setid_filter, bench_result_sets, bench_results, bench_metrics, bench_events, gc_metrics + static_inc_filter = applied["static_inc"] + benchmark_filter = "*" + cycle_budget_ms = cycle_budget_ms_local + autotune_pct = a_autotune + jvm_version = a_jvm + setid_filter = setid_filter_local + + rs = _raw_rs.sort(['set_id']) + br = _raw_br.sort(['set_id']) + bm = _raw_bm + be = _raw_be + + rs, br, bm, be = [filter_benchmarks(t) for t in [rs, br, bm, be]] + bench_result_sets = rs + bench_results = br + bench_metrics = bm + bench_events = be + gc_metrics = rs.join(bm, ['benchmark_name','origin','set_id'], ['name','value']).sort(['set_id']) + + from deephaven import new_table + from deephaven.column import string_col + + def _empty(label="No data"): + return new_table([string_col("benchmark_name", [label])]) + + # Strip filter substring and Static/Inc prefix from names + substr = f"_{a_cycle}_p{a_autotune}_j{a_jvm}" + inc_static = applied["static_inc"] + def _clean_name(name): + import re + name = name.replace(substr, '') + name = re.sub(rf'[\s_-]*{re.escape(inc_static)}[\s_-]*', '_', name) + name = name.strip('_') + return name + def _rename(tbl): + format_cols = [c.name for c in tbl.columns if '__TABLE_STYLE_FORMAT' in c.name] + if format_cols: + tbl = tbl.drop_columns(format_cols) + renames = [] + for c in tbl.columns: + new_name = _clean_name(c.name) + if new_name != c.name: + renames.append(f"{new_name}={c.name}") + if renames: + tbl = tbl.rename_columns(renames) + # Also clean gc_type and benchmark_name cell values + col_names = [c.name for c in tbl.columns] + import re + def _clean_expr(col): + return f'{col} = {col}.replace(`{substr}`, ``).replaceAll(`[\\\\s_-]*{re.escape(inc_static)}[\\\\s_-]*`, `_`).replaceAll(`^_|_$`, ``)' + if "gc_type" in col_names: + tbl = tbl.update_view([_clean_expr('gc_type')]) + if "benchmark_name" in col_names: + tbl = tbl.update_view([_clean_expr('benchmark_name')]) + return tbl + + try: + t_summary = _rename(get_summary_table()) + except Exception: + t_summary = _empty("No data") + + try: + t_throughput = add_summary_rows(_rename(get_throughput_table()), higher_is_better=True) + except Exception as e: + import traceback; traceback.print_exc() + t_throughput = _empty("No data") + + try: + t_jitter = add_summary_rows(_rename(get_jitter_table()), higher_is_better=False) + except Exception as e: + import traceback; traceback.print_exc() + t_jitter = _empty("No data") + + try: + t_on_budget = _rename(get_cycle_onbudget_table()) + t_on_budget = add_summary_rows(t_on_budget, higher_is_better=True, use_geo_mean=False) + cc_cols = [col.name for col in t_on_budget.columns if col.name != 'benchmark_name' and '__TABLE_STYLE_FORMAT' not in col.name] + t_on_budget = t_on_budget.format_columns([f"{c} = Decimal(`0.00'%'`)" for c in cc_cols]) + except Exception as e: + import traceback; traceback.print_exc() + t_on_budget = _empty("No data") + + return t_summary, t_throughput, t_jitter, t_on_budget + + gc_summary, gc_throughput, gc_jitter, gc_on_budget = use_memo( + compute_tables, [applied] + ) + + filter_label = f"{applied['static_inc']} | {a_cycle}ms | p{a_autotune} | j{a_jvm}" + + controls = ui.flex( + ui.picker( + "Static", "Inc", + label="Static/Inc", + selected_key=static_inc_val, + on_selection_change=set_static_inc, + ), + ui.picker( + "100", "1000", + label="Cycle Budget (ms)", + selected_key=cycle_budget_str, + on_selection_change=set_cycle_budget, + ), + ui.picker( + "80", "90", "100", "110", + label="Autotune %", + selected_key=autotune_val, + on_selection_change=set_autotune, + ), + ui.picker( + "17", "25", + label="JVM Version", + selected_key=jvm_val, + on_selection_change=set_jvm, + ), + ui.button("Apply", on_press=handle_apply, variant="accent"), + direction="row", + gap="size-200", + align_items="end", + flex_grow=0, + flex_shrink=0, + ) + + is_inc = applied["static_inc"] != "Static" + + top_row = ui.flex( + ui.flex( + ui.heading(f"GC Summary — {filter_label}", level=4), + ui.table(gc_summary), + direction="column", flex_grow=1, + ), + ui.flex( + ui.heading(f"GC Throughput — {filter_label}", level=4), + ui.table(gc_throughput), + direction="column", flex_grow=1, + ), + direction="row", flex_grow=1, gap="size-200", + ) + + content = [controls, top_row] + + if is_inc: + bottom_row = ui.flex( + ui.flex( + ui.heading(f"GC Jitter — {filter_label}", level=4), + ui.table(gc_jitter), + direction="column", flex_grow=1, + ), + ui.flex( + ui.heading(f"GC On Budget — {filter_label}", level=4), + ui.table(gc_on_budget), + direction="column", flex_grow=1, + ), + direction="row", flex_grow=1, gap="size-200", + ) + content.append(bottom_row) + + return ui.flex( + *content, + direction="column", + flex_grow=1, + gap="size-100", + UNSAFE_style={"height": "100%", "overflow": "auto"}, + ) + + +@ui.component +def gc_throughput_dashboard(): + # Applied state — only this triggers re-render (on form submit) + applied, set_applied = use_state({ + "static_inc": "Inc", + "cycle_budget": "100", "autotune": "100", "jvm": "25", + "benchmark": None, + }) + + def handle_submit(data): + set_applied({ + "static_inc": "Inc", + "cycle_budget": data.get("cycle_budget", "100"), + "autotune": data.get("autotune", "100"), + "jvm": data.get("jvm", "25"), + "benchmark": data.get("benchmark", None), + }) + + a_cycle = applied["cycle_budget"] + a_autotune = applied["autotune"] + a_jvm = applied["jvm"] + a_bench = applied["benchmark"] + setid_filter_local = f'*_{a_cycle}_p{a_autotune}_j{a_jvm}*' + + # Load available benchmarks based on applied filters + def compute_benchmarks(): + sid_filter = applied["static_inc"] + sid_regex = setid_filter_local.replace("*", ".*") + type_regex = "|".join(f".*{t.strip()}.*" for t in sid_filter.split(",")) if sid_filter else ".*" + rs = _raw_rs.where([ + f"set_id.matches(`{sid_regex}`)", + f"benchmark_name.matches(`{type_regex}`)", + ]) + try: + return sorted(dhpd.to_pandas( + rs.select_distinct(["benchmark_name"]) + )["benchmark_name"].tolist()) + except Exception: + return [] + + all_benchmarks = use_memo(compute_benchmarks, [applied]) + + # Use the applied benchmark; auto-select first if it's invalid + effective_bench = a_bench + if all_benchmarks and (not effective_bench or effective_bench not in all_benchmarks): + effective_bench = all_benchmarks[0] + + # Compute CDF/CCDF and time series from applied state + def compute_distribution(): + if not effective_bench: + return None, None, None, None + try: + from deephaven import agg + setid_regex = setid_filter_local.replace("*", ".*") + _replace_expr = f'set_id.replace("{actor}" + "/" + "{prefix}","")' + + # --- CDF/CCDF --- + events = _raw_be \ + .where(['name = `ugp.cycle.time`']) \ + .view(['benchmark_name', 'start', 'cycleCost=value', + f'gc_type=set_id.replace("{actor}" + "/" + "{prefix}","")']) + events = events.where([ + f'benchmark_name = `{effective_bench}`', + f'gc_type.matches(`{setid_regex}`)', + ]) + + n = events.count_by("N", by=["benchmark_name", "gc_type"]) + + cdf_t = ( + events + .sort(["benchmark_name", "gc_type", "cycleCost"]) + .partition_by(["benchmark_name", "gc_type"]).transform(_compute_rank).merge() + .natural_join(n, on=["benchmark_name", "gc_type"]) + .update_view(["Percentile = 100.0 * Rank / N", "cycleCostSec = cycleCost / 1_000_000_000.0"]) + ) + ccdf_t = ( + events + .sort(["benchmark_name", "gc_type", "cycleCost"]) + .partition_by(["benchmark_name", "gc_type"]).transform(_compute_rank).merge() + .natural_join(n, on=["benchmark_name", "gc_type"]) + .update_view(["Percentile = 100.0 * (1.0 - (double)Rank / N)", "cycleCostSec = cycleCost / 1_000_000_000.0"]) + ) + + # --- Throughput time series --- + median_runs = _raw_rs.where([ + f'benchmark_name = `{effective_bench}`', + f'set_id.matches(`{setid_regex}`)', + ]).view(['benchmark_name', 'set_id', 'run_id']) + + tput_raw = _raw_be.where(['name = `duration_rows`']) \ + .where_in(median_runs, cols=['benchmark_name', 'set_id', 'run_id']) + min_starts = tput_raw.agg_by([agg.min_("min_start = start")], by=["benchmark_name", "set_id", "run_id"]) + tput_t = tput_raw \ + .natural_join(min_starts, on=["benchmark_name", "set_id", "run_id"]) \ + .view(['benchmark_name', + 'duration_offset=(start - min_start) / 1_000_000_000.0', + 'throughput=value * 1_000_000_000.0 / duration', + f'gc_type=({_replace_expr}).isEmpty() ? set_id : {_replace_expr}']) \ + .sort("duration_offset") + + # --- GC activity events --- + gc_raw = _raw_be.where(['type = `jdk.GarbageCollection`']) \ + .where_in(median_runs, cols=['benchmark_name', 'set_id', 'run_id']) + min_starts_gc = gc_raw.agg_by([agg.min_("min_start = start")], by=["benchmark_name", "set_id", "run_id"]) + gc_t = gc_raw \ + .natural_join(min_starts_gc, on=["benchmark_name", "set_id", "run_id"]) \ + .view(['benchmark_name', + 'gc_start=(start - min_start) / 1_000_000_000.0', + 'gc_duration=duration / 1_000_000_000.0', + f'gc_type=({_replace_expr}).isEmpty() ? set_id : {_replace_expr}']) \ + .sort("gc_start") + + return cdf_t, ccdf_t, tput_t, gc_t + except Exception: + return None, None, None, None + + cdf_table, ccdf_table, tput_table, gc_table = use_memo(compute_distribution, [applied]) + + # --- Controls: wrapped in ui.form so pickers don't make server calls until Apply --- + controls = ui.form( + ui.flex( + ui.picker( + *all_benchmarks, + label="Benchmark", + name="benchmark", + default_selected_key=all_benchmarks[0] if all_benchmarks else None, + ) if all_benchmarks else ui.text("No benchmarks"), + ui.picker( + "100", "1000", + label="Cycle Budget (ms)", + name="cycle_budget", + default_selected_key="100", + ), + ui.picker( + "80", "90", "100", "110", + label="Autotune %", + name="autotune", + default_selected_key="100", + ), + ui.picker( + "17", "25", + label="JVM Version", + name="jvm", + default_selected_key="25", + ), + ui.button("Apply", type="submit", variant="accent"), + direction="row", + gap="size-200", + align_items="end", + ), + on_submit=handle_submit, + flex_grow=0, + flex_shrink=0, + ) + + # --- Plots --- + has_data = cdf_table is not None and ccdf_table is not None + if has_data: + gc_types = sorted(dhpd.to_pandas( + cdf_table.select_distinct(["gc_type"]) + )["gc_type"].tolist()) + color_map = {gc: COLORS[i % len(COLORS)] for i, gc in enumerate(gc_types)} + + def make_cdf_updater(title): + def update(fig): + fig.update_layout( + title=dict(text=title, font=dict(size=14)), + xaxis=dict(title="Cycle Cost (sec)"), + yaxis=dict(title="Percentile"), + margin=dict(t=40, b=40, l=50, r=10), + colorway=[color_map[gc] for gc in gc_types], + legend=dict(orientation="v", yanchor="bottom", y=0, xanchor="left", x=1.02), + ) + return fig + return update + + cdf_plot = dx.line( + cdf_table, x="cycleCostSec", y="Percentile", + by=["gc_type"], log_x=True, + title=f"{effective_bench} CDF", + unsafe_update_figure=make_cdf_updater(f"{effective_bench} CDF"), + ) + ccdf_plot = dx.line( + ccdf_table, x="cycleCostSec", y="Percentile", + by=["gc_type"], log_x=True, log_y=True, + title=f"{effective_bench} CCDF", + unsafe_update_figure=make_cdf_updater(f"{effective_bench} CCDF"), + ) + left_col = ui.flex(cdf_plot, ccdf_plot, direction="column", flex_grow=1, gap="size-200") + + if tput_table is not None and gc_table is not None: + def make_throughput_updater(fig): + fig.update_traces(line=dict(width=1)) + fig.update_layout( + title=dict(text=f"{effective_bench} - Throughput", font=dict(size=14)), + legend=dict(yanchor="bottom", y=0, xanchor="left", x=1.02), + margin=dict(t=40, b=10, l=60, r=10), + xaxis=dict(title="", showticklabels=False), + colorway=[color_map[gc] for gc in gc_types], + ) + return fig + + def make_gc_updater(fig): + fig.update_traces( + error_x=dict(type="data", symmetric=False, thickness=1), + marker=dict(size=2), + ) + fig.update_layout( + title=dict(text="GC Activity", font=dict(size=14)), + legend=dict(yanchor="bottom", y=0, xanchor="left", x=1.02), + margin=dict(t=40, b=40, l=60, r=10), + xaxis=dict(title="Elapsed Time (s)"), + yaxis=dict(title="GC (s)"), + colorway=[color_map[gc] for gc in gc_types], + ) + return fig + + throughput_plot = dx.line( + tput_table, x="duration_offset", y="throughput", + color="gc_type", + unsafe_update_figure=make_throughput_updater, + ) + gc_plot = dx.scatter( + gc_table, x="gc_start", y="gc_duration", + error_x="gc_duration", color="gc_type", log_y=True, + unsafe_update_figure=make_gc_updater, + ) + right_col = ui.flex(throughput_plot, gc_plot, direction="column", flex_grow=1, gap="size-200") + else: + right_col = ui.text("No throughput/GC data available.") + + plots = ui.flex(left_col, right_col, direction="row", flex_grow=1, gap="size-200") + else: + plots = ui.text("Click Apply to load plots.") + + return ui.flex( + controls, + plots, + direction="column", + flex_grow=1, + gap="size-100", + UNSAFE_style={"height": "100%", "overflow": "auto"}, + ) + + +@ui.component +def gc_rankings_dashboard(): + import math + import pandas as pd + from deephaven import new_table + from deephaven.column import string_col + + def compute_rankings(): + global actor, static_inc_filter, benchmark_filter, cycle_budget_ms, autotune_pct, jvm_version + global setid_filter, bench_result_sets, bench_results, bench_metrics, bench_events, gc_metrics + + si_options = ["Static", "Inc"] + budget_options = ["100", "1000"] + autotune_options = ["80", "90", "100", "110"] + jvm_options = ["17", "25"] + + tput_rows = [] + jitter_rows = [] + budget_rows = [] + + for si in si_options: + for budget in budget_options: + for auto in autotune_options: + for jvm in jvm_options: + # Set globals for this combination + static_inc_filter = si + benchmark_filter = "*" + cycle_budget_ms = int(budget) + autotune_pct = auto + jvm_version = jvm + setid_filter = f'*_{budget}_p{auto}_j{jvm}*' + + settings_label = f"{si}_{budget}_p{auto}_j{jvm}" + + rs = _raw_rs.sort(['set_id']) + br = _raw_br.sort(['set_id']) + bm = _raw_bm + be = _raw_be + rs, br, bm, be = [filter_benchmarks(t) for t in [rs, br, bm, be]] + bench_result_sets = rs + bench_results = br + bench_metrics = bm + bench_events = be + gc_metrics = rs.join(bm, ['benchmark_name','origin','set_id'], ['name','value']).sort(['set_id']) + + suffix = f"_{budget}_p{auto}_j{jvm}" + + import re as _re + def _gc_name(col): + name = col.replace(suffix, '') + name = _re.sub(rf'[\s_-]*{_re.escape(si)}[\s_-]*', '_', name) + name = _re.sub(r'_+', '_', name) + name = name.strip('_') + return name + + def _visible_cols(tbl): + return [c.name for c in tbl.columns if c.name != 'benchmark_name' and '__TABLE_STYLE_FORMAT' not in c.name] + + def _geo_mean(df, col): + vals = df[col].dropna().values.astype(float) + vals = vals[vals > 0] + return float(np.exp(np.log(vals).mean())) if len(vals) > 0 else float('nan') + + def _rank_row(gc_means, settings, higher_is_better): + # Sort GCs: best first + valid = [(gc, v) for gc, v in gc_means.items() if not math.isnan(v)] + if not valid: + return None + valid.sort(key=lambda x: x[1], reverse=higher_is_better) + best_val = valid[0][1] + row = {"Settings": settings} + for i, (gc, val) in enumerate(valid): + rank_col = f"{i+1}" + if i == 0: + row[rank_col] = gc + else: + if best_val != 0: + pct = ((val - best_val) / best_val) * 100.0 + row[rank_col] = f"{gc} ({pct:+.0f}%)" + else: + row[rank_col] = gc + return row + + # Throughput + try: + tbl = get_throughput_table() + cols = _visible_cols(tbl) + df = dhpd.to_pandas(tbl.view(['benchmark_name'] + cols)) + gc_means = {_gc_name(c): _geo_mean(df, c) for c in cols} + r = _rank_row(gc_means, settings_label, higher_is_better=True) + if r: + tput_rows.append(r) + except Exception: + pass + + # Jitter (lower is better) + try: + tbl = get_jitter_table() + cols = _visible_cols(tbl) + df = dhpd.to_pandas(tbl.view(['benchmark_name'] + cols)) + gc_means = {_gc_name(c): _geo_mean(df, c) for c in cols} + r = _rank_row(gc_means, settings_label, higher_is_better=False) + if r: + jitter_rows.append(r) + except Exception: + pass + + # On-budget (higher is better, use mean) + try: + tbl = get_cycle_onbudget_table() + cols = _visible_cols(tbl) + df = dhpd.to_pandas(tbl.view(['benchmark_name'] + cols)) + gc_means = {_gc_name(c): float(df[c].dropna().mean()) for c in cols} + r = _rank_row(gc_means, settings_label, higher_is_better=True) + if r: + budget_rows.append(r) + except Exception: + pass + + def _build_table(rows, label): + if not rows: + return new_table([string_col("Settings", [f"No {label} data"])]) + max_rank = max(len(r) - 1 for r in rows) + settings_vals = [r.get("Settings", "") for r in rows] + columns = [string_col("Settings", settings_vals)] + for i in range(max_rank): + columns.append(string_col(f"r{i+1}", [r.get(str(i+1), "") for r in rows])) + return new_table(columns) + + t_tput = _build_table(tput_rows, "throughput") + t_jitter = _build_table(jitter_rows, "jitter") + t_budget = _build_table(budget_rows, "on-budget") + return t_tput, t_jitter, t_budget + + tput_rank, jitter_rank, budget_rank = use_memo(compute_rankings, []) + + return ui.flex( + ui.flex( + ui.heading("Throughput Rankings (GEO MEAN, higher is better)", level=4), + ui.table(tput_rank), + direction="column", flex_grow=1, + ), + ui.flex( + ui.heading("Jitter Rankings (GEO MEAN, lower is better)", level=4), + ui.table(jitter_rank), + direction="column", flex_grow=1, + ), + ui.flex( + ui.heading("On-Budget Rankings (MEAN, higher is better)", level=4), + ui.table(budget_rank), + direction="column", flex_grow=1, + ), + direction="column", + flex_grow=1, + gap="size-200", + UNSAFE_style={"height": "100%", "overflow": "auto"}, + ) + +@ui.component +def gc_extras_dashboard(): + import math + from deephaven import new_table + from deephaven.column import string_col, long_col, double_col + + def compute_extras(): + global actor, static_inc_filter, benchmark_filter, cycle_budget_ms, autotune_pct, jvm_version + global setid_filter, bench_result_sets, bench_results, bench_metrics, bench_events, gc_metrics + + # Configurations to compare (all p100_j25) + configs = [ + ("Static_1000", "Static", "1000"), + ("Inc_1000", "Inc", "1000"), + ("Inc_100", "Inc", "100"), + ] + + import re as _re + + def _strip_si(name): + """Remove Static/Inc from benchmark name to create a common key.""" + name = _re.sub(r'(?i)\b(Static|Inc)\b', '', name) + name = _re.sub(r'_+', '_', name).strip('_') + name = _re.sub(r'\s*-\s*$', '', name) + return name + + # Collect per-benchmark GEO MEAN throughput for each config + config_data = {} # config_label -> {common_key: geo_mean_throughput} + for label, si, budget in configs: + static_inc_filter = si + benchmark_filter = "*" + cycle_budget_ms = int(budget) + autotune_pct = "100" + jvm_version = "25" + setid_filter = f'*_{budget}_p100_j25*' + + rs = _raw_rs.sort(['set_id']) + br = _raw_br.sort(['set_id']) + bm = _raw_bm + be = _raw_be + rs, br, bm, be = [filter_benchmarks(t) for t in [rs, br, bm, be]] + bench_result_sets = rs + bench_results = br + bench_metrics = bm + bench_events = be + gc_metrics = rs.join(bm, ['benchmark_name','origin','set_id'], ['name','value']).sort(['set_id']) + + try: + df = dhpd.to_pandas( + bench_result_sets.view(['benchmark_name', 'op_rate']) + .where(['!isNull(op_rate)', 'op_rate > 0']) + ) + bench_means = {} + for bname, grp in df.groupby('benchmark_name'): + key = _strip_si(bname) + vals = grp['op_rate'].values.astype(float) + if len(vals) > 0: + bench_means[key] = float(np.exp(np.log(vals).mean())) + config_data[label] = bench_means + except Exception: + config_data[label] = {} + + # Build table using common benchmark keys + baseline_label = "Static_1000" + baseline = config_data.get(baseline_label, {}) + all_keys = sorted(baseline.keys()) + + if not all_keys: + return new_table([string_col("benchmark_name", ["No data"])]) + + bench_names = [] + baseline_vals = [] + inc_1000_vals = [] + inc_100_vals = [] + + for key in all_keys: + base_val = baseline.get(key, float('nan')) + bench_names.append(key) + baseline_vals.append(base_val) + + # Inc_1000 % diff + inc1000 = config_data.get("Inc_1000", {}).get(key, float('nan')) + if not math.isnan(base_val) and base_val > 0 and not math.isnan(inc1000): + inc_1000_vals.append(((inc1000 - base_val) / base_val) * 100.0) + else: + inc_1000_vals.append(float('nan')) + + # Inc_100 % diff + inc100 = config_data.get("Inc_100", {}).get(key, float('nan')) + if not math.isnan(base_val) and base_val > 0 and not math.isnan(inc100): + inc_100_vals.append(((inc100 - base_val) / base_val) * 100.0) + else: + inc_100_vals.append(float('nan')) + + t = new_table([ + string_col("benchmark_name", bench_names), + long_col("Static_1000", [int(round(v)) if not math.isnan(v) else 0 for v in baseline_vals]), + double_col("Inc_1000", [round(v, 1) if not math.isnan(v) else float('nan') for v in inc_1000_vals]), + double_col("Inc_100", [round(v, 1) if not math.isnan(v) else float('nan') for v in inc_100_vals]), + ]).format_columns([ + "Inc_1000 = Decimal(`0.0'%'`)", + "Inc_100 = Decimal(`0.0'%'`)", + ]) + return t + + extras_table = use_memo(compute_extras, []) + + return ui.flex( + ui.flex( + ui.heading("Throughput Comparison — GEO MEAN across GCs (p100, j25)", level=4), + ui.table(extras_table), + direction="column", flex_grow=1, + ), + direction="column", + flex_grow=1, + gap="size-200", + UNSAFE_style={"height": "100%", "overflow": "auto"}, + ) + +gc_rankings = gc_rankings_dashboard() +gc_bench_summary = gc_bench_summary_dashboard() +gc_throughput = gc_throughput_dashboard() +gc_extras = gc_extras_dashboard() diff --git a/src/test/java/io/deephaven/benchmark/api/QueryLogTest.java b/src/test/java/io/deephaven/benchmark/api/QueryLogTest.java index 0d5169b6..14cc37d7 100644 --- a/src/test/java/io/deephaven/benchmark/api/QueryLogTest.java +++ b/src/test/java/io/deephaven/benchmark/api/QueryLogTest.java @@ -1,4 +1,4 @@ -/* Copyright (c) 2022-2023 Deephaven Data Labs and Patent Pending */ +/* Copyright (c) 2022-2026 Deephaven Data Labs and Patent Pending */ package io.deephaven.benchmark.api; import static org.junit.jupiter.api.Assertions.*; @@ -11,7 +11,7 @@ public class QueryLogTest { @Test public void logQuery() throws Exception { Path outParent = Paths.get(getClass().getResource("test-profile.properties").toURI()).getParent(); - Files.deleteIfExists(QueryLog.getLogFile(outParent, QueryLogTest.class)); + Files.deleteIfExists(outParent.resolve("query.md")); var qlog = new QueryLog(outParent, QueryLogTest.class); qlog.setName(getClass().getSimpleName()); @@ -37,21 +37,21 @@ public void logQuery() throws Exception { ## Test - 1st Test ### Query 1 - ```` + ``` setup test - ```` + ``` ### Query 2 - ```` + ``` query1 query line - ```` + ``` ### Query 3 - ```` + ``` query2 query line - ```` + ``` """.replace("\r", "").trim(); var text = Filer.getFileText(qlog.logFile); diff --git a/src/test/java/io/deephaven/benchmark/util/FilerTest.java b/src/test/java/io/deephaven/benchmark/util/FilerTest.java index 773cd46a..53442e79 100644 --- a/src/test/java/io/deephaven/benchmark/util/FilerTest.java +++ b/src/test/java/io/deephaven/benchmark/util/FilerTest.java @@ -1,4 +1,4 @@ -/* Copyright (c) 2022-2023 Deephaven Data Labs and Patent Pending */ +/* Copyright (c) 2022-2026 Deephaven Data Labs and Patent Pending */ package io.deephaven.benchmark.util; import static org.junit.jupiter.api.Assertions.*; @@ -48,4 +48,11 @@ public void getUrlText() throws Exception { assertEquals("One Two Three\nFour Five Six", Filer.getURLText(url), "Wrong file text"); } + @Test + public void getByteSize() throws Exception { + var p = Paths.get(getClass().getResource("filerfolder").toURI()).toFile().toString(); + assertEquals(46, Filer.getByteSize(p), "Wrong byte size"); + p = Paths.get(getClass().getResource("filertest.txt").toURI()).toFile().toString(); + assertEquals(27, Filer.getByteSize(p), "Wrong byte size"); + } } diff --git a/src/test/resources/io/deephaven/benchmark/util/filerfolder/filer1.txt b/src/test/resources/io/deephaven/benchmark/util/filerfolder/filer1.txt new file mode 100644 index 00000000..9f40cf63 --- /dev/null +++ b/src/test/resources/io/deephaven/benchmark/util/filerfolder/filer1.txt @@ -0,0 +1 @@ +This file has bytes \ No newline at end of file diff --git a/src/test/resources/io/deephaven/benchmark/util/filerfolder/folder1/filer2.txt b/src/test/resources/io/deephaven/benchmark/util/filerfolder/folder1/filer2.txt new file mode 100644 index 00000000..58da37b3 --- /dev/null +++ b/src/test/resources/io/deephaven/benchmark/util/filerfolder/folder1/filer2.txt @@ -0,0 +1 @@ +This file has bytes as well \ No newline at end of file