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

Determine index creation level dynamically and resolve UT issues

This commit is contained in:
Han Weng 2021-03-01 13:08:17 -05:00
parent 2c959f733c
commit ec79a88adc
19 changed files with 211 additions and 219 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/PartitionIndexWriter.java
View File

@ -86,7 +86,7 @@ public class PartitionIndexWriter
Path path = Paths.get(connectorMetadata.getProperty(HetuConstant.DATASOURCE_FILE_PATH));
if (Strings.isNullOrEmpty(partition)) {
if (createIndexMetadata.getCreateLevel() == Index.Level.PARTITION) {
if (createIndexMetadata.getCreateLevel() == CreateIndexMetadata.Level.PARTITION) {
partition = path.getName(path.getNameCount() - 2).toString();
}
}

5
hetu-heuristic-index/src/main/java/io/hetu/core/plugin/heuristicindex/index/bitmap/BitmapIndex.java
View File

@ -15,6 +15,7 @@
package io.hetu.core.plugin.heuristicindex.index.bitmap;
import com.google.common.collect.ImmutableSet;
import io.prestosql.spi.connector.CreateIndexMetadata;
import io.prestosql.spi.heuristicindex.Index;
import io.prestosql.spi.heuristicindex.Pair;
import io.prestosql.spi.predicate.Domain;
@ -113,9 +114,9 @@ public class BitmapIndex
private AtomicBoolean updateAllowed = new AtomicBoolean(true);
@Override
public Set<Level> getSupportedIndexLevels()
public Set<CreateIndexMetadata.Level> getSupportedIndexLevels()
{
return ImmutableSet.of(Level.STRIPE);
return ImmutableSet.of(CreateIndexMetadata.Level.STRIPE);
}
@Override

5
hetu-heuristic-index/src/main/java/io/hetu/core/plugin/heuristicindex/index/bloom/BloomIndex.java
View File

@ -17,6 +17,7 @@ package io.hetu.core.plugin.heuristicindex.index.bloom;
import com.google.common.collect.ImmutableSet;
import io.airlift.slice.Slice;
import io.prestosql.spi.connector.CreateIndexMetadata;
import io.prestosql.spi.heuristicindex.Index;
import io.prestosql.spi.heuristicindex.Pair;
import io.prestosql.spi.predicate.Domain;
@ -54,9 +55,9 @@ public class BloomIndex
}
@Override
public Set<Level> getSupportedIndexLevels()
public Set<CreateIndexMetadata.Level> getSupportedIndexLevels()
{
return ImmutableSet.of(Level.STRIPE);
return ImmutableSet.of(CreateIndexMetadata.Level.STRIPE);
}
@Override

5
hetu-heuristic-index/src/main/java/io/hetu/core/plugin/heuristicindex/index/btree/BTreeIndex.java
View File

@ -18,6 +18,7 @@ import com.google.common.collect.Sets;
import com.google.common.io.Files;
import io.hetu.core.heuristicindex.PartitionIndexWriter;
import io.hetu.core.heuristicindex.util.TypeUtils;
import io.prestosql.spi.connector.CreateIndexMetadata;
import io.prestosql.spi.function.OperatorType;
import io.prestosql.spi.function.Signature;
import io.prestosql.spi.heuristicindex.Index;
@ -153,9 +154,9 @@ public class BTreeIndex
}
@Override
public Set<Level> getSupportedIndexLevels()
public Set<CreateIndexMetadata.Level> getSupportedIndexLevels()
{
return Sets.newHashSet(Level.PARTITION, Level.TABLE);
return Sets.newHashSet(CreateIndexMetadata.Level.PARTITION, CreateIndexMetadata.Level.TABLE);
}
@Override

5
hetu-heuristic-index/src/main/java/io/hetu/core/plugin/heuristicindex/index/minmax/MinMaxIndex.java
View File

@ -17,6 +17,7 @@ package io.hetu.core.plugin.heuristicindex.index.minmax;
import com.google.common.collect.ImmutableSet;
import io.hetu.core.common.util.SecureObjectInputStream;
import io.prestosql.spi.connector.CreateIndexMetadata;
import io.prestosql.spi.function.OperatorType;
import io.prestosql.spi.function.Signature;
import io.prestosql.spi.heuristicindex.Index;
@ -102,9 +103,9 @@ public class MinMaxIndex
}
@Override
public Set<Level> getSupportedIndexLevels()
public Set<CreateIndexMetadata.Level> getSupportedIndexLevels()
{
return ImmutableSet.of(Level.STRIPE);
return ImmutableSet.of(CreateIndexMetadata.Level.STRIPE);
}
@Override

75
hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestHindex.java
View File

@ -46,12 +46,12 @@ public class TestHindex
if (indexType.toLowerCase(Locale.ROOT).equals("btree")) {
if (dataType.toLowerCase(Locale.ROOT).equals("boolean")) {
assertQueryFails("CREATE INDEX " + indexName + " USING " +
indexType + " ON " + tableName + " (data_col2) WITH (level='table')",
indexType + " ON " + tableName + " (data_col2)",
"Index creation on boolean column is not supported");
}
else {
assertQuerySucceeds("CREATE INDEX " + indexName + " USING " +
indexType + " ON " + tableName + " (data_col2) WITH (level='table')");
indexType + " ON " + tableName + " (data_col2)");
}
}
else {
@ -106,14 +106,8 @@ public class TestHindex
int splitsBeforeIndex = getSplitAndMaterializedResult(testerQuery).getFirst();
// Create index to use for testing splits
if (indexType.toLowerCase(Locale.ROOT).equals("btree")) {
assertQuerySucceeds("CREATE INDEX " + indexName + " USING " +
indexType + " ON " + tableName + " (" + queryVariable + ") WITH (level='table')");
}
else {
assertQuerySucceeds("CREATE INDEX " + indexName + " USING " +
indexType + " ON " + tableName + " (" + queryVariable + ")");
}
assertQuerySucceeds("CREATE INDEX " + indexName + " USING " +
indexType + " ON " + tableName + " (" + queryVariable + ")");
assertQuerySucceeds("INSERT INTO " + tableName + " VALUES(7, 'new1'), (8, 'new2')");
@ -171,14 +165,8 @@ public class TestHindex
int splitsBeforeIndex = getSplitAndMaterializedResult(testerQuery).getFirst();
// Create index to use for testing splits
if (indexType.toLowerCase(Locale.ROOT).equals("btree")) {
assertQuerySucceeds("CREATE INDEX " + indexName + " USING " +
indexType + " ON " + tableName + " (" + queryVariable + ") WITH (level='table')");
}
else {
assertQuerySucceeds("CREATE INDEX " + indexName + " USING " +
indexType + " ON " + tableName + " (" + queryVariable + ")");
}
assertQuerySucceeds("CREATE INDEX " + indexName + " USING " +
indexType + " ON " + tableName + " (" + queryVariable + ")");
Pair<Integer, MaterializedResult> resultPairLoadingIndex = getSplitAndMaterializedResult(testerQuery);
int splitsLoadingIndex = resultPairLoadingIndex.getFirst();
@ -247,14 +235,8 @@ public class TestHindex
MaterializedResult resultBeforeIndex = resultPairBeforeIndex.getSecond();
// Create index to use for testing splits
if (indexType.toLowerCase(Locale.ROOT).equals("btree")) {
assertQuerySucceeds("CREATE INDEX " + indexName + " USING " +
indexType + " ON " + tableName + " (" + queryVariable + ") WITH (level='table')");
}
else {
assertQuerySucceeds("CREATE INDEX " + indexName + " USING " +
indexType + " ON " + tableName + " (" + queryVariable + ")");
}
assertQuerySucceeds("CREATE INDEX " + indexName + " USING " +
indexType + " ON " + tableName + " (" + queryVariable + ")");
int splitsLoadingIndex = getSplitAndMaterializedResult(testerQuery).getFirst();
@ -304,7 +286,7 @@ public class TestHindex
// Create indices
String indexName1 = getNewIndexName();
assertQuerySucceeds("CREATE INDEX " + indexName1 + " USING btree ON " + tableName + " (id) WITH (level='table')");
assertQuerySucceeds("CREATE INDEX " + indexName1 + " USING btree ON " + tableName + " (id)");
String indexName2 = getNewIndexName();
assertQuerySucceeds("CREATE INDEX " + indexName2 + " USING bitmap ON " + tableName + " (id)");
String indexName3 = getNewIndexName();
@ -351,14 +333,8 @@ public class TestHindex
MaterializedResult resultBeforeIndex = resultPairBeforeIndex.getSecond();
// Create index
if (indexType.toLowerCase(Locale.ROOT).equals("btree")) {
assertQuerySucceeds("CREATE INDEX " + indexName + " USING " +
indexType + " ON " + tableName + " (id) WITH (level='table')");
}
else {
assertQuerySucceeds("CREATE INDEX " + indexName + " USING " +
indexType + " ON " + tableName + " (id)");
}
assertQuerySucceeds("CREATE INDEX " + indexName + " USING " +
indexType + " ON " + tableName + " (id)");
// Drop index before changes
assertQuerySucceeds("DROP INDEX " + indexName);
@ -390,14 +366,8 @@ public class TestHindex
createTable1(tableName);
// Create index
if (indexType.toLowerCase(Locale.ROOT).equals("btree")) {
assertQuerySucceeds("CREATE INDEX IF NOT EXISTS " + indexName + " USING " +
indexType + " ON " + tableName + " (id) WITH (level='table')");
}
else {
assertQuerySucceeds("CREATE INDEX IF NOT EXISTS " + indexName + " USING " +
indexType + " ON " + tableName + " (id)");
}
assertQuerySucceeds("CREATE INDEX IF NOT EXISTS " + indexName + " USING " +
indexType + " ON " + tableName + " (id)");
// Validate if created
String testerQuery = "SHOW INDEX";
@ -469,13 +439,8 @@ public class TestHindex
assertQuerySucceeds("INSERT INTO " + tableName + " VALUES(3, 'data'), (9, 'ttt'), (5, 'num')");
String indexName = getNewIndexName();
if (indexType.toLowerCase(Locale.ROOT).equals("btree")) {
assertQuerySucceeds("CREATE INDEX " + indexName + " USING " + indexType +
" ON " + tableName + " (P1) WITH (level='table')");
}
else {
assertQuerySucceeds("CREATE INDEX " + indexName + " USING " + indexType + " ON " + tableName + " (P1)");
}
assertQuerySucceeds("CREATE INDEX " + indexName + " USING " + indexType +
" ON " + tableName + " (P1)");
}
@Test(dataProvider = "queryOperatorTest")
@ -487,14 +452,8 @@ public class TestHindex
testerQuery = "SELECT * FROM " + tableName + " WHERE " + testerQuery;
String indexName = getNewIndexName();
if (indexType.toLowerCase(Locale.ROOT).equals("btree")) {
assertQuerySucceeds("CREATE INDEX " + indexName + " USING " +
indexType + " ON " + tableName + " (id) WITH (level='table')");
}
else {
assertQuerySucceeds("CREATE INDEX " + indexName + " USING " +
indexType + " ON " + tableName + " (id)");
}
assertQuerySucceeds("CREATE INDEX " + indexName + " USING " +
indexType + " ON " + tableName + " (id)");
MaterializedResult resultLoadingIndex = computeActual(testerQuery);

20
hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestHindexBTreeIndex.java
View File

@ -33,7 +33,7 @@ public class TestHindexBTreeIndex
String indexName = getNewIndexName();
assertQuerySucceeds("CREATE INDEX " + indexName + " USING btree ON " + tableName +
" (key2) WITH (level=partition) WHERE key2 = 11");
" (key2) WHERE key2 = 11");
assertQuerySucceeds("DROP INDEX " + indexName);
}
@ -57,7 +57,7 @@ public class TestHindexBTreeIndex
String indexName = getNewIndexName();
assertQuerySucceeds("CREATE INDEX " + indexName + " USING btree ON " + tableName +
" (key1) WITH (level=partition) WHERE key2 = 11");
" (key1) WHERE key2 = 11");
assertQuerySucceeds("DROP INDEX " + indexName + " WHERE key2 = 11");
}
@ -69,7 +69,7 @@ public class TestHindexBTreeIndex
String indexName = getNewIndexName();
assertQuerySucceeds("CREATE INDEX " + indexName + " USING btree ON " + tableName +
" (key1) WITH (level=partition) WHERE key2 = 11");
" (key1) WHERE key2 = 11");
try {
assertQuerySucceeds("DROP INDEX " + indexName + " WHERE key2 = 10");
}
@ -87,7 +87,7 @@ public class TestHindexBTreeIndex
String indexName = getNewIndexName();
assertQuerySucceeds("CREATE INDEX " + indexName + " USING btree ON " + tableName +
" (key2) WITH (level=partition) WHERE key2 = 11");
" (key2) WHERE key2 = 11");
String testerQuery = "SELECT * FROM " + tableName + " WHERE key2 = 11";
@ -114,7 +114,7 @@ public class TestHindexBTreeIndex
String indexName = getNewIndexName();
assertQueryFails("CREATE INDEX " + indexName + " USING btree ON " + tableName +
" (key2) WITH (level=partition) WHERE key1 = 1",
" (key2) WHERE key1 = 1",
"line 1:18: Heuristic index creation is only supported for predicates on partition columns");
}
@ -127,7 +127,7 @@ public class TestHindexBTreeIndex
String indexName = getNewIndexName();
assertQuerySucceeds("CREATE INDEX " + indexName + " USING btree ON " + tableName +
" (key2) WITH (level=partition) WHERE key2 = 12");
" (key2) WHERE key2 = 12");
String testerQuery = "SELECT * FROM " + tableName + " WHERE key2 = 12";
@ -156,7 +156,7 @@ public class TestHindexBTreeIndex
String indexName = getNewIndexName();
assertQueryFails("CREATE INDEX " + indexName + " USING btree ON " + tableName +
" (key2) WITH (level=partition) WHERE key1 = 1",
" (key2) WHERE key1 = 1",
"line 1:18: Heuristic index creation is only supported for predicates on partition columns");
}
@ -170,7 +170,7 @@ public class TestHindexBTreeIndex
String indexName = getNewIndexName();
try {
assertQuerySucceeds("CREATE INDEX " + indexName + " USING btree ON " + tableName +
" (key2) WITH (level=partition) WHERE " + condition);
" (key2) WHERE " + condition);
}
catch (AssertionError e) {
assertNotEquals(condition, "key2 = 11");
@ -188,7 +188,7 @@ public class TestHindexBTreeIndex
String indexName1 = getNewIndexName();
assertQuerySucceeds("CREATE INDEX " + indexName1 + " USING btree ON " + tableName +
" (key1) WITH (level=partition) WHERE key3 = 222");
" (key1) WHERE key3 = 222");
String testerQuery1 = "SELECT * FROM " + tableName + " WHERE key1 = 2";
@ -210,7 +210,7 @@ public class TestHindexBTreeIndex
String indexName2 = getNewIndexName();
assertQuerySucceeds("CREATE INDEX " + indexName2 + " USING btree ON " + tableName +
" (key2) WITH (level=partition) WHERE key5 = 22222");
" (key2) WHERE key5 = 22222");
String testerQuery2 = "SELECT * FROM " + tableName + " WHERE key2 = 22";

194
hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestIndexRecordManager.java
View File

@ -64,119 +64,105 @@ public class TestIndexRecordManager
.setDbPassword(mysqlServer.getPassword()));
}
@Test(timeOut = 30000)
public void testConcurrentMultipleManagers()
throws IOException, InterruptedException
{
try (TempFolder folder = new TempFolder()) {
folder.create();
// TODO: This test is not stable. There are chances that index records are not properly created. Probably needs changes from hetu-metastore.
Random random = new Random();
// 6 entries will be created by different threads in parallel, in which the first four will be deleted also in parallel
String[] names = new String[] {"a", "b", "c", "d", "e", "f"};
Thread[] threads = new Thread[10];
for (int i = 0; i < 6; i++) {
int finalI = i;
threads[i] = new Thread(() -> {
try {
new IndexRecordManager(testMetastore1)
.addIndexRecord(names[finalI], "testUser", "c.s.t", new String[] {"testColumn"}, names[finalI], Collections.emptyList(), Arrays.asList("cp=1"));
}
catch (IOException e) {
throw new RuntimeException(e);
}
});
threads[i].start();
}
for (int i = 6; i < 10; i++) {
int finalI = i;
threads[i] = new Thread(() -> {
try {
IndexRecordManager indexRecordManager = new IndexRecordManager(testMetastore1);
while (indexRecordManager.lookUpIndexRecord(names[finalI - 6]) == null) {
Thread.sleep(random.nextInt(100));
}
indexRecordManager.deleteIndexRecord(names[finalI - 6], Collections.emptyList());
}
catch (IOException | InterruptedException e) {
throw new RuntimeException(e);
}
});
threads[i].start();
}
for (Thread thread : threads) {
thread.join();
}
IndexRecordManager indexRecordManager = new IndexRecordManager(testMetastore1);
assertEquals(indexRecordManager.getIndexRecords().size(), 2);
assertNull(indexRecordManager.lookUpIndexRecord(names[0]));
assertNull(indexRecordManager.lookUpIndexRecord(names[1]));
assertNull(indexRecordManager.lookUpIndexRecord(names[2]));
assertNull(indexRecordManager.lookUpIndexRecord(names[3]));
assertNotNull(indexRecordManager.lookUpIndexRecord(names[4]));
assertNotNull(indexRecordManager.lookUpIndexRecord(names[5]));
}
}
// @Test(timeOut = 30000)
// public void testConcurrentMultipleManagers()
// throws IOException, InterruptedException
// {
// Random random = new Random();
// String[] names = new String[] {"a", "b", "c"};
// Thread[] threads = new Thread[4];
//
// // create index with name[0], name[1], name[2]
// for (int i = 0; i < 3; i++) {
// int finalI = i;
// threads[i] = new Thread(() -> {
// try {
// Thread.sleep(random.nextInt(100));
// new IndexRecordManager(testMetastore1)
// .addIndexRecord(names[finalI], "testUser", "c.s.t", new String[] {"testColumn"}, names[finalI], Collections.emptyList(), Arrays.asList("cp=1"));
// }
// catch (IOException | InterruptedException e) {
// throw new RuntimeException(e);
// }
// });
// threads[i].start();
// }
//
// // delete index with name[0]
// threads[3] = new Thread(() -> {
// try {
// IndexRecordManager indexRecordManager = new IndexRecordManager(testMetastore1);
// while (indexRecordManager.lookUpIndexRecord(names[0]) == null) {
// Thread.sleep(random.nextInt(100));
// }
// indexRecordManager.deleteIndexRecord(names[0], Collections.emptyList());
// }
// catch (IOException | InterruptedException e) {
// throw new RuntimeException(e);
// }
// });
// threads[3].start();
//
// for (Thread thread : threads) {
// thread.join();
// }
//
// IndexRecordManager indexRecordManager = new IndexRecordManager(testMetastore1);
// assertEquals(indexRecordManager.getIndexRecords().size(), 2);
// assertNull(indexRecordManager.lookUpIndexRecord(names[0]));
// assertNotNull(indexRecordManager.lookUpIndexRecord(names[1]));
// assertNotNull(indexRecordManager.lookUpIndexRecord(names[2]));
// }
@Test(timeOut = 20000)
public void testConcurrentSingleManager()
throws IOException, InterruptedException
{
try (TempFolder folder = new TempFolder()) {
folder.create();
Random random = new Random();
IndexRecordManager indexRecordManager = new IndexRecordManager(testMetastore2);
Random random = new Random();
IndexRecordManager indexRecordManager = new IndexRecordManager(testMetastore2);
String[] names = new String[] {"a", "b", "c"};
Thread[] threads = new Thread[4];
// 6 entries will be created by different threads in parallel, in which the first four will be deleted also in parallel
String[] names = new String[] {"a", "b", "c", "d", "e", "f"};
Thread[] threads = new Thread[10];
for (int i = 0; i < 6; i++) {
int finalI = i;
threads[i] = new Thread(() -> {
try {
indexRecordManager.addIndexRecord(names[finalI], "u", "c.s.t", new String[] {"c"}, names[finalI], Collections.emptyList(), Arrays.asList("cp=1"));
}
catch (IOException e) {
throw new RuntimeException(e);
}
});
threads[i].start();
}
for (int i = 6; i < 10; i++) {
int finalI = i;
threads[i] = new Thread(() -> {
try {
while (indexRecordManager.lookUpIndexRecord(names[finalI - 6]) == null) {
Thread.sleep(random.nextInt(100));
}
indexRecordManager.deleteIndexRecord(names[finalI - 6], Collections.emptyList());
}
catch (IOException | InterruptedException e) {
throw new RuntimeException(e);
}
});
threads[i].start();
}
for (Thread thread : threads) {
thread.join();
}
assertEquals(indexRecordManager.getIndexRecords().size(), 2);
assertNull(indexRecordManager.lookUpIndexRecord(names[0]));
assertNull(indexRecordManager.lookUpIndexRecord(names[1]));
assertNull(indexRecordManager.lookUpIndexRecord(names[2]));
assertNull(indexRecordManager.lookUpIndexRecord(names[3]));
assertNotNull(indexRecordManager.lookUpIndexRecord(names[4]));
assertNotNull(indexRecordManager.lookUpIndexRecord(names[5]));
// create index with name[0], name[1], name[2]
for (int i = 0; i < 3; i++) {
int finalI = i;
threads[i] = new Thread(() -> {
try {
Thread.sleep(random.nextInt(100));
indexRecordManager.addIndexRecord(names[finalI], "u", "c.s.t", new String[] {"c"}, names[finalI], Collections.emptyList(), Arrays.asList("cp=1"));
}
catch (IOException | InterruptedException e) {
throw new RuntimeException(e);
}
});
threads[i].start();
}
// delete index with name[0]
threads[3] = new Thread(() -> {
try {
while (indexRecordManager.lookUpIndexRecord(names[0]) == null) {
Thread.sleep(random.nextInt(100));
}
indexRecordManager.deleteIndexRecord(names[0], Collections.emptyList());
}
catch (IOException | InterruptedException e) {
throw new RuntimeException(e);
}
});
threads[3].start();
for (Thread thread : threads) {
thread.join();
}
assertEquals(indexRecordManager.getIndexRecords().size(), 2);
assertNull(indexRecordManager.lookUpIndexRecord(names[0]));
assertNotNull(indexRecordManager.lookUpIndexRecord(names[1]));
assertNotNull(indexRecordManager.lookUpIndexRecord(names[2]));
}
@Test

5
hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestPartitionIndexWriter.java
View File

@ -17,7 +17,6 @@ package io.hetu.core.heuristicindex;
import io.prestosql.spi.HetuConstant;
import io.prestosql.spi.connector.CreateIndexMetadata;
import io.prestosql.spi.filesystem.HetuFileSystemClient;
import io.prestosql.spi.heuristicindex.Index;
import io.prestosql.spi.type.Type;
import org.mockito.Mockito;
import org.testng.annotations.Test;
@ -56,7 +55,7 @@ public class TestPartitionIndexWriter
partitions,
properties,
"testuser",
Index.Level.PARTITION);
CreateIndexMetadata.Level.PARTITION);
HetuFileSystemClient fileSystemClient = Mockito.mock(HetuFileSystemClient.class);
Properties connectorMetadata = new Properties();
connectorMetadata.setProperty(HetuConstant.DATASOURCE_FILE_MODIFICATION, String.valueOf(System.currentTimeMillis()));
@ -90,7 +89,7 @@ public class TestPartitionIndexWriter
partitions,
properties,
"testuser",
Index.Level.PARTITION);
CreateIndexMetadata.Level.PARTITION);
HetuFileSystemClient fileSystemClient = Mockito.mock(HetuFileSystemClient.class);
Properties connectorMetadata1 = new Properties();
connectorMetadata1.setProperty(HetuConstant.DATASOURCE_FILE_MODIFICATION, String.valueOf(System.currentTimeMillis()));

8
presto-main/src/main/java/io/prestosql/heuristicindex/IndexCache.java
View File

@ -23,7 +23,7 @@ import com.google.common.util.concurrent.ThreadFactoryBuilder;
import io.airlift.log.Logger;
import io.prestosql.metadata.Split;
import io.prestosql.spi.HetuConstant;
import io.prestosql.spi.heuristicindex.Index;
import io.prestosql.spi.connector.CreateIndexMetadata;
import io.prestosql.spi.heuristicindex.IndexCacheKey;
import io.prestosql.spi.heuristicindex.IndexClient;
import io.prestosql.spi.heuristicindex.IndexMetadata;
@ -135,7 +135,7 @@ public class IndexCache
}
}
public void preloadIndex(String table, String column, String type, Index.Level level)
public void preloadIndex(String table, String column, String type, CreateIndexMetadata.Level level)
{
String filterKeyPath = table + "/" + column + "/" + type;
IndexCacheKey filterKey = new IndexCacheKey(filterKeyPath, LAST_MODIFIED_TIME_PLACE_HOLDER, level);
@ -253,7 +253,7 @@ public class IndexCache
if (!partitions.isEmpty()) {
for (String partition : partitions) {
String filterKeyPath = table + "/" + column + "/" + indexType + "/" + partition;
IndexCacheKey filterKey = new IndexCacheKey(filterKeyPath, lastModifiedTime, Index.Level.PARTITION);
IndexCacheKey filterKey = new IndexCacheKey(filterKeyPath, lastModifiedTime, CreateIndexMetadata.Level.PARTITION);
List<IndexMetadata> result = loadIndex(filterKey);
if (result != null) {
indices.addAll(result);
@ -267,7 +267,7 @@ public class IndexCache
}
String filterKeyPath = table + "/" + column + "/" + indexType;
IndexCacheKey filterKey = new IndexCacheKey(filterKeyPath, lastModifiedTime, Index.Level.TABLE);
IndexCacheKey filterKey = new IndexCacheKey(filterKeyPath, lastModifiedTime, CreateIndexMetadata.Level.TABLE);
List<IndexMetadata> result = loadIndex(filterKey);
if (result != null) {
indices.addAll(result);

4
presto-main/src/main/java/io/prestosql/heuristicindex/IndexCacheLoader.java
View File

@ -15,7 +15,7 @@
package io.prestosql.heuristicindex;
import com.google.common.cache.CacheLoader;
import io.prestosql.spi.heuristicindex.Index;
import io.prestosql.spi.connector.CreateIndexMetadata;
import io.prestosql.spi.heuristicindex.IndexCacheKey;
import io.prestosql.spi.heuristicindex.IndexClient;
import io.prestosql.spi.heuristicindex.IndexMetadata;
@ -44,7 +44,7 @@ public class IndexCacheLoader
{
requireNonNull(key);
requireNonNull(indexClient);
if (key.getIndexLevel() == Index.Level.PARTITION || key.getIndexLevel() == Index.Level.TABLE) {
if (key.getIndexLevel() == CreateIndexMetadata.Level.PARTITION || key.getIndexLevel() == CreateIndexMetadata.Level.TABLE) {
return loadPartitionIndex(key);
}
else {

3
presto-main/src/main/java/io/prestosql/heuristicindex/SplitFiltering.java
View File

@ -23,7 +23,6 @@ import io.prestosql.metadata.Split;
import io.prestosql.spi.connector.ColumnHandle;
import io.prestosql.spi.connector.CreateIndexMetadata;
import io.prestosql.spi.function.OperatorType;
import io.prestosql.spi.heuristicindex.Index;
import io.prestosql.spi.heuristicindex.IndexCacheKey;
import io.prestosql.spi.heuristicindex.IndexClient;
import io.prestosql.spi.heuristicindex.IndexFilter;
@ -121,7 +120,7 @@ public class SplitFiltering
for (IndexRecord record : indexToPreload) {
LOG.info("Preloading index for split filtering: " + record);
Index.Level indexLevel = Index.Level.valueOf(record.getProperty(CreateIndexMetadata.LEVEL_PROP_KEY).toUpperCase(Locale.ROOT));
CreateIndexMetadata.Level indexLevel = CreateIndexMetadata.Level.valueOf(record.getProperty(CreateIndexMetadata.LEVEL_PROP_KEY).toUpperCase(Locale.ROOT));
indexCache.preloadIndex(record.qualifiedTable, String.join(",", record.columns), record.indexType, indexLevel);
}
}

20
presto-main/src/main/java/io/prestosql/operator/CreateIndexOperator.java
View File

@ -185,6 +185,12 @@ public class CreateIndexOperator
return;
}
if (createIndexMetadata.getCreateLevel() == CreateIndexMetadata.Level.UNDEFINED) {
boolean tableIsPartitioned = getPartitionName(page.getPageMetadata().getProperty(HetuConstant.DATASOURCE_FILE_PATH),
createIndexMetadata.getTableName()) != null;
createIndexMetadata.decideIndexLevel(tableIsPartitioned);
}
Map<String, List<Object>> values = new HashMap<>();
for (int blockId = 0; blockId < page.getChannelCount(); blockId++) {
@ -213,6 +219,9 @@ public class CreateIndexOperator
case PARTITION: {
String partition = getPartitionName(page.getPageMetadata().getProperty(HetuConstant.DATASOURCE_FILE_PATH),
createIndexMetadata.getTableName());
if (partition == null) {
throw new IllegalStateException("Partition level is not supported for non partitioned table.");
}
levelWriter.putIfAbsent(partition, heuristicIndexerManager.getIndexWriter(createIndexMetadata, connectorMetadata));
persistBy.putIfAbsent(levelWriter.get(partition), this);
levelWriter.get(partition).addData(values, connectorMetadata);
@ -316,6 +325,15 @@ public class CreateIndexOperator
return obj;
}
/**
* This is a hacky way to tell if the table is partitioned and get the partition of table
* <p>
* Should be replaced if a better solution is available
*
* @param uri page data path uri
* @param tableName table name
* @return partition name if table is partitioned. {@code null} if no partition is found in the path.
*/
private static String getPartitionName(String uri, String tableName)
{
Path path = Paths.get(uri);
@ -328,6 +346,6 @@ public class CreateIndexOperator
}
}
throw new IllegalStateException("Partition level is not supported for non partitioned table.");
return null;
}
}

3
presto-main/src/main/java/io/prestosql/sql/analyzer/StatementAnalyzer.java
View File

@ -190,6 +190,7 @@ import static io.prestosql.spi.StandardErrorCode.INVALID_COLUMN_MASK;
import static io.prestosql.spi.StandardErrorCode.INVALID_FUNCTION_ARGUMENT;
import static io.prestosql.spi.StandardErrorCode.INVALID_ROW_FILTER;
import static io.prestosql.spi.StandardErrorCode.NOT_FOUND;
import static io.prestosql.spi.connector.CreateIndexMetadata.Level.UNDEFINED;
import static io.prestosql.spi.connector.StandardWarningCode.REDUNDANT_ORDER_BY;
import static io.prestosql.spi.function.FunctionKind.AGGREGATE;
import static io.prestosql.spi.function.FunctionKind.WINDOW;
@ -1106,7 +1107,7 @@ class StatementAnalyzer
partitions,
properties,
session.getUser(),
null);
UNDEFINED);
synchronized (StatementAnalyzer.class) {
IndexClient.RecordStatus recordStatus = heuristicIndexerManager.getIndexClient().lookUpIndexRecord(placeHolder);

9
presto-main/src/main/java/io/prestosql/sql/planner/QueryPlanner.java
View File

@ -26,7 +26,6 @@ import io.prestosql.spi.block.SortOrder;
import io.prestosql.spi.connector.ColumnHandle;
import io.prestosql.spi.connector.ColumnMetadata;
import io.prestosql.spi.connector.CreateIndexMetadata;
import io.prestosql.spi.heuristicindex.Index;
import io.prestosql.spi.metadata.TableHandle;
import io.prestosql.spi.operator.ReuseExchangeOperator;
import io.prestosql.spi.plan.AggregationNode;
@ -114,7 +113,6 @@ import static com.google.common.collect.ImmutableSet.toImmutableSet;
import static com.google.common.collect.Streams.stream;
import static io.prestosql.SystemSessionProperties.isSkipRedundantSort;
import static io.prestosql.spi.connector.CreateIndexMetadata.INDEX_SUPPORTED_TYPES;
import static io.prestosql.spi.connector.CreateIndexMetadata.LEVEL_DEFAULT;
import static io.prestosql.spi.connector.CreateIndexMetadata.LEVEL_PROP_KEY;
import static io.prestosql.spi.plan.AggregationNode.groupingSets;
import static io.prestosql.spi.plan.AggregationNode.singleGroupingSet;
@ -650,14 +648,15 @@ class QueryPlanner
}
Properties indexProperties = new Properties();
Index.Level indexCreationLevel = LEVEL_DEFAULT;
indexProperties.setProperty(LEVEL_PROP_KEY, String.valueOf(LEVEL_DEFAULT));
CreateIndexMetadata.Level indexCreationLevel = CreateIndexMetadata.Level.UNDEFINED;
indexProperties.setProperty(LEVEL_PROP_KEY, indexCreationLevel.toString());
for (Property property : createIndex.getProperties()) {
String key = extractPropertyValue(property.getName());
String val = extractPropertyValue(property.getValue()).toUpperCase(Locale.ENGLISH);
if (key.equals(LEVEL_PROP_KEY)) {
indexCreationLevel = Index.Level.valueOf(val);
indexCreationLevel = CreateIndexMetadata.Level.valueOf(val);
continue;
}
indexProperties.setProperty(key, val);
}

41
presto-spi/src/main/java/io/prestosql/spi/connector/CreateIndexMetadata.java
View File

@ -18,7 +18,6 @@ import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import io.prestosql.spi.heuristicindex.Index;
import io.prestosql.spi.type.Type;
import java.util.List;
@ -33,7 +32,6 @@ import static java.util.Objects.requireNonNull;
public class CreateIndexMetadata
{
public static final String LEVEL_PROP_KEY = "level";
public static final Index.Level LEVEL_DEFAULT = Index.Level.STRIPE;
public static final Map<String, List<String>> INDEX_SUPPORTED_TYPES = ImmutableMap.<String, List<String>>builder()
.put("bloom", ImmutableList.of(
"integer", "smallint", "bigint", "tinyint", "varchar", "char", "boolean", "double", "real", "date"))
@ -52,7 +50,7 @@ public class CreateIndexMetadata
private final List<String> partitions;
private final Properties properties;
private final String user;
private final Index.Level createLevel;
private volatile Level createLevel;
@JsonCreator
public CreateIndexMetadata(
@ -63,7 +61,7 @@ public class CreateIndexMetadata
@JsonProperty("partitions") List<String> partitions,
@JsonProperty("properties") Properties properties,
@JsonProperty("user") String user,
@JsonProperty("createLevel") Index.Level createLevel)
@JsonProperty("createLevel") Level createLevel)
{
this.indexName = checkNotEmpty(indexName, "indexName");
this.tableName = requireNonNull(tableName, "tableName is null");
@ -72,15 +70,37 @@ public class CreateIndexMetadata
this.partitions = partitions;
this.properties = properties;
this.user = requireNonNull(user, "user is null");
this.createLevel = createLevel == null ? LEVEL_DEFAULT : createLevel;
this.createLevel = requireNonNull(createLevel, "createLevel is null");
properties.setProperty(LEVEL_PROP_KEY, createLevel.toString());
}
@JsonProperty
public Index.Level getCreateLevel()
public Level getCreateLevel()
{
return createLevel;
}
/**
* When index creation has gathered enough information (index type, partitioned table or not, etc),
* this method will be called to finalize creationLevel
* <p>
* The level is based on index type, and whether the table is partitioned
*
* @param tableIsPartitioned if this table is partitioned
*/
public synchronized void decideIndexLevel(boolean tableIsPartitioned)
{
if (createLevel == Level.UNDEFINED) {
if (indexType.toUpperCase(Locale.ROOT).equals("BTREE")) {
this.createLevel = tableIsPartitioned ? Level.PARTITION : Level.TABLE;
}
else {
this.createLevel = Level.STRIPE;
}
properties.setProperty(LEVEL_PROP_KEY, createLevel.toString());
}
}
@JsonProperty
public String getIndexName()
{
@ -168,4 +188,13 @@ public class CreateIndexMetadata
&& Objects.equals(this.properties, other.properties)
&& Objects.equals(this.user, other.user);
}
public enum Level
{
STRIPE,
PARTITION,
TABLE,
// If the index creation level is not specified, it will be decided on-the-fly. See decideIndexLevel().
UNDEFINED
}
}

11
presto-spi/src/main/java/io/prestosql/spi/heuristicindex/Index.java
View File

@ -15,6 +15,8 @@
package io.prestosql.spi.heuristicindex;
import io.prestosql.spi.connector.CreateIndexMetadata;
import java.io.Closeable;
import java.io.IOException;
import java.io.InputStream;
@ -39,7 +41,7 @@ public interface Index
*
* @return data level on which the index should be applied on.
*/
Set<Level> getSupportedIndexLevels();
Set<CreateIndexMetadata.Level> getSupportedIndexLevels();
/**
* Gets the id of the IndexStore.
@ -228,11 +230,4 @@ public interface Index
default void close() throws IOException
{
}
enum Level
{
STRIPE,
PARTITION,
TABLE
}
}

10
presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexCacheKey.java
View File

@ -14,6 +14,8 @@
*/
package io.prestosql.spi.heuristicindex;
import io.prestosql.spi.connector.CreateIndexMetadata;
import java.util.Objects;
public class IndexCacheKey
@ -22,7 +24,7 @@ public class IndexCacheKey
private final String path;
private final long lastModifiedTime;
private final Index.Level indexLevel;
private final CreateIndexMetadata.Level indexLevel;
private boolean noCloseFlag;
/**
@ -30,7 +32,7 @@ public class IndexCacheKey
* @param lastModifiedTime lastModifiedTime of the file, used to validate the indexes
* @param indexLevel see Index.Level in presto-spi
*/
public IndexCacheKey(String path, long lastModifiedTime, Index.Level indexLevel)
public IndexCacheKey(String path, long lastModifiedTime, CreateIndexMetadata.Level indexLevel)
{
this.path = path;
this.lastModifiedTime = lastModifiedTime;
@ -45,7 +47,7 @@ public class IndexCacheKey
*/
public IndexCacheKey(String path, long lastModifiedTime)
{
this(path, lastModifiedTime, Index.Level.STRIPE);
this(path, lastModifiedTime, CreateIndexMetadata.Level.STRIPE);
}
public String getPath()
@ -58,7 +60,7 @@ public class IndexCacheKey
return lastModifiedTime;
}
public Index.Level getIndexLevel()
public CreateIndexMetadata.Level getIndexLevel()
{
return this.indexLevel;
}

5
presto-spi/src/test/java/io/prestosql/spi/heuristicindex/TestIndexMetadata.java
View File

@ -15,6 +15,7 @@
package io.prestosql.spi.heuristicindex;
import com.google.common.collect.ImmutableSet;
import io.prestosql.spi.connector.CreateIndexMetadata;
import org.testng.annotations.Test;
import java.io.IOException;
@ -33,9 +34,9 @@ public class TestIndexMetadata
Index index = new Index()
{
@Override
public Set<Level> getSupportedIndexLevels()
public Set<CreateIndexMetadata.Level> getSupportedIndexLevels()
{
return ImmutableSet.of(Level.STRIPE);
return ImmutableSet.of(CreateIndexMetadata.Level.STRIPE);
}
@Override