From 78dfd71c94e9c65df41e12dc9891fbc86b51323d Mon Sep 17 00:00:00 2001 From: Han Weng Date: Wed, 25 Nov 2020 16:43:13 -0500 Subject: [PATCH] Check index record and evict entries from cache when updated --- .../core/heuristicindex/FileIndexWriter.java | 5 +- .../heuristicindex/IndexRecordManager.java | 15 +- .../util/IndexCommandUtils.java | 145 ------------------ .../TestIndexRecordManager.java | 60 +------- .../plugin/hive/util/IndexCache.java | 62 ++++++-- .../prestosql/plugin/hive/HiveTestUtils.java | 3 +- .../plugin/hive/util/TestIndexCache.java | 11 +- .../prestosql/heuristicindex/IndexCache.java | 61 +++++++- .../heuristicindex/SplitFiltering.java | 2 +- .../operator/CreateIndexOperator.java | 10 +- .../sql/analyzer/StatementAnalyzer.java | 6 +- .../prestosql/utils/HeuristicIndexUtils.java | 40 +++-- .../heuristicindex/TestIndexCache.java | 9 +- .../spi/heuristicindex/IndexRecord.java | 36 ++++- 14 files changed, 194 insertions(+), 271 deletions(-) delete mode 100644 hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/util/IndexCommandUtils.java diff --git a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/FileIndexWriter.java b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/FileIndexWriter.java index f21f994a0..2e37e6b3b 100644 --- a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/FileIndexWriter.java +++ b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/FileIndexWriter.java @@ -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> for (Map.Entry, 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 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); } diff --git a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/IndexRecordManager.java b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/IndexRecordManager.java index 1efb6638b..9768c04ee 100644 --- a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/IndexRecordManager.java +++ b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/IndexRecordManager.java @@ -61,12 +61,12 @@ public class IndexRecordManager throws IOException { Path recordFile = root.resolve(RECORD_FILE_NAME); - List records = new ArrayList<>(); + ImmutableList.Builder 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 records = getIndexRecords(); + List records = new ArrayList<>(getIndexRecords()); // read from records and make a copy Iterator 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 records = getIndexRecords(); + List 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()); diff --git a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/util/IndexCommandUtils.java b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/util/IndexCommandUtils.java deleted file mode 100644 index 27d68cadf..000000000 --- a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/util/IndexCommandUtils.java +++ /dev/null @@ -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 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; - } - } -} diff --git a/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestIndexRecordManager.java b/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestIndexRecordManager.java index 76dc7c279..f10e094cf 100644 --- a/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestIndexRecordManager.java +++ b/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestIndexRecordManager.java @@ -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 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 indexProperties, List 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); } } } diff --git a/presto-hive/src/main/java/io/prestosql/plugin/hive/util/IndexCache.java b/presto-hive/src/main/java/io/prestosql/plugin/hive/util/IndexCache.java index 4480c4d78..3d0b62c7d 100644 --- a/presto-hive/src/main/java/io/prestosql/plugin/hive/util/IndexCache.java +++ b/presto-hive/src/main/java/io/prestosql/plugin/hive/util/IndexCache.java @@ -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> cache; + private List 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) e.getValue()).stream().forEach(i -> { + CacheBuilder> cacheBuilder = CacheBuilder.newBuilder() + .removalListener(e -> ((List) 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, 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 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> 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); + } + } + } } diff --git a/presto-hive/src/test/java/io/prestosql/plugin/hive/HiveTestUtils.java b/presto-hive/src/test/java/io/prestosql/plugin/hive/HiveTestUtils.java index 3d3909e47..6835db21c 100644 --- a/presto-hive/src/test/java/io/prestosql/plugin/hive/HiveTestUtils.java +++ b/presto-hive/src/test/java/io/prestosql/plugin/hive/HiveTestUtils.java @@ -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 getIndices(String catalog, String table, HiveSplit hiveSplit, TupleDomain effectivePredicate, List partitions) diff --git a/presto-hive/src/test/java/io/prestosql/plugin/hive/util/TestIndexCache.java b/presto-hive/src/test/java/io/prestosql/plugin/hive/util/TestIndexCache.java index 82a7305ba..e660fa7ae 100644 --- a/presto-hive/src/test/java/io/prestosql/plugin/hive/util/TestIndexCache.java +++ b/presto-hive/src/test/java/io/prestosql/plugin/hive/util/TestIndexCache.java @@ -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 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 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 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 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 diff --git a/presto-main/src/main/java/io/prestosql/heuristicindex/IndexCache.java b/presto-main/src/main/java/io/prestosql/heuristicindex/IndexCache.java index 294f96414..38632b368 100644 --- a/presto-main/src/main/java/io/prestosql/heuristicindex/IndexCache.java +++ b/presto-main/src/main/java/io/prestosql/heuristicindex/IndexCache.java @@ -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> cache; + private final List 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> cacheBuilder = CacheBuilder.newBuilder() .removalListener(e -> ((List) 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, 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 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); + } + } + } } diff --git a/presto-main/src/main/java/io/prestosql/heuristicindex/SplitFiltering.java b/presto-main/src/main/java/io/prestosql/heuristicindex/SplitFiltering.java index 309a057bb..c18273b54 100644 --- a/presto-main/src/main/java/io/prestosql/heuristicindex/SplitFiltering.java +++ b/presto-main/src/main/java/io/prestosql/heuristicindex/SplitFiltering.java @@ -82,7 +82,7 @@ public class SplitFiltering { if (indexCache == null) { CacheLoader> cacheLoader = new IndexCacheLoader(indexClient); - indexCache = new IndexCache(cacheLoader); + indexCache = new IndexCache(cacheLoader, indexClient); } } diff --git a/presto-main/src/main/java/io/prestosql/operator/CreateIndexOperator.java b/presto-main/src/main/java/io/prestosql/operator/CreateIndexOperator.java index 48aede277..8ad0f3c1d 100644 --- a/presto-main/src/main/java/io/prestosql/operator/CreateIndexOperator.java +++ b/presto-main/src/main/java/io/prestosql/operator/CreateIndexOperator.java @@ -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: { diff --git a/presto-main/src/main/java/io/prestosql/sql/analyzer/StatementAnalyzer.java b/presto-main/src/main/java/io/prestosql/sql/analyzer/StatementAnalyzer.java index 02c3b6b6f..cbf6969d5 100644 --- a/presto-main/src/main/java/io/prestosql/sql/analyzer/StatementAnalyzer.java +++ b/presto-main/src/main/java/io/prestosql/sql/analyzer/StatementAnalyzer.java @@ -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 partitionColumns = partitions.stream().map(k -> k.substring(0, k.indexOf("="))).collect(Collectors.toList()); + Set 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 = metadata.getTableHandle(session, tableFullName); diff --git a/presto-main/src/main/java/io/prestosql/utils/HeuristicIndexUtils.java b/presto-main/src/main/java/io/prestosql/utils/HeuristicIndexUtils.java index 5a67ab0b1..1e7c9dadb 100644 --- a/presto-main/src/main/java/io/prestosql/utils/HeuristicIndexUtils.java +++ b/presto-main/src/main/java/io/prestosql/utils/HeuristicIndexUtils.java @@ -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 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) diff --git a/presto-main/src/test/java/io/prestosql/heuristicindex/TestIndexCache.java b/presto-main/src/test/java/io/prestosql/heuristicindex/TestIndexCache.java index 63975875a..9ebf8cf28 100644 --- a/presto-main/src/test/java/io/prestosql/heuristicindex/TestIndexCache.java +++ b/presto-main/src/test/java/io/prestosql/heuristicindex/TestIndexCache.java @@ -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 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 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 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); diff --git a/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexRecord.java b/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexRecord.java index 1bbbd59ce..3dc923876 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexRecord.java +++ b/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexRecord.java @@ -29,6 +29,7 @@ public class IndexRecord public final String indexType; public final List properties; public final List partitions; + private long lastModifiedTime; public IndexRecord(String name, String user, String table, String[] columns, String indexType, List properties, List 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; } }