diff --git a/hetu-docs/en/indexer/indexer-cli.md b/hetu-docs/en/indexer/indexer-cli.md index d50da7fda..a8e53e687 100644 --- a/hetu-docs/en/indexer/indexer-cli.md +++ b/hetu-docs/en/indexer/indexer-cli.md @@ -11,7 +11,8 @@ For example, `/bin/index` and must be executed fr Usage: index [-v] [--debug] [--disableLocking] --table= [-c=] [--column=[,...]]... [--partition=[,...]]... - [--type=[,...]]... + [--type=[,...]]... + [-I=[,...]] Using this index tool, you can CREATE, SHOW and DELETE indexes. diff --git a/hetu-docs/zh/indexer/indexer-cli.md b/hetu-docs/zh/indexer/indexer-cli.md index 59b8639b2..652164aa8 100644 --- a/hetu-docs/zh/indexer/indexer-cli.md +++ b/hetu-docs/zh/indexer/indexer-cli.md @@ -9,8 +9,8 @@ 使用方法：index [-v] [--debug] [--disableLocking] --table=
[-c=] [--column=[,...]]... [--partition=[,...]]... - [--type=[,...]]...[-p=[, - ...]]... + [--type=[,...]]... + [-I=[,...]] 使用此索引工具，您可以创建、显示和删除索引。 diff --git a/presto-hive/src/main/java/io/prestosql/plugin/hive/HivePageSourceProvider.java b/presto-hive/src/main/java/io/prestosql/plugin/hive/HivePageSourceProvider.java index 28a7cfef1..414a0fc48 100644 --- a/presto-hive/src/main/java/io/prestosql/plugin/hive/HivePageSourceProvider.java +++ b/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 indexes = null; - if (indexCache != null) { + if (indexCache != null && session.isHeuristicIndexFilterEnabled()) { indexes = indexCache.getIndices( session.getCatalog().orElse(null), hiveTable.getSchemaTableName().toString(), hiveSplit, hiveTable.getCompactEffectivePredicate(), diff --git a/presto-main/src/main/java/io/prestosql/FullConnectorSession.java b/presto-main/src/main/java/io/prestosql/FullConnectorSession.java index 64401fbc2..87e362d18 100644 --- a/presto-main/src/main/java/io/prestosql/FullConnectorSession.java +++ b/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); + } } diff --git a/presto-main/src/main/java/io/prestosql/SystemSessionProperties.java b/presto-main/src/main/java/io/prestosql/SystemSessionProperties.java index 73789c425..da0d53304 100644 --- a/presto-main/src/main/java/io/prestosql/SystemSessionProperties.java +++ b/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> 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); + } } diff --git a/presto-main/src/main/java/io/prestosql/execution/scheduler/FixedSourcePartitionedScheduler.java b/presto-main/src/main/java/io/prestosql/execution/scheduler/FixedSourcePartitionedScheduler.java index 0568a7e5e..43bf9b1a5 100644 --- a/presto-main/src/main/java/io/prestosql/execution/scheduler/FixedSourcePartitionedScheduler.java +++ b/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 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) { diff --git a/presto-main/src/main/java/io/prestosql/execution/scheduler/SourcePartitionedScheduler.java b/presto-main/src/main/java/io/prestosql/execution/scheduler/SourcePartitionedScheduler.java index e9fde80f4..b9df31b16 100644 --- a/presto-main/src/main/java/io/prestosql/execution/scheduler/SourcePartitionedScheduler.java +++ b/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 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 overallNewTasks = ImmutableSet.builder(); List> overallBlockedFutures = new ArrayList<>(); + boolean anyBlockedOnPlacements = false; boolean anyBlockedOnNextSplitBatch = false; boolean anyNotBlocked = false; - - boolean applyFilter = PropertyService.getBooleanProperty(HetuConstant.FILTER_ENABLED) - && PredicateExtractor.isSplitFilterApplicable(stage); + boolean applyFilter = isHeuristicIndexFilterEnabled(session) && PredicateExtractor.isSplitFilterApplicable(stage); for (Entry 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 filteredSplit = applyFilter ? SplitUtils.getFilteredSplit(PredicateExtractor.getExpression(stage), PredicateExtractor.getFullyQualifiedName(stage), nextSplits, heuristicIndexerManager) : nextSplits.getSplits(); diff --git a/presto-main/src/main/java/io/prestosql/execution/scheduler/SqlQueryScheduler.java b/presto-main/src/main/java/io/prestosql/execution/scheduler/SqlQueryScheduler.java index fe56bf8c1..68bad1077 100644 --- a/presto-main/src/main/java/io/prestosql/execution/scheduler/SqlQueryScheduler.java +++ b/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 { diff --git a/presto-main/src/main/java/io/prestosql/metadata/SessionPropertyManager.java b/presto-main/src/main/java/io/prestosql/metadata/SessionPropertyManager.java index 2aea28842..02ad5400c 100644 --- a/presto-main/src/main/java/io/prestosql/metadata/SessionPropertyManager.java +++ b/presto-main/src/main/java/io/prestosql/metadata/SessionPropertyManager.java @@ -83,8 +83,8 @@ public final class SessionPropertyManager public SessionPropertyManager(List> systemSessionProperties, HetuConfig hetuConfig) { SessionPropertyManager.hetuConfig = hetuConfig; - this.addSystemSessionProperties(systemSessionProperties); this.loadConfigToService(hetuConfig); + this.addSystemSessionProperties(systemSessionProperties); } public void addSystemSessionProperties(List> systemSessionProperties) diff --git a/presto-main/src/main/java/io/prestosql/testing/LocalQueryRunner.java b/presto-main/src/main/java/io/prestosql/testing/LocalQueryRunner.java index c69776596..4864116c8 100644 --- a/presto-main/src/main/java/io/prestosql/testing/LocalQueryRunner.java +++ b/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(), diff --git a/presto-main/src/test/java/io/prestosql/execution/scheduler/TestSourcePartitionedScheduler.java b/presto-main/src/test/java/io/prestosql/execution/scheduler/TestSourcePartitionedScheduler.java index f708ba25a..1079a0bf0 100644 --- a/presto-main/src/test/java/io/prestosql/execution/scheduler/TestSourcePartitionedScheduler.java +++ b/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) diff --git a/presto-main/src/test/java/io/prestosql/sql/analyzer/TestAnalyzer.java b/presto-main/src/test/java/io/prestosql/sql/analyzer/TestAnalyzer.java index 14cd0a4ae..3c9079c2d 100644 --- a/presto-main/src/test/java/io/prestosql/sql/analyzer/TestAnalyzer.java +++ b/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" + diff --git a/presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorSession.java b/presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorSession.java index f7a4fdbc2..4921eb092 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorSession.java +++ b/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; + } }