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

Check index record and evict entries from cache when updated

This commit is contained in:
Han Weng 2020-11-25 16:43:13 -05:00
parent dfedbb0526
commit 78dfd71c94
14 changed files with 194 additions and 271 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/FileIndexWriter.java
View File

@ -32,6 +32,7 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.AbstractMap;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
@ -128,10 +129,10 @@ public class FileIndexWriter
// each entry represents a mapping from column name -> list<entry<page values, page number>>
for (Map.Entry<String, List<Map.Entry<List<Object>, Integer>>> entry : indexPages.get(stripeOffset).entrySet()) {
// sort the page values lists based on page numbers
Collections.sort(entry.getValue(), Comparator.comparingInt(o -> o.getValue()));
entry.getValue().sort(Comparator.comparingInt(Map.Entry::getValue));
// collect all page values lists into a single list
List<Object> columnValues = entry.getValue().stream()
.map(Map.Entry::getKey).flatMap(i -> i.stream()).collect(Collectors.toList());
.map(Map.Entry::getKey).flatMap(Collection::stream).collect(Collectors.toList());
columnValuesMap.put(entry.getKey(), columnValues);
}

15
hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/IndexRecordManager.java
View File

@ -61,12 +61,12 @@ public class IndexRecordManager
throws IOException
{
Path recordFile = root.resolve(RECORD_FILE_NAME);
List<IndexRecord> records = new ArrayList<>();
ImmutableList.Builder<IndexRecord> records = ImmutableList.builder();
if (!fs.exists(recordFile)) {
synchronized (cacheLock) {
// invalidate cache
cache = records;
cache = records.build();
cacheLastModifiedTime = 0;
}
return cache;
@ -91,7 +91,7 @@ public class IndexRecordManager
throw new IllegalArgumentException(
"Error reading index record. Index record storage has been updated. Please delete old index directory and recreate the indices.");
}
cache = records;
cache = records.build();
cacheLastModifiedTime = modifiedTime;
}
}
@ -138,7 +138,7 @@ public class IndexRecordManager
FileBasedLock lock = new FileBasedLock(fs, root);
try {
lock.lock();
List<IndexRecord> records = getIndexRecords();
List<IndexRecord> records = new ArrayList<>(getIndexRecords()); // read from records and make a copy
Iterator<IndexRecord> iterator = records.iterator();
while (iterator.hasNext()) {
IndexRecord record = iterator.next();
@ -162,7 +162,7 @@ public class IndexRecordManager
FileBasedLock lock = new FileBasedLock(fs, root);
try {
lock.lock();
List<IndexRecord> records = getIndexRecords();
List<IndexRecord> records = new ArrayList<>(getIndexRecords()); // read from records and make a copy
if (partitionsToRemove.isEmpty()) {
// remove record
records.removeIf(record -> record.name.equals(name));
@ -174,6 +174,7 @@ public class IndexRecordManager
IndexRecord record = iterator.next();
if (record.name.equals(name)) {
record.partitions.removeAll(partitionsToRemove);
record.setLastModifiedTime(System.currentTimeMillis());
if (record.partitions.isEmpty()) {
iterator.remove();
}
@ -196,11 +197,9 @@ public class IndexRecordManager
{
Path recordFile = root.resolve(RECORD_FILE_NAME);
boolean writeHead = false;
try (OutputStream os = fs.newOutputStream(recordFile)) {
// Use IndexRecord to generate a special "entry" as table head so it's easier to maintain when csv format changes
String head = new IndexRecord("Name", "User", "Table", new String[] {"Columns"}, "IndexType",
ImmutableList.of("Properties"), ImmutableList.of("Partitions")).toCsvRecord();
String head = IndexRecord.getHeader();
os.write(head.getBytes());
for (IndexRecord record : records) {
os.write(record.toCsvRecord().getBytes());

145
hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/util/IndexCommandUtils.java
View File

@ -1,145 +0,0 @@
/*
* Copyright (C) 2018-2020. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.hetu.core.heuristicindex.util;
import com.google.common.collect.ImmutableSet;
import io.hetu.core.common.util.SecurePathWhiteList;
import io.hetu.core.filesystem.HdfsFileSystemClientFactory;
import io.hetu.core.filesystem.LocalFileSystemClientFactory;
import io.hetu.core.heuristicindex.HeuristicIndexFactory;
import io.prestosql.spi.filesystem.HetuFileSystemClient;
import io.prestosql.spi.filesystem.HetuFileSystemClientFactory;
import io.prestosql.spi.heuristicindex.IndexFactory;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Properties;
import java.util.Set;
import static com.google.common.base.Preconditions.checkArgument;
import static java.util.Objects.requireNonNull;
public class IndexCommandUtils
{
// Strong coupling with filesystem client
private static final Set<HetuFileSystemClientFactory> availableFactories = ImmutableSet.of(new LocalFileSystemClientFactory(), new HdfsFileSystemClientFactory());
private IndexCommandUtils()
{
}
public static IndexFactory getIndexFactory()
{
return new HeuristicIndexFactory();
}
/**
* Loads the properties from the corresponding catalog properties file
*
* @param fullyQualifiedTableName Fully qualified table name used to search for the catalog file,
* i.e. catalog.schema.table
* @param configRootDir The path to the directory that contains the configuration files
* @return Properties object with the corresponding DataSource properties loaded from the configuration files
* @throws IOException Thrown by reading the configuration files
*/
public static Properties loadDataSourceProperties(String fullyQualifiedTableName, String configRootDir)
throws IOException
{
String[] parts = IndexServiceUtils.getTableParts(fullyQualifiedTableName);
String catalog = parts[0];
// load the catalog properties in catalog dir
File catalogPropertiesFile = Paths.get(configRootDir, IndexConstants.CATALOG_CONFIGS_DIR, catalog + ".properties").toFile();
return IndexServiceUtils.loadProperties(catalogPropertiesFile);
}
/**
* Load the properties from config.properties related to indexstore, including the root dir and filesystem profile
*
* @param configRootDir The path to the directory that contains the configuration files
* @return Properties object with global indexer properties loaded from the configuration files
* @throws IOException When IOException occurs during reading config files or initializing filesystem client
*/
public static IndexStore loadIndexStore(String configRootDir)
throws IOException
{
// load all properties from config.properties
File configPropertiesFile = Paths.get(configRootDir, IndexConstants.CONFIG_FILE).toFile();
Properties properties = IndexServiceUtils.loadProperties(configPropertiesFile);
Path root = Paths.get(requireNonNull(properties.getProperty(IndexConstants.INDEXSTORE_URI_KEY),
IndexConstants.INDEXSTORE_URI_KEY + " is not set in config.properties"));
try {
checkArgument(!root.toString().contains("../"), "Index store directory path must be absolute");
checkArgument(SecurePathWhiteList.isSecurePath(root.toString()),
"Index store directory path must be at user workspace " + SecurePathWhiteList.getSecurePathWhiteList().toString());
}
catch (IOException e) {
throw new IllegalArgumentException("Failed to get secure path list.", e);
}
String fileSystemProfileName = requireNonNull(properties.getProperty(IndexConstants.INDEXSTORE_FILESYSTEM_PROFILE_KEY),
IndexConstants.INDEXSTORE_FILESYSTEM_PROFILE_KEY + " is not set in config.properties");
File fileSystemProfile = new File(String.format("%s/filesystem/%s.properties", configRootDir, fileSystemProfileName));
if (!fileSystemProfile.exists()) {
throw new FileNotFoundException(String.format("Filesystem profile '%s' not found", fileSystemProfileName));
}
// Strong coupling with filesystem client, change when modifying filesystem client profile
try (InputStream is = new FileInputStream(fileSystemProfile)) {
Properties fsConfig = new Properties();
fsConfig.load(is);
String fsType = fsConfig.getProperty("fs.client.type");
for (HetuFileSystemClientFactory factory : availableFactories) {
if (fsType.equalsIgnoreCase(factory.getName())) {
return new IndexStore(factory.getFileSystemClient(fsConfig, root), root);
}
}
throw new IllegalArgumentException(String.format("fs.client.type '%s' has no registered factory", fsType));
}
}
public static class IndexStore
{
final HetuFileSystemClient fs;
final Path root;
public IndexStore(HetuFileSystemClient fs, Path root)
{
this.fs = fs;
this.root = root;
}
public HetuFileSystemClient getFs()
{
return fs;
}
public Path getRoot()
{
return root;
}
}
}

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

@ -23,16 +23,13 @@ import io.prestosql.spi.heuristicindex.IndexRecord;
import org.testng.annotations.Test;
import java.io.IOException;
import java.lang.reflect.Field;
import java.nio.file.Paths;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Properties;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNotSame;
import static org.testng.Assert.assertNull;
@ -227,48 +224,8 @@ public class TestIndexRecordManager
testIndexRecordAddLookUpHelper("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList(), ImmutableList.of("12", "123"));
}
@Test
public void testRecordEqualAndHash()
{
IndexRecord r1 = new IndexRecord("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList(), Collections.emptyList());
IndexRecord r2 = new IndexRecord("testName", "testUser", "testTable", new String[] {
"testColumn"}, "minmax", Collections.emptyList(), ImmutableList.of("note"));
IndexRecord r3 = new IndexRecord("testName", "testUser", "testTable", new String[] {"testColumn"}, "bloom", Collections.emptyList(), Collections.emptyList());
IndexRecord r4 = new IndexRecord("testName", "testUser", "testTable", new String[] {"testColumn",
"testColumn2"}, "minmax", Collections.emptyList(), Collections.emptyList());
assertEquals(r1, r1);
assertEquals(r1, r2);
assertNotEquals(r1, r3);
assertNotEquals(r1, r4);
HashSet<IndexRecord> testSet = new HashSet<>();
testSet.add(r1);
assertEquals(testSet.size(), 1);
testSet.add(r2);
assertEquals(testSet.size(), 1);
testSet.add(r3);
assertEquals(testSet.size(), 2);
testSet.add(r4);
assertEquals(testSet.size(), 3);
}
@Test(expectedExceptions = AssertionError.class)
public void testAddAndLookUpDifferentPartitions()
throws IOException, IllegalAccessException
{
try (TempFolder folder = new TempFolder()) {
folder.create();
IndexRecordManager indexRecordManager = new IndexRecordManager(FILE_SYSTEM_CLIENT, folder.getRoot().toPath());
IndexRecord expected = new IndexRecord("testName", "testUser", "testTable", new String[] {
"testColumn"}, "minmax", Collections.emptyList(), ImmutableList.of(""));
indexRecordManager.addIndexRecord("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList(), Arrays.asList("cp=1"));
IndexRecord actual = indexRecordManager.lookUpIndexRecord("testName");
assertIndexRecordFullyEqual(actual, expected);
}
}
private void testIndexRecordAddLookUpHelper(String name, String user, String table, String[] columns, String indexType, List<String> indexProperties, List<String> partitions)
throws IOException, IllegalAccessException
throws IOException
{
try (TempFolder folder = new TempFolder()) {
folder.create();
@ -278,22 +235,11 @@ public class TestIndexRecordManager
IndexRecord actual1 = indexRecordManager.lookUpIndexRecord(name);
assertNotNull(actual1);
assertIndexRecordFullyEqual(actual1, expected);
assertEquals(actual1, expected);
IndexRecord actual2 = indexRecordManager.lookUpIndexRecord(table, columns, indexType);
assertNotNull(actual2);
assertIndexRecordFullyEqual(actual2, expected);
}
}
// Compare two IndexRecord objects and assert all fields are equal.
// Unlike the equals() method of IndexRecord, this method compares ALL fields for testing.
private void assertIndexRecordFullyEqual(IndexRecord actual, IndexRecord expected)
throws IllegalAccessException
{
for (Field field : actual.getClass().getDeclaredFields()) {
field.setAccessible(true);
assertEquals(field.get(actual), field.get(expected));
assertEquals(actual2, expected);
}
}
}

62
presto-hive/src/main/java/io/prestosql/plugin/hive/util/IndexCache.java
View File

@ -18,7 +18,6 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.cache.Weigher;
import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.google.inject.Inject;
@ -27,8 +26,10 @@ import io.hetu.core.common.heuristicindex.IndexCacheKey;
import io.prestosql.plugin.hive.HiveColumnHandle;
import io.prestosql.plugin.hive.HiveSplit;
import io.prestosql.spi.HetuConstant;
import io.prestosql.spi.heuristicindex.IndexClient;
import io.prestosql.spi.heuristicindex.IndexMetadata;
import io.prestosql.spi.heuristicindex.IndexNotCreatedException;
import io.prestosql.spi.heuristicindex.IndexRecord;
import io.prestosql.spi.predicate.TupleDomain;
import io.prestosql.spi.service.PropertyService;
import org.apache.hadoop.fs.Path;
@ -56,18 +57,21 @@ public class IndexCache
private static ScheduledExecutorService executor;
private Long loadDelay; // in millisecond
private Long refreshRate; // in millisecond
private LoadingCache<IndexCacheKey, List<IndexMetadata>> cache;
private List<IndexRecord> indexRecords;
@Inject
public IndexCache(CacheLoader loader)
public IndexCache(CacheLoader loader, IndexClient indexClient)
{
// If the static variables have not been initialized
if (PropertyService.getBooleanProperty(HetuConstant.FILTER_ENABLED)) {
loadDelay = PropertyService.getDurationProperty(HetuConstant.FILTER_CACHE_LOADING_DELAY).toMillis();
refreshRate = Math.min(loadDelay / 2, 5000L);
int numThreads = Math.min(Runtime.getRuntime().availableProcessors(), PropertyService.getLongProperty(HetuConstant.FILTER_CACHE_LOADING_THREADS).intValue());
executor = Executors.newScheduledThreadPool(numThreads, threadFactory);
CacheBuilder cacheBuilder = CacheBuilder.newBuilder()
.removalListener(e -> ((List<IndexMetadata>) e.getValue()).stream().forEach(i -> {
CacheBuilder<IndexCacheKey, List<IndexMetadata>> cacheBuilder = CacheBuilder.newBuilder()
.removalListener(e -> ((List<IndexMetadata>) e.getValue()).forEach(i -> {
try {
i.getIndex().close();
}
@ -77,7 +81,7 @@ public class IndexCache
}))
.expireAfterWrite(PropertyService.getDurationProperty(HetuConstant.FILTER_CACHE_TTL).toMillis(), TimeUnit.MILLISECONDS)
.maximumWeight(PropertyService.getLongProperty(HetuConstant.FILTER_CACHE_MAX_MEMORY))
.weigher((Weigher<IndexCacheKey, List<IndexMetadata>>) (indexCacheKey, indices) -> {
.weigher((indexCacheKey, indices) -> {
int memorySize = 0;
for (IndexMetadata indexMetadata : indices) {
// HetuConstant.FILTER_CACHE_MAX_MEMORY is set in KBs
@ -89,15 +93,45 @@ public class IndexCache
if (PropertyService.getBooleanProperty(HetuConstant.FILTER_CACHE_SOFT_REFERENCE)) {
cacheBuilder.softValues();
}
executor.scheduleAtFixedRate(() -> {
try {
List<IndexRecord> newRecords = indexClient.getAllIndexRecords();
if (indexRecords != null) {
for (IndexRecord old : indexRecords) {
boolean found = false;
for (IndexRecord now : newRecords) {
if (now.name.equals(old.name)) {
found = true;
if (now.getLastModifiedTime() != old.getLastModifiedTime()) {
// index record has been updated. evict
evictFromCache(old);
LOG.debug("Index for {%s} has been evicted from cache because the index has been updated.", old);
}
}
}
// old record is gone. evict from cache
if (!found) {
evictFromCache(old);
LOG.debug("Index for {%s} has been evicted from cache because the index has been dropped.", old);
}
}
}
indexRecords = newRecords;
}
catch (Exception e) {
LOG.debug(e, "Error using index records to refresh cache");
}
}, loadDelay, refreshRate, TimeUnit.MILLISECONDS);
cache = cacheBuilder.build(loader);
}
}
// Override the loadDelay, for testing
public IndexCache(CacheLoader loader, Long loadDelay)
public IndexCache(CacheLoader<IndexCacheKey, List<IndexMetadata>> loader, Long loadDelay, IndexClient indexClient)
{
this(loader);
this(loader, indexClient);
this.loadDelay = loadDelay;
}
@ -168,9 +202,7 @@ public class IndexCache
}
// cache contained the key
if (predicateIndexes != null) {
splitIndexes.addAll(predicateIndexes);
}
splitIndexes.addAll(predicateIndexes);
}
}
});
@ -183,4 +215,14 @@ public class IndexCache
{
return cache.size();
}
private void evictFromCache(IndexRecord record)
{
String recordInCacheKey = String.format("%s/%s/%s", record.table, String.join(",", record.columns), record.indexType);
for (IndexCacheKey key : cache.asMap().keySet()) {
if (key.getPath().startsWith(recordInCacheKey)) {
cache.invalidate(key);
}
}
}
}

3
presto-hive/src/test/java/io/prestosql/plugin/hive/HiveTestUtils.java
View File

@ -49,6 +49,7 @@ import io.prestosql.spi.type.Type;
import io.prestosql.spi.type.TypeManager;
import io.prestosql.spi.type.TypeSignatureParameter;
import io.prestosql.spi.util.BloomFilter;
import io.prestosql.testing.NoOpIndexClient;
import io.prestosql.testing.TestingConnectorSession;
import io.prestosql.type.InternalTypeManager;
@ -125,7 +126,7 @@ public final class HiveTestUtils
public static IndexCache getNoOpIndexCache()
{
return new IndexCache(new IndexCacheLoader(null))
return new IndexCache(new IndexCacheLoader(null), new NoOpIndexClient())
{
@Override
public List<IndexMetadata> getIndices(String catalog, String table, HiveSplit hiveSplit, TupleDomain<HiveColumnHandle> effectivePredicate, List<HiveColumnHandle> partitions)

11
presto-hive/src/test/java/io/prestosql/plugin/hive/util/TestIndexCache.java
View File

@ -27,6 +27,7 @@ import io.prestosql.spi.predicate.Domain;
import io.prestosql.spi.predicate.TupleDomain;
import io.prestosql.spi.predicate.ValueSet;
import io.prestosql.spi.service.PropertyService;
import io.prestosql.testing.NoOpIndexClient;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
@ -102,7 +103,7 @@ public class TestIndexCache
IndexCacheLoader indexCacheLoader = mock(IndexCacheLoader.class);
when(indexCacheLoader.load(any())).thenReturn(expectedIndices);
IndexCache indexCache = new IndexCache(indexCacheLoader);
IndexCache indexCache = new IndexCache(indexCacheLoader, new NoOpIndexClient());
List<IndexMetadata> actualSplitIndex = indexCache.getIndices(catalog, table, testHiveSplit,
effectivePredicate, testPartitions);
assertEquals(actualSplitIndex.size(), 0);
@ -128,7 +129,7 @@ public class TestIndexCache
IndexCacheLoader indexCacheLoader = mock(IndexCacheLoader.class);
when(indexCacheLoader.load(any())).thenThrow(ExecutionException.class);
IndexCache indexCache = new IndexCache(indexCacheLoader, loadDelay);
IndexCache indexCache = new IndexCache(indexCacheLoader, loadDelay, new NoOpIndexClient());
List<IndexMetadata> actualSplitIndex = indexCache.getIndices(catalog, table, testHiveSplit,
effectivePredicate, testPartitions);
assertEquals(actualSplitIndex.size(), 0);
@ -155,7 +156,7 @@ public class TestIndexCache
IndexCacheLoader indexCacheLoader = mock(IndexCacheLoader.class);
when(indexCacheLoader.load(any())).thenReturn(expectedIndices);
IndexCache indexCache = new IndexCache(indexCacheLoader, loadDelay);
IndexCache indexCache = new IndexCache(indexCacheLoader, loadDelay, new NoOpIndexClient());
List<IndexMetadata> actualSplitIndex = indexCache.getIndices(catalog, table, testHiveSplit,
effectivePredicate, testPartitions);
assertEquals(actualSplitIndex.size(), 0);
@ -190,7 +191,7 @@ public class TestIndexCache
IndexCacheLoader indexCacheLoader = mock(IndexCacheLoader.class);
when(indexCacheLoader.load(any())).thenReturn(expectedIndices);
IndexCache indexCache = new IndexCache(indexCacheLoader, loadDelay);
IndexCache indexCache = new IndexCache(indexCacheLoader, loadDelay, new NoOpIndexClient());
List<IndexMetadata> actualSplitIndex = indexCache.getIndices(catalog, table, testHiveSplit,
effectivePredicateForPartition, partitionColumns);
assertEquals(actualSplitIndex.size(), 0);
@ -207,7 +208,7 @@ public class TestIndexCache
{
synchronized (this) {
IndexCacheLoader indexCacheLoader = mock(IndexCacheLoader.class);
IndexCache indexCache = new IndexCache(indexCacheLoader, loadDelay);
IndexCache indexCache = new IndexCache(indexCacheLoader, loadDelay, new NoOpIndexClient());
when(testHiveSplit.getLastModifiedTime()).thenReturn(testLastModifiedTime);
//get index for split1

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

@ -18,7 +18,6 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.cache.Weigher;
import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import io.airlift.log.Logger;
@ -26,12 +25,15 @@ import io.hetu.core.common.heuristicindex.IndexCacheKey;
import io.prestosql.metadata.Split;
import io.prestosql.spi.HetuConstant;
import io.prestosql.spi.heuristicindex.Index;
import io.prestosql.spi.heuristicindex.IndexClient;
import io.prestosql.spi.heuristicindex.IndexMetadata;
import io.prestosql.spi.heuristicindex.IndexNotCreatedException;
import io.prestosql.spi.heuristicindex.IndexRecord;
import io.prestosql.spi.service.PropertyService;
import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
@ -54,15 +56,18 @@ public class IndexCache
private Long loadDelay; // in millisecond
private LoadingCache<IndexCacheKey, List<IndexMetadata>> cache;
private final List<IndexRecord> indexRecords = new ArrayList<>();
public IndexCache(CacheLoader loader)
public IndexCache(CacheLoader loader, IndexClient indexClient)
{
// If the static variables have not been initialized
if (PropertyService.getBooleanProperty(HetuConstant.FILTER_ENABLED)) {
loadDelay = PropertyService.getDurationProperty(HetuConstant.FILTER_CACHE_LOADING_DELAY).toMillis();
// in millisecond
long refreshRate = Math.min(loadDelay / 2, 5000L);
int numThreads = Math.min(Runtime.getRuntime().availableProcessors(), PropertyService.getLongProperty(HetuConstant.FILTER_CACHE_LOADING_THREADS).intValue());
executor = Executors.newScheduledThreadPool(numThreads, threadFactory);
CacheBuilder cacheBuilder = CacheBuilder.newBuilder()
CacheBuilder<IndexCacheKey, List<IndexMetadata>> cacheBuilder = CacheBuilder.newBuilder()
.removalListener(e -> ((List<IndexMetadata>) e.getValue()).stream().forEach(i -> {
try {
i.getIndex().close();
@ -73,7 +78,7 @@ public class IndexCache
}))
.expireAfterWrite(PropertyService.getDurationProperty(HetuConstant.FILTER_CACHE_TTL).toMillis(), TimeUnit.MILLISECONDS)
.maximumWeight(PropertyService.getLongProperty(HetuConstant.FILTER_CACHE_MAX_MEMORY))
.weigher((Weigher<IndexCacheKey, List<IndexMetadata>>) (indexCacheKey, indices) -> {
.weigher((indexCacheKey, indices) -> {
int memorySize = 0;
for (IndexMetadata indexMetadata : indices) {
// HetuConstant.FILTER_CACHE_MAX_MEMORY is set in KBs
@ -85,6 +90,44 @@ public class IndexCache
if (PropertyService.getBooleanProperty(HetuConstant.FILTER_CACHE_SOFT_REFERENCE)) {
cacheBuilder.softValues();
}
executor.scheduleAtFixedRate(() -> {
try {
synchronized (indexRecords) {
List<IndexRecord> newRecords = indexClient.getAllIndexRecords();
if (indexRecords.isEmpty()) {
indexRecords.addAll(newRecords);
}
else {
for (IndexRecord old : indexRecords) {
boolean found = false;
for (IndexRecord now : newRecords) {
if (now.name.equals(old.name)) {
found = true;
if (now.getLastModifiedTime() != old.getLastModifiedTime()) {
// index record has been updated. evict
evictFromCache(old);
LOG.debug("Index for {%s} has been evicted from cache because the index has been updated.", old);
indexRecords.clear();
indexRecords.addAll(newRecords);
}
}
}
// old record is gone. evict from cache
if (!found) {
evictFromCache(old);
LOG.debug("Index for {%s} has been evicted from cache because the index has been dropped.", old);
indexRecords.clear();
indexRecords.addAll(newRecords);
}
}
}
}
}
catch (Exception e) {
LOG.debug(e, "Error using index records to refresh cache");
}
}, loadDelay, refreshRate, TimeUnit.MILLISECONDS);
cache = cacheBuilder.build(loader);
}
}
@ -206,4 +249,14 @@ public class IndexCache
{
return cache.size();
}
private void evictFromCache(IndexRecord record)
{
String recordInCacheKey = String.format("%s/%s/%s", record.table, String.join(",", record.columns), record.indexType);
for (IndexCacheKey key : cache.asMap().keySet()) {
if (key.getPath().startsWith(recordInCacheKey)) {
cache.invalidate(key);
}
}
}
}

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

@ -82,7 +82,7 @@ public class SplitFiltering
{
if (indexCache == null) {
CacheLoader<IndexCacheKey, List<IndexMetadata>> cacheLoader = new IndexCacheLoader(indexClient);
indexCache = new IndexCache(cacheLoader);
indexCache = new IndexCache(cacheLoader, indexClient);
}
}

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

@ -180,13 +180,9 @@ public class CreateIndexOperator
switch (createIndexMetadata.getCreateLevel()) {
case STRIPE: {
String filePath = page.getPageMetadata().getProperty(HetuConstant.DATASOURCE_FILE_PATH);
IndexWriter indexWriter = levelWriter.computeIfAbsent(filePath,
k -> {
IndexWriter writer = heuristicIndexerManager.getIndexWriter(createIndexMetadata, connectorMetadata);
persistBy.putIfAbsent(writer, this);
return writer;
});
indexWriter.addData(values, connectorMetadata);
levelWriter.computeIfAbsent(filePath, k -> heuristicIndexerManager.getIndexWriter(createIndexMetadata, connectorMetadata));
persistBy.putIfAbsent(levelWriter.get(filePath), this);
levelWriter.get(filePath).addData(values, connectorMetadata);
break;
}
case PARTITION: {

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

@ -1044,11 +1044,13 @@ class StatementAnalyzer
partitions = HeuristicIndexUtils.extractPartitions(createIndex.getExpression().get());
// check partition name validate, create index …… where pt_d = xxx;
// pt_d must be partition column
List<String> partitionColumns = partitions.stream().map(k -> k.substring(0, k.indexOf("="))).collect(Collectors.toList());
Set<String> partitionColumns = partitions.stream().map(k -> k.substring(0, k.indexOf("="))).collect(Collectors.toSet());
if (partitionColumns.size() > 1) {
// currently only support one partition column
throw new IllegalArgumentException("Heuristic index only supports predicates on one column");
}
partitionColumn = partitionColumns.get(0);
// The only entry in set should be the only partition column name
partitionColumn = partitionColumns.iterator().next();
}
Optional<TableHandle> tableHandle = metadata.getTableHandle(session, tableFullName);

40
presto-main/src/main/java/io/prestosql/utils/HeuristicIndexUtils.java
View File

@ -17,9 +17,12 @@ package io.prestosql.utils;
import io.prestosql.sql.parser.ParsingException;
import io.prestosql.sql.tree.ComparisonExpression;
import io.prestosql.sql.tree.Expression;
import io.prestosql.sql.tree.InListExpression;
import io.prestosql.sql.tree.InPredicate;
import io.prestosql.sql.tree.LogicalBinaryExpression;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@ -35,30 +38,33 @@ public class HeuristicIndexUtils
if (expression instanceof ComparisonExpression) {
ComparisonExpression exp = (ComparisonExpression) expression;
if (exp.getOperator() != ComparisonExpression.Operator.EQUAL) {
throw new ParsingException("Unsupported WHERE expression. Only equality expressions are supported with OR operator, " +
"e.g. partition=1, partition=1 OR partition=2");
if (exp.getOperator() == ComparisonExpression.Operator.EQUAL) {
return Collections.singletonList(exp.getLeft().toString() + "=" + parseSpecialPartitionValues(exp.getRight().toString()));
}
// check predicate column is validate partition column.
return Collections.singletonList(exp.getLeft().toString() + "=" + parseSpecialPartitionValues(exp.getRight().toString()));
}
else if (expression instanceof LogicalBinaryExpression) {
LogicalBinaryExpression exp = (LogicalBinaryExpression) expression;
if (exp.getOperator() != LogicalBinaryExpression.Operator.OR) {
throw new ParsingException("Unsupported WHERE expression. Only equality expressions are supported with OR operator. " +
"e.g. partition=1, partition=1 OR partition=2");
if (exp.getOperator() == LogicalBinaryExpression.Operator.OR) {
Expression left = exp.getLeft();
Expression right = exp.getRight();
return Stream.concat(extractPartitions(left).stream(), extractPartitions(right).stream()).collect(Collectors.toList());
}
}
else if (expression instanceof InPredicate) {
Expression valueList = ((InPredicate) expression).getValueList();
if (valueList instanceof InListExpression) {
InListExpression inListExpression = (InListExpression) valueList;
List<String> res = new LinkedList<>();
for (Expression expr : inListExpression.getValues()) {
res.add(((InPredicate) expression).getValue().toString() + "=" + parseSpecialPartitionValues(expr.toString()));
}
return res;
}
}
Expression left = exp.getLeft();
Expression right = exp.getRight();
return Stream.concat(extractPartitions(left).stream(), extractPartitions(right).stream()).collect(Collectors.toList());
}
else {
throw new ParsingException("Unsupported WHERE expression. Only equality expressions are supported with OR operator. " +
"e.g. partition=1, partition=1 OR partition=2");
}
throw new ParsingException("Unsupported WHERE expression. Only in-predicate/equality-expression are supported" +
"e.g. partition=1, partition in (1,2)");
}
private static String parseSpecialPartitionValues(String rightVal)

9
presto-main/src/test/java/io/prestosql/heuristicindex/TestIndexCache.java
View File

@ -24,6 +24,7 @@ import io.prestosql.spi.connector.ConnectorSplit;
import io.prestosql.spi.heuristicindex.Index;
import io.prestosql.spi.heuristicindex.IndexMetadata;
import io.prestosql.spi.service.PropertyService;
import io.prestosql.testing.NoOpIndexClient;
import org.mockito.internal.stubbing.answers.Returns;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
@ -86,7 +87,7 @@ public class TestIndexCache
IndexCacheLoader indexCacheLoader = mock(IndexCacheLoader.class);
when(indexCacheLoader.load(any())).then(new Returns(expectedIndices));
IndexCache indexCache = new IndexCache(indexCacheLoader);
IndexCache indexCache = new IndexCache(indexCacheLoader, new NoOpIndexClient());
List<IndexMetadata> actualSplitIndex = indexCache.getIndices(table, column, split);
assertEquals(actualSplitIndex.size(), 0);
Thread.sleep(loadDelay + 1000);
@ -108,7 +109,7 @@ public class TestIndexCache
IndexCacheLoader indexCacheLoader = mock(IndexCacheLoader.class);
when(indexCacheLoader.load(any())).thenThrow(ExecutionException.class);
IndexCache indexCache = new IndexCache(indexCacheLoader);
IndexCache indexCache = new IndexCache(indexCacheLoader, new NoOpIndexClient());
List<IndexMetadata> actualSplitIndex = indexCache.getIndices(table, column, split);
assertEquals(actualSplitIndex.size(), 0);
Thread.sleep(loadDelay + 500);
@ -132,7 +133,7 @@ public class TestIndexCache
IndexCacheLoader indexCacheLoader = mock(IndexCacheLoader.class);
when(indexCacheLoader.load(any())).then(new Returns(expectedIndices));
IndexCache indexCache = new IndexCache(indexCacheLoader);
IndexCache indexCache = new IndexCache(indexCacheLoader, new NoOpIndexClient());
List<IndexMetadata> actualSplitIndex = indexCache.getIndices(table, column, split);
assertEquals(actualSplitIndex.size(), 0);
Thread.sleep(loadDelay + 500);
@ -150,7 +151,7 @@ public class TestIndexCache
{
when(connectorSplit.getLastModifiedTime()).thenReturn(testLastModifiedTime);
IndexCacheLoader indexCacheLoader = mock(IndexCacheLoader.class);
IndexCache indexCache = new IndexCache(indexCacheLoader);
IndexCache indexCache = new IndexCache(indexCacheLoader, new NoOpIndexClient());
//get index for split1
IndexMetadata indexMetadata1 = mock(IndexMetadata.class);

36
presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexRecord.java
View File

@ -29,6 +29,7 @@ public class IndexRecord
public final String indexType;
public final List<String> properties;
public final List<String> partitions;
private long lastModifiedTime;
public IndexRecord(String name, String user, String table, String[] columns, String indexType, List<String> properties, List<String> partitions)
{
@ -39,6 +40,7 @@ public class IndexRecord
this.indexType = indexType;
this.properties = properties;
this.partitions = partitions;
this.lastModifiedTime = System.currentTimeMillis();
}
public IndexRecord(String csvRecord)
@ -51,12 +53,18 @@ public class IndexRecord
this.indexType = records[4];
this.properties = Arrays.stream(records[5].split(",")).filter(s -> !s.equals("")).collect(Collectors.toList());
this.partitions = Arrays.stream(records[6].split(",")).filter(s -> !s.equals("")).collect(Collectors.toList());
this.lastModifiedTime = Long.parseLong(records[7]);
}
public static String getHeader()
{
return String.format("%s|%s|%s|%s|%s|%s|%s|%s\n", "name", "user", "table", "columns", "indexType", "properties", "partitions", "lastModifiedTime");
}
public String toCsvRecord()
{
return String.format("%s|%s|%s|%s|%s|%s|%s\n", name, user, table, String.join(COLUMN_DELIMITER, columns), indexType,
String.join(",", properties), String.join(",", partitions));
return String.format("%s|%s|%s|%s|%s|%s|%s|%s\n", name, user, table, String.join(COLUMN_DELIMITER, columns), indexType,
String.join(",", properties), String.join(",", partitions), lastModifiedTime);
}
@Override
@ -79,7 +87,7 @@ public class IndexRecord
@Override
public int hashCode()
{
int result = Objects.hash(name, user, table, indexType);
int result = Objects.hash(name, user, table, columns, indexType);
result = 31 * result + Arrays.hashCode(columns);
return result;
}
@ -87,10 +95,22 @@ public class IndexRecord
@Override
public String toString()
{
return name + ","
+ user + ","
+ table + ","
+ "[" + String.join(",", columns) + "],"
+ indexType;
return "IndexRecord{" +
"name='" + name + '\'' +
", table='" + table + '\'' +
", columns=" + Arrays.toString(columns) +
", indexType='" + indexType + '\'' +
", lastModifiedTime=" + lastModifiedTime +
'}';
}
public long getLastModifiedTime()
{
return lastModifiedTime;
}
public void setLastModifiedTime(long lastModifiedTime)
{
this.lastModifiedTime = lastModifiedTime;
}
}