diff --git a/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryMetadata.java b/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryMetadata.java index 75f8e4530..862c4b595 100644 --- a/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryMetadata.java +++ b/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryMetadata.java @@ -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, diff --git a/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryPageSinkProvider.java b/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryPageSinkProvider.java index 9184d58ed..3b9bbc046 100644 --- a/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryPageSinkProvider.java +++ b/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryPageSinkProvider.java @@ -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 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> 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); } } diff --git a/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryTableProperties.java b/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryTableProperties.java index b7a96de51..c26464f45 100644 --- a/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryTableProperties.java +++ b/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryTableProperties.java @@ -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 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; diff --git a/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryWriteTableHandle.java b/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryWriteTableHandle.java index 24f8be0b7..12a168a73 100644 --- a/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryWriteTableHandle.java +++ b/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryWriteTableHandle.java @@ -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 activeTableIds; private final List columns; private final List 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 activeTableIds, @JsonProperty("columns") List columns, @JsonProperty("sortedBy") List 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 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 getActiveTableIds() { diff --git a/presto-memory/src/main/java/io/prestosql/plugin/memory/data/MemoryTableManager.java b/presto-memory/src/main/java/io/prestosql/plugin/memory/data/MemoryTableManager.java index bf46e9823..942254631 100644 --- a/presto-memory/src/main/java/io/prestosql/plugin/memory/data/MemoryTableManager.java +++ b/presto-memory/src/main/java/io/prestosql/plugin/memory/data/MemoryTableManager.java @@ -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 columns, List sortedBy, List indexColumns) + public synchronized void initialize(long tableId, boolean compressionEnabled, boolean asyncProcessingEnabled, List columns, List sortedBy, List 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. - * + *

* 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. - * + *

* 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) diff --git a/presto-memory/src/main/java/io/prestosql/plugin/memory/data/Table.java b/presto-memory/src/main/java/io/prestosql/plugin/memory/data/Table.java index 169eeabe0..37420b072 100644 --- a/presto-memory/src/main/java/io/prestosql/plugin/memory/data/Table.java +++ b/presto-memory/src/main/java/io/prestosql/plugin/memory/data/Table.java @@ -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 columns; private final List sortedBy; private final List indexColumns; @@ -107,21 +110,23 @@ public class Table private final List 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 columns, List sortedBy, + public Table(long id, boolean compressionEnabled, boolean asyncEnabled, Path tableDataRoot, List columns, List sortedBy, List 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> 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); } } } diff --git a/presto-memory/src/test/java/io/prestosql/plugin/memory/MemoryQueryRunner.java b/presto-memory/src/test/java/io/prestosql/plugin/memory/MemoryQueryRunner.java index 1bba91b3f..bdf4f0c76 100644 --- a/presto-memory/src/test/java/io/prestosql/plugin/memory/MemoryQueryRunner.java +++ b/presto-memory/src/test/java/io/prestosql/plugin/memory/MemoryQueryRunner.java @@ -52,7 +52,7 @@ public final class MemoryQueryRunner memoryProperties.put(FOLDER_PROPERTY_KEY, folder.newFolder("memory-connector").getAbsolutePath()); for (Map.Entry entry : newConfigs.entrySet()) { - memoryProperties.replace(entry.getKey(), entry.getValue()); + memoryProperties.put(entry.getKey(), entry.getValue()); } return memoryProperties; diff --git a/presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemorySelection.java b/presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemorySelection.java index c1370941c..d738bac30 100644 --- a/presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemorySelection.java +++ b/presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemorySelection.java @@ -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 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 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 data1 = new ArrayList<>(); -// ArrayList 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 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 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 data1 = new ArrayList<>(); -// ArrayList 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 getResults(@Language("SQL") String sql) -// { -// MaterializedResult rows = computeActual(sql); -// -// List 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 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 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 data1 = new ArrayList<>(); + ArrayList 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 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 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 data1 = new ArrayList<>(); + ArrayList 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 getResults(@Language("SQL") String sql) + { + MaterializedResult rows = computeActual(sql); + + List 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; + } } diff --git a/presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemorySpilling.java b/presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemorySpilling.java index 2c0138562..88d684e1f 100644 --- a/presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemorySpilling.java +++ b/presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemorySpilling.java @@ -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; + } + } + } }