!126 Add session control for heuristic index

Merge pull request !126 from Han_Weng/add-indexer-session-control

hetu-docs/en/indexer/indexer-cli.md

@@ -11,7 +11,8 @@ For example, `<path to installtion directory>/bin/index` and must be executed fr
 Usage: index [-v] [--debug] [--disableLocking] --table=<table>
          [-c=<configDirPath>] [--column=<columns>[,<columns>...]]...
          [--partition=<partitions>[,<partitions>...]]...
-         [--type=<indexTypes>[,<indexTypes>...]]... <command>
+         [--type=<indexTypes>[,<indexTypes>...]]...
+         [-I=<indexproperties>[,<indexproperties>...]]<command>
 
 Using this index tool, you can CREATE, SHOW and DELETE indexes.
 

hetu-docs/zh/indexer/indexer-cli.md

@@ -9,8 +9,8 @@
 使用方法：index [-v] [--debug] [--disableLocking] --table=<table>
          [-c=<configDirPath>] [--column=<columns>[,<columns>...]]...
          [--partition=<partitions>[,<partitions>...]]...
-         [--type=<indexTypes>[,<indexTypes>...]]...[-p=<plugins>[,
+         [--type=<indexTypes>[,<indexTypes>...]]...
-         <plugins>...]]...<command>
+         [-I=<indexproperties>[,<indexproperties>...]]<command>
 
 使用此索引工具，您可以创建、显示和删除索引。
 

presto-hive/src/main/java/io/prestosql/plugin/hive/HivePageSourceProvider.java

@@ -127,7 +127,7 @@ public class HivePageSourceProvider
                 new HdfsEnvironment.HdfsContext(session, hiveSplit.getDatabase(), hiveSplit.getTable()), path);
 
         List<IndexMetadata> indexes = null;
-        if (indexCache != null) {
+        if (indexCache != null && session.isHeuristicIndexFilterEnabled()) {
             indexes = indexCache.getIndices(
                     session.getCatalog().orElse(null),
                     hiveTable.getSchemaTableName().toString(), hiveSplit, hiveTable.getCompactEffectivePredicate(),

presto-main/src/main/java/io/prestosql/FullConnectorSession.java

@@ -167,4 +167,10 @@ public class FullConnectorSession
     {
         return SystemSessionProperties.getDynamicFilteringWaitTime(session);
     }
+
+    @Override
+    public boolean isHeuristicIndexFilterEnabled()
+    {
+        return SystemSessionProperties.isHeuristicIndexFilterEnabled(session);
+    }
 }

presto-main/src/main/java/io/prestosql/SystemSessionProperties.java

@@ -26,6 +26,7 @@ import io.prestosql.sql.analyzer.FeaturesConfig.DynamicFilterDataType;
 import io.prestosql.sql.analyzer.FeaturesConfig.JoinDistributionType;
 import io.prestosql.sql.analyzer.FeaturesConfig.JoinReorderingStrategy;
 import io.prestosql.sql.analyzer.FeaturesConfig.RedistributeWritesType;
+import io.prestosql.utils.HetuConfig;
 
 import javax.inject.Inject;
 
@@ -141,12 +142,13 @@ public final class SystemSessionProperties
     public static final String DYNAMIC_FILTERING_BLOOM_FILTER_FPP = "dynamic_filtering_bloom_filter_fpp";
     public static final String ENABLE_EXECUTION_PLAN_CACHE = "enable_execution_plan_cache";
     public static final String ENABLE_CROSS_REGION_DYNAMIC_FILTER = "cross-region-dynamic-filter-enabled";
+    public static final String ENABLE_HEURISTICINDEX_FILTER = "heuristicindex_filter_enabled";
 
     private final List<PropertyMetadata<?>> sessionProperties;
 
     public SystemSessionProperties()
     {
-        this(new QueryManagerConfig(), new TaskManagerConfig(), new MemoryManagerConfig(), new FeaturesConfig());
+        this(new QueryManagerConfig(), new TaskManagerConfig(), new MemoryManagerConfig(), new FeaturesConfig(), new HetuConfig());
     }
 
     @Inject
@@ -154,7 +156,8 @@ public final class SystemSessionProperties
             QueryManagerConfig queryManagerConfig,
             TaskManagerConfig taskManagerConfig,
             MemoryManagerConfig memoryManagerConfig,
-            FeaturesConfig featuresConfig)
+            FeaturesConfig featuresConfig,
+            HetuConfig hetuConfig)
     {
         sessionProperties = ImmutableList.of(
                 stringProperty(
@@ -635,6 +638,11 @@ public final class SystemSessionProperties
                         ENABLE_EXECUTION_PLAN_CACHE,
                         "Enable execution plan caching",
                         featuresConfig.isEnableExecutionPlanCache(),
+                        false),
+                booleanProperty(
+                        ENABLE_HEURISTICINDEX_FILTER,
+                        "Enable heuristic index filter",
+                        hetuConfig.isFilterEnabled(),
                         false));
     }
 
@@ -1123,4 +1131,9 @@ public final class SystemSessionProperties
     {
         return session.getSystemProperty(ENABLE_EXECUTION_PLAN_CACHE, Boolean.class);
     }
+
+    public static boolean isHeuristicIndexFilterEnabled(Session session)
+    {
+        return session.getSystemProperty(ENABLE_HEURISTICINDEX_FILTER, Boolean.class);
+    }
 }

presto-main/src/main/java/io/prestosql/execution/scheduler/FixedSourcePartitionedScheduler.java

@@ -18,6 +18,7 @@ import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Streams;
 import com.google.common.util.concurrent.ListenableFuture;
 import io.airlift.log.Logger;
+import io.prestosql.Session;
 import io.prestosql.execution.Lifespan;
 import io.prestosql.execution.RemoteTask;
 import io.prestosql.execution.SqlStageExecution;
@@ -75,6 +76,7 @@ public class FixedSourcePartitionedScheduler
             OptionalInt concurrentLifespansPerTask,
             NodeSelector nodeSelector,
             List<ConnectorPartitionHandle> partitionHandles,
+            Session session,
             HeuristicIndexerManager heuristicIndexerManager)
     {
         requireNonNull(stage, "stage is null");
@@ -109,6 +111,7 @@ public class FixedSourcePartitionedScheduler
         for (PlanNodeId planNodeId : schedulingOrder) {
             SplitSource splitSource = splitSources.get(planNodeId);
             boolean groupedExecutionForScanNode = stageExecutionDescriptor.isScanGroupedExecution(planNodeId);
+
             SourceScheduler sourceScheduler = newSourcePartitionedSchedulerAsSourceScheduler(
                     stage,
                     planNodeId,
@@ -116,6 +119,7 @@ public class FixedSourcePartitionedScheduler
                     splitPlacementPolicy,
                     Math.max(splitBatchSize / concurrentLifespans, 1),
                     groupedExecutionForScanNode,
+                    session,
                     heuristicIndexerManager);
 
             if (stageExecutionDescriptor.isStageGroupedExecution() && !groupedExecutionForScanNode) {

presto-main/src/main/java/io/prestosql/execution/scheduler/SourcePartitionedScheduler.java

@@ -20,6 +20,7 @@ import com.google.common.collect.Multimap;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
+import io.prestosql.Session;
 import io.prestosql.execution.Lifespan;
 import io.prestosql.execution.RemoteTask;
 import io.prestosql.execution.SqlStageExecution;
@@ -27,9 +28,7 @@ import io.prestosql.execution.scheduler.FixedSourcePartitionedScheduler.Bucketed
 import io.prestosql.heuristicindex.HeuristicIndexerManager;
 import io.prestosql.metadata.InternalNode;
 import io.prestosql.metadata.Split;
-import io.prestosql.spi.HetuConstant;
 import io.prestosql.spi.connector.ConnectorPartitionHandle;
-import io.prestosql.spi.service.PropertyService;
 import io.prestosql.split.EmptySplit;
 import io.prestosql.split.SplitSource;
 import io.prestosql.split.SplitSource.SplitBatch;
@@ -54,6 +53,7 @@ import static com.google.common.util.concurrent.Futures.nonCancellationPropagati
 import static io.airlift.concurrent.MoreFutures.addSuccessCallback;
 import static io.airlift.concurrent.MoreFutures.getFutureValue;
 import static io.airlift.concurrent.MoreFutures.whenAnyComplete;
+import static io.prestosql.SystemSessionProperties.isHeuristicIndexFilterEnabled;
 import static io.prestosql.execution.scheduler.ScheduleResult.BlockedReason.MIXED_SPLIT_QUEUES_FULL_AND_WAITING_FOR_SOURCE;
 import static io.prestosql.execution.scheduler.ScheduleResult.BlockedReason.NO_ACTIVE_DRIVER_GROUP;
 import static io.prestosql.execution.scheduler.ScheduleResult.BlockedReason.SPLIT_QUEUES_FULL;
@@ -95,6 +95,7 @@ public class SourcePartitionedScheduler
     private final int splitBatchSize;
     private final PlanNodeId partitionedNode;
     private final boolean groupedExecution;
+    private final Session session;
     private final HeuristicIndexerManager heuristicIndexerManager;
 
     private final Map<Lifespan, ScheduleGroup> scheduleGroups = new HashMap<>();
@@ -110,12 +111,14 @@ public class SourcePartitionedScheduler
             SplitPlacementPolicy splitPlacementPolicy,
             int splitBatchSize,
             boolean groupedExecution,
+            Session session,
             HeuristicIndexerManager heuristicIndexerManager)
     {
         this.stage = requireNonNull(stage, "stage is null");
         this.partitionedNode = requireNonNull(partitionedNode, "partitionedNode is null");
         this.splitSource = requireNonNull(splitSource, "splitSource is null");
         this.splitPlacementPolicy = requireNonNull(splitPlacementPolicy, "splitPlacementPolicy is null");
+        this.session = requireNonNull(session, "session is null");
         this.heuristicIndexerManager = requireNonNull(heuristicIndexerManager, "heuristicIndexerManager is null");
 
         checkArgument(splitBatchSize > 0, "splitBatchSize must be at least one");
@@ -141,10 +144,11 @@ public class SourcePartitionedScheduler
             SplitSource splitSource,
             SplitPlacementPolicy splitPlacementPolicy,
             int splitBatchSize,
+            Session session,
             HeuristicIndexerManager heuristicIndexerManager)
     {
         SourcePartitionedScheduler sourcePartitionedScheduler = new SourcePartitionedScheduler(stage, partitionedNode, splitSource,
-                splitPlacementPolicy, splitBatchSize, false, heuristicIndexerManager);
+                splitPlacementPolicy, splitBatchSize, false, session, heuristicIndexerManager);
         sourcePartitionedScheduler.startLifespan(Lifespan.taskWide(), NOT_PARTITIONED);
         sourcePartitionedScheduler.noMoreLifespans();
 
@@ -184,10 +188,11 @@ public class SourcePartitionedScheduler
             SplitPlacementPolicy splitPlacementPolicy,
             int splitBatchSize,
             boolean groupedExecution,
+            Session session,
             HeuristicIndexerManager heuristicIndexerManager)
     {
         return new SourcePartitionedScheduler(stage, partitionedNode, splitSource, splitPlacementPolicy,
-                splitBatchSize, groupedExecution, heuristicIndexerManager);
+                splitBatchSize, groupedExecution, session, heuristicIndexerManager);
     }
 
     @Override
@@ -219,12 +224,11 @@ public class SourcePartitionedScheduler
         int overallSplitAssignmentCount = 0;
         ImmutableSet.Builder<RemoteTask> overallNewTasks = ImmutableSet.builder();
         List<ListenableFuture<?>> overallBlockedFutures = new ArrayList<>();
+
         boolean anyBlockedOnPlacements = false;
         boolean anyBlockedOnNextSplitBatch = false;
         boolean anyNotBlocked = false;
-
+        boolean applyFilter = isHeuristicIndexFilterEnabled(session) && PredicateExtractor.isSplitFilterApplicable(stage);
-        boolean applyFilter = PropertyService.getBooleanProperty(HetuConstant.FILTER_ENABLED)
-                && PredicateExtractor.isSplitFilterApplicable(stage);
 
         for (Entry<Lifespan, ScheduleGroup> entry : scheduleGroups.entrySet()) {
             Lifespan lifespan = entry.getKey();
@@ -247,7 +251,7 @@ public class SourcePartitionedScheduler
                     SplitBatch nextSplits = getFutureValue(scheduleGroup.nextSplitBatchFuture);
                     scheduleGroup.nextSplitBatchFuture = null;
 
-                    //add split filter to filter out split has no valid rows
+                    // add split filter to filter out splits that do not contain valid rows
                     List<Split> filteredSplit = applyFilter ? SplitUtils.getFilteredSplit(PredicateExtractor.getExpression(stage),
                             PredicateExtractor.getFullyQualifiedName(stage), nextSplits, heuristicIndexerManager) : nextSplits.getSplits();
 

presto-main/src/main/java/io/prestosql/execution/scheduler/SqlQueryScheduler.java

@@ -342,8 +342,10 @@ public class SqlQueryScheduler
             SplitPlacementPolicy placementPolicy = new DynamicSplitPlacementPolicy(nodeSelector, stage::getAllTasks);
 
             checkArgument(!plan.getFragment().getStageExecutionDescriptor().isStageGroupedExecution());
+
             stageSchedulers.put(stageId, newSourcePartitionedSchedulerAsStageScheduler(stage, planNodeId, splitSource,
-                    placementPolicy, splitBatchSize, heuristicIndexerManager));
+                    placementPolicy, splitBatchSize, session, heuristicIndexerManager));
+
             bucketToPartition = Optional.of(new int[1]);
         }
         else if (partitioningHandle.equals(SCALED_WRITER_DISTRIBUTION)) {
@@ -404,6 +406,7 @@ public class SqlQueryScheduler
                         getConcurrentLifespansPerNode(session),
                         nodeScheduler.createNodeSelector(catalogName),
                         connectorPartitionHandles,
+                        session,
                         heuristicIndexerManager));
             }
             else {

presto-main/src/main/java/io/prestosql/metadata/SessionPropertyManager.java

@@ -83,8 +83,8 @@ public final class SessionPropertyManager
     public SessionPropertyManager(List<PropertyMetadata<?>> systemSessionProperties, HetuConfig hetuConfig)
     {
         SessionPropertyManager.hetuConfig = hetuConfig;
-        this.addSystemSessionProperties(systemSessionProperties);
         this.loadConfigToService(hetuConfig);
+        this.addSystemSessionProperties(systemSessionProperties);
     }
 
     public void addSystemSessionProperties(List<PropertyMetadata<?>> systemSessionProperties)

presto-main/src/main/java/io/prestosql/testing/LocalQueryRunner.java

@@ -317,7 +317,8 @@ public class LocalQueryRunner
 
         this.metadata = new MetadataManager(
                 featuresConfig,
-                new SessionPropertyManager(new SystemSessionProperties(new QueryManagerConfig(), taskManagerConfig, new MemoryManagerConfig(), featuresConfig)),
+                // new HetuConfig object passed, if split filtering is needed in the runner, a modified HetuConfig object with filter settings manually set must be used.
+                new SessionPropertyManager(new SystemSessionProperties(new QueryManagerConfig(), taskManagerConfig, new MemoryManagerConfig(), featuresConfig, new HetuConfig())),
                 new SchemaPropertyManager(),
                 new TablePropertyManager(),
                 new ColumnPropertyManager(),

presto-main/src/test/java/io/prestosql/execution/scheduler/TestSourcePartitionedScheduler.java

@@ -17,6 +17,7 @@ import com.google.common.base.Supplier;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Iterables;
+import io.prestosql.Session;
 import io.prestosql.client.NodeVersion;
 import io.prestosql.connector.CatalogName;
 import io.prestosql.cost.StatsAndCosts;
@@ -89,6 +90,7 @@ import static io.prestosql.sql.planner.SystemPartitioningHandle.SOURCE_DISTRIBUT
 import static io.prestosql.sql.planner.plan.ExchangeNode.Type.GATHER;
 import static io.prestosql.sql.planner.plan.JoinNode.Type.INNER;
 import static io.prestosql.testing.TestingHandles.TEST_TABLE_HANDLE;
+import static io.prestosql.testing.TestingSession.testSessionBuilder;
 import static io.prestosql.testing.assertions.PrestoExceptionAssert.assertPrestoExceptionThrownBy;
 import static java.lang.Integer.min;
 import static java.util.Objects.requireNonNull;
@@ -108,6 +110,7 @@ public class TestSourcePartitionedScheduler
     private final LocationFactory locationFactory = new MockLocationFactory();
     private final InMemoryNodeManager nodeManager = new InMemoryNodeManager();
     private final FinalizerService finalizerService = new FinalizerService();
+    private static final Session session = testSessionBuilder().build();
     SeedStoreManager seedStoreManager = new SeedStoreManager(new FileSystemClientManager());
 
     public TestSourcePartitionedScheduler()
@@ -324,7 +327,7 @@ public class TestSourcePartitionedScheduler
                     Iterables.getOnlyElement(plan.getSplitSources().keySet()),
                     Iterables.getOnlyElement(plan.getSplitSources().values()),
                     new DynamicSplitPlacementPolicy(nodeScheduler.createNodeSelector(CONNECTOR_ID), stage::getAllTasks),
-                    2, new HeuristicIndexerManager(new FileSystemClientManager()));
+                    2, session, new HeuristicIndexerManager(new FileSystemClientManager()));
             scheduler.schedule();
         }).hasErrorCode(NO_NODES_AVAILABLE);
     }
@@ -449,7 +452,7 @@ public class TestSourcePartitionedScheduler
         SplitSource splitSource = Iterables.getOnlyElement(plan.getSplitSources().values());
         SplitPlacementPolicy placementPolicy = new DynamicSplitPlacementPolicy(nodeScheduler.createNodeSelector(splitSource.getCatalogName()), stage::getAllTasks);
         return newSourcePartitionedSchedulerAsStageScheduler(stage, sourceNode, splitSource,
-                placementPolicy, splitBatchSize, new HeuristicIndexerManager(new FileSystemClientManager()));
+                placementPolicy, splitBatchSize, session, new HeuristicIndexerManager(new FileSystemClientManager()));
     }
 
     private static StageExecutionPlan createPlan(ConnectorSplitSource splitSource)

presto-main/src/test/java/io/prestosql/sql/analyzer/TestAnalyzer.java

@@ -52,6 +52,7 @@ import io.prestosql.testing.TestingMetadata;
 import io.prestosql.transaction.TransactionId;
 import io.prestosql.transaction.TransactionInfo;
 import io.prestosql.transaction.TransactionManager;
+import io.prestosql.utils.HetuConfig;
 import org.intellij.lang.annotations.Language;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
@@ -581,7 +582,8 @@ public class TestAnalyzer
                 new QueryManagerConfig(),
                 new TaskManagerConfig(),
                 new MemoryManagerConfig(),
-                new FeaturesConfig().setMaxGroupingSets(2048)))).build();
+                new FeaturesConfig().setMaxGroupingSets(2048),
+                new HetuConfig()))).build();
         analyze(session, "SELECT a, b, c, d, e, f, g, h, i, j, k, SUM(l)" +
                 "FROM (VALUES (1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12))\n" +
                 "t (a, b, c, d, e, f, g, h, i, j, k, l)\n" +

presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorSession.java

@@ -74,4 +74,9 @@ public interface ConnectorSession
     {
         return new Duration(0, TimeUnit.MILLISECONDS);
     }
+
+    default boolean isHeuristicIndexFilterEnabled()
+    {
+        return true;
+    }
 }