-
Notifications
You must be signed in to change notification settings - Fork 98
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
IGNITE-23944 Sql. Add TPC DS suite (#4873)
- Loading branch information
Showing
132 changed files
with
6,601 additions
and
92 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
133 changes: 133 additions & 0 deletions
133
...r/src/integrationTest/java/org/apache/ignite/internal/benchmark/AbstractTpcBenchmark.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,133 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.ignite.internal.benchmark; | ||
|
|
||
| import java.io.IOException; | ||
| import java.nio.file.Files; | ||
| import java.nio.file.Path; | ||
| import java.time.Duration; | ||
| import java.util.concurrent.Semaphore; | ||
| import java.util.concurrent.atomic.AtomicInteger; | ||
| import java.util.concurrent.atomic.AtomicReference; | ||
| import org.apache.ignite.internal.sql.engine.util.TpcTable; | ||
| import org.apache.ignite.sql.IgniteSql; | ||
| import org.openjdk.jmh.annotations.Scope; | ||
| import org.openjdk.jmh.annotations.Setup; | ||
| import org.openjdk.jmh.annotations.State; | ||
|
|
||
| /** | ||
| * Abstract benchmark class that initializes schema and fills up tables for TPC suite. | ||
| */ | ||
| @State(Scope.Benchmark) | ||
| @SuppressWarnings({"WeakerAccess", "unused"}) | ||
| public abstract class AbstractTpcBenchmark extends AbstractMultiNodeBenchmark { | ||
| private static final String DATASET_READY_MARK_FILE_NAME = "ready.txt"; | ||
|
|
||
| protected IgniteSql sql; | ||
|
|
||
| abstract TpcTable[] tablesToInit(); | ||
|
|
||
| abstract Path pathToDataset(); | ||
|
|
||
| /** Initializes a schema and fills tables with data. */ | ||
| @Setup | ||
| public void initSchema() throws Throwable { | ||
| try { | ||
| sql = publicIgnite.sql(); | ||
|
|
||
| if (!Files.exists(workDir().resolve(DATASET_READY_MARK_FILE_NAME))) { | ||
| Path pathToDataset = pathToDataset(); | ||
|
|
||
| if (pathToDataset == null) { | ||
| throw new IllegalStateException("Path do dataset is not provided. Please read the comment" | ||
| + " in the beginning of " + this.getClass().getSimpleName() + ".class"); | ||
| } | ||
|
|
||
| System.out.println("Going to create schema..."); | ||
|
|
||
| for (TpcTable table : tablesToInit()) { | ||
| System.out.println("Going to create table \"" + table.tableName() + "\"..."); | ||
| sql.executeScript(table.ddlScript()); | ||
| System.out.println("Done"); | ||
|
|
||
| fillTable(table, pathToDataset); | ||
| } | ||
|
|
||
| Files.createFile(workDir().resolve(DATASET_READY_MARK_FILE_NAME)); | ||
| } | ||
| } catch (Throwable e) { | ||
| nodeTearDown(); | ||
|
|
||
| throw e; | ||
| } | ||
| } | ||
|
|
||
| private void fillTable(TpcTable table, Path pathToDataset) throws Throwable { | ||
| System.out.println("Going to fill table \"" + table.tableName() + "\"..."); | ||
| long start = System.nanoTime(); | ||
| Iterable<Object[]> dataProvider = () -> { | ||
| try { | ||
| return table.dataProvider(pathToDataset); | ||
| } catch (IOException e) { | ||
| throw new RuntimeException(e); | ||
| } | ||
| }; | ||
|
|
||
| Semaphore semaphore = new Semaphore(1024); // 1024 was chosen empirically | ||
| AtomicReference<Throwable> exceptionHolder = new AtomicReference<>(null); | ||
| AtomicInteger inserted = new AtomicInteger(); | ||
| int expectedCount = 0; | ||
| for (Object[] params : dataProvider) { | ||
| semaphore.acquire(); | ||
| sql.executeAsync(null, table.insertPrepareStatement(), params) | ||
| .whenComplete((ignored, ex) -> { | ||
| semaphore.release(); | ||
|
|
||
| int val = inserted.incrementAndGet(); | ||
| if (val % 10_000 == 0) { | ||
| System.out.println(val + " rows uploaded to \"" + table.tableName() + "\""); | ||
| } | ||
|
|
||
| if (ex != null) { | ||
| exceptionHolder.compareAndSet(null, ex); | ||
| } | ||
| }); | ||
|
|
||
| if (exceptionHolder.get() != null) { | ||
| throw exceptionHolder.get(); | ||
| } | ||
|
|
||
| expectedCount++; | ||
| } | ||
|
|
||
| while (expectedCount != inserted.intValue()) { | ||
| if (exceptionHolder.get() != null) { | ||
| throw exceptionHolder.get(); | ||
| } | ||
|
|
||
| Thread.sleep(100); | ||
| } | ||
|
|
||
| System.out.println("Table \"" + table.tableName() + "\" filled in " + Duration.ofNanos(System.nanoTime() - start)); | ||
| } | ||
|
|
||
| @Override | ||
| protected void createTable(String tableName) { | ||
| // NO-OP | ||
| } | ||
| } |
112 changes: 112 additions & 0 deletions
112
.../runner/src/integrationTest/java/org/apache/ignite/internal/benchmark/TpcdsBenchmark.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,112 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.ignite.internal.benchmark; | ||
|
|
||
| import java.nio.file.Path; | ||
| import java.util.concurrent.TimeUnit; | ||
| import org.apache.ignite.internal.sql.engine.util.TpcTable; | ||
| import org.apache.ignite.internal.sql.engine.util.tpcds.TpcdsHelper; | ||
| import org.apache.ignite.internal.sql.engine.util.tpcds.TpcdsTables; | ||
| import org.openjdk.jmh.annotations.Benchmark; | ||
| import org.openjdk.jmh.annotations.BenchmarkMode; | ||
| import org.openjdk.jmh.annotations.Fork; | ||
| import org.openjdk.jmh.annotations.Measurement; | ||
| import org.openjdk.jmh.annotations.Mode; | ||
| import org.openjdk.jmh.annotations.OutputTimeUnit; | ||
| import org.openjdk.jmh.annotations.Param; | ||
| import org.openjdk.jmh.annotations.Scope; | ||
| import org.openjdk.jmh.annotations.Setup; | ||
| import org.openjdk.jmh.annotations.State; | ||
| import org.openjdk.jmh.annotations.Threads; | ||
| import org.openjdk.jmh.annotations.Warmup; | ||
| import org.openjdk.jmh.infra.Blackhole; | ||
| import org.openjdk.jmh.runner.Runner; | ||
| import org.openjdk.jmh.runner.RunnerException; | ||
| import org.openjdk.jmh.runner.options.Options; | ||
| import org.openjdk.jmh.runner.options.OptionsBuilder; | ||
|
|
||
| /** | ||
| * Benchmark that runs sql queries from TPC-DS suite via embedded client. | ||
| */ | ||
| @State(Scope.Benchmark) | ||
| @Fork(1) | ||
| @Threads(1) | ||
| @Warmup(iterations = 10, time = 2) | ||
| @Measurement(iterations = 20, time = 2) | ||
| @BenchmarkMode(Mode.AverageTime) | ||
| @OutputTimeUnit(TimeUnit.MILLISECONDS) | ||
| @SuppressWarnings({"WeakerAccess", "unused"}) | ||
| public class TpcdsBenchmark extends AbstractTpcBenchmark { | ||
| /* | ||
| Minimal configuration of this benchmark requires specifying pathToDataset. Dataset is set of CSV | ||
| files with name `{$tableName}.dat` per each table and character `|` as separator. | ||
| By default, cluster's work directory will be created as a temporary folder. This implies, | ||
| that all data generated by benchmark will be cleared automatically. However, this also implies | ||
| that cluster will be recreated on EVERY RUN. To initialize cluster once and then reuse it state | ||
| override `AbstractMultiNodeBenchmark.workDir()` method. Don't forget to clear that directory afterwards. | ||
| */ | ||
|
|
||
| @Override | ||
| TpcTable[] tablesToInit() { | ||
| return TpcdsTables.values(); | ||
| } | ||
|
|
||
| @Override | ||
| Path pathToDataset() { | ||
| throw new RuntimeException("Provide path to directory containing <table_name>.dat files"); | ||
| } | ||
|
|
||
| @Param("1") | ||
| private String queryId; | ||
|
|
||
| private String queryString; | ||
|
|
||
| /** Initializes a query string. */ | ||
| @Setup | ||
| public void setUp() throws Throwable { | ||
| try { | ||
| queryString = TpcdsHelper.getQuery(queryId); | ||
| } catch (Throwable e) { | ||
| nodeTearDown(); | ||
|
|
||
| throw e; | ||
| } | ||
| } | ||
|
|
||
| /** Benchmark that measures performance of queries from TPC-DS suite. */ | ||
| @Benchmark | ||
| public void run(Blackhole bh) { | ||
| try (var rs = sql.execute(null, queryString)) { | ||
| while (rs.hasNext()) { | ||
| bh.consume(rs.next()); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Benchmark's entry point. | ||
| */ | ||
| public static void main(String[] args) throws RunnerException { | ||
| Options opt = new OptionsBuilder() | ||
| .include(".*" + TpcdsBenchmark.class.getSimpleName() + ".*") | ||
| .build(); | ||
|
|
||
| new Runner(opt).run(); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.