jffwang
/
hetu-core
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Add session control for heuristic index

This commit is contained in:
Han Weng 2020-08-07 16:10:53 -04:00
parent 51f4b9d7e2
commit cf91091522
13 changed files with 62 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
hetu-docs/en/indexer/indexer-cli.md
View File

@ -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.

4
hetu-docs/zh/indexer/indexer-cli.md
View File

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

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

@ -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(),

6
presto-main/src/main/java/io/prestosql/FullConnectorSession.java
View File

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

17
presto-main/src/main/java/io/prestosql/SystemSessionProperties.java
View File

@ -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);
}
}

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

@ -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) {

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

@ -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 = PropertyService.getBooleanProperty(HetuConstant.FILTER_ENABLED)
&& PredicateExtractor.isSplitFilterApplicable(stage);
boolean applyFilter = isHeuristicIndexFilterEnabled(session) && 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();

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

@ -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 {

2
presto-main/src/main/java/io/prestosql/metadata/SessionPropertyManager.java
View File

@ -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)

3
presto-main/src/main/java/io/prestosql/testing/LocalQueryRunner.java
View File

@ -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(),

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

@ -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)

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

@ -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" +

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

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