jffwang
/
hetu-core
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Memory connector fixes/Add async processing property/UT fixes

This commit is contained in:
Ayush Gupta 2021-08-13 12:57:44 -04:00
parent f2ef01ffe0
commit c137254422
9 changed files with 765 additions and 740 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryMetadata.java
View File

@ -300,6 +300,7 @@ public class MemoryMetadata
newTableName.getSchemaName(),
newTableName.getTableName(),
oldTableHandle.isCompressionEnabled(),
oldTableHandle.isAsyncProcessingEnabled(),
oldTableHandle.getActiveTableIds(),
oldTableHandle.getColumns(),
oldTableHandle.getSortedBy(),
@ -408,12 +409,14 @@ public class MemoryMetadata
System.currentTimeMillis()));
boolean spillCompressionEnabled = MemoryTableProperties.getSpillCompressionEnabled(tableMetadata.getProperties());
boolean asyncProcessingEnabled = MemoryTableProperties.getAsyncProcessingEnabled(tableMetadata.getProperties());
return new MemoryWriteTableHandle(
nextId,
tableMetadata.getTable().getSchemaName(),
tableMetadata.getTable().getTableName(),
spillCompressionEnabled,
asyncProcessingEnabled,
getTableIdSet(nextId),
columnHandles,
sortedBy,

30
presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryPageSinkProvider.java
View File

@ -33,6 +33,8 @@ import javax.inject.Inject;
import java.io.FileNotFoundException;
import java.util.Collection;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import static com.google.common.base.Preconditions.checkState;
import static io.prestosql.plugin.memory.MemoryErrorCode.MISSING_DATA;
@ -45,6 +47,7 @@ public class MemoryPageSinkProvider
{
private final MemoryTableManager pagesStore;
private final HostAddress currentHostAddress;
private final ConcurrentHashMap<Long, AtomicInteger> tableSinkCount;
@Inject
public MemoryPageSinkProvider(MemoryTableManager pagesStore, NodeManager nodeManager)
@ -57,6 +60,7 @@ public class MemoryPageSinkProvider
{
this.pagesStore = requireNonNull(pagesStore, "pagesStore is null");
this.currentHostAddress = requireNonNull(currentHostAddress, "currentHostAddress is null");
this.tableSinkCount = new ConcurrentHashMap<>();
}
/**
@ -70,9 +74,13 @@ public class MemoryPageSinkProvider
long tableId = memoryOutputTableHandle.getTable();
checkState(memoryOutputTableHandle.getActiveTableIds().contains(tableId));
AtomicInteger sinkCount = tableSinkCount.computeIfAbsent(tableId, (k) -> new AtomicInteger(0));
sinkCount.incrementAndGet();
pagesStore.refreshTables(memoryOutputTableHandle.getActiveTableIds());
pagesStore.initialize(tableId,
memoryOutputTableHandle.isCompressionEnabled(),
memoryOutputTableHandle.isAsyncProcessingEnabled(),
memoryOutputTableHandle.getColumns(),
memoryOutputTableHandle.getSortedBy(),
memoryOutputTableHandle.getIndexColumns());
@ -84,7 +92,7 @@ public class MemoryPageSinkProvider
throw new PrestoException(GENERIC_USER_ERROR, "Failed writing data to memory.spill-path, ensure directory has correct permissions and free space is available.", e);
}
return new MemoryPageSink(pagesStore, currentHostAddress, tableId);
return new MemoryPageSink(pagesStore, currentHostAddress, tableId, sinkCount);
}
/**
@ -98,6 +106,9 @@ public class MemoryPageSinkProvider
long tableId = memoryOutputTableHandle.getTable();
checkState(memoryOutputTableHandle.getActiveTableIds().contains(tableId));
AtomicInteger sinkCount = tableSinkCount.computeIfAbsent(tableId, (k) -> new AtomicInteger(0));
sinkCount.incrementAndGet();
pagesStore.refreshTables(memoryOutputTableHandle.getActiveTableIds());
// Try restore since table was created and may have data
@ -113,6 +124,7 @@ public class MemoryPageSinkProvider
//
pagesStore.initialize(tableId,
memoryOutputTableHandle.isCompressionEnabled(),
memoryOutputTableHandle.isAsyncProcessingEnabled(),
memoryOutputTableHandle.getColumns(),
memoryOutputTableHandle.getSortedBy(),
memoryOutputTableHandle.getIndexColumns());
@ -128,7 +140,7 @@ public class MemoryPageSinkProvider
throw new PrestoException(GENERIC_USER_ERROR, "Failed writing data to memory.spill-path, ensure directory has correct permissions and free space is available.", e);
}
return new MemoryPageSink(pagesStore, currentHostAddress, tableId);
return new MemoryPageSink(pagesStore, currentHostAddress, tableId, sinkCount);
}
private static class MemoryPageSink
@ -138,12 +150,14 @@ public class MemoryPageSinkProvider
private final HostAddress currentHostAddress;
private final long tableId;
private long addedRows;
private AtomicInteger sinkCount;
public MemoryPageSink(MemoryTableManager tablesManager, HostAddress currentHostAddress, long tableId)
public MemoryPageSink(MemoryTableManager tablesManager, HostAddress currentHostAddress, long tableId, AtomicInteger sinkCount)
{
this.tablesManager = requireNonNull(tablesManager, "pagesStore is null");
this.currentHostAddress = requireNonNull(currentHostAddress, "currentHostAddress is null");
this.tableId = tableId;
this.sinkCount = sinkCount;
}
@Override
@ -157,14 +171,18 @@ public class MemoryPageSinkProvider
@Override
public CompletableFuture<Collection<Slice>> finish()
{
tablesManager.finishUpdatingTable(tableId);
int logicalPartCount = tablesManager.getTableLpCount(tableId);
return completedFuture(ImmutableList.of(new MemoryDataFragment(currentHostAddress, addedRows, logicalPartCount).toSlice()));
if (sinkCount.decrementAndGet() == 0) {
tablesManager.finishUpdatingTable(tableId);
int logicalPartCount = tablesManager.getTableLpCount(tableId);
return completedFuture(ImmutableList.of(new MemoryDataFragment(currentHostAddress, addedRows, logicalPartCount).toSlice()));
}
return completedFuture(ImmutableList.of(new MemoryDataFragment(currentHostAddress, addedRows, 0).toSlice()));
}
@Override
public void abort()
{
sinkCount.decrementAndGet();
tablesManager.cleanTable(tableId);
}
}

13
presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryTableProperties.java
View File

@ -33,6 +33,8 @@ public class MemoryTableProperties
{
public static final String SPILL_COMPRESSION_PROPERTY = "spill_compression";
public static final boolean SPILL_COMPRESSION_DEFAULT_VALUE = true;
public static final String ASYNC_PROCESSING = "async_processing";
public static final boolean ASYNC_PROCESSING_DEFAULT_VALUE = true;
public static final String PARTITIONED_BY_PROPERTY = "partitioned_by";
public static final String SORTED_BY_PROPERTY = "sorted_by";
public static final String INDEX_COLUMNS_PROPERTY = "index_columns";
@ -85,6 +87,11 @@ public class MemoryTableProperties
SPILL_COMPRESSION_PROPERTY,
"Whether to enable page compression during spilling",
SPILL_COMPRESSION_DEFAULT_VALUE,
false),
PropertyMetadata.booleanProperty(
ASYNC_PROCESSING,
"Whether to process LogicalParts asynchronously",
ASYNC_PROCESSING_DEFAULT_VALUE,
false));
}
@ -122,6 +129,12 @@ public class MemoryTableProperties
return spillCompressionEnabled == null ? SPILL_COMPRESSION_DEFAULT_VALUE : spillCompressionEnabled;
}
public static boolean getAsyncProcessingEnabled(Map<String, Object> tableProperties)
{
Boolean asyncProcessingEnabled = (Boolean) tableProperties.get(ASYNC_PROCESSING);
return asyncProcessingEnabled == null ? ASYNC_PROCESSING_DEFAULT_VALUE : asyncProcessingEnabled;
}
private static SortingColumn sortingColumnFromString(String name)
{
SortOrder order = SortOrder.ASC_NULLS_FIRST;

12
presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryWriteTableHandle.java
View File

@ -24,6 +24,7 @@ import java.util.List;
import java.util.Set;
import static com.google.common.base.MoreObjects.toStringHelper;
import static io.prestosql.plugin.memory.MemoryTableProperties.ASYNC_PROCESSING_DEFAULT_VALUE;
import static io.prestosql.plugin.memory.MemoryTableProperties.SPILL_COMPRESSION_DEFAULT_VALUE;
import static java.util.Objects.requireNonNull;
@ -32,6 +33,7 @@ public final class MemoryWriteTableHandle
{
private final long table;
private final boolean compressionEnabled;
private final boolean asyncProcessingEnabled;
private final Set<Long> activeTableIds;
private final List<MemoryColumnHandle> columns;
private final List<SortingColumn> sortedBy;
@ -45,6 +47,7 @@ public final class MemoryWriteTableHandle
@JsonProperty("schemaName") String schemaName,
@JsonProperty("tableName") String tableName,
@JsonProperty("compressEnabled") boolean compressionEnabled,
@JsonProperty("asyncProcessingEnabled") boolean asyncProcessingEnabled,
@JsonProperty("activeTableIds") Set<Long> activeTableIds,
@JsonProperty("columns") List<MemoryColumnHandle> columns,
@JsonProperty("sortedBy") List<SortingColumn> sortedBy,
@ -58,12 +61,13 @@ public final class MemoryWriteTableHandle
this.columns = requireNonNull(columns, "columns is null");
this.sortedBy = requireNonNull(sortedBy, "sortedBy is null");
this.indexColumns = requireNonNull(indexColumns, "indexColumns is null");
this.asyncProcessingEnabled = asyncProcessingEnabled;
}
@VisibleForTesting
MemoryWriteTableHandle(long table, Set<Long> activeTableIds)
{
this(table, "", "", SPILL_COMPRESSION_DEFAULT_VALUE, activeTableIds, Collections.emptyList(), Collections.emptyList(), Collections.emptyList());
this(table, "", "", SPILL_COMPRESSION_DEFAULT_VALUE, ASYNC_PROCESSING_DEFAULT_VALUE, activeTableIds, Collections.emptyList(), Collections.emptyList(), Collections.emptyList());
}
@JsonProperty
@ -78,6 +82,12 @@ public final class MemoryWriteTableHandle
return compressionEnabled;
}
@JsonProperty
public boolean isAsyncProcessingEnabled()
{
return asyncProcessingEnabled;
}
@JsonProperty
public Set<Long> getActiveTableIds()
{

50
presto-memory/src/main/java/io/prestosql/plugin/memory/data/MemoryTableManager.java
View File

@ -55,8 +55,6 @@ import java.util.Map;
import java.util.OptionalDouble;
import java.util.OptionalLong;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
@ -99,7 +97,8 @@ public class MemoryTableManager
this.pagesSerde = requireNonNull(pagesSerde, "pagesSerde is null");
}
public void validateSpillRoot() throws IOException
public void validateSpillRoot()
throws IOException
{
RandomAccessFile testFile;
String name = UUID.randomUUID().toString();
@ -116,42 +115,33 @@ public class MemoryTableManager
}
}
public synchronized void finishUpdatingTable(long id)
public void finishUpdatingTable(long id)
{
tables.get(id).finishCreation();
Timer timer = new Timer(true);
timer.scheduleAtFixedRate(new TimerTask()
{
@Override
public void run()
{
if (tables.containsKey(id) && tables.get(id).allProcessed()) {
try {
if (tables.get(id).isSpilled()) {
timer.cancel();
return;
}
spillTable(id);
// processing has finished. Creation overhead can be released
releaseMemory(tables.get(id).getByteSize() * (CREATION_SCALE_FACTOR - 1), "Finish processing table " + id);
}
catch (Exception e) {
LOG.error("Failed to serialize table " + id, e);
}
tables.get(id).finishCreation(() -> {
// this should only be called once entire table has been processed
if (tables.containsKey(id) && tables.get(id).allProcessed()) {
try {
// first spill the table to disk
spillTable(id);
// release memory overhead used during processing
releaseMemory(tables.get(id).getByteSize() * (CREATION_SCALE_FACTOR - 1), "Finish processing table " + id);
}
catch (Exception e) {
LOG.error("Failed to serialize table " + id, e);
}
}
}, 0, 3000);
});
}
/**
* Initialize a table and store it in memory
*/
public synchronized void initialize(long tableId, boolean compressionEnabled, List<MemoryColumnHandle> columns, List<SortingColumn> sortedBy, List<String> indexColumns)
public synchronized void initialize(long tableId, boolean compressionEnabled, boolean asyncProcessingEnabled, List<MemoryColumnHandle> columns, List<SortingColumn> sortedBy, List<String> indexColumns)
{
if (!tables.containsKey(tableId)) {
tables.put(tableId, new Table(tableId,
compressionEnabled,
asyncProcessingEnabled,
spillRoot.resolve(String.valueOf(tableId)),
columns,
sortedBy,
@ -334,7 +324,7 @@ public class MemoryTableManager
/**
* Spill table to disk.
*
* <p>
* Table object (metadata) is serialized into one file. Pages are serialized separately in logical part.
*
* @param id table id to spill
@ -367,7 +357,7 @@ public class MemoryTableManager
/**
* Restore the table from disk to load into tables map.
*
* <p>
* Only the skeleton of the table and the logical parts in it will be restored at this time.
* (pages won't be loaded until used)
*
@ -445,7 +435,7 @@ public class MemoryTableManager
}
currentBytes.set(newSize);
onSuccess.run();
logNumFormat("Fulfilled %s bytes. Current: %s", bytes, currentBytes.get());
logNumFormat("Fulfilled %s bytes for Table %s. Current: %s", bytes, reserved, currentBytes.get());
}
private synchronized void releaseMemory(long bytes, String reason)

128
presto-memory/src/main/java/io/prestosql/plugin/memory/data/Table.java
View File

@ -23,6 +23,7 @@ import io.prestosql.plugin.memory.MemoryThreadManager;
import io.prestosql.plugin.memory.SortingColumn;
import io.prestosql.spi.Page;
import io.prestosql.spi.PageSorter;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.block.SortOrder;
import io.prestosql.spi.connector.ColumnHandle;
import io.prestosql.spi.predicate.TupleDomain;
@ -38,11 +39,14 @@ import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.TreeMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import static io.prestosql.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR;
import static java.util.Objects.requireNonNull;
public class Table
@ -73,32 +77,31 @@ public class Table
the new class must be added to the whitelist below.
*/
public static final String[] TYPES_WHITELIST = ImmutableList.of(
Number.class.getCanonicalName(),
Integer.class.getCanonicalName(),
Long.class.getCanonicalName(),
Table.class.getName(),
AtomicInteger.class.getName(),
List.class.getName(),
ArrayList.class.getName(),
TableState.class.getName(),
MemoryColumnHandle.class.getName(),
SortingColumn.class.getName(),
SortOrder.class.getName(),
Enum.class.getName(),
HashMap.class.getName(),
HashSet.class.getName(),
AbstractMap.SimpleEntry.class.getName(),
BloomFilter.class.getName(),
BloomFilter.BitSet.class.getName(),
LogicalPart.class.getName(),
LogicalPart.LogicalPartState.class.getName(),
TreeMap.class.getName(),
LogicalPart.SparseValue.class.getName(),
AtomicReference.class.getName(),
long[].class.getName())
Number.class.getCanonicalName(),
Integer.class.getCanonicalName(),
Long.class.getCanonicalName(),
Table.class.getName(),
AtomicInteger.class.getName(),
List.class.getName(),
ArrayList.class.getName(),
TableState.class.getName(),
MemoryColumnHandle.class.getName(),
SortingColumn.class.getName(),
SortOrder.class.getName(),
Enum.class.getName(),
HashMap.class.getName(),
HashSet.class.getName(),
AbstractMap.SimpleEntry.class.getName(),
BloomFilter.class.getName(),
BloomFilter.BitSet.class.getName(),
LogicalPart.class.getName(),
LogicalPart.LogicalPartState.class.getName(),
TreeMap.class.getName(),
LogicalPart.SparseValue.class.getName(),
AtomicReference.class.getName(),
long[].class.getName())
.toArray(new String[0]);
private final long processingDelay;
private final List<MemoryColumnHandle> columns;
private final List<SortingColumn> sortedBy;
private final List<String> indexColumns;
@ -107,21 +110,23 @@ public class Table
private final List<LogicalPart> logicalParts; // actual data (pages) stored here
private final boolean compressionEnabled;
private TableState tableState;
private long lastModified = System.currentTimeMillis();
private long byteSize;
private final long id;
private final boolean asyncEnabled;
private transient Path tableDataRoot;
private transient PagesSerde pagesSerde;
private transient PageSorter pageSorter;
private transient TypeManager typeManager;
public Table(long id, boolean compressionEnabled, Path tableDataRoot, List<MemoryColumnHandle> columns, List<SortingColumn> sortedBy,
public Table(long id, boolean compressionEnabled, boolean asyncEnabled, Path tableDataRoot, List<MemoryColumnHandle> columns, List<SortingColumn> sortedBy,
List<String> indexColumns, PageSorter pageSorter, MemoryConfig config, TypeManager typeManager, PagesSerde pagesSerde)
{
this.id = id;
this.tableDataRoot = tableDataRoot;
this.maxLogicalPartBytes = config.getMaxLogicalPartSize().toBytes();
this.maxPageSizeBytes = Long.valueOf(config.getMaxPageSize().toBytes()).intValue();
this.processingDelay = config.getProcessingDelay().toMillis();
this.compressionEnabled = compressionEnabled;
this.columns = requireNonNull(columns, "columns is null");
this.sortedBy = requireNonNull(sortedBy, "sortedBy is null");
@ -129,29 +134,9 @@ public class Table
this.pageSorter = requireNonNull(pageSorter, "pageSorter is null");
this.typeManager = requireNonNull(typeManager, "typeManager is null");
this.pagesSerde = requireNonNull(pagesSerde, "pagesSerde is null");
this.asyncEnabled = asyncEnabled;
this.logicalParts = new ArrayList<>();
MemoryThreadManager.getSharedThreadPool().scheduleWithFixedDelay(() -> {
if ((System.currentTimeMillis() - lastModified) > processingDelay) {
for (int i = 0; i < logicalParts.size(); i++) {
LogicalPart logicalPart = logicalParts.get(i);
if (logicalPart.getProcessingState().get() == LogicalPart.LogicalPartState.FINISHED_ADDING) {
int finalI = i;
MemoryThreadManager.getSharedThreadPool().execute(() -> {
LOG.info("Processing Table %d :: logicalPart %d", id, finalI + 1);
try {
logicalPart.process();
}
catch (Exception e) {
LOG.warn("Failed to process Table %d :: logicalPart %d", id, finalI + 1);
}
LOG.info("Processed Table %d :: logicalPart %d", id, finalI + 1);
});
}
}
}
}, 5, 2, TimeUnit.SECONDS);
}
/**
@ -180,7 +165,6 @@ public class Table
}
logicalParts.get(logicalParts.size() - 1).add(page);
byteSize += page.getSizeInBytes();
lastModified = System.currentTimeMillis();
tableState = TableState.MODIFIED;
}
@ -217,13 +201,51 @@ public class Table
return tableState == TableState.SPILLED;
}
public void finishCreation()
public void finishCreation(Runnable cleanup)
{
tableState = TableState.COMMITTED;
for (LogicalPart logicalPart : logicalParts) {
List<Future<?>> futuresList = new ArrayList<>(logicalParts.size());
for (int i = 0; i < logicalParts.size(); i++) {
// for all new logical parts, set state to finished adding pages
if (logicalPart.getProcessingState().get() == LogicalPart.LogicalPartState.ACCEPTING_PAGES) {
logicalPart.finishAdding();
if (logicalParts.get(i).getProcessingState().get() == LogicalPart.LogicalPartState.ACCEPTING_PAGES) {
logicalParts.get(i).finishAdding();
}
int finalI = i;
Runnable runnable = () -> {
LOG.info("Processing Table %d :: logicalPart %d", id, finalI + 1);
try {
logicalParts.get(finalI).process();
// run manager's cleanup, this does two things
// 1. spills the table to disk, the manager handles this because the Table itself doesn't know how/where to spill
// 2. manager handles memory release
// only once all LPs are processed this cleanup will be called by the last LP
if (allProcessed()) {
cleanup.run();
}
}
catch (Exception e) {
LOG.warn("Failed to process Table %d :: logicalPart %d", id, finalI + 1);
}
LOG.info("Processed Table %d :: logicalPart %d", id, finalI + 1);
};
if (asyncEnabled) {
MemoryThreadManager.getSharedThreadPool().schedule(runnable, 5, TimeUnit.SECONDS);
}
else {
futuresList.add(MemoryThreadManager.getSharedThreadPool().submit(runnable));
}
}
// Used to synchronize processing, it will only run for sync processing since for async the list is empty
for (Future<?> future : futuresList) {
try {
future.get();
}
catch (ExecutionException | InterruptedException e) {
throw new PrestoException(GENERIC_INTERNAL_ERROR, "Failed to process table", e);
}
}
}

2
presto-memory/src/test/java/io/prestosql/plugin/memory/MemoryQueryRunner.java
View File

@ -52,7 +52,7 @@ public final class MemoryQueryRunner
memoryProperties.put(FOLDER_PROPERTY_KEY, folder.newFolder("memory-connector").getAbsolutePath());
for (Map.Entry<String, String> entry : newConfigs.entrySet()) {
memoryProperties.replace(entry.getKey(), entry.getValue());
memoryProperties.put(entry.getKey(), entry.getValue());
}
return memoryProperties;

963
presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemorySelection.java
View File

@ -15,502 +15,479 @@
package io.prestosql.plugin.memory;
import com.google.common.collect.ImmutableMap;
import io.prestosql.testing.MaterializedResult;
import io.prestosql.testing.MaterializedRow;
import io.prestosql.tests.AbstractTestQueryFramework;
import org.intellij.lang.annotations.Language;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Locale;
import static io.prestosql.testing.assertions.Assert.assertEquals;
import static java.lang.String.format;
import static org.testng.Assert.assertTrue;
@Test(singleThreaded = true)
public class TestMemorySelection
// extends AbstractTestQueryFramework
extends AbstractTestQueryFramework
{
// // after a table is created, a background process processes each of the logiparts (sorts, creates index)
// // this can take some time. if tests fails try increasing this value
// long processingWait = 10000;
//
// public TestMemorySelection()
// {
// super(() -> MemoryQueryRunner.createQueryRunner(1, ImmutableMap.of(), ImmutableMap.of(), true));
// }
//
// @AfterMethod
// public void dropAllTables()
// throws InterruptedException
// {
// MaterializedResult tables = computeActual("SHOW TABLES");
// for (MaterializedRow row : tables.getMaterializedRows()) {
// assertQuerySucceeds("DROP TABLE IF EXISTS " + row.getField(0));
// }
// // Used to force workers to get an updated list of valid tables, so the old tables are deleted
// assertQuerySucceeds("CREATE TABLE nation AS SELECT * FROM tpch.tiny.nation");
//
// // Give workers time to delete their data
// Thread.sleep(processingWait);
//
// assertQuerySucceeds("DROP TABLE nation");
// }
//
// @Test
// public void testSortedBySelect()
// {
// assertUpdate("CREATE TABLE test_sort_select WITH (sorted_by=ARRAY['nationkey']) AS SELECT * FROM tpch.tiny.nation", "SELECT count(*) FROM nation");
//
// assertQuery("SELECT * FROM test_sort_select ORDER BY nationkey", "SELECT * FROM nation ORDER BY nationkey");
//
// assertQuery("SELECT * FROM test_sort_select WHERE nationkey = 3", "SELECT * FROM nation WHERE nationkey = 3");
//
// assertQueryResult("INSERT INTO test_sort_select SELECT * FROM tpch.tiny.nation", 25L);
//
// assertQueryResult("SELECT count(*) FROM test_sort_select", 50L);
//
// assertQuery("SELECT * FROM test_sort_select WHERE nationkey = 3",
// "SELECT * FROM nation WHERE nationkey = 3 UNION ALL SELECT * FROM nation WHERE nationkey = 3");
//
// assertQueryResult("INSERT INTO test_sort_select SELECT * FROM tpch.tiny.nation", 25L);
//
// assertQueryResult("SELECT count(*) FROM test_sort_select", 75L);
//
// assertQuery("SELECT * FROM test_sort_select WHERE nationkey = 3",
// "SELECT * FROM nation WHERE nationkey = 3 UNION ALL SELECT * FROM nation WHERE nationkey = 3 UNION ALL SELECT * FROM nation WHERE nationkey = 3");
// }
//
// @Test
// public void testIndexColumnsSelect()
// {
// assertUpdate("CREATE TABLE test_index_select WITH (index_columns=ARRAY['nationkey']) AS SELECT * FROM tpch.tiny.nation", "SELECT count(*) FROM nation");
//
// assertQuery("SELECT * FROM test_index_select ORDER BY nationkey", "SELECT * FROM nation ORDER BY nationkey");
//
// assertQuery("SELECT * FROM test_index_select WHERE nationkey = 3", "SELECT * FROM nation WHERE nationkey = 3");
//
// assertQueryResult("INSERT INTO test_index_select SELECT * FROM tpch.tiny.nation", 25L);
//
// assertQueryResult("SELECT count(*) FROM test_index_select", 50L);
//
// assertQuery("SELECT * FROM test_index_select WHERE nationkey = 3",
// "SELECT * FROM nation WHERE nationkey = 3 UNION ALL SELECT * FROM nation WHERE nationkey = 3");
//
// assertQueryResult("INSERT INTO test_index_select SELECT * FROM tpch.tiny.nation", 25L);
//
// assertQueryResult("SELECT count(*) FROM test_index_select", 75L);
//
// assertQuery("SELECT * FROM test_index_select WHERE nationkey = 3",
// "SELECT * FROM nation WHERE nationkey = 3 UNION ALL SELECT * FROM nation WHERE nationkey = 3 UNION ALL SELECT * FROM nation WHERE nationkey = 3");
// }
//
// @DataProvider(name = "memConIndexSingleOperators")
// public Object[][] memConIndexSingleOperators()
// {
// return new Object[][] {{"=", 1}, {">", 1}, {"<", 1}, {">=", 1}, {"<=", 1}, {"IN", 1}, {"BETWEEN", 2}};
// }
//
// @Test(dataProvider = "memConIndexSingleOperators")
// public void testIndexSingleOperators(String queryOperator, int testKeys)
// throws InterruptedException
// {
// assertQuerySucceeds("CREATE TABLE test_indexOperations WITH (sorted_by=ARRAY['custkey']) AS SELECT * FROM tpch.tiny.orders");
//
// // get custkeys
// List<Long> custkeys = new ArrayList<>();
// custkeys.add(1000L);
// custkeys.add(5000L);
// custkeys.add(7000L);
//
// String predicateQuery = "SELECT count(*) FROM test_indexOperations WHERE custkey ";
//
// if (queryOperator.toLowerCase(Locale.ROOT).contains("in")) {
// predicateQuery += queryOperator + " (";
// int custkeySize = custkeys.size();
// for (int i = 0; i < custkeySize; i++) {
// predicateQuery += " " + custkeys.get(i);
// if (i < custkeySize - 1) {
// predicateQuery += ", ";
// }
// }
// predicateQuery += ")";
// }
// else if (queryOperator.toLowerCase(Locale.ROOT).contains("between")) {
// predicateQuery += queryOperator + " ";
// long minkey = Math.min(custkeys.get(0), custkeys.get(1));
// long maxkey = Math.max(custkeys.get(0), custkeys.get(1));
// predicateQuery += minkey + " AND " + maxkey;
// }
// else {
// predicateQuery += queryOperator + " " + custkeys.get(0);
// }
//
// // apply one predicate and get input rows read, this should read all the rows
// long inputRowCountBefore = assertQuerySucceedsGetInputRows(predicateQuery);
//
// // wait for sorting and indexing to complete
// Thread.sleep(processingWait);
//
// // apply one predicate and get input rows read, this time since predicate is on sort column, rows should be reduced
// long inputRowCountOnePredicate = assertQuerySucceedsGetInputRows(predicateQuery);
//
// // Drop table for next use
// assertQuerySucceeds("DROP TABLE test_indexOperations");
//
// System.out.println(predicateQuery);
//
// assertTrue(inputRowCountBefore > inputRowCountOnePredicate, "inputRowCountBefore=" + inputRowCountBefore + " inputRowCountOnePredicate=" + inputRowCountOnePredicate);
// }
//
// @Test(dataProvider = "memConIndexSingleOperators")
// public void testIndexSingleOperatorsVerifyResults(String queryOperator, int testKeys)
// throws InterruptedException
// {
// assertQuerySucceeds("CREATE TABLE test_indexOperations WITH (sorted_by=ARRAY['custkey']) AS SELECT * FROM tpch.tiny.orders");
//
// // get custkeys
// List<Long> custkeys = new ArrayList<>();
// for (int i = 0; i < testKeys; i++) {
// custkeys.add((long) getSingleResult("SELECT custkey FROM test_indexOperations limit 1"));
// }
//
// String predicateQuery = "SELECT custkey FROM test_indexOperations WHERE custkey ";
//
// if (queryOperator.toLowerCase(Locale.ROOT).contains("in")) {
// predicateQuery += queryOperator + " (";
// int custkeySize = custkeys.size();
// for (int i = 0; i < custkeySize; i++) {
// predicateQuery += " " + custkeys.get(i);
// if (i < custkeySize - 1) {
// predicateQuery += ", ";
// }
// }
// predicateQuery += ")";
// }
// else if (queryOperator.toLowerCase(Locale.ROOT).contains("between")) {
// predicateQuery += queryOperator + " ";
// long minkey = Math.min(custkeys.get(0), custkeys.get(1));
// long maxkey = Math.max(custkeys.get(0), custkeys.get(1));
// predicateQuery += minkey + " AND " + maxkey;
// }
// else {
// predicateQuery += queryOperator + " " + custkeys.get(0);
// }
//
// MaterializedResult result1 = computeActual(predicateQuery);
//
// // wait for sorting and indexing to complete
// Thread.sleep(processingWait);
//
// MaterializedResult result2 = computeActual(predicateQuery);
//
// // Drop table for next use
// assertQuerySucceeds("DROP TABLE test_indexOperations");
//
// System.out.println(predicateQuery);
//
// ArrayList<String> data1 = new ArrayList<>();
// ArrayList<String> data2 = new ArrayList<>();
// for (MaterializedRow item1 : result1.getMaterializedRows()) {
// data1.add(item1.toString());
// }
// for (MaterializedRow item2 : result2.getMaterializedRows()) {
// data2.add(item2.toString());
// }
// Collections.sort(data1);
// Collections.sort(data2);
// System.out.println(data1.size());
// System.out.println();
// System.out.println(data2.size());
//
// assertEquals(data1, data2);
// }
//
// @DataProvider(name = "memConIndexSingleColMultiOperators")
// public Object[][] memConIndexSingleColMultiOperators()
// {
// return new Object[][] {
// {"custkey < # OR custkey > #"}, {"custkey > # AND custkey < #"},
// {"custkey <= # OR custkey >= #"}, {"custkey = # OR custkey IN (#)"},
// {"custkey = # AND custkey NOT IN (#)"}, {"custkey <> # AND custkey IN (#)"},
// {"custkey BETWEEN # AND # OR custkey = #"}, {"custkey BETWEEN # AND # OR custkey IN (#)"},
// {"custkey BETWEEN # AND # OR custkey BETWEEN # AND #"}};
// }
//
// @Test(dataProvider = "memConIndexSingleColMultiOperators")
// public void testIndexSingleColMultiOperators(String queryOperator)
// throws InterruptedException
// {
// assertQuerySucceeds("CREATE TABLE test_indexOperations WITH (sorted_by=ARRAY['custkey']) AS SELECT * FROM tpch.tiny.orders");
//
// // get how many test keys there are
// long testKeys = queryOperator.chars().filter(c -> c == '#').count();
//
// // get custkeys
// List<Long> custkeys = new ArrayList<>();
// custkeys.add(1000L);
// custkeys.add(5000L);
// custkeys.add(7000L);
// custkeys.add(9000L);
//
// for (Long key : custkeys) {
// queryOperator = queryOperator.replaceFirst("#", String.valueOf(key));
// }
//
// String predicateQuery = "SELECT count(distinct custkey) FROM test_indexOperations WHERE " + queryOperator;
//
// // apply one predicate and get input rows read, this should read all the rows
// long inputRowCountBefore = assertQuerySucceedsGetInputRows(predicateQuery);
//
// // wait for sorting and indexing to complete
// Thread.sleep(processingWait);
//
// // apply one predicate and get input rows read, this time since predicate is on sort column, rows should be reduced
// long inputRowCountOnePredicate = assertQuerySucceedsGetInputRows(predicateQuery);
//
// // Drop table for next use
// assertQuerySucceeds("DROP TABLE test_indexOperations");
//
// assertTrue(inputRowCountBefore > inputRowCountOnePredicate, "inputRowCountBefore=" + inputRowCountBefore + " inputRowCountOnePredicate=" + inputRowCountOnePredicate);
// }
//
// @Test(dataProvider = "memConIndexSingleColMultiOperators")
// public void testIndexSingleColMultiOperatorsVerifyResults(String queryOperator)
// throws InterruptedException
// {
// assertQuerySucceeds("CREATE TABLE test_indexOperations WITH (sorted_by=ARRAY['custkey']) AS SELECT * FROM tpch.tiny.orders");
//
// // get how many test keys there are
// long testKeys = queryOperator.chars().filter(c -> c == '#').count();
//
// // get custkeys
// List<Object> results = getResults("SELECT * from (SELECT custkey FROM test_indexOperations LIMIT " + testKeys + ") ORDER BY custkey ASC");
// for (Object key : results) {
// queryOperator = queryOperator.replaceFirst("#", String.valueOf((long) key));
// }
//
// String predicateQuery = "SELECT count(distinct custkey) FROM test_indexOperations WHERE " + queryOperator;
//
// MaterializedResult result1 = computeActual(predicateQuery);
//
// // wait for sorting and indexing to complete
// Thread.sleep(processingWait);
//
// MaterializedResult result2 = computeActual(predicateQuery);
//
// // Drop table for next use
// assertQuerySucceeds("DROP TABLE test_indexOperations");
//
// System.out.println(predicateQuery);
//
// ArrayList<String> data1 = new ArrayList<>();
// ArrayList<String> data2 = new ArrayList<>();
// for (MaterializedRow item1 : result1.getMaterializedRows()) {
// data1.add(item1.toString());
// }
// for (MaterializedRow item2 : result2.getMaterializedRows()) {
// data2.add(item2.toString());
// }
// Collections.sort(data1);
// Collections.sort(data2);
// System.out.println(data1.size());
// System.out.println();
// System.out.println(data2.size());
//
// assertEquals(data1, data2);
// }
//
// @Test
// public void testSortByInputRowCount()
// throws InterruptedException
// {
// assertQuerySucceeds("CREATE TABLE test_sorting WITH (sorted_by=ARRAY['orderkey']) AS SELECT * FROM tpch.tiny.orders");
//
// // get one of the orderkey
// Object val = getSingleResult("SELECT orderkey FROM test_sorting limit 1");
// long orderkey = (long) val;
//
// // apply predicate and get input rows read, this should read all the rows
// long inputRowCountBefore = assertQueryResultGetInputRows("SELECT count(*) FROM test_sorting WHERE orderkey = " + orderkey, 1L);
//
// // wait for sorting and indexing to complete
// Thread.sleep(processingWait);
//
// // apply predicate and get input rows read, this time since predicate is on sorted column, rows should be reduced
// long inputRowCountAfter = assertQueryResultGetInputRows("SELECT count(*) FROM test_sorting WHERE orderkey = " + orderkey, 1L);
//
// assertTrue(inputRowCountBefore > inputRowCountAfter, "inputRowCountBefore=" + inputRowCountBefore + " inputRowCountAfter=" + inputRowCountAfter);
//
// // apply another predicate and get input rows read
// // the predicate is on sort column, but its outside the value range so minmax index should reduce rows further
// long inputRowCountInvalidValue = assertQueryResultGetInputRows("SELECT count(*) FROM test_sorting WHERE orderkey = 1000000000", 0L);
//
// assertTrue(inputRowCountAfter > inputRowCountInvalidValue, "inputRowCountAfter=" + inputRowCountAfter + " inputRowCountInvalidValue=" + inputRowCountInvalidValue);
//
// System.out.println("inputRowCountBefore=" + inputRowCountBefore + " inputRowCountAfter=" + inputRowCountAfter + " inputRowCountInvalidValue=" + inputRowCountInvalidValue);
// }
//
// @Test
// public void testIndexInputRowCount()
// throws InterruptedException
// {
// assertQuerySucceeds("CREATE TABLE test_index WITH (index_columns=ARRAY['orderkey']) AS SELECT * FROM tpch.tiny.orders");
//
// // get one of the orderkey
// Object val = getSingleResult("SELECT orderkey FROM test_index limit 1");
// long orderkey = (long) val;
//
// // apply predicate and get input rows read, this should read all the rows
// long inputRowCountBefore = assertQueryResultGetInputRows("SELECT count(*) FROM test_index WHERE orderkey = " + orderkey, 1L);
//
// // wait for indexing to complete
// Thread.sleep(processingWait);
//
// // apply predicate and get input rows read, this time since predicate is on index column, rows should be reduced
// long inputRowCountAfter = assertQueryResultGetInputRows("SELECT count(*) FROM test_index WHERE orderkey = " + orderkey, 1L);
//
// assertTrue(inputRowCountBefore > inputRowCountAfter, "inputRowCountBefore=" + inputRowCountBefore + " inputRowCountAfter=" + inputRowCountAfter);
//
// // apply another predicate and get input rows read
// // the predicate is on index column, but its outside the value range so minmax index should reduce rows further
// long inputRowCountInvalidValue = assertQueryResultGetInputRows("SELECT count(*) FROM test_index WHERE orderkey = 1000000000", 0L);
//
// assertTrue(inputRowCountAfter > inputRowCountInvalidValue, "inputRowCountAfter=" + inputRowCountAfter + " inputRowCountInvalidValue=" + inputRowCountInvalidValue);
//
// System.out.println("inputRowCountBefore=" + inputRowCountBefore + " inputRowCountAfter=" + inputRowCountAfter + " inputRowCountInvalidValue=" + inputRowCountInvalidValue);
// }
//
// @Test
// public void testSortAndIndexInputRowCount()
// throws InterruptedException
// {
// assertQuerySucceeds("CREATE TABLE test_sortindex WITH (sorted_by=ARRAY['custkey'], index_columns=ARRAY['orderkey']) AS SELECT * FROM tpch.tiny.orders");
//
// // get one of the custkey
// Object val = getSingleResult("SELECT custkey FROM test_sortindex limit 1");
// long custkey = (long) val;
//
// // apply one predicate and get input rows read, this should read all the rows
// long inputRowCountBefore = assertQuerySucceedsGetInputRows(
// "SELECT count(*) FROM test_sortindex WHERE custkey = " + custkey);
//
// // wait for sorting and indexing to complete
// Thread.sleep(processingWait);
//
// // apply one predicate and get input rows read, this time since predicate is on sort column, rows should be reduced
// long inputRowCountOnePredicate = assertQuerySucceedsGetInputRows(
// "SELECT count(*) FROM test_sortindex WHERE custkey = " + custkey);
//
// assertTrue(inputRowCountBefore > inputRowCountOnePredicate, "inputRowCountBefore=" + inputRowCountBefore + " inputRowCountOnePredicate=" + inputRowCountOnePredicate);
//
// // get one of the orderkeys for the custkey
// Object val2 = getSingleResult("SELECT orderkey FROM test_sortindex WHERE custkey = " + custkey);
// long orderKey = (long) val2;
//
// // apply two predicates and get input rows read, this time since predicates are on both sort column and index column, rows should be reduced further
// long inputRowCountTwoPredicates = assertQuerySucceedsGetInputRows(
// String.format("SELECT count(*) FROM test_sortindex WHERE custkey = %s and orderKey = %s", custkey, orderKey));
//
// assertTrue(inputRowCountOnePredicate > inputRowCountTwoPredicates, "inputRowCountOnePredicate=" + inputRowCountOnePredicate + " inputRowCountTwoPredicates=" + inputRowCountTwoPredicates);
// }
//
// @Test
// public void testFilteringBenchmark()
// throws InterruptedException
// {
// assertQuerySucceeds("CREATE TABLE test_filter_bench WITH (sorted_by=ARRAY['custkey'], index_columns=ARRAY['orderkey']) AS SELECT * FROM tpch.tiny.orders");
//
// // get one of the custkey
// Object val = getSingleResult("SELECT custkey FROM test_filter_bench limit 1");
// long custkey = (long) val;
//
// // get one of the orderkeys for the custkey
// Object val2 = getSingleResult("SELECT orderkey FROM test_filter_bench WHERE custkey = " + custkey);
// long orderKey = (long) val2;
//
// // wait for sorting and indexing to complete
// Thread.sleep(processingWait);
//
// Thread[] threads = new Thread[25];
//
// for (int i = 0; i < threads.length; i++) {
// threads[i] = new Thread(() ->
// assertQuerySucceedsGetInputRows(String.format("SELECT count(*) FROM test_filter_bench WHERE custkey = %s and orderKey = %s", custkey, orderKey)));
// }
//
// long before = System.currentTimeMillis();
// for (Thread thread : threads) {
// thread.start();
// }
//
// for (Thread thread : threads) {
// thread.join();
// }
//
// System.out.println(Double.valueOf(System.currentTimeMillis() - before) / threads.length);
// }
//
// private void assertQueryResult(@Language("SQL") String sql, Object... expected)
// {
// MaterializedResult rows = computeActual(sql);
// assertEquals(rows.getRowCount(), expected.length);
//
// for (int i = 0; i < expected.length; i++) {
// MaterializedRow materializedRow = rows.getMaterializedRows().get(i);
// int fieldCount = materializedRow.getFieldCount();
// assertTrue(fieldCount == 1, format("Expected only one column, but got '%d'", fieldCount));
// Object value = materializedRow.getField(0);
// assertEquals(value, expected[i]);
// assertTrue(materializedRow.getFieldCount() == 1);
// }
// }
//
// private long assertQueryResultGetInputRows(@Language("SQL") String sql, Object... expected)
// {
// assertQueryResult(sql, expected);
//
// return getInputRowsOfLastQueryExecution(sql);
// }
//
// private long assertQuerySucceedsGetInputRows(@Language("SQL") String sql, Object... expected)
// {
// assertQuerySucceeds(sql);
//
// return getInputRowsOfLastQueryExecution(sql);
// }
//
// private long getInputRowsOfLastQueryExecution(@Language("SQL") String sql)
// {
// String inputRowsSql = "select sum(raw_input_rows) from system.runtime.tasks where query_id in (select query_id from system.runtime.queries where query='" + sql + "' order by created desc limit 1)";
//
// MaterializedResult rows = computeActual(inputRowsSql);
//
// assertEquals(rows.getRowCount(), 1);
//
// MaterializedRow materializedRow = rows.getMaterializedRows().get(0);
// int fieldCount = materializedRow.getFieldCount();
// assertTrue(fieldCount == 1, format("Expected only one column, but got '%d'", fieldCount));
// Object value = materializedRow.getField(0);
//
// return (long) value;
// }
//
// private Object getSingleResult(@Language("SQL") String sql)
// {
// MaterializedResult rows = computeActual(sql);
// assertTrue(rows.getRowCount() > 0);
//
// MaterializedRow materializedRow = rows.getMaterializedRows().get(0);
// int fieldCount = materializedRow.getFieldCount();
// assertTrue(fieldCount == 1, format("Expected only one column, but got '%d'", fieldCount));
// Object value = materializedRow.getField(0);
//
// return value;
// }
//
// private List<Object> getResults(@Language("SQL") String sql)
// {
// MaterializedResult rows = computeActual(sql);
//
// List<Object> values = new ArrayList<>();
//
// for (MaterializedRow materializedRow : rows.getMaterializedRows()) {
// int fieldCount = materializedRow.getFieldCount();
// assertTrue(fieldCount == 1, format("Expected only one column, but got '%d'", fieldCount));
// values.add(materializedRow.getField(0));
// }
//
// return values;
// }
private static final String maxLocalPartSize = "64MB";
public TestMemorySelection()
{
super(() -> MemoryQueryRunner.createQueryRunner(2, ImmutableMap.of("task.writer-count", "8"), ImmutableMap.of("memory.max-logical-part-size", maxLocalPartSize), true));
}
@AfterMethod
public void dropAllTables()
{
MaterializedResult tables = computeActual("SHOW TABLES");
for (MaterializedRow row : tables.getMaterializedRows()) {
assertQuerySucceeds("DROP TABLE IF EXISTS " + row.getField(0));
}
}
@Test
public void testSortedBySelect()
{
assertUpdate("CREATE TABLE test_sort_select WITH (sorted_by=ARRAY['nationkey']) AS SELECT * FROM tpch.tiny.nation", "SELECT count(*) FROM nation");
assertQuery("SELECT * FROM test_sort_select ORDER BY nationkey", "SELECT * FROM nation ORDER BY nationkey");
assertQuery("SELECT * FROM test_sort_select WHERE nationkey = 3", "SELECT * FROM nation WHERE nationkey = 3");
assertQueryResult("INSERT INTO test_sort_select SELECT * FROM tpch.tiny.nation", 25L);
assertQueryResult("SELECT count(*) FROM test_sort_select", 50L);
assertQuery("SELECT * FROM test_sort_select WHERE nationkey = 3",
"SELECT * FROM nation WHERE nationkey = 3 UNION ALL SELECT * FROM nation WHERE nationkey = 3");
assertQueryResult("INSERT INTO test_sort_select SELECT * FROM tpch.tiny.nation", 25L);
assertQueryResult("SELECT count(*) FROM test_sort_select", 75L);
assertQuery("SELECT * FROM test_sort_select WHERE nationkey = 3",
"SELECT * FROM nation WHERE nationkey = 3 UNION ALL SELECT * FROM nation WHERE nationkey = 3 UNION ALL SELECT * FROM nation WHERE nationkey = 3");
}
@Test
public void testIndexColumnsSelect()
{
assertUpdate("CREATE TABLE test_index_select WITH (index_columns=ARRAY['nationkey']) AS SELECT * FROM tpch.tiny.nation", "SELECT count(*) FROM nation");
assertQuery("SELECT * FROM test_index_select ORDER BY nationkey", "SELECT * FROM nation ORDER BY nationkey");
assertQuery("SELECT * FROM test_index_select WHERE nationkey = 3", "SELECT * FROM nation WHERE nationkey = 3");
assertQueryResult("INSERT INTO test_index_select SELECT * FROM tpch.tiny.nation", 25L);
assertQueryResult("SELECT count(*) FROM test_index_select", 50L);
assertQuery("SELECT * FROM test_index_select WHERE nationkey = 3",
"SELECT * FROM nation WHERE nationkey = 3 UNION ALL SELECT * FROM nation WHERE nationkey = 3");
assertQueryResult("INSERT INTO test_index_select SELECT * FROM tpch.tiny.nation", 25L);
assertQueryResult("SELECT count(*) FROM test_index_select", 75L);
assertQuery("SELECT * FROM test_index_select WHERE nationkey = 3",
"SELECT * FROM nation WHERE nationkey = 3 UNION ALL SELECT * FROM nation WHERE nationkey = 3 UNION ALL SELECT * FROM nation WHERE nationkey = 3");
}
@DataProvider(name = "memConIndexSingleOperators")
public Object[][] memConIndexSingleOperators()
{
return new Object[][] {{"=", 1}, {">", 1}, {"<", 1}, {">=", 1}, {"<=", 1}, {"IN", 1}, {"BETWEEN", 2}};
}
@Test(dataProvider = "memConIndexSingleOperators")
public void testIndexSingleOperators(String queryOperator, int testKeys)
{
assertQuerySucceeds("CREATE TABLE test_indexOperations WITH (sorted_by=ARRAY['custkey'], async_processing=false) AS SELECT * FROM tpch.tiny.orders");
// get custkeys
List<Long> custkeys = new ArrayList<>();
custkeys.add(1000L);
custkeys.add(5000L);
custkeys.add(7000L);
String predicateQuery = "SELECT count(*) FROM test_indexOperations WHERE custkey ";
if (queryOperator.toLowerCase(Locale.ROOT).contains("in")) {
predicateQuery += queryOperator + " (";
int custkeySize = custkeys.size();
for (int i = 0; i < custkeySize; i++) {
predicateQuery += " " + custkeys.get(i);
if (i < custkeySize - 1) {
predicateQuery += ", ";
}
}
predicateQuery += ")";
}
else if (queryOperator.toLowerCase(Locale.ROOT).contains("between")) {
predicateQuery += queryOperator + " ";
long minkey = Math.min(custkeys.get(0), custkeys.get(1));
long maxkey = Math.max(custkeys.get(0), custkeys.get(1));
predicateQuery += minkey + " AND " + maxkey;
}
else {
predicateQuery += queryOperator + " " + custkeys.get(0);
}
// get total number of rows
long totalRows = assertQuerySucceedsGetInputRows("SELECT count(*) FROM test_indexOperations");
// apply one predicate and get input rows read, this time since predicate is on sort column, rows should be reduced
long inputRowCountOnePredicate = assertQuerySucceedsGetInputRows(predicateQuery);
// Drop table for next use
assertQuerySucceeds("DROP TABLE test_indexOperations");
System.out.println(predicateQuery);
assertTrue(totalRows > inputRowCountOnePredicate, "totalRows=" + totalRows + " inputRowCountOnePredicate=" + inputRowCountOnePredicate);
}
@Test(dataProvider = "memConIndexSingleOperators")
public void testIndexSingleOperatorsVerifyResults(String queryOperator, int testKeys)
{
assertQuerySucceeds("CREATE TABLE test_indexOperations WITH (sorted_by=ARRAY['custkey'], async_processing=false) AS SELECT * FROM tpch.tiny.orders");
// get custkeys
List<Long> custkeys = new ArrayList<>();
for (int i = 0; i < testKeys; i++) {
custkeys.add((long) getSingleResult("SELECT custkey FROM test_indexOperations limit 1"));
}
String predicateQuery = "SELECT custkey FROM test_indexOperations WHERE custkey ";
if (queryOperator.toLowerCase(Locale.ROOT).contains("in")) {
predicateQuery += queryOperator + " (";
int custkeySize = custkeys.size();
for (int i = 0; i < custkeySize; i++) {
predicateQuery += " " + custkeys.get(i);
if (i < custkeySize - 1) {
predicateQuery += ", ";
}
}
predicateQuery += ")";
}
else if (queryOperator.toLowerCase(Locale.ROOT).contains("between")) {
predicateQuery += queryOperator + " ";
long minkey = Math.min(custkeys.get(0), custkeys.get(1));
long maxkey = Math.max(custkeys.get(0), custkeys.get(1));
predicateQuery += minkey + " AND " + maxkey;
}
else {
predicateQuery += queryOperator + " " + custkeys.get(0);
}
MaterializedResult result1 = computeActual(predicateQuery.replace("test_indexOperations", "tpch.tiny.orders"));
MaterializedResult result2 = computeActual(predicateQuery);
// Drop table for next use
assertQuerySucceeds("DROP TABLE test_indexOperations");
System.out.println(predicateQuery);
ArrayList<String> data1 = new ArrayList<>();
ArrayList<String> data2 = new ArrayList<>();
for (MaterializedRow item1 : result1.getMaterializedRows()) {
data1.add(item1.toString());
}
for (MaterializedRow item2 : result2.getMaterializedRows()) {
data2.add(item2.toString());
}
Collections.sort(data1);
Collections.sort(data2);
System.out.println(data1.size());
System.out.println();
System.out.println(data2.size());
assertEquals(data1, data2);
}
@DataProvider(name = "memConIndexSingleColMultiOperators")
public Object[][] memConIndexSingleColMultiOperators()
{
return new Object[][] {
{"custkey < # OR custkey > #"}, {"custkey > # AND custkey < #"},
{"custkey <= # OR custkey >= #"}, {"custkey = # OR custkey IN (#)"},
{"custkey = # AND custkey NOT IN (#)"}, {"custkey <> # AND custkey IN (#)"},
{"custkey BETWEEN # AND # OR custkey = #"}, {"custkey BETWEEN # AND # OR custkey IN (#)"},
{"custkey BETWEEN # AND # OR custkey BETWEEN # AND #"}};
}
@Test(dataProvider = "memConIndexSingleColMultiOperators")
public void testIndexSingleColMultiOperators(String queryOperator)
throws InterruptedException
{
assertQuerySucceeds("CREATE TABLE test_indexOperations WITH (sorted_by=ARRAY['custkey'], async_processing=false) AS SELECT * FROM tpch.tiny.orders");
// get how many test keys there are
long testKeys = queryOperator.chars().filter(c -> c == '#').count();
// get custkeys
List<Long> custkeys = new ArrayList<>();
custkeys.add(1000L);
custkeys.add(5000L);
custkeys.add(7000L);
custkeys.add(9000L);
for (Long key : custkeys) {
queryOperator = queryOperator.replaceFirst("#", String.valueOf(key));
}
String predicateQuery = "SELECT count(distinct custkey) FROM test_indexOperations WHERE " + queryOperator;
// get total number rows
long totalRows = assertQuerySucceedsGetInputRows("SELECT count(*) FROM test_indexOperations");
// apply one predicate and get input rows read, this time since predicate is on sort column, rows should be reduced
long inputRowCountOnePredicate = assertQuerySucceedsGetInputRows(predicateQuery);
// Drop table for next use
assertQuerySucceeds("DROP TABLE test_indexOperations");
assertTrue(totalRows > inputRowCountOnePredicate, "totalRows=" + totalRows + " inputRowCountOnePredicate=" + inputRowCountOnePredicate);
}
@Test(dataProvider = "memConIndexSingleColMultiOperators")
public void testIndexSingleColMultiOperatorsVerifyResults(String queryOperator)
throws InterruptedException
{
assertQuerySucceeds("CREATE TABLE test_indexOperations WITH (sorted_by=ARRAY['custkey'], async_processing=false) AS SELECT * FROM tpch.tiny.orders");
// get how many test keys there are
long testKeys = queryOperator.chars().filter(c -> c == '#').count();
// get custkeys
List<Object> results = getResults("SELECT * from (SELECT custkey FROM test_indexOperations LIMIT " + testKeys + ") ORDER BY custkey ASC");
for (Object key : results) {
queryOperator = queryOperator.replaceFirst("#", String.valueOf((long) key));
}
String predicateQuery = "SELECT count(distinct custkey) FROM test_indexOperations WHERE " + queryOperator;
MaterializedResult result1 = computeActual(predicateQuery.replace("test_indexOperations", "tpch.tiny.orders"));
MaterializedResult result2 = computeActual(predicateQuery);
// Drop table for next use
assertQuerySucceeds("DROP TABLE test_indexOperations");
System.out.println(predicateQuery);
ArrayList<String> data1 = new ArrayList<>();
ArrayList<String> data2 = new ArrayList<>();
for (MaterializedRow item1 : result1.getMaterializedRows()) {
data1.add(item1.toString());
}
for (MaterializedRow item2 : result2.getMaterializedRows()) {
data2.add(item2.toString());
}
Collections.sort(data1);
Collections.sort(data2);
System.out.println(data1.size());
System.out.println();
System.out.println(data2.size());
assertEquals(data1, data2);
}
@Test
public void testSortByInputRowCount()
throws InterruptedException
{
assertQuerySucceeds("CREATE TABLE test_sorting WITH (sorted_by=ARRAY['orderkey'], async_processing=false) AS SELECT * FROM tpch.tiny.orders");
// get one of the orderkey
Object val = getSingleResult("SELECT orderkey FROM test_sorting limit 1");
long orderkey = (long) val;
// get total number of rows
long totalRows = (long) getSingleResult("SELECT count(*) FROM test_sorting");
// apply predicate and get input rows read, this time since predicate is on sorted column, rows should be reduced
long inputRowCountAfter = assertQueryResultGetInputRows("SELECT count(*) FROM test_sorting WHERE orderkey = " + orderkey, 1L);
assertTrue(totalRows > inputRowCountAfter, "totalRows=" + totalRows + " inputRowCountAfter=" + inputRowCountAfter);
// apply another predicate and get input rows read
// the predicate is on sort column, but its outside the value range so minmax index should reduce rows further
long inputRowCountInvalidValue = assertQueryResultGetInputRows("SELECT count(*) FROM test_sorting WHERE orderkey = 1000000000", 0L);
assertTrue(inputRowCountAfter > inputRowCountInvalidValue, "inputRowCountAfter=" + inputRowCountAfter + " inputRowCountInvalidValue=" + inputRowCountInvalidValue);
System.out.println("totalRows=" + totalRows + " inputRowCountAfter=" + inputRowCountAfter + " inputRowCountInvalidValue=" + inputRowCountInvalidValue);
}
@Test
public void testIndexInputRowCount()
{
assertQuerySucceeds("CREATE TABLE test_index WITH (index_columns=ARRAY['orderkey'], async_processing=false) AS SELECT * FROM tpch.sf1.orders");
// get one of the orderkey
Object val = getSingleResult("SELECT orderkey FROM test_index limit 1");
long orderkey = (long) val;
// get total number of rows
long totalRows = (long) getSingleResult("SELECT count(*) FROM test_index");
// apply predicate and get input rows read, this time since predicate is on index column, rows should be reduced
long inputRowCountAfter = assertQueryResultGetInputRows("SELECT count(*) FROM test_index WHERE orderkey = " + orderkey, 1L);
assertTrue(totalRows > inputRowCountAfter, "totalRows=" + totalRows + " inputRowCountAfter=" + inputRowCountAfter);
// apply another predicate and get input rows read
// the predicate is on index column, but its outside the value range so minmax index should reduce rows further
long inputRowCountInvalidValue = assertQueryResultGetInputRows("SELECT count(*) FROM test_index WHERE orderkey = 1000000000", 0L);
assertTrue(inputRowCountAfter > inputRowCountInvalidValue, "inputRowCountAfter=" + inputRowCountAfter + " inputRowCountInvalidValue=" + inputRowCountInvalidValue);
System.out.println("totalRows=" + totalRows + " inputRowCountAfter=" + inputRowCountAfter + " inputRowCountInvalidValue=" + inputRowCountInvalidValue);
}
@Test
public void testSortAndIndexInputRowCount()
{
assertQuerySucceeds("CREATE TABLE test_sortindex WITH (sorted_by=ARRAY['custkey'], index_columns=ARRAY['orderkey'], async_processing=false) AS SELECT * FROM tpch.sf1.orders");
// get one of the orderkey
Object val = getSingleResult("SELECT orderkey FROM test_sortindex limit 1");
long orderkey = (long) val;
// get total input rows, this should read all the rows
long totalRows = (long) getSingleResult("SELECT count(*) FROM test_sortindex");
// apply one predicate and get input rows read, this time since predicate is on index column, rows should be reduced
long inputRowCountOnePredicate = assertQuerySucceedsGetInputRows(
"SELECT count(*) FROM test_sortindex WHERE orderkey = " + orderkey);
assertTrue(totalRows > inputRowCountOnePredicate, "totalRows=" + totalRows + " inputRowCountOnePredicate=" + inputRowCountOnePredicate);
// get one of the custkeys for the orderkey
Object val2 = getSingleResult("SELECT custkey FROM test_sortindex WHERE orderkey = " + orderkey);
long custkey = (long) val2;
// apply two predicates and get input rows read, this time since predicates are on both sort column and index column, rows should be reduced further
long inputRowCountTwoPredicates = assertQuerySucceedsGetInputRows(
String.format("SELECT count(*) FROM test_sortindex WHERE custkey = %s and orderKey = %s", custkey, orderkey));
assertTrue(inputRowCountOnePredicate > inputRowCountTwoPredicates, "inputRowCountOnePredicate=" + inputRowCountOnePredicate + " inputRowCountTwoPredicates=" + inputRowCountTwoPredicates);
}
@Test
public void testFilteringBenchmark()
throws InterruptedException
{
assertQuerySucceeds("CREATE TABLE test_filter_bench WITH (sorted_by=ARRAY['custkey'], index_columns=ARRAY['orderkey'], async_processing=false) AS SELECT * FROM tpch.tiny.orders");
// get one of the custkey
Object val = getSingleResult("SELECT custkey FROM test_filter_bench limit 1");
long custkey = (long) val;
// get one of the orderkeys for the custkey
Object val2 = getSingleResult("SELECT orderkey FROM test_filter_bench WHERE custkey = " + custkey);
long orderKey = (long) val2;
Thread[] threads = new Thread[25];
for (int i = 0; i < threads.length; i++) {
threads[i] = new Thread(() ->
assertQuerySucceedsGetInputRows(String.format("SELECT count(*) FROM test_filter_bench WHERE custkey = %s and orderKey = %s", custkey, orderKey)));
}
long before = System.currentTimeMillis();
for (Thread thread : threads) {
thread.start();
}
for (Thread thread : threads) {
thread.join();
}
System.out.println(Double.valueOf(System.currentTimeMillis() - before) / threads.length);
}
private void assertQueryResult(@Language("SQL") String sql, Object... expected)
{
MaterializedResult rows = computeActual(sql);
assertEquals(rows.getRowCount(), expected.length);
for (int i = 0; i < expected.length; i++) {
MaterializedRow materializedRow = rows.getMaterializedRows().get(i);
int fieldCount = materializedRow.getFieldCount();
assertTrue(fieldCount == 1, format("Expected only one column, but got '%d'", fieldCount));
Object value = materializedRow.getField(0);
assertEquals(value, expected[i]);
assertTrue(materializedRow.getFieldCount() == 1);
}
}
private long assertQueryResultGetInputRows(@Language("SQL") String sql, Object... expected)
{
assertQueryResult(sql, expected);
return getInputRowsOfLastQueryExecution(sql);
}
private long assertQuerySucceedsGetInputRows(@Language("SQL") String sql, Object... expected)
{
assertQuerySucceeds(sql);
return getInputRowsOfLastQueryExecution(sql);
}
private long getInputRowsOfLastQueryExecution(@Language("SQL") String sql)
{
String inputRowsSql = "select sum(raw_input_rows) from system.runtime.tasks where query_id in (select query_id from system.runtime.queries where query='" + sql + "' order by created desc limit 1)";
MaterializedResult rows = computeActual(inputRowsSql);
assertEquals(rows.getRowCount(), 1);
MaterializedRow materializedRow = rows.getMaterializedRows().get(0);
int fieldCount = materializedRow.getFieldCount();
assertTrue(fieldCount == 1, format("Expected only one column, but got '%d'", fieldCount));
Object value = materializedRow.getField(0);
return (long) value;
}
private Object getSingleResult(@Language("SQL") String sql)
{
MaterializedResult rows = computeActual(sql);
assertTrue(rows.getRowCount() > 0);
MaterializedRow materializedRow = rows.getMaterializedRows().get(0);
int fieldCount = materializedRow.getFieldCount();
assertTrue(fieldCount == 1, format("Expected only one column, but got '%d'", fieldCount));
Object value = materializedRow.getField(0);
return value;
}
private List<Object> getResults(@Language("SQL") String sql)
{
MaterializedResult rows = computeActual(sql);
List<Object> values = new ArrayList<>();
for (MaterializedRow materializedRow : rows.getMaterializedRows()) {
int fieldCount = materializedRow.getFieldCount();
assertTrue(fieldCount == 1, format("Expected only one column, but got '%d'", fieldCount));
values.add(materializedRow.getField(0));
}
return values;
}
}

304
presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemorySpilling.java
View File

@ -14,165 +14,157 @@
*/
package io.prestosql.plugin.memory;
import com.google.common.collect.ImmutableMap;
import io.prestosql.testing.MaterializedResult;
import io.prestosql.testing.MaterializedRow;
import io.prestosql.tests.AbstractTestQueryFramework;
import org.intellij.lang.annotations.Language;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@Test(singleThreaded = true)
public class TestMemorySpilling
// extends AbstractTestQueryFramework
extends AbstractTestQueryFramework
{
// private static final String maxDataPerNode = "8MB";
// private static final int processingDelay = 10000;
// private static final int numLoops = 3;
//
// public TestMemorySpilling()
// {
// super(() -> MemoryQueryRunner.createQueryRunner(1, ImmutableMap.of(), ImmutableMap.of("memory.max-data-per-node", maxDataPerNode), false));
// }
//
// @AfterMethod
// public void dropAllTables()
// throws InterruptedException
// {
// MaterializedResult tables = computeActual("SHOW TABLES");
// for (MaterializedRow row : tables.getMaterializedRows()) {
// assertQuerySucceeds("DROP TABLE IF EXISTS " + row.getField(0));
// }
// // Used to force workers to get an updated list of valid tables, so the old tables are deleted
// assertQuerySucceeds("CREATE TABLE nation AS SELECT * FROM tpch.tiny.nation");
//
// // Give workers time to delete their data
// Thread.sleep(processingDelay);
//
// assertQuerySucceeds("DROP TABLE nation");
// }
//
// @Test
// public void testEvictionCreate()
// throws InterruptedException
// {
// synchronized (this) {
// try {
// // create one orders table
// assertQuerySucceeds("CREATE TABLE test_t1_1 AS SELECT * FROM tpch.tiny.orders");
// Thread.sleep(processingDelay);
//
// // without memory eviction this table creation would not succeed
// // if this query passes then the eviction is functioning well
// assertQuerySucceeds("CREATE TABLE test_t1_2 AS SELECT * FROM tpch.tiny.orders");
//
// // creation of a table over total limit. even with eviction this should fail
// assertQueryFails("CREATE TABLE test_t1_3 AS SELECT * FROM tpch.tiny.lineitem", "Memory limit \\[.+\\] for memory connector exceeded. Current: \\[.+\\]. Requested: \\[.+\\]");
// }
// finally {
// assertQuerySucceeds("DROP TABLE IF EXISTS test_t1_1");
// assertQuerySucceeds("DROP TABLE IF EXISTS test_t1_2");
// assertQuerySucceeds("DROP TABLE IF EXISTS test_t1_3");
// Thread.sleep(processingDelay);
// }
// }
// }
//
// @Test
// public void testEvictionSelect()
// throws InterruptedException
// {
// synchronized (this) {
// try {
// assertQuerySucceeds("CREATE TABLE test_t2_1 AS SELECT * FROM tpch.tiny.orders");
// Thread.sleep(processingDelay);
//
// assertQuerySucceeds("CREATE TABLE test_t2_2 AS SELECT * FROM tpch.tiny.orders");
// Thread.sleep(processingDelay);
//
// assertQuerySucceeds("SELECT COUNT(*) FROM test_t2_1");
// assertQuerySucceeds("SELECT COUNT(*) FROM test_t2_2");
// }
// finally {
// assertQuerySucceeds("DROP TABLE IF EXISTS test_t2_1");
// assertQuerySucceeds("DROP TABLE IF EXISTS test_t2_2");
// }
// }
// }
//
// @Test
// public void testConcurrentEvictionSelect()
// throws Throwable
// {
// int numTables = 3;
// synchronized (this) {
// try {
// Thread.sleep(processingDelay);
// // create the tables
// for (int i = 0; i < numTables; i++) {
// assertQuerySucceeds("CREATE TABLE test_t3_" + i + " AS SELECT * FROM tpch.tiny.orders");
// Thread.sleep(processingDelay);
// }
// QueryThread[] threads = new QueryThread[numTables];
// QueryThreadExceptionHandler[] threadExceptionHandlers = new QueryThreadExceptionHandler[numTables];
// // define the query threads
// for (int i = 0; i < numTables; i++) {
// threads[i] = new QueryThread("SELECT COUNT(*) FROM test_t3_" + i);
// threadExceptionHandlers[i] = new QueryThreadExceptionHandler();
// threads[i].setUncaughtExceptionHandler(threadExceptionHandlers[i]);
// }
//
// // start the query threads
// for (int i = 0; i < numTables; i++) {
// threads[i].start();
// }
//
// // wait till all the query threads are finished
// for (int i = 0; i < numTables; i++) {
// threads[i].join();
// }
//
// // catch any exceptions from the threads
// for (int i = 0; i < numTables; i++) {
// if (threadExceptionHandlers[i].exception != null) {
// throw threads[i].exception.getCause();
// }
// }
// }
// finally {
// // delete tables
// for (int i = 0; i < numTables; i++) {
// assertQuerySucceeds("DROP TABLE IF EXISTS test_t3_" + i);
// }
// }
// }
// }
//
// class QueryThreadExceptionHandler
// implements Thread.UncaughtExceptionHandler
// {
// Throwable exception;
//
// @Override
// public void uncaughtException(Thread th, Throwable ex)
// {
// this.exception = ex;
// }
// }
//
// class QueryThread
// extends Thread
// {
// String query;
// Exception exception;
//
// public QueryThread(@Language("SQL") String query)
// {
// this.query = query;
// }
//
// public void run()
// {
// try {
// assertQuerySucceeds(this.query);
// }
// catch (Exception ex) {
// this.exception = ex;
// }
// }
// }
private static final String maxDataPerNode = "9MB";
private static final int processingDelay = 10000;
private static final int numLoops = 3;
public TestMemorySpilling()
{
super(() -> MemoryQueryRunner.createQueryRunner(1, ImmutableMap.of(), ImmutableMap.of("memory.max-data-per-node", maxDataPerNode), false));
}
@AfterMethod
@BeforeMethod
public void dropAllTables()
{
MaterializedResult tables = computeActual("SHOW TABLES");
for (MaterializedRow row : tables.getMaterializedRows()) {
assertQuerySucceeds("DROP TABLE IF EXISTS " + row.getField(0));
}
}
@Test
public void testEvictionCreate()
{
synchronized (this) {
try {
// create one orders table
assertQuerySucceeds("CREATE TABLE test_t1_1 WITH (async_processing=false) AS SELECT * FROM tpch.tiny.orders");
// without memory eviction this table creation would not succeed
// if this query passes then the eviction is functioning well
assertQuerySucceeds("CREATE TABLE test_t1_2 WITH (async_processing=false) AS SELECT * FROM tpch.tiny.orders");
// creation of a table over total limit. even with eviction this should fail
assertQueryFails("CREATE TABLE test_t1_3 WITH (async_processing=false) AS SELECT * FROM tpch.tiny.lineitem", "Memory limit \\[.+\\] for memory connector exceeded. Current: \\[.+\\]. Requested: \\[.+\\]");
}
finally {
assertQuerySucceeds("DROP TABLE IF EXISTS test_t1_1");
assertQuerySucceeds("DROP TABLE IF EXISTS test_t1_2");
assertQuerySucceeds("DROP TABLE IF EXISTS test_t1_3");
}
}
}
@Test
public void testEvictionSelect()
{
synchronized (this) {
try {
assertQuerySucceeds("CREATE TABLE test_t2_1 WITH (async_processing=false) AS SELECT * FROM tpch.tiny.orders");
assertQuerySucceeds("CREATE TABLE test_t2_2 WITH (async_processing=false) AS SELECT * FROM tpch.tiny.orders");
assertQuerySucceeds("SELECT COUNT(*) FROM test_t2_1");
assertQuerySucceeds("SELECT COUNT(*) FROM test_t2_2");
}
finally {
assertQuerySucceeds("DROP TABLE IF EXISTS test_t2_1");
assertQuerySucceeds("DROP TABLE IF EXISTS test_t2_2");
}
}
}
@Test
public void testConcurrentEvictionSelect()
throws Throwable
{
int numTables = 3;
synchronized (this) {
try {
// create the tables
for (int i = 0; i < numTables; i++) {
assertQuerySucceeds("CREATE TABLE test_t3_" + i + " WITH (async_processing=false) AS SELECT * FROM tpch.tiny.orders");
}
QueryThread[] threads = new QueryThread[numTables];
QueryThreadExceptionHandler[] threadExceptionHandlers = new QueryThreadExceptionHandler[numTables];
// define the query threads
for (int i = 0; i < numTables; i++) {
threads[i] = new QueryThread("SELECT COUNT(*) FROM test_t3_" + i);
threadExceptionHandlers[i] = new QueryThreadExceptionHandler();
threads[i].setUncaughtExceptionHandler(threadExceptionHandlers[i]);
}
// start the query threads
for (int i = 0; i < numTables; i++) {
threads[i].start();
}
// wait till all the query threads are finished
for (int i = 0; i < numTables; i++) {
threads[i].join();
}
// catch any exceptions from the threads
for (int i = 0; i < numTables; i++) {
if (threadExceptionHandlers[i].exception != null) {
throw threads[i].exception.getCause();
}
}
}
finally {
// delete tables
for (int i = 0; i < numTables; i++) {
assertQuerySucceeds("DROP TABLE IF EXISTS test_t3_" + i);
}
}
}
}
class QueryThreadExceptionHandler
implements Thread.UncaughtExceptionHandler
{
Throwable exception;
@Override
public void uncaughtException(Thread th, Throwable ex)
{
this.exception = ex;
}
}
class QueryThread
extends Thread
{
String query;
Exception exception;
public QueryThread(@Language("SQL") String query)
{
this.query = query;
}
public void run()
{
try {
assertQuerySucceeds(this.query);
}
catch (Exception ex) {
this.exception = ex;
}
}
}
}