diff --git a/hetu-docs/en/indexer/overview.md b/hetu-docs/en/indexer/overview.md index ddaa39b27..0a6ab6ff7 100644 --- a/hetu-docs/en/indexer/overview.md +++ b/hetu-docs/en/indexer/overview.md @@ -123,7 +123,7 @@ Subsequent queries will utilize the index to reduce the amount of data read | hetu.heuristicindex.indexstore.uri | /opt/hetu/indices/ | No | Directory under which all index files are stored| | hetu.heuristicindex.indexstore.filesystem.profile | local-config-default| No | This property defines the filesystem profile used to read and write index| -Heuristic indexer now uses Hetu Metastore to manage its metadata. Please check [vdm](../connector/vdm.md) for more information. +Heuristic indexer now uses Hetu Metastore to manage its metadata. Please check [Hetu Metastore](../admin/meta-store.md) for more information. ## Index Statements diff --git a/hetu-docs/zh/indexer/overview.md b/hetu-docs/zh/indexer/overview.md index 3222358b9..0e40b5a3d 100644 --- a/hetu-docs/zh/indexer/overview.md +++ b/hetu-docs/zh/indexer/overview.md @@ -107,7 +107,7 @@ | hetu.heuristicindex.indexstore.uri | /opt/hetu/indices/ | 否 | 所有索引文件存储的目录| | hetu.heuristicindex.indexstore.filesystem.profile | local-config-default| 否 | 用于存储索引文件的文件系统属性描述文件名称| -索引功能现使用Hetu Metastore管理元数据。请参阅 [vdm](../connector/vdm.md) 获取关于如何配置的更多信息。 +索引功能现使用Hetu Metastore管理元数据。请参阅 [Hetu Metastore](../admin/meta-store.md) 获取关于如何配置的更多信息。 ## 索引语句 diff --git a/presto-main/src/main/java/io/prestosql/event/QueryMonitor.java b/presto-main/src/main/java/io/prestosql/event/QueryMonitor.java index babd86358..a6d2551b2 100644 --- a/presto-main/src/main/java/io/prestosql/event/QueryMonitor.java +++ b/presto-main/src/main/java/io/prestosql/event/QueryMonitor.java @@ -28,10 +28,12 @@ import io.prestosql.execution.Column; import io.prestosql.execution.ExecutionFailureInfo; import io.prestosql.execution.Input; import io.prestosql.execution.QueryInfo; +import io.prestosql.execution.QueryState; import io.prestosql.execution.QueryStats; import io.prestosql.execution.StageInfo; import io.prestosql.execution.TaskInfo; import io.prestosql.execution.TaskState; +import io.prestosql.heuristicindex.HeuristicIndexerManager; import io.prestosql.metadata.Metadata; import io.prestosql.metadata.SessionPropertyManager; import io.prestosql.operator.OperatorStats; @@ -63,6 +65,7 @@ import java.util.Map; import java.util.Optional; import java.util.stream.Collectors; +import static io.prestosql.execution.QueryState.FAILED; import static io.prestosql.execution.QueryState.QUEUED; import static io.prestosql.sql.planner.planprinter.PlanPrinter.textDistributedPlan; import static java.lang.Math.max; @@ -86,6 +89,7 @@ public class QueryMonitor private final SessionPropertyManager sessionPropertyManager; private final Metadata metadata; private final int maxJsonLimit; + private final HeuristicIndexerManager heuristicIndexerManager; @Inject public QueryMonitor( @@ -98,7 +102,8 @@ public class QueryMonitor NodeVersion nodeVersion, SessionPropertyManager sessionPropertyManager, Metadata metadata, - QueryMonitorConfig config) + QueryMonitorConfig config, + HeuristicIndexerManager heuristicIndexerManager) { this.eventListenerManager = requireNonNull(eventListenerManager, "eventListenerManager is null"); this.stageInfoCodec = requireNonNull(stageInfoCodec, "stageInfoCodec is null"); @@ -111,6 +116,7 @@ public class QueryMonitor this.sessionPropertyManager = requireNonNull(sessionPropertyManager, "sessionPropertyManager is null"); this.metadata = requireNonNull(metadata, "metadata is null"); this.maxJsonLimit = toIntExact(requireNonNull(config, "config is null").getMaxOutputStageJsonSize().toBytes()); + this.heuristicIndexerManager = requireNonNull(heuristicIndexerManager, "heuristicIndexerManager is null"); } public void queryCreatedEvent(BasicQueryInfo queryInfo) @@ -181,6 +187,13 @@ public class QueryMonitor logQueryTimeline(queryInfo); } + public void indexCreationStateChangeEvent(QueryState state, QueryInfo queryInfo) + { + if (state == FAILED) { + heuristicIndexerManager.cleanUpIndexRecord(queryInfo); + } + } + public void queryCompletedEvent(QueryInfo queryInfo) { QueryStats queryStats = queryInfo.getQueryStats(); 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 b44eb0912..05727c65b 100644 --- a/presto-main/src/main/java/io/prestosql/execution/SqlQueryManager.java +++ b/presto-main/src/main/java/io/prestosql/execution/SqlQueryManager.java @@ -49,6 +49,7 @@ import javax.annotation.concurrent.ThreadSafe; import javax.inject.Inject; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.NoSuchElementException; import java.util.Objects; @@ -246,6 +247,18 @@ public class SqlQueryManager throw new PrestoException(GENERIC_INTERNAL_ERROR, format("Query %s already registered", queryExecution.getQueryId())); } + if (isIndexCreationQuery(queryExecution.getQueryInfo())) { + queryExecution.addStateChangeListener(state -> { + try { + queryMonitor.indexCreationStateChangeEvent(state, queryExecution.getQueryInfo()); + } + finally { + // execution MUST be added to the expiration queue or there will be a leak + queryTracker.expireQuery(queryExecution.getQueryId()); + } + }); + } + queryExecution.addFinalQueryInfoListener(finalQueryInfo -> { try { queryMonitor.queryCompletedEvent(finalQueryInfo); @@ -421,4 +434,9 @@ public class SqlQueryManager queryTracker.removeQuery(queryId); } } + + private boolean isIndexCreationQuery(QueryInfo queryInfo) + { + return queryInfo.getQuery().toUpperCase(Locale.ROOT).startsWith("CREATE INDEX"); + } } diff --git a/presto-main/src/main/java/io/prestosql/heuristicindex/HeuristicIndexerManager.java b/presto-main/src/main/java/io/prestosql/heuristicindex/HeuristicIndexerManager.java index 051d18c00..f3b379458 100644 --- a/presto-main/src/main/java/io/prestosql/heuristicindex/HeuristicIndexerManager.java +++ b/presto-main/src/main/java/io/prestosql/heuristicindex/HeuristicIndexerManager.java @@ -17,6 +17,7 @@ package io.prestosql.heuristicindex; import com.google.common.annotations.VisibleForTesting; import com.google.inject.Inject; import io.airlift.log.Logger; +import io.prestosql.execution.QueryInfo; import io.prestosql.filesystem.FileSystemClientManager; import io.prestosql.metastore.HetuMetaStoreManager; import io.prestosql.spi.HetuConstant; @@ -26,6 +27,7 @@ import io.prestosql.spi.heuristicindex.IndexClient; import io.prestosql.spi.heuristicindex.IndexFactory; import io.prestosql.spi.heuristicindex.IndexFilter; import io.prestosql.spi.heuristicindex.IndexMetadata; +import io.prestosql.spi.heuristicindex.IndexRecord; import io.prestosql.spi.heuristicindex.IndexWriter; import io.prestosql.spi.metastore.HetuMetastore; import io.prestosql.spi.service.PropertyService; @@ -37,6 +39,7 @@ import java.nio.file.FileSystemException; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Properties; @@ -127,4 +130,20 @@ public class HeuristicIndexerManager } } } + + public void cleanUpIndexRecord(QueryInfo queryInfo) + { + try { + String query = queryInfo.getQuery(); + LOG.debug("Clean up index record after this query failed: %s", query); + String indexName = query.split(" ")[2]; + IndexRecord record = indexClient.lookUpIndexRecord(indexName); + if (record != null && record.isInProgressRecord()) { + indexClient.deleteIndex(indexName, Collections.emptyList()); + } + } + catch (Exception e) { + LOG.debug("Failed to clean index record for : %s", queryInfo.getQuery()); + } + } }