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 823daa3a4..dd92ba6eb 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 @@ -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 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 6f8e47cfc..20ca89213 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 @@ -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 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 indexProperties, List 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 partitionsToRemove) - throws IOException { getIndexRecords().stream().filter(record -> record.name.equals(name)) .forEach(record -> { diff --git a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/PartitionIndexWriter.java b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/PartitionIndexWriter.java index 842d56919..0708e84c3 100644 --- a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/PartitionIndexWriter.java +++ b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/PartitionIndexWriter.java @@ -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 -> "::" + 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 symbolToIdMap; private final Map>, String> dataMap; @@ -121,6 +131,12 @@ public class PartitionIndexWriter Comparable> comparableKey = (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 diff --git a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/filter/HeuristicIndexFilter.java b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/filter/HeuristicIndexFilter.java index 809dab8e0..4b23bdd37 100644 --- a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/filter/HeuristicIndexFilter.java +++ b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/filter/HeuristicIndexFilter.java @@ -39,6 +39,11 @@ public class HeuristicIndexFilter { Map> 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> 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(), 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 e27638e6a..b9da13644 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 @@ -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); } }); 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 e60da01f8..ef1cc6d23 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 @@ -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) { diff --git a/presto-main/src/main/java/io/prestosql/execution/SqlQueryManager.java b/presto-main/src/main/java/io/prestosql/execution/SqlQueryManager.java index d73ab804b..d5ea8dc69 100644 --- a/presto-main/src/main/java/io/prestosql/execution/SqlQueryManager.java +++ b/presto-main/src/main/java/io/prestosql/execution/SqlQueryManager.java @@ -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 { 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 5eaaae573..5f2c064f9 100644 --- a/presto-main/src/main/java/io/prestosql/heuristicindex/IndexCache.java +++ b/presto-main/src/main/java/io/prestosql/heuristicindex/IndexCache.java @@ -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) { diff --git a/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryMetadata.java b/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryMetadata.java index 22991778f..771555ed7 100644 --- a/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryMetadata.java +++ b/presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryMetadata.java @@ -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 tables = new ConcurrentHashMap<>(); - private final Map views = new ConcurrentHashMap<>(); private final HetuMetastore metastore; private final MemoryConfig config; + // tables and views are cached here + private final Map tables = new ConcurrentHashMap<>(); + private final Map 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) { diff --git a/presto-memory/src/main/java/io/prestosql/plugin/memory/TableInfo.java b/presto-memory/src/main/java/io/prestosql/plugin/memory/TableInfo.java index 2b108fd62..5d7b61a87 100644 --- a/presto-memory/src/main/java/io/prestosql/plugin/memory/TableInfo.java +++ b/presto-memory/src/main/java/io/prestosql/plugin/memory/TableInfo.java @@ -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 TABLE_INFO_JSON_CODEC = jsonCodec(TableInfo.class); private final long id; diff --git a/presto-memory/src/main/java/io/prestosql/plugin/memory/data/LogicalPart.java b/presto-memory/src/main/java/io/prestosql/plugin/memory/data/LogicalPart.java index a495be2ec..f1c3b1e57 100644 --- a/presto-memory/src/main/java/io/prestosql/plugin/memory/data/LogicalPart.java +++ b/presto-memory/src/main/java/io/prestosql/plugin/memory/data/LogicalPart.java @@ -131,6 +131,7 @@ public class LogicalPart private transient PageSorter pageSorter; private transient List typeSignatures; private transient List types; + // Using majority of memory and disk space. Serialized and deserialized separately. Only loaded when used. private transient List 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 { diff --git a/presto-memory/src/main/java/io/prestosql/plugin/memory/data/MemoryTableManager.java b/presto-memory/src/main/java/io/prestosql/plugin/memory/data/MemoryTableManager.java index b59fc31c6..22ba3aed2 100644 --- a/presto-memory/src/main/java/io/prestosql/plugin/memory/data/MemoryTableManager.java +++ b/presto-memory/src/main/java/io/prestosql/plugin/memory/data/MemoryTableManager.java @@ -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 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 columns, List sortedBy, List 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 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 { diff --git a/presto-memory/src/main/java/io/prestosql/plugin/memory/data/Table.java b/presto-memory/src/main/java/io/prestosql/plugin/memory/data/Table.java index e366f0860..fc4f9194d 100644 --- a/presto-memory/src/main/java/io/prestosql/plugin/memory/data/Table.java +++ b/presto-memory/src/main/java/io/prestosql/plugin/memory/data/Table.java @@ -104,7 +104,7 @@ public class Table private final List indexColumns; private final long maxLogicalPartBytes; private final int maxPageSizeBytes; - private final List logicalParts; + private final List 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()) { diff --git a/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexCacheKey.java b/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexCacheKey.java index 496b66395..5ad293cb9 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexCacheKey.java +++ b/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexCacheKey.java @@ -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) diff --git a/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexFilter.java b/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexFilter.java index 9540a14d0..05abfb793 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexFilter.java +++ b/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexFilter.java @@ -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); diff --git a/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexNotCreatedException.java b/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexNotCreatedException.java index 63656edf7..4f976a852 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexNotCreatedException.java +++ b/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexNotCreatedException.java @@ -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 {