Improve index record to cache entries and pre-screen indices to load
This commit is contained in:
Han Weng
parent
91a1dc9953
commit
ec8ea30f70
|
|
@ -47,7 +47,7 @@ import java.util.function.Function;
|
|||
import java.util.stream.Collectors;
|
||||
import java.util.stream.Stream;
|
||||
|
||||
import static io.hetu.core.heuristicindex.util.IndexConstants.COLUMN_DELIMITER;
|
||||
import static io.hetu.core.heuristicindex.IndexRecord.COLUMN_DELIMITER;
|
||||
import static java.util.Objects.requireNonNull;
|
||||
|
||||
/**
|
||||
|
|
@ -64,12 +64,14 @@ public class HeuristicIndexClient
|
|||
|
||||
private HetuFileSystemClient fs;
|
||||
private Path root;
|
||||
private IndexRecordManager indexRecordManager;
|
||||
private Map<String, Index> indexTypesMap;
|
||||
|
||||
public HeuristicIndexClient(Set<Index> indexTypes, HetuFileSystemClient fs, Path root)
|
||||
{
|
||||
this.fs = fs;
|
||||
this.root = root;
|
||||
this.indexRecordManager = new IndexRecordManager(fs, root);
|
||||
indexTypesMap = indexTypes.stream().collect(Collectors.toMap(
|
||||
type -> type.getId().toLowerCase(Locale.ENGLISH),
|
||||
Function.identity()));
|
||||
|
|
@ -83,6 +85,19 @@ public class HeuristicIndexClient
|
|||
|
||||
List<IndexMetadata> indexes = new LinkedList<>();
|
||||
|
||||
Path indexKeyPath = Paths.get(path);
|
||||
try {
|
||||
if (indexRecordManager.lookUpIndexRecord(indexKeyPath.subpath(0, 1).toString(),
|
||||
new String[] {indexKeyPath.subpath(1, 2).toString()}, indexKeyPath.subpath(2, 3).toString()) == null) {
|
||||
// Use index record file to pre-screen. If record does not contain the index, skip loading
|
||||
return null;
|
||||
}
|
||||
}
|
||||
catch (Exception e) {
|
||||
// On exception, log and continue reading from disk
|
||||
LOG.debug("Error reading index records: " + path);
|
||||
}
|
||||
|
||||
for (Map.Entry<String, Index> entry : readIndexMap(path).entrySet()) {
|
||||
String absolutePath = entry.getKey();
|
||||
Path remainder = Paths.get(absolutePath.replaceFirst(root.toString(), ""));
|
||||
|
|
|
|||
|
|
@ -77,13 +77,14 @@ public class IndexCommand
|
|||
this.user = user;
|
||||
}
|
||||
|
||||
public IndexRecordManager.IndexRecord getIndex()
|
||||
public IndexRecord getIndex()
|
||||
{
|
||||
try {
|
||||
validatePaths();
|
||||
IndexFactory factory = IndexCommandUtils.getIndexFactory();
|
||||
IndexCommandUtils.IndexStore indexStore = loadIndexStore(configDirPath);
|
||||
return IndexRecordManager.lookUpIndexRecord(indexStore.getFs(), indexStore.getRoot(), indexName);
|
||||
IndexRecordManager indexRecordManager = new IndexRecordManager(indexStore.getFs(), indexStore.getRoot());
|
||||
return indexRecordManager.lookUpIndexRecord(indexName);
|
||||
}
|
||||
catch (IOException e) {
|
||||
e.printStackTrace(System.err);
|
||||
|
|
@ -91,18 +92,19 @@ public class IndexCommand
|
|||
}
|
||||
}
|
||||
|
||||
public List<IndexRecordManager.IndexRecord> getIndexes()
|
||||
public List<IndexRecord> getIndexes()
|
||||
{
|
||||
try {
|
||||
validatePaths();
|
||||
IndexCommandUtils.IndexStore indexStore = loadIndexStore(configDirPath);
|
||||
IndexRecordManager indexRecordManager = new IndexRecordManager(indexStore.getFs(), indexStore.getRoot());
|
||||
|
||||
if (indexName.equals("")) {
|
||||
return IndexRecordManager.readAllIndexRecords(indexStore.getFs(), indexStore.getRoot());
|
||||
return indexRecordManager.getIndexRecords();
|
||||
}
|
||||
else {
|
||||
List<IndexRecordManager.IndexRecord> records = Collections.singletonList(
|
||||
IndexRecordManager.lookUpIndexRecord(indexStore.getFs(), indexStore.getRoot(), indexName));
|
||||
List<IndexRecord> records = Collections.singletonList(
|
||||
indexRecordManager.lookUpIndexRecord(indexName));
|
||||
if (records.get(0) == null) {
|
||||
return Collections.emptyList();
|
||||
}
|
||||
|
|
@ -122,7 +124,8 @@ public class IndexCommand
|
|||
IndexFactory factory = IndexCommandUtils.getIndexFactory();
|
||||
IndexCommandUtils.IndexStore indexStore = loadIndexStore(configDirPath);
|
||||
IndexClient deleteClient = factory.getIndexClient(indexStore.getFs(), indexStore.getRoot());
|
||||
IndexRecordManager.IndexRecord record = IndexRecordManager.lookUpIndexRecord(indexStore.getFs(), indexStore.getRoot(), indexName);
|
||||
IndexRecordManager indexRecordManager = new IndexRecordManager(indexStore.getFs(), indexStore.getRoot());
|
||||
IndexRecord record = indexRecordManager.lookUpIndexRecord(indexName);
|
||||
if (record == null) {
|
||||
System.out.printf("Index with name [%s] does not exist.%n%n", indexName);
|
||||
return;
|
||||
|
|
@ -132,7 +135,7 @@ public class IndexCommand
|
|||
return;
|
||||
}
|
||||
deleteClient.deleteIndex(record.table, record.columns, record.indexType);
|
||||
IndexRecordManager.deleteIndexRecord(indexStore.getFs(), indexStore.getRoot(), indexName);
|
||||
indexRecordManager.deleteIndexRecord(indexName);
|
||||
System.out.format("Deleted index [%s].%n%n", indexName);
|
||||
}
|
||||
catch (IOException e) {
|
||||
|
|
@ -156,8 +159,9 @@ public class IndexCommand
|
|||
validatePaths();
|
||||
IndexCommandUtils.IndexStore indexStore = loadIndexStore(configDirPath);
|
||||
IndexFactory factory = IndexCommandUtils.getIndexFactory();
|
||||
IndexRecordManager.IndexRecord sameNameRecord = IndexRecordManager.lookUpIndexRecord(indexStore.getFs(), indexStore.getRoot(), indexName);
|
||||
IndexRecordManager.IndexRecord sameIndexRecord = IndexRecordManager.lookUpIndexRecord(indexStore.getFs(), indexStore.getRoot(), table, columns, indexType);
|
||||
IndexRecordManager indexRecordManager = new IndexRecordManager(indexStore.getFs(), indexStore.getRoot());
|
||||
IndexRecord sameNameRecord = indexRecordManager.lookUpIndexRecord(indexName);
|
||||
IndexRecord sameIndexRecord = indexRecordManager.lookUpIndexRecord(table, columns, indexType);
|
||||
|
||||
if (sameNameRecord == null) {
|
||||
if (sameIndexRecord != null) {
|
||||
|
|
@ -205,7 +209,7 @@ public class IndexCommand
|
|||
requireNonNull(columns, "No columns specified for create command");
|
||||
IndexWriter writer = factory.getIndexWriter(dsProperties, ixProperties, indexStore.getFs(), indexStore.getRoot());
|
||||
writer.createIndex(table, columns, partitions, indexType);
|
||||
IndexRecordManager.addIndexRecord(indexStore.getFs(), indexStore.getRoot(), indexName, user, table, columns, indexType, partitions);
|
||||
indexRecordManager.addIndexRecord(indexName, user, table, columns, indexType, partitions);
|
||||
if (!verbose) {
|
||||
System.out.print("\n");
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,92 @@
|
|||
/*
|
||||
* 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;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Objects;
|
||||
|
||||
public class IndexRecord
|
||||
{
|
||||
public static final String COLUMN_DELIMITER = ",";
|
||||
public final String name;
|
||||
public final String user;
|
||||
public final String table;
|
||||
public final String[] columns;
|
||||
public final String indexType;
|
||||
public final List<String> partitions;
|
||||
|
||||
public IndexRecord(String name, String user, String table, String[] columns, String indexType, List<String> partitions)
|
||||
{
|
||||
this.name = name;
|
||||
this.user = user == null ? "" : user;
|
||||
this.table = table;
|
||||
this.columns = columns;
|
||||
this.indexType = indexType;
|
||||
this.partitions = partitions;
|
||||
}
|
||||
|
||||
public IndexRecord(String csvRecord)
|
||||
{
|
||||
String[] records = csvRecord.split("\\t");
|
||||
this.name = records[0];
|
||||
this.user = records[1];
|
||||
this.table = records[2];
|
||||
this.columns = records[3].split(COLUMN_DELIMITER);
|
||||
this.indexType = records[4];
|
||||
this.partitions = records.length > 5 ? Arrays.asList(records[5].split(",")) : Collections.emptyList();
|
||||
}
|
||||
|
||||
public String toCsvRecord()
|
||||
{
|
||||
return String.format("%s\t%s\t%s\t%s\t%s\t%s\n", name, user, table, String.join(COLUMN_DELIMITER, columns), indexType, String.join(",", partitions));
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object o)
|
||||
{
|
||||
if (this == o) {
|
||||
return true;
|
||||
}
|
||||
if (!(o instanceof IndexRecord)) {
|
||||
return false;
|
||||
}
|
||||
IndexRecord that = (IndexRecord) o;
|
||||
return Objects.equals(name, that.name) &&
|
||||
Objects.equals(user, that.user) &&
|
||||
Objects.equals(table, that.table) &&
|
||||
Arrays.equals(columns, that.columns) &&
|
||||
Objects.equals(indexType, that.indexType);
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode()
|
||||
{
|
||||
int result = Objects.hash(name, user, table, indexType);
|
||||
result = 31 * result + Arrays.hashCode(columns);
|
||||
return result;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString()
|
||||
{
|
||||
return name + ","
|
||||
+ user + ","
|
||||
+ table + ","
|
||||
+ "[" + String.join(",", columns) + "],"
|
||||
+ indexType;
|
||||
}
|
||||
}
|
||||
|
|
@ -16,6 +16,7 @@ package io.hetu.core.heuristicindex;
|
|||
|
||||
import com.google.common.collect.ImmutableList;
|
||||
import io.hetu.core.common.util.SecurePathWhiteList;
|
||||
import io.hetu.core.filesystem.SupportedFileAttributes;
|
||||
import io.prestosql.spi.filesystem.FileBasedLock;
|
||||
import io.prestosql.spi.filesystem.HetuFileSystemClient;
|
||||
|
||||
|
|
@ -26,51 +27,72 @@ import java.io.OutputStream;
|
|||
import java.nio.file.Path;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collections;
|
||||
import java.util.Iterator;
|
||||
import java.util.LinkedList;
|
||||
import java.util.List;
|
||||
import java.util.Objects;
|
||||
|
||||
import static com.google.common.base.Preconditions.checkArgument;
|
||||
import static io.hetu.core.heuristicindex.util.IndexConstants.COLUMN_DELIMITER;
|
||||
|
||||
public class IndexRecordManager
|
||||
{
|
||||
private static final String RECORD_FILE_NAME = "INDEX_RECORDS";
|
||||
private final HetuFileSystemClient fs;
|
||||
private final Path root;
|
||||
|
||||
private IndexRecordManager() {}
|
||||
private List<IndexRecord> cache;
|
||||
private long cacheLastModifiedTime;
|
||||
|
||||
public static List<IndexRecord> readAllIndexRecords(HetuFileSystemClient fs, Path root)
|
||||
public IndexRecordManager(HetuFileSystemClient fs, Path root)
|
||||
{
|
||||
this.fs = fs;
|
||||
this.root = root;
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
public List<IndexRecord> getIndexRecords()
|
||||
throws IOException
|
||||
{
|
||||
validatePath(root);
|
||||
Path recordFile = root.resolve(RECORD_FILE_NAME);
|
||||
List<IndexRecord> records = new ArrayList<>();
|
||||
|
||||
if (!fs.exists(recordFile)) {
|
||||
return records;
|
||||
// invalidate cache
|
||||
cache = records;
|
||||
cacheLastModifiedTime = 0;
|
||||
return cache;
|
||||
}
|
||||
|
||||
try (BufferedReader reader = new BufferedReader(new InputStreamReader(fs.newInputStream(recordFile)))) {
|
||||
reader.readLine(); // skip header
|
||||
while (true) {
|
||||
String line = reader.readLine();
|
||||
if (line == null) {
|
||||
break;
|
||||
long modifiedTime = (long) fs.getAttribute(recordFile, SupportedFileAttributes.LAST_MODIFIED_TIME);
|
||||
if (modifiedTime != cacheLastModifiedTime) {
|
||||
// invalidate cache
|
||||
try (BufferedReader reader = new BufferedReader(new InputStreamReader(fs.newInputStream(recordFile)))) {
|
||||
reader.readLine(); // skip header
|
||||
while (true) {
|
||||
String line = reader.readLine();
|
||||
if (line == null) {
|
||||
break;
|
||||
}
|
||||
records.add(new IndexRecord(line));
|
||||
}
|
||||
records.add(new IndexRecord(line));
|
||||
}
|
||||
cache = records;
|
||||
cacheLastModifiedTime = modifiedTime;
|
||||
}
|
||||
|
||||
return records;
|
||||
return cache;
|
||||
}
|
||||
|
||||
public static IndexRecord lookUpIndexRecord(HetuFileSystemClient fs, Path root, String name)
|
||||
public IndexRecord lookUpIndexRecord(String name)
|
||||
throws IOException
|
||||
{
|
||||
validatePath(root);
|
||||
List<IndexRecord> records = readAllIndexRecords(fs, root);
|
||||
List<IndexRecord> records = getIndexRecords();
|
||||
|
||||
for (IndexRecord record : records) {
|
||||
if (record.name.equals(name)) {
|
||||
|
|
@ -81,11 +103,10 @@ public class IndexRecordManager
|
|||
return null;
|
||||
}
|
||||
|
||||
public static IndexRecord lookUpIndexRecord(HetuFileSystemClient fs, Path root, String table, String[] columns, String indexType)
|
||||
public IndexRecord lookUpIndexRecord(String table, String[] columns, String indexType)
|
||||
throws IOException
|
||||
{
|
||||
validatePath(root);
|
||||
List<IndexRecord> records = readAllIndexRecords(fs, root);
|
||||
List<IndexRecord> records = getIndexRecords();
|
||||
|
||||
for (IndexRecord record : records) {
|
||||
if (record.table.equals(table) && Arrays.equals(record.columns, columns) && record.indexType.equals(indexType)) {
|
||||
|
|
@ -100,15 +121,14 @@ public class IndexRecordManager
|
|||
* Add IndexRecord into record file. If the method is called with a name that already exists,
|
||||
* it will OVERWRITE the existing entry but combine the note part
|
||||
*/
|
||||
public static synchronized void addIndexRecord(HetuFileSystemClient fs, Path root, String name, String user, String table, String[] columns, String indexType, String... partitions)
|
||||
public synchronized void addIndexRecord(String name, String user, String table, String[] columns, String indexType, String... partitions)
|
||||
throws IOException
|
||||
{
|
||||
validatePath(root);
|
||||
// Protect root directory
|
||||
FileBasedLock lock = new FileBasedLock(fs, root);
|
||||
try {
|
||||
lock.lock();
|
||||
List<IndexRecord> records = readAllIndexRecords(fs, root);
|
||||
List<IndexRecord> records = getIndexRecords();
|
||||
Iterator<IndexRecord> iterator = records.iterator();
|
||||
List<String> partitionsToWrite = new LinkedList<>(Arrays.asList(partitions));
|
||||
while (iterator.hasNext()) {
|
||||
|
|
@ -119,24 +139,23 @@ public class IndexRecordManager
|
|||
}
|
||||
}
|
||||
records.add(new IndexRecord(name, user, table, columns, indexType, partitionsToWrite));
|
||||
writeIndexRecords(fs, root, records);
|
||||
writeIndexRecords(records);
|
||||
}
|
||||
finally {
|
||||
lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
public static synchronized void deleteIndexRecord(HetuFileSystemClient fs, Path root, String name)
|
||||
public synchronized void deleteIndexRecord(String name)
|
||||
throws IOException
|
||||
{
|
||||
validatePath(root);
|
||||
// Protect root directory
|
||||
FileBasedLock lock = new FileBasedLock(fs, root);
|
||||
try {
|
||||
lock.lock();
|
||||
List<IndexRecord> records = readAllIndexRecords(fs, root);
|
||||
List<IndexRecord> records = getIndexRecords();
|
||||
records.removeIf(record -> record.name.equals(name));
|
||||
writeIndexRecords(fs, root, records);
|
||||
writeIndexRecords(records);
|
||||
}
|
||||
finally {
|
||||
lock.unlock();
|
||||
|
|
@ -147,10 +166,9 @@ public class IndexRecordManager
|
|||
* Write the given records into the record file. This operation OVERWRITES the existing file and is NOT atomoc.
|
||||
* Therefore it should only be called from lock-protected block to avoid overwriting data.
|
||||
*/
|
||||
private static void writeIndexRecords(HetuFileSystemClient fs, Path root, List<IndexRecord> records)
|
||||
private void writeIndexRecords(List<IndexRecord> records)
|
||||
throws IOException
|
||||
{
|
||||
validatePath(root);
|
||||
Path recordFile = root.resolve(RECORD_FILE_NAME);
|
||||
|
||||
boolean writeHead = false;
|
||||
|
|
@ -163,87 +181,4 @@ public class IndexRecordManager
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static void validatePath(Path root)
|
||||
{
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
public static class IndexRecord
|
||||
{
|
||||
public final String name;
|
||||
public final String user;
|
||||
public final String table;
|
||||
public final String[] columns;
|
||||
public final String indexType;
|
||||
public final List<String> partitions;
|
||||
|
||||
public IndexRecord(String name, String user, String table, String[] columns, String indexType, List<String> partitions)
|
||||
{
|
||||
this.name = name;
|
||||
this.user = user == null ? "" : user;
|
||||
this.table = table;
|
||||
this.columns = columns;
|
||||
this.indexType = indexType;
|
||||
this.partitions = partitions;
|
||||
}
|
||||
|
||||
public IndexRecord(String csvRecord)
|
||||
{
|
||||
String[] records = csvRecord.split("\\t");
|
||||
this.name = records[0];
|
||||
this.user = records[1];
|
||||
this.table = records[2];
|
||||
this.columns = records[3].split(COLUMN_DELIMITER);
|
||||
this.indexType = records[4];
|
||||
this.partitions = records.length > 5 ? Arrays.asList(records[5].split(",")) : Collections.emptyList();
|
||||
}
|
||||
|
||||
public String toCsvRecord()
|
||||
{
|
||||
return String.format("%s\t%s\t%s\t%s\t%s\t%s\n", name, user, table, String.join(COLUMN_DELIMITER, columns), indexType, String.join(",", partitions));
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object o)
|
||||
{
|
||||
if (this == o) {
|
||||
return true;
|
||||
}
|
||||
if (!(o instanceof IndexRecord)) {
|
||||
return false;
|
||||
}
|
||||
IndexRecord that = (IndexRecord) o;
|
||||
return Objects.equals(name, that.name) &&
|
||||
Objects.equals(user, that.user) &&
|
||||
Objects.equals(table, that.table) &&
|
||||
Arrays.equals(columns, that.columns) &&
|
||||
Objects.equals(indexType, that.indexType);
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode()
|
||||
{
|
||||
int result = Objects.hash(name, user, table, indexType);
|
||||
result = 31 * result + Arrays.hashCode(columns);
|
||||
return result;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString()
|
||||
{
|
||||
return name + ","
|
||||
+ user + ","
|
||||
+ table + ","
|
||||
+ "[" + String.join(",", columns) + "],"
|
||||
+ indexType;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -30,8 +30,6 @@ public class IndexConstants
|
|||
*/
|
||||
public static final String LAST_MODIFIED_FILE_PREFIX = "lastModified=";
|
||||
|
||||
public static final String COLUMN_DELIMITER = ",";
|
||||
|
||||
public static final String CONFIG_FILE = "config.properties";
|
||||
|
||||
public static final String CATALOG_CONFIGS_DIR = "catalog";
|
||||
|
|
|
|||
|
|
@ -70,6 +70,7 @@ public class TestHeuristicIndexFactory
|
|||
indexes.add(new BloomIndex());
|
||||
|
||||
HetuFileSystemClient fs = new HetuLocalFileSystemClient(new LocalConfig(new Properties()), folder.getRoot().toPath());
|
||||
IndexRecordManager indexRecordManager = new IndexRecordManager(fs, folder.getRoot().toPath());
|
||||
|
||||
HeuristicIndexWriter writer = new HeuristicIndexWriter(dataSource, indexes, fs, folder.getRoot().toPath());
|
||||
|
||||
|
|
@ -78,6 +79,10 @@ public class TestHeuristicIndexFactory
|
|||
String[] partitons = new String[] {"p=bar"};
|
||||
writer.createIndex(table, columns, partitons, "bloom");
|
||||
writer.createIndex(table, columns, partitons, "minmax");
|
||||
indexRecordManager.addIndexRecord("i1", "testUser", table, new String[] {"0"}, "bloom", partitons);
|
||||
indexRecordManager.addIndexRecord("i2", "testUser", table, new String[] {"2"}, "bloom", partitons);
|
||||
indexRecordManager.addIndexRecord("i3", "testUser", table, new String[] {"0"}, "minmax", partitons);
|
||||
indexRecordManager.addIndexRecord("i4", "testUser", table, new String[] {"2"}, "minmax", partitons);
|
||||
|
||||
IndexClient client = new HeuristicIndexFactory().getIndexClient(fs, folder.getRoot().toPath());
|
||||
List<IndexMetadata> splits = client.readSplitIndex(table);
|
||||
|
|
|
|||
|
|
@ -17,9 +17,7 @@ package io.hetu.core.heuristicindex;
|
|||
import io.hetu.core.common.filesystem.TempFolder;
|
||||
import io.hetu.core.heuristicindex.util.IndexCommandUtils;
|
||||
import io.hetu.core.heuristicindex.util.IndexConstants;
|
||||
import io.prestosql.spi.heuristicindex.IndexClient;
|
||||
import io.prestosql.spi.heuristicindex.IndexFactory;
|
||||
import io.prestosql.spi.heuristicindex.IndexWriter;
|
||||
import org.powermock.core.classloader.annotations.PowerMockIgnore;
|
||||
import org.powermock.core.classloader.annotations.PrepareForTest;
|
||||
import org.powermock.modules.testng.PowerMockTestCase;
|
||||
|
|
@ -29,23 +27,17 @@ import java.io.File;
|
|||
import java.io.FileOutputStream;
|
||||
import java.io.IOException;
|
||||
import java.io.OutputStream;
|
||||
import java.nio.file.Path;
|
||||
import java.nio.file.Paths;
|
||||
import java.util.Properties;
|
||||
|
||||
import static io.hetu.core.heuristicindex.util.IndexCommandUtils.loadDataSourceProperties;
|
||||
import static io.hetu.core.heuristicindex.util.IndexCommandUtils.loadIndexStore;
|
||||
import static org.mockito.Matchers.any;
|
||||
import static org.mockito.Matchers.anyString;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.times;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.powermock.api.mockito.PowerMockito.mockStatic;
|
||||
import static org.powermock.api.mockito.PowerMockito.when;
|
||||
import static org.testng.Assert.assertNotNull;
|
||||
import static org.testng.Assert.assertTrue;
|
||||
|
||||
@PrepareForTest({IndexCommandUtils.class, IndexRecordManager.class})
|
||||
@PrepareForTest(IndexCommandUtils.class)
|
||||
@PowerMockIgnore("javax.management.*")
|
||||
@Test(singleThreaded = true)
|
||||
public class TestIndexCommand
|
||||
|
|
@ -85,56 +77,6 @@ public class TestIndexCommand
|
|||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testCreateCommand()
|
||||
throws IOException
|
||||
{
|
||||
try (TempFolder testFolder = new TempFolder()) {
|
||||
testFolder.create();
|
||||
|
||||
IndexFactory factory = mock(IndexFactory.class);
|
||||
IndexWriter writer = mock(IndexWriter.class);
|
||||
when(factory.getIndexWriter(any(), any(), any(), any())).thenReturn(writer);
|
||||
mockStatic(IndexRecordManager.class);
|
||||
when(IndexRecordManager.readAllIndexRecords(any(), any())).thenReturn(null);
|
||||
mockStatic(IndexCommandUtils.class);
|
||||
when(loadDataSourceProperties(anyString(), anyString())).thenReturn(new Properties());
|
||||
when(loadIndexStore(anyString())).thenReturn(new IndexCommandUtils.IndexStore(null, null));
|
||||
when(IndexCommandUtils.getIndexFactory()).thenReturn(factory);
|
||||
|
||||
IndexCommand indexCommand = new IndexCommand(testFolder.getRoot().getAbsolutePath(), "abc", "catalog.schema.table", new String[] {"column"}, null,
|
||||
"bloom", null, false, null);
|
||||
indexCommand.createIndex();
|
||||
|
||||
verify(writer, times(1)).createIndex(any(), any(), any(), any());
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDeleteCommand()
|
||||
throws IOException
|
||||
{
|
||||
try (TempFolder testFolder = new TempFolder()) {
|
||||
testFolder.create();
|
||||
|
||||
IndexFactory factory = mock(IndexFactory.class);
|
||||
IndexClient client = mock(IndexClient.class);
|
||||
when(factory.getIndexClient(any(), any())).thenReturn(client);
|
||||
mockStatic(IndexRecordManager.class);
|
||||
when(IndexRecordManager.lookUpIndexRecord(any(), any(), anyString())).thenReturn(new IndexRecordManager.IndexRecord(null, null, null, null, null, null));
|
||||
mockStatic(IndexCommandUtils.class);
|
||||
when(loadDataSourceProperties(anyString(), anyString())).thenReturn(new Properties());
|
||||
when(loadIndexStore(anyString())).thenReturn(new IndexCommandUtils.IndexStore(null, null));
|
||||
when(IndexCommandUtils.getIndexFactory()).thenReturn(factory);
|
||||
|
||||
IndexCommand indexCommand = new IndexCommand(testFolder.getRoot().getAbsolutePath(), "abc", "catalog.schema.table", new String[] {"column"}, null,
|
||||
"bloom", null, false, null);
|
||||
indexCommand.deleteIndex();
|
||||
|
||||
verify(client, times(1)).deleteIndex(any(), any(), any());
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testLoadIndexWriterFromConfigFile()
|
||||
throws IOException
|
||||
|
|
@ -142,7 +84,6 @@ public class TestIndexCommand
|
|||
IndexFactory factory = new HeuristicIndexFactory();
|
||||
|
||||
Properties dsProps = new Properties();
|
||||
Path root = Paths.get("/tmp");
|
||||
dsProps.setProperty("connector.name", "empty");
|
||||
|
||||
Properties ixProps = new Properties();
|
||||
|
|
|
|||
|
|
@ -36,7 +36,7 @@ import static org.testng.Assert.assertNull;
|
|||
|
||||
public class TestIndexRecordManager
|
||||
{
|
||||
private static final HetuFileSystemClient FILE_SYSTEM_CLIENT = new HetuLocalFileSystemClient(new LocalConfig(new Properties()), Paths.get("/"));
|
||||
private static final HetuFileSystemClient FILE_SYSTEM_CLIENT = new HetuLocalFileSystemClient(new LocalConfig(new Properties()), Paths.get(System.getProperty("java.io.tmpdir")));
|
||||
|
||||
@Test
|
||||
public void testDelete()
|
||||
|
|
@ -44,27 +44,28 @@ public class TestIndexRecordManager
|
|||
{
|
||||
try (TempFolder folder = new TempFolder()) {
|
||||
folder.create();
|
||||
IndexRecordManager.addIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "1", "testUser", "testTable", new String[] {"testColumn"}, "minmax", "cp=1");
|
||||
IndexRecordManager.addIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "2", "testUser", "testTable", new String[] {"testColumn"}, "minmax", "cp=1");
|
||||
assertNotNull(IndexRecordManager.lookUpIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "1"));
|
||||
assertEquals(IndexRecordManager.readAllIndexRecords(FILE_SYSTEM_CLIENT, folder.getRoot().toPath()).size(), 2);
|
||||
IndexRecordManager indexRecordManager = new IndexRecordManager(FILE_SYSTEM_CLIENT, folder.getRoot().toPath());
|
||||
indexRecordManager.addIndexRecord("1", "testUser", "testTable", new String[] {"testColumn"}, "minmax", "cp=1");
|
||||
indexRecordManager.addIndexRecord("2", "testUser", "testTable", new String[] {"testColumn"}, "minmax", "cp=1");
|
||||
assertNotNull(indexRecordManager.lookUpIndexRecord("1"));
|
||||
assertEquals(indexRecordManager.getIndexRecords().size(), 2);
|
||||
|
||||
// Delete 1
|
||||
IndexRecordManager.deleteIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "1");
|
||||
assertNull(IndexRecordManager.lookUpIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "1"));
|
||||
assertNotNull(IndexRecordManager.lookUpIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "2"));
|
||||
assertEquals(IndexRecordManager.readAllIndexRecords(FILE_SYSTEM_CLIENT, folder.getRoot().toPath()).size(), 1);
|
||||
indexRecordManager.deleteIndexRecord("1");
|
||||
assertNull(indexRecordManager.lookUpIndexRecord("1"));
|
||||
assertNotNull(indexRecordManager.lookUpIndexRecord("2"));
|
||||
assertEquals(indexRecordManager.getIndexRecords().size(), 1);
|
||||
|
||||
// Delete 1 again
|
||||
IndexRecordManager.deleteIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "1");
|
||||
assertNull(IndexRecordManager.lookUpIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "1"));
|
||||
assertNotNull(IndexRecordManager.lookUpIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "2"));
|
||||
assertEquals(IndexRecordManager.readAllIndexRecords(FILE_SYSTEM_CLIENT, folder.getRoot().toPath()).size(), 1);
|
||||
indexRecordManager.deleteIndexRecord("1");
|
||||
assertNull(indexRecordManager.lookUpIndexRecord("1"));
|
||||
assertNotNull(indexRecordManager.lookUpIndexRecord("2"));
|
||||
assertEquals(indexRecordManager.getIndexRecords().size(), 1);
|
||||
|
||||
// Delete 2
|
||||
IndexRecordManager.deleteIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "2");
|
||||
assertNull(IndexRecordManager.lookUpIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "2"));
|
||||
assertEquals(IndexRecordManager.readAllIndexRecords(FILE_SYSTEM_CLIENT, folder.getRoot().toPath()).size(), 0);
|
||||
indexRecordManager.deleteIndexRecord("2");
|
||||
assertNull(indexRecordManager.lookUpIndexRecord("2"));
|
||||
assertEquals(indexRecordManager.getIndexRecords().size(), 0);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -81,18 +82,18 @@ public class TestIndexRecordManager
|
|||
@Test
|
||||
public void testRecordEqualAndHash()
|
||||
{
|
||||
IndexRecordManager.IndexRecord r1 = new IndexRecordManager.IndexRecord("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList());
|
||||
IndexRecordManager.IndexRecord r2 = new IndexRecordManager.IndexRecord("testName", "testUser", "testTable", new String[] {
|
||||
IndexRecord r1 = new IndexRecord("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList());
|
||||
IndexRecord r2 = new IndexRecord("testName", "testUser", "testTable", new String[] {
|
||||
"testColumn"}, "minmax", ImmutableList.of("note"));
|
||||
IndexRecordManager.IndexRecord r3 = new IndexRecordManager.IndexRecord("testName", "testUser", "testTable", new String[] {"testColumn"}, "bloom", Collections.emptyList());
|
||||
IndexRecordManager.IndexRecord r4 = new IndexRecordManager.IndexRecord("testName", "testUser", "testTable", new String[] {"testColumn",
|
||||
IndexRecord r3 = new IndexRecord("testName", "testUser", "testTable", new String[] {"testColumn"}, "bloom", Collections.emptyList());
|
||||
IndexRecord r4 = new IndexRecord("testName", "testUser", "testTable", new String[] {"testColumn",
|
||||
"testColumn2"}, "minmax", Collections.emptyList());
|
||||
assertEquals(r1, r1);
|
||||
assertEquals(r1, r2);
|
||||
assertNotEquals(r1, r3);
|
||||
assertNotEquals(r1, r4);
|
||||
|
||||
HashSet<IndexRecordManager.IndexRecord> testSet = new HashSet<>();
|
||||
HashSet<IndexRecord> testSet = new HashSet<>();
|
||||
testSet.add(r1);
|
||||
assertEquals(testSet.size(), 1);
|
||||
testSet.add(r2);
|
||||
|
|
@ -109,10 +110,11 @@ public class TestIndexRecordManager
|
|||
{
|
||||
try (TempFolder folder = new TempFolder()) {
|
||||
folder.create();
|
||||
IndexRecordManager.IndexRecord expected = new IndexRecordManager.IndexRecord("testName", "testUser", "testTable", new String[] {
|
||||
IndexRecordManager indexRecordManager = new IndexRecordManager(FILE_SYSTEM_CLIENT, folder.getRoot().toPath());
|
||||
IndexRecord expected = new IndexRecord("testName", "testUser", "testTable", new String[] {
|
||||
"testColumn"}, "minmax", ImmutableList.of(""));
|
||||
IndexRecordManager.addIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", "cp=1");
|
||||
IndexRecordManager.IndexRecord actual = IndexRecordManager.lookUpIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), "testName");
|
||||
indexRecordManager.addIndexRecord("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", "cp=1");
|
||||
IndexRecord actual = indexRecordManager.lookUpIndexRecord("testName");
|
||||
assertIndexRecordFullyEqual(actual, expected);
|
||||
}
|
||||
}
|
||||
|
|
@ -122,14 +124,15 @@ public class TestIndexRecordManager
|
|||
{
|
||||
try (TempFolder folder = new TempFolder()) {
|
||||
folder.create();
|
||||
IndexRecordManager.IndexRecord expected = new IndexRecordManager.IndexRecord(name, user, table, columns, indexType, note);
|
||||
IndexRecordManager.addIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), name, user, table, columns, indexType, note.toArray(new String[0]));
|
||||
IndexRecordManager indexRecordManager = new IndexRecordManager(FILE_SYSTEM_CLIENT, folder.getRoot().toPath());
|
||||
IndexRecord expected = new IndexRecord(name, user, table, columns, indexType, note);
|
||||
indexRecordManager.addIndexRecord(name, user, table, columns, indexType, note.toArray(new String[0]));
|
||||
|
||||
IndexRecordManager.IndexRecord actual1 = IndexRecordManager.lookUpIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), name);
|
||||
IndexRecord actual1 = indexRecordManager.lookUpIndexRecord(name);
|
||||
assertNotNull(actual1);
|
||||
assertIndexRecordFullyEqual(actual1, expected);
|
||||
|
||||
IndexRecordManager.IndexRecord actual2 = IndexRecordManager.lookUpIndexRecord(FILE_SYSTEM_CLIENT, folder.getRoot().toPath(), table, columns, indexType);
|
||||
IndexRecord actual2 = indexRecordManager.lookUpIndexRecord(table, columns, indexType);
|
||||
assertNotNull(actual2);
|
||||
assertIndexRecordFullyEqual(actual2, expected);
|
||||
}
|
||||
|
|
@ -137,7 +140,7 @@ public class TestIndexRecordManager
|
|||
|
||||
// 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(IndexRecordManager.IndexRecord actual, IndexRecordManager.IndexRecord expected)
|
||||
private void assertIndexRecordFullyEqual(IndexRecord actual, IndexRecord expected)
|
||||
throws IllegalAccessException
|
||||
{
|
||||
for (Field field : actual.getClass().getDeclaredFields()) {
|
||||
|
|
|
|||
|
|
@ -23,7 +23,7 @@ import io.airlift.log.Logging;
|
|||
import io.airlift.log.LoggingConfiguration;
|
||||
import io.airlift.units.Duration;
|
||||
import io.hetu.core.heuristicindex.IndexCommand;
|
||||
import io.hetu.core.heuristicindex.IndexRecordManager;
|
||||
import io.hetu.core.heuristicindex.IndexRecord;
|
||||
import io.prestosql.client.ClientSelectedRole;
|
||||
import io.prestosql.client.ClientSession;
|
||||
import io.prestosql.client.ClientTypeSignature;
|
||||
|
|
@ -343,10 +343,10 @@ public class Console
|
|||
.add(new Column("Index Type", VARCHAR, new ClientTypeSignature(VARCHAR)))
|
||||
.add(new Column("Partitions", VARCHAR, new ClientTypeSignature(VARCHAR)))
|
||||
.build();
|
||||
List<IndexRecordManager.IndexRecord> records = command.getIndexes();
|
||||
List<IndexRecord> records = command.getIndexes();
|
||||
|
||||
List<List<?>> rows = new ArrayList<>();
|
||||
for (IndexRecordManager.IndexRecord v : records) {
|
||||
for (IndexRecord v : records) {
|
||||
if (!verifyAccess(queryRunner, exiting, v.table)) {
|
||||
continue;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -28,6 +28,7 @@ import io.prestosql.plugin.hive.HiveColumnHandle;
|
|||
import io.prestosql.plugin.hive.HiveSplit;
|
||||
import io.prestosql.spi.HetuConstant;
|
||||
import io.prestosql.spi.heuristicindex.IndexMetadata;
|
||||
import io.prestosql.spi.heuristicindex.IndexNotRegisteredException;
|
||||
import io.prestosql.spi.predicate.TupleDomain;
|
||||
import io.prestosql.spi.service.PropertyService;
|
||||
import org.apache.hadoop.fs.Path;
|
||||
|
|
@ -136,6 +137,9 @@ public class IndexCache
|
|||
LOG.debug("Loaded index for %s.", indexCacheKeyPath);
|
||||
}
|
||||
catch (ExecutionException e) {
|
||||
if (e.getCause() instanceof IndexNotRegisteredException) {
|
||||
// Do nothing. Index not registered.
|
||||
}
|
||||
if (LOG.isDebugEnabled()) {
|
||||
LOG.debug(e, "Unable to load index for %s. ", indexCacheKeyPath);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ import com.google.inject.Inject;
|
|||
import io.hetu.core.common.heuristicindex.IndexCacheKey;
|
||||
import io.prestosql.spi.heuristicindex.IndexClient;
|
||||
import io.prestosql.spi.heuristicindex.IndexMetadata;
|
||||
import io.prestosql.spi.heuristicindex.IndexNotRegisteredException;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.stream.Collectors;
|
||||
|
|
@ -67,6 +68,11 @@ public class IndexCacheLoader
|
|||
throw new Exception("No valid index files found for key " + key, e);
|
||||
}
|
||||
|
||||
// null indicates that the index is not registered in index records
|
||||
if (indices == null) {
|
||||
throw new IndexNotRegisteredException();
|
||||
}
|
||||
|
||||
// lastModified file was valid, but no index files for the given types
|
||||
if (indices.isEmpty()) {
|
||||
throw new Exception("No index files found for key " + key);
|
||||
|
|
|
|||
|
|
@ -26,6 +26,7 @@ import io.hetu.core.common.heuristicindex.IndexCacheKey;
|
|||
import io.prestosql.metadata.Split;
|
||||
import io.prestosql.spi.HetuConstant;
|
||||
import io.prestosql.spi.heuristicindex.IndexMetadata;
|
||||
import io.prestosql.spi.heuristicindex.IndexNotRegisteredException;
|
||||
import io.prestosql.spi.service.PropertyService;
|
||||
|
||||
import java.net.URI;
|
||||
|
|
@ -105,7 +106,10 @@ public class IndexCache
|
|||
LOG.debug("Loaded index for %s.", filterKey);
|
||||
}
|
||||
catch (ExecutionException e) {
|
||||
if (LOG.isDebugEnabled()) {
|
||||
if (e.getCause() instanceof IndexNotRegisteredException) {
|
||||
// Do nothing. Index not registered.
|
||||
}
|
||||
else if (LOG.isDebugEnabled()) {
|
||||
LOG.debug(e, "Unable to load index for %s. ", filterKey);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ import com.google.common.cache.CacheLoader;
|
|||
import io.hetu.core.common.heuristicindex.IndexCacheKey;
|
||||
import io.prestosql.spi.heuristicindex.IndexClient;
|
||||
import io.prestosql.spi.heuristicindex.IndexMetadata;
|
||||
import io.prestosql.spi.heuristicindex.IndexNotRegisteredException;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.stream.Collectors;
|
||||
|
|
@ -65,6 +66,11 @@ public class IndexCacheLoader
|
|||
throw new Exception("No valid index file found for key " + key, e);
|
||||
}
|
||||
|
||||
// null indicates that the index is not registered in index records
|
||||
if (indices == null) {
|
||||
throw new IndexNotRegisteredException();
|
||||
}
|
||||
|
||||
// lastModified file was valid, but no index files for the given types
|
||||
if (indices.isEmpty()) {
|
||||
throw new Exception("No index files found for key " + key);
|
||||
|
|
|
|||
|
|
@ -0,0 +1,20 @@
|
|||
/*
|
||||
* 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.prestosql.spi.heuristicindex;
|
||||
|
||||
public class IndexNotRegisteredException
|
||||
extends Exception
|
||||
{
|
||||
}
|
||||
Loading…
Reference in New Issue