Clean up in progress record if index creation fails

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
 

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) 获取关于如何配置的更多信息。
 
 ## 索引语句
 

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();

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");
+    }
 }

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());
+        }
+    }
 }