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

!1044 Knowledge transfer by Han - Improve document on memory connector and heuristic index 

Merge pull request !1044 from Han_Weng/han-code-doc
This commit is contained in:
i-robot 2021-08-03 14:31:21 +00:00 committed by Gitee
parent ff7522019e ce974fad5c
commit 6e00cb448d
16 changed files with 120 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -73,7 +73,7 @@ public class FileIndexWriter
private Path tmpPath;
/**
* Constructor
* Constructor. The file index writer is per ORC file, marked by `dataSourceFileName`.
*
* @param createIndexMetadata metadata of create index, includes indexName, tableName, indexType, indexColumns and partitions
* @param fs filesystem client to access filesystem where the indexes are persisted/stored

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

@ -22,13 +22,16 @@ import io.prestosql.spi.metastore.model.DatabaseEntity;
import io.prestosql.spi.metastore.model.TableEntity;
import io.prestosql.spi.metastore.model.TableEntityType;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Optional;
/**
* Index record is stored in hetu metastore as parameters at table level. For each table entity,
* See {@link IndexRecord} for more details.
*/
public class IndexRecordManager
{
private static final Logger LOG = Logger.get(IndexRecordManager.class);
@ -40,6 +43,13 @@ public class IndexRecordManager
this.metastore = metastore;
}
/**
* List all parameters at table level and filter those with hindex prefix.
*
* Construct {@code IndexRecord} objects from them.
*
* @return a list of deserialized {@code IndexRecord} objects.
*/
public List<IndexRecord> getIndexRecords()
{
long startTime = System.currentTimeMillis();
@ -61,12 +71,17 @@ public class IndexRecordManager
return records;
}
/**
* Look up index record according to name.
*/
public IndexRecord lookUpIndexRecord(String name)
throws IOException
{
return getIndexRecords().stream().filter(indexRecord -> indexRecord.name.equals(name)).findFirst().orElse(null);
}
/**
* Look up index record according to what it is for (triplet of [table, column, type]).
*/
public IndexRecord lookUpIndexRecord(String table, String[] columns, String indexType)
{
String[] tableQualified = table.split("\\.");
@ -89,10 +104,10 @@ 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 partition column
*
* it will OVERWRITE the existing entry but COMBINE the partition columns (if it previously was partitioned)
*/
public synchronized void addIndexRecord(String name, String user, String table, String[] columns, String indexType, List<String> indexProperties, List<String> partitions)
throws IOException
{
IndexRecord record = new IndexRecord(name, user, table, columns, indexType, indexProperties, partitions);
IndexRecord old = lookUpIndexRecord(name);
@ -135,8 +150,13 @@ public class IndexRecordManager
record.serializeValue());
}
/**
* Delete index record from metastore according to name. Also allows partial deletion.
*
* @param name name of index to delete
* @param partitionsToRemove the partitions to remove. If this list is empty, remove all.
*/
public synchronized void deleteIndexRecord(String name, List<String> partitionsToRemove)
throws IOException
{
getIndexRecords().stream().filter(record -> record.name.equals(name))
.forEach(record -> {

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

@ -46,7 +46,7 @@ import static io.prestosql.spi.heuristicindex.SerializationUtils.serializeStripe
/**
* Indexes which needs to be created at table or partition level
* needs to use this writer.
* needs to use this writer. E.g. BTREE index.
*/
public class PartitionIndexWriter
implements IndexWriter
@ -62,6 +62,16 @@ public class PartitionIndexWriter
private final Properties properties;
private final HetuFileSystemClient fs;
private final Path root;
/*
Each stripe from pages is mapped to a auto-incremental integer, starting from 0.
This mapping is stored in symbolToIdMap like 0 -> "<ORCFileName>:<stripeStart>:<stripeEnd>"
See more about serialization at {@link io.prestosql.spi.heuristicindex.SerializationUtils}
The inverted index is stored in data map like {10 -> "1,2", 2 -> "2,3"}, meaning that
value 10 occurs in stripe 1 and 2, and value 2 occurs in stripe 2 and 3. The majority of memory
is used by this map.
*/
private final AtomicInteger counter = new AtomicInteger(0); // symbol table counter
private final Map<String, String> symbolToIdMap;
private final Map<Comparable<? extends Comparable<?>>, String> dataMap;
@ -121,6 +131,12 @@ public class PartitionIndexWriter
Comparable<? extends Comparable<?>> comparableKey = (Comparable<? extends Comparable<?>>) key;
String existing = dataMap.putIfAbsent(comparableKey, code);
if (existing != null) {
// replace old string with new values added. e.g. "1,2,2" -> "1,2,2,3"
// while loop used to allow concurrent modification.
// THIS IS UGLY BUT IS THE WORKING SOLUTION TO SAVE MEMORY.
// Tried to use collections like List[1,2,2,3] or Set[1,2,3] but both
// crash node with too high memory usage.
// TODO: Resolved memory issue in a more decent way
String newData = getNewData(key, code);
boolean done = dataMap.replace(comparableKey, existing, newData);
while (!done) {
@ -139,6 +155,9 @@ public class PartitionIndexWriter
return output + "," + splitData;
}
/**
* Persists the data into an Index object and serialize it to disk.
*/
@Override
public void persist()
throws IOException

7
hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/filter/HeuristicIndexFilter.java
View File

@ -39,6 +39,11 @@ public class HeuristicIndexFilter
{
Map<String, List<IndexMetadata>> indices;
/**
* Construct the filter with indexes.
*
* @param indices A map of column name to list of index. Only one column is supported for now.
*/
public HeuristicIndexFilter(Map<String, List<IndexMetadata>> indices)
{
this.indices = indices;
@ -65,7 +70,7 @@ public class HeuristicIndexFilter
// todo remote udf, we should get FunctionHandle from FunctionAndTypeManager
CallExpression left = new CallExpression(OperatorType.GREATER_THAN_OR_EQUAL.name(), new BuiltInFunctionHandle(sigLeft), specialForm.getType(), ImmutableList.of(specialForm.getArguments().get(0), specialForm.getArguments().get(1)), Optional.empty());
CallExpression right = new CallExpression(OperatorType.LESS_THAN_OR_EQUAL.name(), new BuiltInFunctionHandle(sigRight), specialForm.getType(), ImmutableList.of(specialForm.getArguments().get(0), specialForm.getArguments().get(2)), Optional.empty());
return matches(left) && matches(right);
return matches(left) && matches(right); // break it to (>= left and <= right)
case IN:
Signature sigEqual = Signature.internalOperator(OperatorType.EQUAL,
specialForm.getType().getTypeSignature(),

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

@ -66,7 +66,7 @@ public class TestIndexRecordManager
@Test(timeOut = 30000)
public void testConcurrentMultipleManagers()
throws IOException, InterruptedException
throws InterruptedException
{
Random random = new Random();
String[] names = new String[] {"a", "b", "c"};
@ -81,7 +81,7 @@ public class TestIndexRecordManager
new IndexRecordManager(testMetastore1)
.addIndexRecord(names[finalI], "testUser", "c.s.t", new String[] {"testColumn"}, names[finalI], Collections.emptyList(), Arrays.asList("cp=1"));
}
catch (IOException | InterruptedException e) {
catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
@ -97,7 +97,7 @@ public class TestIndexRecordManager
}
indexRecordManager.deleteIndexRecord(names[0], Collections.emptyList());
}
catch (IOException | InterruptedException e) {
catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
@ -116,7 +116,7 @@ public class TestIndexRecordManager
@Test(timeOut = 20000)
public void testConcurrentSingleManager()
throws IOException, InterruptedException
throws InterruptedException
{
Random random = new Random();
IndexRecordManager indexRecordManager = new IndexRecordManager(testMetastore2);
@ -132,7 +132,7 @@ public class TestIndexRecordManager
Thread.sleep(random.nextInt(100));
indexRecordManager.addIndexRecord(names[finalI], "u", "c.s.t", new String[] {"c"}, names[finalI], Collections.emptyList(), Arrays.asList("cp=1"));
}
catch (IOException | InterruptedException e) {
catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
@ -147,7 +147,7 @@ public class TestIndexRecordManager
}
indexRecordManager.deleteIndexRecord(names[0], Collections.emptyList());
}
catch (IOException | InterruptedException e) {
catch (InterruptedException e) {
throw new RuntimeException(e);
}
});

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

@ -94,6 +94,7 @@ public class IndexCache
if (PropertyService.getBooleanProperty(HetuConstant.FILTER_CACHE_SOFT_REFERENCE)) {
cacheBuilder.softValues();
}
// Refresh cache according to index records in the background. Evict index from cache if it's dropped.
executor.scheduleAtFixedRate(() -> {
try {
if (cache.size() > 0) {

2
presto-main/src/main/java/io/prestosql/execution/SqlQueryManager.java
View File

@ -257,6 +257,8 @@ public class SqlQueryManager
throw new PrestoException(GENERIC_INTERNAL_ERROR, format("Query %s already registered", queryExecution.getQueryId()));
}
// CreateIndex operations could not register cleanup operations like connectors
// Therefore a listener is added here to clean up index records on failure
if (isIndexCreationQuery(queryExecution.getQueryInfo())) {
queryExecution.addStateChangeListener(state -> {
try {

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

@ -98,6 +98,7 @@ public class IndexCache
if (PropertyService.getBooleanProperty(HetuConstant.FILTER_CACHE_SOFT_REFERENCE)) {
cacheBuilder.softValues();
}
// Refresh cache according to index records in the background. Evict index from cache if it's dropped.
executor.scheduleAtFixedRate(() -> {
try {
if (cache.size() > 0) {

9
presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryMetadata.java
View File

@ -81,7 +81,7 @@ import static java.util.stream.Collectors.toMap;
public class MemoryMetadata
implements ConnectorMetadata
{
public static final String MEM_KEY = "memory";
public static final String MEM_KEY = "memory"; // memory metadata all under this catalog
public static final String DEFAULT_SCHEMA = "default";
public static final String TABLE_ID_KEY = "id"; // used as param key in TableEntity
public static final String TABLE_OUTPUT_HANDLE = "output_handle"; // used as param key in TableEntity
@ -92,10 +92,11 @@ public class MemoryMetadata
private final NodeManager nodeManager;
private final TypeManager typeManager;
private final AtomicLong nextTableId;
private final Map<Long, TableInfo> tables = new ConcurrentHashMap<>();
private final Map<SchemaTableName, ConnectorViewDefinition> views = new ConcurrentHashMap<>();
private final HetuMetastore metastore;
private final MemoryConfig config;
// tables and views are cached here
private final Map<Long, TableInfo> tables = new ConcurrentHashMap<>();
private final Map<SchemaTableName, ConnectorViewDefinition> views = new ConcurrentHashMap<>();
@Inject
public MemoryMetadata(TypeManager typeManager, NodeManager nodeManager, HetuMetastore metastore, MemoryTableManager tableManager, MemoryConfig memoryConfig)
@ -248,8 +249,6 @@ public class MemoryMetadata
updateTableInfo(handle.getId(), null);
}
// CONTINUE HERE
@Override
public void renameTable(ConnectorSession session, ConnectorTableHandle tableHandle, SchemaTableName newTableName)
{

1
presto-memory/src/main/java/io/prestosql/plugin/memory/TableInfo.java
View File

@ -34,6 +34,7 @@ import static java.util.Objects.requireNonNull;
public class TableInfo
{
// Codec object to serialize/deserialize this class. Stored in metadata.
private static final JsonCodec<TableInfo> TABLE_INFO_JSON_CODEC = jsonCodec(TableInfo.class);
private final long id;

8
presto-memory/src/main/java/io/prestosql/plugin/memory/data/LogicalPart.java
View File

@ -131,6 +131,7 @@ public class LogicalPart
private transient PageSorter pageSorter;
private transient List<TypeSignature> typeSignatures;
private transient List<Type> types;
// Using majority of memory and disk space. Serialized and deserialized separately. Only loaded when used.
private transient List<Page> pages;
public LogicalPart(
@ -701,6 +702,9 @@ public class LogicalPart
return "logicalPart" + lpNum;
}
/**
* Deserialize pages from disk
*/
private synchronized void readPages()
throws IOException
{
@ -720,6 +724,10 @@ public class LogicalPart
LOG.debug("[Load] %s completed. Time elapsed: %dms", pagesFile.toString(), dur);
}
/**
* Serialize pages to disk
* @throws IOException
*/
private synchronized void writePages()
throws IOException
{

30
presto-memory/src/main/java/io/prestosql/plugin/memory/data/MemoryTableManager.java
View File

@ -84,6 +84,8 @@ public class MemoryTableManager
@GuardedBy("this")
private final AtomicLong currentBytes = new AtomicLong();
// in-memory LRU map of tableId -> table
private final Map<Long, Table> tables = new LinkedHashMap<>(16, 0.75f, true);
@Inject
@ -142,6 +144,9 @@ public class MemoryTableManager
}, 0, 3000);
}
/**
* Initialize a table and store it in memory
*/
public synchronized void initialize(long tableId, boolean compressionEnabled, List<MemoryColumnHandle> columns, List<SortingColumn> sortedBy, List<String> indexColumns)
{
if (!tables.containsKey(tableId)) {
@ -249,6 +254,11 @@ public class MemoryTableManager
return tables.containsKey(tableId);
}
/**
* Rollback table creation/insertion
*
* @param tableId the id of table to be cleaned
*/
public synchronized void cleanTable(Long tableId)
{
if (tables.containsKey(tableId)) {
@ -264,6 +274,11 @@ public class MemoryTableManager
}
}
/**
* Clean local tables. All non-active tables stored locally in tables map, and the data serialized on disk will be cleaned.
*
* @param activeTableIds the ids of active tables. all tables not in this set will be cleaned.
*/
public synchronized void refreshTables(Set<Long> activeTableIds)
{
// We have to remember that there might be some race conditions when there are two tables created at once.
@ -317,6 +332,13 @@ public class MemoryTableManager
return new Page(page.getPositionCount(), outputBlocks);
}
/**
* Spill table to disk.
*
* Table object (metadata) is serialized into one file. Pages are serialized separately in logical part.
*
* @param id table id to spill
*/
private synchronized void spillTable(long id)
throws IOException
{
@ -343,6 +365,14 @@ public class MemoryTableManager
LOG.debug("[Spill] Table " + id + " has been serialized to disk. Time elapsed: " + dur + "ms");
}
/**
* Restore the table from disk to load into tables map.
*
* Only the skeleton of the table and the logical parts in it will be restored at this time.
* (pages won't be loaded until used)
*
* @param id table to be restored
*/
public synchronized void restoreTable(long id)
throws IOException, ClassNotFoundException
{

11
presto-memory/src/main/java/io/prestosql/plugin/memory/data/Table.java
View File

@ -104,7 +104,7 @@ public class Table
private final List<String> indexColumns;
private final long maxLogicalPartBytes;
private final int maxPageSizeBytes;
private final List<LogicalPart> logicalParts;
private final List<LogicalPart> logicalParts; // actual data (pages) stored here
private final boolean compressionEnabled;
private TableState tableState;
private long lastModified = System.currentTimeMillis();
@ -154,7 +154,9 @@ public class Table
}, 5, 2, TimeUnit.SECONDS);
}
// used for deserialization
/**
* used for deserialization. these objects are per-runtime so must be restored separately after loading from disk.
*/
public void restoreTransientObjects(PageSorter pageSorter, TypeManager typeManager, PagesSerde pagesSerde, Path tableDataRoot)
{
this.pageSorter = pageSorter;
@ -166,6 +168,11 @@ public class Table
}
}
/**
* Add page to splits in a robin-robin way.
*
* @param page page to add
*/
public void add(Page page)
{
if (logicalParts.isEmpty() || !logicalParts.get(logicalParts.size() - 1).canAdd()) {

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

@ -25,7 +25,7 @@ public class IndexCacheKey
private final String path;
private final long lastModifiedTime;
private final CreateIndexMetadata.Level indexLevel;
private boolean noCloseFlag;
private boolean noCloseFlag; // Indicate that this index should not be closed at removal
/**
* @param path path to the file the index files should be read for
@ -75,7 +75,7 @@ public class IndexCacheKey
return noCloseFlag;
}
// only the path should be used as the key
// only the path is used as the key
// the lastModifiedTime time is only used to check if index is valid
@Override
public boolean equals(Object o)

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

@ -23,7 +23,7 @@ public interface IndexFilter
* Apply the filter on a given expression to check if the index matches the expression.
*
* @param expression the expression used to filter the result. e.g. col_a = 10 AND col_b in ("a", "b")
* @return if the indices in this filter matches the expression.
* @return if ANY index in this filter matches the expression.
*/
public boolean matches(Object expression);

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

@ -14,6 +14,9 @@
*/
package io.prestosql.spi.heuristicindex;
/**
* Special marker exception to indicate that a index is not created for the given data.
*/
public class IndexNotCreatedException
extends Exception
{