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

!1436 support omniruntime 

Merge pull request !1436 from zhousipei/support_omniruntime
This commit is contained in:
i-robot 2022-04-16 07:15:09 +00:00 committed by Gitee
parent 6ff4d8264c 4b0edc0804
commit db1c66c8e5
No known key found for this signature in database
GPG Key ID: 173E9B9CA92EEF8F
38 changed files with 721 additions and 133 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -37,6 +37,9 @@ import java.util.Optional;
import java.util.OptionalInt;
import static com.google.common.base.Preconditions.checkArgument;
import static io.prestosql.spi.HetuConstant.EXTENSION_EXECUTION_PLANNER_CLASS_PATH;
import static io.prestosql.spi.HetuConstant.EXTENSION_EXECUTION_PLANNER_ENABLED;
import static io.prestosql.spi.HetuConstant.EXTENSION_EXECUTION_PLANNER_JAR_PATH;
import static io.prestosql.spi.StandardErrorCode.INVALID_SESSION_PROPERTY;
import static io.prestosql.spi.session.PropertyMetadata.booleanProperty;
import static io.prestosql.spi.session.PropertyMetadata.dataSizeProperty;
@ -810,6 +813,22 @@ public final class SystemSessionProperties
SKIP_NON_APPLICABLE_RULES_ENABLED,
"Whether to skip applying some selected rules based on query pattern",
featuresConfig.isSkipNonApplicableRulesEnabled(),
false),
// add extension execution planner and operator
stringProperty(
EXTENSION_EXECUTION_PLANNER_JAR_PATH,
"extension execution planner jar path",
hetuConfig.getExtensionExecutionPlannerJarPath(),
false),
stringProperty(
EXTENSION_EXECUTION_PLANNER_CLASS_PATH,
"extension execution planner class path",
hetuConfig.getExtensionExecutionPlannerClassPath(),
false),
booleanProperty(
EXTENSION_EXECUTION_PLANNER_ENABLED,
"extension execution planner enabled",
hetuConfig.getExtensionExecutionPlannerEnabled(),
false));
}
@ -1418,4 +1437,19 @@ public final class SystemSessionProperties
{
return session.getSystemProperty(SKIP_NON_APPLICABLE_RULES_ENABLED, Boolean.class);
}
public static Boolean isExtensionExecutionPlannerEnabled(Session session)
{
return session.getSystemProperty(EXTENSION_EXECUTION_PLANNER_ENABLED, Boolean.class);
}
public static String getExtensionExecutionPlannerJarPath(Session session)
{
return session.getSystemProperty(EXTENSION_EXECUTION_PLANNER_JAR_PATH, String.class);
}
public static String getExtensionExecutionPlannerClassPath(Session session)
{
return session.getSystemProperty(EXTENSION_EXECUTION_PLANNER_CLASS_PATH, String.class);
}
}

93
presto-main/src/main/java/io/prestosql/execution/SqlTaskExecutionFactory.java
View File

@ -14,6 +14,7 @@
package io.prestosql.execution;
import io.airlift.concurrent.SetThreadName;
import io.airlift.log.Logger;
import io.hetu.core.transport.execution.buffer.PagesSerdeFactory;
import io.prestosql.Session;
import io.prestosql.event.SplitMonitor;
@ -23,12 +24,19 @@ import io.prestosql.memory.QueryContext;
import io.prestosql.metadata.Metadata;
import io.prestosql.operator.CommonTableExecutionContext;
import io.prestosql.operator.TaskContext;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.plan.PlanNodeId;
import io.prestosql.sql.planner.LocalExecutionPlanner;
import io.prestosql.sql.planner.LocalExecutionPlanner.LocalExecutionPlan;
import io.prestosql.sql.planner.PlanFragment;
import io.prestosql.sql.planner.TypeProvider;
import javax.annotation.Nullable;
import java.lang.reflect.Constructor;
import java.net.MalformedURLException;
import java.net.URL;
import java.net.URLClassLoader;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@ -36,12 +44,19 @@ import java.util.OptionalInt;
import java.util.concurrent.Executor;
import static com.google.common.base.Throwables.throwIfUnchecked;
import static io.prestosql.SystemSessionProperties.getExtensionExecutionPlannerClassPath;
import static io.prestosql.SystemSessionProperties.getExtensionExecutionPlannerJarPath;
import static io.prestosql.SystemSessionProperties.isExchangeCompressionEnabled;
import static io.prestosql.SystemSessionProperties.isExtensionExecutionPlannerEnabled;
import static io.prestosql.execution.SqlTaskExecution.createSqlTaskExecution;
import static io.prestosql.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR;
import static java.util.Objects.requireNonNull;
public class SqlTaskExecutionFactory
{
private static final Logger log = Logger.get(SqlTaskExecutionFactory.class);
private static LocalExecutionPlanner extensionPlanner;
private static boolean extensionPlannerInitialized;
private final Executor taskNotificationExecutor;
private final TaskExecutor taskExecutor;
@ -83,20 +98,32 @@ public class SqlTaskExecutionFactory
consumer,
new PagesSerdeFactory(metadata.getFunctionAndTypeManager().getBlockEncodingSerde(), isExchangeCompressionEnabled(session)));
LocalExecutionPlan localExecutionPlan;
LocalExecutionPlan localExecutionPlan = null;
try (SetThreadName ignored = new SetThreadName("Task-%s", taskStateMachine.getTaskId())) {
try {
localExecutionPlan = planner.plan(
taskContext,
fragment.getRoot(),
TypeProvider.copyOf(fragment.getSymbols()),
fragment.getPartitioningScheme(),
fragment.getStageExecutionDescriptor(),
fragment.getPartitionedSources(),
outputBuffer,
fragment.getFeederCTEId(),
fragment.getFeederCTEParentId(),
cteCtx);
if (isExtensionExecutionPlannerEnabled(session)) {
String jarPath = getExtensionExecutionPlannerJarPath(session);
String classPath = getExtensionExecutionPlannerClassPath(session);
if (jarPath != null && !jarPath.equals("") && classPath != null && !classPath.equals("")) {
localExecutionPlan = loadExtensionLocalExecutionPlan(outputBuffer, fragment, taskContext, cteCtx, jarPath, classPath);
}
else {
throw new PrestoException(GENERIC_INTERNAL_ERROR, "Extension execution planner jar path or class path isn't configured correctly");
}
}
if (localExecutionPlan == null) {
localExecutionPlan = planner.plan(
taskContext,
fragment.getRoot(),
TypeProvider.copyOf(fragment.getSymbols()),
fragment.getPartitioningScheme(),
fragment.getStageExecutionDescriptor(),
fragment.getPartitionedSources(),
outputBuffer,
fragment.getFeederCTEId(),
fragment.getFeederCTEParentId(),
cteCtx);
}
}
catch (Throwable e) {
// planning failed
@ -115,4 +142,46 @@ public class SqlTaskExecutionFactory
taskNotificationExecutor,
splitMonitor);
}
@Nullable
private LocalExecutionPlan loadExtensionLocalExecutionPlan(OutputBuffer outputBuffer, PlanFragment fragment, TaskContext taskContext, Map<String, CommonTableExecutionContext> cteCtx, String jarPath, String classPath)
{
if (!extensionPlannerInitialized) {
try {
ExtensionClassLoader extensionClassLoader = new ExtensionClassLoader(jarPath, Thread.currentThread().getContextClassLoader());
Thread.currentThread().setContextClassLoader(extensionClassLoader);
Class<?> aClass = extensionClassLoader.loadClass(classPath);
Constructor<?> constructor = aClass.getConstructor(LocalExecutionPlanner.class);
extensionPlanner = (LocalExecutionPlanner) constructor.newInstance(planner);
extensionPlannerInitialized = true;
}
catch (Throwable e) {
log.warn("get extension LocalExecutionPlanner failed: %s", e.toString());
throw new PrestoException(GENERIC_INTERNAL_ERROR, e);
}
}
if (extensionPlanner != null) {
return extensionPlanner.plan(
taskContext,
fragment.getRoot(),
TypeProvider.copyOf(fragment.getSymbols()),
fragment.getPartitioningScheme(),
fragment.getStageExecutionDescriptor(),
fragment.getPartitionedSources(),
outputBuffer,
fragment.getFeederCTEId(),
fragment.getFeederCTEParentId(),
cteCtx);
}
return null;
}
public static class ExtensionClassLoader
extends URLClassLoader
{
public ExtensionClassLoader(final String path, ClassLoader parent) throws MalformedURLException
{
super(new URL[] {new URL(path)}, parent);
}
}
}

13
presto-main/src/main/java/io/prestosql/execution/StateMachine.java
View File

@ -49,6 +49,7 @@ public class StateMachine<T>
private final Executor executor;
private final Object lock = new Object();
private final Set<T> terminalStates;
private StateChangeListener tailStateChangeListener;
@GuardedBy("lock")
private volatile T state;
@ -279,6 +280,12 @@ public class StateMachine<T>
inTerminalState = isTerminalState(currentState);
if (!inTerminalState) {
stateChangeListeners.add(stateChangeListener);
if (tailStateChangeListener != null) {
if (stateChangeListeners.contains(tailStateChangeListener)) {
stateChangeListeners.remove(tailStateChangeListener);
}
stateChangeListeners.add(tailStateChangeListener);
}
}
}
@ -287,6 +294,12 @@ public class StateMachine<T>
safeExecute(() -> stateChangeListener.stateChanged(currentState));
}
public void addStateChangeListenerToTail(StateChangeListener<T> stateChangeListener)
{
tailStateChangeListener = stateChangeListener;
addStateChangeListener(stateChangeListener);
}
@VisibleForTesting
boolean isTerminalState(T state)
{

9
presto-main/src/main/java/io/prestosql/execution/TaskStateMachine.java
View File

@ -121,6 +121,15 @@ public class TaskStateMachine
taskState.addStateChangeListener(stateChangeListener);
}
/**
* Add listener to the tail, this listener will be notified at last when state changed.
* @param stateChangeListener listener of state change.
*/
public void addStateChangeListenerToTail(StateChangeListener<TaskState> stateChangeListener)
{
taskState.addStateChangeListenerToTail(stateChangeListener);
}
@Override
public String toString()
{

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

@ -24,7 +24,7 @@ import java.util.Optional;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Objects.requireNonNull;
final class InternalBlockEncodingSerde
public class InternalBlockEncodingSerde
implements BlockEncodingSerde
{
private final FunctionAndTypeManager functionAndTypeManager;

2
presto-main/src/main/java/io/prestosql/operator/EmptyLookupSource.java
View File

@ -16,7 +16,7 @@ package io.prestosql.operator;
import io.prestosql.spi.Page;
import io.prestosql.spi.PageBuilder;
final class EmptyLookupSource
public final class EmptyLookupSource
implements LookupSource
{
@Override

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

@ -24,6 +24,7 @@ import io.airlift.log.Logger;
import io.airlift.units.DataSize;
import io.airlift.units.Duration;
import io.hetu.core.transport.execution.buffer.PageCodecMarker;
import io.hetu.core.transport.execution.buffer.PagesSerde;
import io.hetu.core.transport.execution.buffer.SerializedPage;
import io.prestosql.failuredetector.FailureDetector;
import io.prestosql.memory.context.LocalMemoryContext;
@ -31,6 +32,7 @@ import io.prestosql.operator.HttpPageBufferClient.ClientCallback;
import io.prestosql.operator.WorkProcessor.ProcessState;
import io.prestosql.snapshot.MultiInputSnapshotState;
import io.prestosql.snapshot.QuerySnapshotManager;
import io.prestosql.spi.Page;
import io.prestosql.spi.snapshot.BlockEncodingSerdeProvider;
import org.apache.commons.lang3.tuple.Pair;
@ -122,6 +124,8 @@ public class ExchangeClient
@GuardedBy("this")
private long averageBytesPerRequest;
private List<Page> pages = new ArrayList<>();
private final AtomicBoolean closed = new AtomicBoolean();
private final AtomicReference<Throwable> failure = new AtomicReference<>();
@ -343,6 +347,19 @@ public class ExchangeClient
});
}
public List<Page> getPages(String target, PagesSerde pagesSerde)
{
SerializedPage serializedPage = pollPage(target).getLeft();
if (serializedPage == null) {
if (isFinished()) {
return pages;
}
return null;
}
pages.add(pagesSerde.deserialize(serializedPage));
return null;
}
@Nullable
public Pair<SerializedPage, String> pollPage(String target)
{

2
presto-main/src/main/java/io/prestosql/operator/HashAggregationOperator.java
View File

@ -117,7 +117,7 @@ public class HashAggregationOperator
}
@VisibleForTesting
HashAggregationOperatorFactory(
public HashAggregationOperatorFactory(
int operatorId,
PlanNodeId planNodeId,
List<? extends Type> groupByTypes,

5
presto-main/src/main/java/io/prestosql/operator/Operator.java
View File

@ -66,7 +66,10 @@ public interface Operator
/**
* For Snapshot - If next output is a marker page, then return it, otherwise return null
*/
Page pollMarker();
default Page pollMarker()
{
return null;
}
/**
* After calling this method operator should revoke all reserved revocable memory.

14
presto-main/src/main/java/io/prestosql/operator/OperatorFactory.java
View File

@ -13,7 +13,11 @@
*/
package io.prestosql.operator;
import com.google.common.collect.ImmutableList;
import io.prestosql.execution.Lifespan;
import io.prestosql.spi.type.Type;
import java.util.List;
public interface OperatorFactory
{
@ -47,4 +51,14 @@ public interface OperatorFactory
}
OperatorFactory duplicate();
default boolean isExtensionOperatorFactory()
{
return false;
}
default List<Type> getSourceTypes()
{
return ImmutableList.of();
}
}

4
presto-main/src/main/java/io/prestosql/operator/PageUtils.java
View File

@ -19,13 +19,13 @@ import io.prestosql.spi.block.LazyBlock;
import java.util.function.LongConsumer;
final class PageUtils
public final class PageUtils
{
private PageUtils()
{
}
static <T> Page recordMaterializedBytes(Page page, LongConsumer sizeInBytesConsumer)
public static <T> Page recordMaterializedBytes(Page page, LongConsumer sizeInBytesConsumer)
{
// account processed bytes from lazy blocks only when they are loaded
Block<T>[] blocks = new Block[page.getChannelCount()];

14
presto-main/src/main/java/io/prestosql/operator/TaskContext.java
View File

@ -40,7 +40,9 @@ import org.joda.time.DateTime;
import javax.annotation.concurrent.GuardedBy;
import javax.annotation.concurrent.ThreadSafe;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.OptionalInt;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
@ -111,6 +113,8 @@ public class TaskContext
private final PagesSerdeFactory serdeFactory;
private final TaskSnapshotManager snapshotManager;
private final Map<String, Object> taskExtendProperties = new HashMap<>();
public static TaskContext createTaskContext(
QueryContext queryContext,
TaskStateMachine taskStateMachine,
@ -185,6 +189,16 @@ public class TaskContext
return snapshotManager;
}
public TaskStateMachine getTaskStateMachine()
{
return taskStateMachine;
}
public Map<String, Object> getTaskExtendProperties()
{
return taskExtendProperties;
}
public PipelineContext addPipelineContext(int pipelineId, boolean inputPipeline, boolean outputPipeline, boolean partitioned)
{
PipelineContext pipelineContext = new PipelineContext(

10
presto-main/src/main/java/io/prestosql/operator/WindowFunctionDefinition.java
View File

@ -64,6 +64,16 @@ public class WindowFunctionDefinition
return type;
}
public WindowFunctionSupplier getFunctionSupplier()
{
return functionSupplier;
}
public List<Integer> getArgumentChannels()
{
return argumentChannels;
}
public WindowFunction createWindowFunction()
{
return functionSupplier.createWindowFunction(argumentChannels);

34
presto-main/src/main/java/io/prestosql/operator/exchange/LocalExchange.java
View File

@ -66,11 +66,11 @@ public class LocalExchange
{
private static final Logger LOG = Logger.get(LocalExchange.class);
private final Supplier<LocalExchanger> exchangerSupplier;
protected Supplier<LocalExchanger> exchangerSupplier;
private final List<LocalExchangeSource> sources;
protected final List<LocalExchangeSource> sources;
private final LocalExchangeMemoryManager memoryManager;
protected final LocalExchangeMemoryManager memoryManager;
@GuardedBy("this")
private boolean allSourcesFinished;
@ -352,27 +352,27 @@ public class LocalExchange
@ThreadSafe
public static class LocalExchangeFactory
{
private final PartitioningHandle partitioning;
private final List<Type> types;
private final List<Integer> partitionChannels;
private final Optional<Integer> partitionHashChannel;
private final PipelineExecutionStrategy exchangeSourcePipelineExecutionStrategy;
private final DataSize maxBufferedBytes;
private final int bufferCount;
private final boolean isForMerge;
private final AggregationNode.AggregationType aggregationType;
protected final PartitioningHandle partitioning;
protected final List<Type> types;
protected final List<Integer> partitionChannels;
protected final Optional<Integer> partitionHashChannel;
protected final PipelineExecutionStrategy exchangeSourcePipelineExecutionStrategy;
protected final DataSize maxBufferedBytes;
protected final int bufferCount;
protected final boolean isForMerge;
protected final AggregationNode.AggregationType aggregationType;
@GuardedBy("this")
private boolean noMoreSinkFactories;
protected boolean noMoreSinkFactories;
// The number of total sink factories are tracked at planning time
// so that the exact number of sink factory is known by the time execution starts.
@GuardedBy("this")
private int numSinkFactories;
protected int numSinkFactories;
@GuardedBy("this")
private final Map<Lifespan, LocalExchange> localExchangeMap = new HashMap<>();
protected final Map<Lifespan, LocalExchange> localExchangeMap = new HashMap<>();
@GuardedBy("this")
private final List<LocalExchangeSinkFactoryId> closedSinkFactories = new ArrayList<>();
protected final List<LocalExchangeSinkFactoryId> closedSinkFactories = new ArrayList<>();
public LocalExchangeFactory(
PartitioningHandle partitioning,
@ -514,7 +514,7 @@ public class LocalExchange
{
private final LocalExchange exchange;
private LocalExchangeSinkFactory(LocalExchange exchange)
public LocalExchangeSinkFactory(LocalExchange exchange)
{
this.exchange = requireNonNull(exchange, "exchange is null");
}

2
presto-main/src/main/java/io/prestosql/operator/exchange/LocalExchangeSinkOperator.java
View File

@ -117,7 +117,7 @@ public class LocalExchangeSinkOperator
private final Function<Page, Page> pagePreprocessor;
private final SingleInputSnapshotState snapshotState;
LocalExchangeSinkOperator(String id, OperatorContext operatorContext, LocalExchangeSink sink, Function<Page, Page> pagePreprocessor)
public LocalExchangeSinkOperator(String id, OperatorContext operatorContext, LocalExchangeSink sink, Function<Page, Page> pagePreprocessor)
{
this.id = id;
this.operatorContext = requireNonNull(operatorContext, "operatorContext is null");

17
presto-main/src/main/java/io/prestosql/operator/exchange/LocalExchangeSource.java
View File

@ -66,6 +66,8 @@ public class LocalExchangeSource
private final Object lock = new Object();
private List<Page> pages = new ArrayList<>();
@GuardedBy("lock")
private SettableFuture<?> notEmptyFuture = NOT_EMPTY;
@ -99,7 +101,7 @@ public class LocalExchangeSource
return Collections.unmodifiableSet(inputChannels);
}
void addPage(PageReference pageReference, String origin)
public void addPage(PageReference pageReference, String origin)
{
checkNotHoldsLock();
@ -238,6 +240,19 @@ public class LocalExchangeSource
return Pair.of(page, origin.orElse(null));
}
public List<Page> getPages()
{
Page page = removePage().getLeft();
if (page == null) {
if (isFinished()) {
return pages;
}
return null;
}
pages.add(page);
return null;
}
public ListenableFuture<?> waitForReading()
{
checkNotHoldsLock();

2
presto-main/src/main/java/io/prestosql/operator/exchange/PageReference.java
View File

@ -23,7 +23,7 @@ import static com.google.common.base.Preconditions.checkArgument;
import static java.util.Objects.requireNonNull;
@ThreadSafe
class PageReference
public class PageReference
{
private final Page page;
private final Runnable onFree;

1
presto-main/src/main/java/io/prestosql/server/PluginManager.java
View File

@ -89,6 +89,7 @@ public class PluginManager
.add("io.airlift.units.")
.add("org.openjdk.jol.")
.add("io.prestosql.sql.tree.")
.add("nova.hetu.omniruntime.vector.")
.build();
private static final Logger log = Logger.get(PluginManager.class);

304
presto-main/src/main/java/io/prestosql/sql/planner/LocalExecutionPlanner.java
View File

@ -342,37 +342,188 @@ public class LocalExecutionPlanner
{
private static final Logger log = Logger.get(LocalExecutionPlanner.class);
private final Metadata metadata;
private final TypeAnalyzer typeAnalyzer;
private final Optional<ExplainAnalyzeContext> explainAnalyzeContext;
private final PageSourceProvider pageSourceProvider;
private final IndexManager indexManager;
private final NodePartitioningManager nodePartitioningManager;
private final PageSinkManager pageSinkManager;
private final ExchangeClientSupplier exchangeClientSupplier;
private final ExpressionCompiler expressionCompiler;
private final PageFunctionCompiler pageFunctionCompiler;
private final JoinFilterFunctionCompiler joinFilterFunctionCompiler;
private final DataSize maxIndexMemorySize;
private final IndexJoinLookupStats indexJoinLookupStats;
private final DataSize maxPartialAggregationMemorySize;
private final DataSize maxPagePartitioningBufferSize;
private final DataSize maxLocalExchangeBufferSize;
private final SpillerFactory spillerFactory;
private final SingleStreamSpillerFactory singleStreamSpillerFactory;
private final PartitioningSpillerFactory partitioningSpillerFactory;
private final PagesIndex.Factory pagesIndexFactory;
private final JoinCompiler joinCompiler;
private final LookupJoinOperators lookupJoinOperators;
private final OrderingCompiler orderingCompiler;
private final StateStoreProvider stateStoreProvider;
private final NodeInfo nodeInfo;
private final CubeManager cubeManager;
private final StateStoreListenerManager stateStoreListenerManager;
private final DynamicFilterCacheManager dynamicFilterCacheManager;
private final HeuristicIndexerManager heuristicIndexerManager;
private final FunctionResolution functionResolution;
private final LogicalRowExpressions logicalRowExpressions;
protected final Metadata metadata;
protected final TypeAnalyzer typeAnalyzer;
protected final Optional<ExplainAnalyzeContext> explainAnalyzeContext;
protected final PageSourceProvider pageSourceProvider;
protected final IndexManager indexManager;
protected final NodePartitioningManager nodePartitioningManager;
protected final PageSinkManager pageSinkManager;
protected final ExchangeClientSupplier exchangeClientSupplier;
protected final ExpressionCompiler expressionCompiler;
protected final PageFunctionCompiler pageFunctionCompiler;
protected final JoinFilterFunctionCompiler joinFilterFunctionCompiler;
protected final DataSize maxIndexMemorySize;
protected final IndexJoinLookupStats indexJoinLookupStats;
protected final DataSize maxPartialAggregationMemorySize;
protected final DataSize maxPagePartitioningBufferSize;
protected final DataSize maxLocalExchangeBufferSize;
protected final SpillerFactory spillerFactory;
protected final SingleStreamSpillerFactory singleStreamSpillerFactory;
protected final PartitioningSpillerFactory partitioningSpillerFactory;
protected final PagesIndex.Factory pagesIndexFactory;
protected final JoinCompiler joinCompiler;
protected final LookupJoinOperators lookupJoinOperators;
protected final OrderingCompiler orderingCompiler;
protected final StateStoreProvider stateStoreProvider;
protected final NodeInfo nodeInfo;
protected final CubeManager cubeManager;
protected final StateStoreListenerManager stateStoreListenerManager;
protected final DynamicFilterCacheManager dynamicFilterCacheManager;
protected final HeuristicIndexerManager heuristicIndexerManager;
protected final FunctionResolution functionResolution;
protected final LogicalRowExpressions logicalRowExpressions;
protected final TaskManagerConfig taskManagerConfig;
public Metadata getMetadata()
{
return metadata;
}
public TypeAnalyzer getTypeAnalyzer()
{
return typeAnalyzer;
}
public Optional<ExplainAnalyzeContext> getExplainAnalyzeContext()
{
return explainAnalyzeContext;
}
public PageSourceProvider getPageSourceProvider()
{
return pageSourceProvider;
}
public IndexManager getIndexManager()
{
return indexManager;
}
public NodePartitioningManager getNodePartitioningManager()
{
return nodePartitioningManager;
}
public PageSinkManager getPageSinkManager()
{
return pageSinkManager;
}
public ExchangeClientSupplier getExchangeClientSupplier()
{
return exchangeClientSupplier;
}
public ExpressionCompiler getExpressionCompiler()
{
return expressionCompiler;
}
public PageFunctionCompiler getPageFunctionCompiler()
{
return pageFunctionCompiler;
}
public JoinFilterFunctionCompiler getJoinFilterFunctionCompiler()
{
return joinFilterFunctionCompiler;
}
public DataSize getMaxIndexMemorySize()
{
return maxIndexMemorySize;
}
public IndexJoinLookupStats getIndexJoinLookupStats()
{
return indexJoinLookupStats;
}
public DataSize getMaxPartialAggregationMemorySize()
{
return maxPartialAggregationMemorySize;
}
public DataSize getMaxPagePartitioningBufferSize()
{
return maxPagePartitioningBufferSize;
}
public DataSize getMaxLocalExchangeBufferSize()
{
return maxLocalExchangeBufferSize;
}
public SpillerFactory getSpillerFactory()
{
return spillerFactory;
}
public SingleStreamSpillerFactory getSingleStreamSpillerFactory()
{
return singleStreamSpillerFactory;
}
public PartitioningSpillerFactory getPartitioningSpillerFactory()
{
return partitioningSpillerFactory;
}
public PagesIndex.Factory getPagesIndexFactory()
{
return pagesIndexFactory;
}
public JoinCompiler getJoinCompiler()
{
return joinCompiler;
}
public LookupJoinOperators getLookupJoinOperators()
{
return lookupJoinOperators;
}
public OrderingCompiler getOrderingCompiler()
{
return orderingCompiler;
}
public StateStoreProvider getStateStoreProvider()
{
return stateStoreProvider;
}
public NodeInfo getNodeInfo()
{
return nodeInfo;
}
public CubeManager getCubeManager()
{
return cubeManager;
}
public StateStoreListenerManager getStateStoreListenerManager()
{
return stateStoreListenerManager;
}
public DynamicFilterCacheManager getDynamicFilterCacheManager()
{
return dynamicFilterCacheManager;
}
public HeuristicIndexerManager getHeuristicIndexerManager()
{
return heuristicIndexerManager;
}
public TaskManagerConfig getTaskManagerConfig()
{
return taskManagerConfig;
}
@Inject
public LocalExecutionPlanner(
@ -415,6 +566,7 @@ public class LocalExecutionPlanner
this.pageFunctionCompiler = requireNonNull(pageFunctionCompiler, "pageFunctionCompiler is null");
this.joinFilterFunctionCompiler = requireNonNull(joinFilterFunctionCompiler, "compiler is null");
this.indexJoinLookupStats = requireNonNull(indexJoinLookupStats, "indexJoinLookupStats is null");
this.taskManagerConfig = taskManagerConfig;
this.maxIndexMemorySize = requireNonNull(taskManagerConfig, "taskManagerConfig is null").getMaxIndexMemoryUsage();
this.spillerFactory = requireNonNull(spillerFactory, "spillerFactory is null");
this.singleStreamSpillerFactory = requireNonNull(singleStreamSpillerFactory, "singleStreamSpillerFactory is null");
@ -633,7 +785,7 @@ public class LocalExecutionPlanner
return first instanceof LookupOuterOperatorFactory && isTableScanPipeline(context.outerToJoinMap.get(driverFactory));
}
private static void addLookupOuterDrivers(LocalExecutionPlanContext context)
protected static void addLookupOuterDrivers(LocalExecutionPlanContext context)
{
// For an outer join on the lookup side (RIGHT or FULL) add an additional
// driver to output the unused rows in the lookup source
@ -664,29 +816,29 @@ public class LocalExecutionPlanner
}
}
private static class LocalExecutionPlanContext
public static class LocalExecutionPlanContext
{
private final TaskContext taskContext;
private final TypeProvider types;
private final List<DriverFactory> driverFactories;
private final Optional<IndexSourceContext> indexSourceContext;
protected final TaskContext taskContext;
protected final TypeProvider types;
protected List<DriverFactory> driverFactories;
protected final Optional<IndexSourceContext> indexSourceContext;
// the collector is shared with all subContexts to allow local dynamic filtering
// with multiple table scans (e.g. co-located joins).
private final LocalDynamicFiltersCollector dynamicFiltersCollector;
protected final LocalDynamicFiltersCollector dynamicFiltersCollector;
// this is shared with all subContexts
private final AtomicInteger nextPipelineId;
protected final AtomicInteger nextPipelineId;
private int nextOperatorId;
private boolean inputDriver = true;
private OptionalInt driverInstanceCount = OptionalInt.empty();
private Map<PlanNodeId, OperatorFactory> cteOperationMap = new HashMap<>();
private Map<String, CommonTableExecutionContext> cteCtx;
protected Map<String, CommonTableExecutionContext> cteCtx;
private static Map<String, PhysicalOperation> sourceInitialized = new ConcurrentHashMap<>();
private final PlanNodeId consumerId;
private final Optional<PlanFragmentId> feederCTEId;
private final Optional<PlanNodeId> feederCTEParentId;
protected final Optional<PlanFragmentId> feederCTEId;
protected final Optional<PlanNodeId> feederCTEParentId;
// Snapshot: record pipeline that corresponds to the lookup-outer pipeline.
// This is used to help determine if a lookup-outer pipeline should be treated as a tabel-scan pipeine.
@ -700,7 +852,7 @@ public class LocalExecutionPlanner
this(taskContext, types, new ArrayList<>(), Optional.empty(), new LocalDynamicFiltersCollector(taskContext, Optional.of(metadata), dynamicFilterCacheManager), new AtomicInteger(0), feederCTEId, feederCTEParentId, cteCtx);
}
private LocalExecutionPlanContext(
protected LocalExecutionPlanContext(
TaskContext taskContext,
TypeProvider types,
List<DriverFactory> driverFactories,
@ -745,11 +897,16 @@ public class LocalExecutionPlanner
return driverFactory;
}
private List<DriverFactory> getDriverFactories()
public List<DriverFactory> getDriverFactories()
{
return ImmutableList.copyOf(driverFactories);
}
public void setDriverFactories(List<DriverFactory> driverFactories)
{
this.driverFactories = driverFactories;
}
public Session getSession()
{
return taskContext.getSession();
@ -780,22 +937,27 @@ public class LocalExecutionPlanner
return indexSourceContext;
}
private int getNextPipelineId()
private AtomicInteger getPipelineId()
{
return nextPipelineId;
}
public int getNextPipelineId()
{
return nextPipelineId.getAndIncrement();
}
private int getNextOperatorId()
public int getNextOperatorId()
{
return nextOperatorId++;
}
private boolean isInputDriver()
public boolean isInputDriver()
{
return inputDriver;
}
private void setInputDriver(boolean inputDriver)
public void setInputDriver(boolean inputDriver)
{
this.inputDriver = inputDriver;
}
@ -866,7 +1028,7 @@ public class LocalExecutionPlanner
}
}
private static class IndexSourceContext
public static class IndexSourceContext
{
private final SetMultimap<Symbol, Integer> indexLookupToProbeInput;
@ -918,13 +1080,13 @@ public class LocalExecutionPlanner
}
}
private class Visitor
public class Visitor
extends InternalPlanVisitor<PhysicalOperation, LocalExecutionPlanContext>
{
private final Session session;
private final StageExecutionDescriptor stageExecutionDescriptor;
protected final Session session;
protected final StageExecutionDescriptor stageExecutionDescriptor;
private Visitor(Session session, StageExecutionDescriptor stageExecutionDescriptor)
public Visitor(Session session, StageExecutionDescriptor stageExecutionDescriptor)
{
this.session = session;
this.stageExecutionDescriptor = stageExecutionDescriptor;
@ -1618,7 +1780,7 @@ public class LocalExecutionPlanner
}
}
private Supplier<List<Map<ColumnHandle, DynamicFilter>>> getDynamicFilterSupplier(Optional<List<List<DynamicFilters.Descriptor>>> dynamicFilters, PlanNode sourceNode, LocalExecutionPlanContext context)
protected Supplier<List<Map<ColumnHandle, DynamicFilter>>> getDynamicFilterSupplier(Optional<List<List<DynamicFilters.Descriptor>>> dynamicFilters, PlanNode sourceNode, LocalExecutionPlanContext context)
{
if (dynamicFilters.isPresent() && !dynamicFilters.get().isEmpty()) {
log.debug("[TableScan] Dynamic filters: %s", dynamicFilters);
@ -1643,7 +1805,7 @@ public class LocalExecutionPlanner
return null;
}
private RowExpression bindChannels(RowExpression inputExpression, Map<Symbol, Integer> sourceLayout, TypeProvider types)
public RowExpression bindChannels(RowExpression inputExpression, Map<Symbol, Integer> sourceLayout, TypeProvider types)
{
RowExpression expression = inputExpression;
Type type = expression.getType();
@ -1864,12 +2026,12 @@ public class LocalExecutionPlanner
stageExecutionDescriptor.isScanGroupedExecution(node.getId()) ? GROUPED_EXECUTION : UNGROUPED_EXECUTION);
}
private ImmutableMap<Symbol, Integer> makeLayout(PlanNode node)
protected ImmutableMap<Symbol, Integer> makeLayout(PlanNode node)
{
return makeLayoutFromOutputSymbols(node.getOutputSymbols());
}
private ImmutableMap<Symbol, Integer> makeLayoutFromOutputSymbols(List<Symbol> outputSymbols)
protected ImmutableMap<Symbol, Integer> makeLayoutFromOutputSymbols(List<Symbol> outputSymbols)
{
ImmutableMap.Builder<Symbol, Integer> outputMappings = ImmutableMap.builder();
int channel = 0;
@ -2231,7 +2393,7 @@ public class LocalExecutionPlanner
return symbols.stream().map(SymbolUtils::toSymbolReference).collect(toImmutableSet());
}
private PhysicalOperation createNestedLoopJoin(JoinNode node, LocalExecutionPlanContext context)
protected PhysicalOperation createNestedLoopJoin(JoinNode node, LocalExecutionPlanContext context)
{
PhysicalOperation probeSource = node.getLeft().accept(this, context);
@ -2459,7 +2621,7 @@ public class LocalExecutionPlanner
return new PhysicalOperation(operator, outputMappings.build(), context, probeSource);
}
private Optional<LocalDynamicFilter> createDynamicFilter(JoinNode node, LocalExecutionPlanContext context, int partitionCount)
protected Optional<LocalDynamicFilter> createDynamicFilter(JoinNode node, LocalExecutionPlanContext context, int partitionCount)
{
if (!isEnableDynamicFiltering(context.getSession())) {
return Optional.empty();
@ -2625,7 +2787,7 @@ public class LocalExecutionPlanner
return lookupSourceFactoryManager;
}
private JoinFilterFunctionFactory compileJoinFilterFunction(
protected JoinFilterFunctionFactory compileJoinFilterFunction(
RowExpression filterExpression,
Map<Symbol, Integer> probeLayout,
Map<Symbol, Integer> buildLayout,
@ -2636,7 +2798,7 @@ public class LocalExecutionPlanner
return joinFilterFunctionCompiler.compileJoinFilterFunction(bindChannels(filterExpression, joinSourcesLayout, types), buildLayout.size());
}
private int sortExpressionAsSortChannel(
public int sortExpressionAsSortChannel(
RowExpression sortExpression,
Map<Symbol, Integer> probeLayout,
Map<Symbol, Integer> buildLayout,
@ -2682,7 +2844,7 @@ public class LocalExecutionPlanner
}
}
private Map<Symbol, Integer> createJoinSourcesLayout(Map<Symbol, Integer> lookupSourceLayout, Map<Symbol, Integer> probeSourceLayout)
protected Map<Symbol, Integer> createJoinSourcesLayout(Map<Symbol, Integer> lookupSourceLayout, Map<Symbol, Integer> probeSourceLayout)
{
ImmutableMap.Builder<Symbol, Integer> joinSourcesLayout = ImmutableMap.builder();
joinSourcesLayout.putAll(lookupSourceLayout);
@ -3203,19 +3365,19 @@ public class LocalExecutionPlanner
throw new UnsupportedOperationException("not yet implemented");
}
private List<Type> getSourceOperatorTypes(PlanNode node, TypeProvider types)
protected List<Type> getSourceOperatorTypes(PlanNode node, TypeProvider types)
{
return getSymbolTypes(node.getOutputSymbols(), types);
}
private List<Type> getSymbolTypes(List<Symbol> symbols, TypeProvider types)
protected List<Type> getSymbolTypes(List<Symbol> symbols, TypeProvider types)
{
return symbols.stream()
.map(types::get)
.collect(toImmutableList());
}
private AccumulatorFactory buildAccumulatorFactory(
protected AccumulatorFactory buildAccumulatorFactory(
PhysicalOperation source,
Aggregation aggregation)
{
@ -3558,7 +3720,7 @@ public class LocalExecutionPlanner
};
}
private static Function<Page, Page> enforceLayoutProcessor(List<Symbol> expectedLayout, Map<Symbol, Integer> inputLayout)
protected static Function<Page, Page> enforceLayoutProcessor(List<Symbol> expectedLayout, Map<Symbol, Integer> inputLayout)
{
int[] channels = expectedLayout.stream()
.peek(symbol -> checkArgument(inputLayout.containsKey(symbol), "channel not found for symbol: %s", symbol))
@ -3573,7 +3735,7 @@ public class LocalExecutionPlanner
return new PageChannelSelector(channels);
}
private static List<Integer> getChannelsForSymbols(List<Symbol> symbols, Map<Symbol, Integer> layout)
protected static List<Integer> getChannelsForSymbols(List<Symbol> symbols, Map<Symbol, Integer> layout)
{
ImmutableList.Builder<Integer> builder = ImmutableList.builder();
for (Symbol symbol : symbols) {
@ -3582,7 +3744,7 @@ public class LocalExecutionPlanner
return builder.build();
}
private static Function<Symbol, Integer> channelGetter(PhysicalOperation source)
protected static Function<Symbol, Integer> channelGetter(PhysicalOperation source)
{
return input -> {
checkArgument(source.getLayout().containsKey(input));
@ -3593,7 +3755,7 @@ public class LocalExecutionPlanner
/**
* Encapsulates an physical operator plus the mapping of logical symbols to channel/field
*/
private static class PhysicalOperation
public static class PhysicalOperation
{
private final List<OperatorFactory> operatorFactories;
private final Map<Symbol, Integer> layout;
@ -3664,7 +3826,7 @@ public class LocalExecutionPlanner
return layout;
}
private List<OperatorFactory> getOperatorFactories()
public List<OperatorFactory> getOperatorFactories()
{
return operatorFactories;
}
@ -3675,7 +3837,7 @@ public class LocalExecutionPlanner
}
}
private static class DriverFactoryParameters
protected static class DriverFactoryParameters
{
private final LocalExecutionPlanContext subContext;
private final PhysicalOperation source;

43
presto-main/src/main/java/io/prestosql/utils/HetuConfig.java
View File

@ -62,10 +62,53 @@ public class HetuConfig
private Duration splitCacheStateUpdateInterval = new Duration(2, TimeUnit.SECONDS);
private boolean isTraceStackVisible;
private String extensionExecutionPlannerJarPath;
private String extensionExecutionPlannerClassPath;
private boolean extensionExecutionPlannerEnabled;
public HetuConfig()
{
}
public boolean getExtensionExecutionPlannerEnabled()
{
return extensionExecutionPlannerEnabled;
}
@Config(HetuConstant.EXTENSION_EXECUTION_PLANNER_ENABLED)
@ConfigDescription("extension execution planner enable from config")
public HetuConfig setExtensionExecutionPlannerEnabled(boolean extensionExecutionPlannerEnabled)
{
this.extensionExecutionPlannerEnabled = extensionExecutionPlannerEnabled;
return this;
}
public String getExtensionExecutionPlannerJarPath()
{
return extensionExecutionPlannerJarPath;
}
@Config(HetuConstant.EXTENSION_EXECUTION_PLANNER_JAR_PATH)
@ConfigDescription("extension execution planner jar path from config")
public HetuConfig setExtensionExecutionPlannerJarPath(String extensionExecutionPlannerJarPath)
{
this.extensionExecutionPlannerJarPath = extensionExecutionPlannerJarPath;
return this;
}
public String getExtensionExecutionPlannerClassPath()
{
return extensionExecutionPlannerClassPath;
}
@Config(HetuConstant.EXTENSION_EXECUTION_PLANNER_CLASS_PATH)
@ConfigDescription("extension execution planner class path from config")
public HetuConfig setExtensionExecutionPlannerClassPath(String extensionExecutionPlannerClassPath)
{
this.extensionExecutionPlannerClassPath = extensionExecutionPlannerClassPath;
return this;
}
@NotNull
public boolean isFilterEnabled()
{

2
presto-main/src/test/java/io/prestosql/operator/window/AbstractTestWindowFunction.java
View File

@ -30,7 +30,7 @@ public abstract class AbstractTestWindowFunction
protected LocalQueryRunner queryRunner;
@BeforeClass
public final void initTestWindowFunction()
public void initTestWindowFunction()
{
queryRunner = new LocalQueryRunner(TEST_SESSION);
}

13
presto-main/src/test/java/io/prestosql/utils/TestHetuConfig.java
View File

@ -54,7 +54,10 @@ public class TestHetuConfig
.setSplitCacheMapEnabled(false)
.setSplitCacheStateUpdateInterval(new Duration(2, TimeUnit.SECONDS))
.setTraceStackVisible(false)
.setIndexToPreload(""));
.setIndexToPreload("")
.setExtensionExecutionPlannerEnabled(false)
.setExtensionExecutionPlannerJarPath(null)
.setExtensionExecutionPlannerClassPath(null));
}
@Test
@ -85,6 +88,9 @@ public class TestHetuConfig
.put("hetu.split-cache-map.state-update-interval", "5s")
.put("stack-trace-visible", "true")
.put("hetu.heuristicindex.filter.cache.preload-indices", "idx1,idx2")
.put("extension_execution_planner_enabled", "true")
.put("extension_execution_planner_jar_path", "")
.put("extension_execution_planner_class_path", "")
.build();
HetuConfig expected = new HetuConfig()
@ -111,7 +117,10 @@ public class TestHetuConfig
.setSplitCacheMapEnabled(true)
.setSplitCacheStateUpdateInterval(new Duration(5, TimeUnit.SECONDS))
.setTraceStackVisible(true)
.setIndexToPreload("idx1,idx2");
.setIndexToPreload("idx1,idx2")
.setExtensionExecutionPlannerEnabled(true)
.setExtensionExecutionPlannerJarPath("")
.setExtensionExecutionPlannerClassPath("");
ConfigAssertions.assertFullMapping(properties, expected);
}

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

@ -58,4 +58,9 @@ public class HetuConstant
// error message
public static final String HINDEX_CONFIG_ERROR_MSG = "Heuristic Index is not enabled in config.properties or is configured incorrectly.";
// extension support message
public static final String EXTENSION_EXECUTION_PLANNER_ENABLED = "extension_execution_planner_enabled";
public static final String EXTENSION_EXECUTION_PLANNER_JAR_PATH = "extension_execution_planner_jar_path";
public static final String EXTENSION_EXECUTION_PLANNER_CLASS_PATH = "extension_execution_planner_class_path";
}

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

@ -370,4 +370,9 @@ public class Page
{
pageMetadata.setProperty(key, value);
}
public Block[] getBlocks()
{
return blocks;
}
}

7
presto-spi/src/main/java/io/prestosql/spi/block/AbstractSingleRowBlock.java
View File

@ -70,6 +70,13 @@ public abstract class AbstractSingleRowBlock<T>
return getRawFieldBlock(position).getLong(rowIndex, offset);
}
@Override
public double getDouble(int position, int offset)
{
checkFieldIndex(position);
return getRawFieldBlock(position).getDouble(rowIndex, offset);
}
@Override
public Slice getSlice(int position, int offset, int length)
{

36
presto-spi/src/main/java/io/prestosql/spi/block/Block.java
View File

@ -69,6 +69,14 @@ public interface Block<T>
throw new UnsupportedOperationException(getClass().getName());
}
/**
* Gets a little endian double at {@code offset} in the value at {@code position}.
*/
default double getDouble(int position, int offset)
{
throw new UnsupportedOperationException(getClass().getName());
}
/**
* Gets a slice at {@code offset} in the value at {@code position}.
*/
@ -340,4 +348,32 @@ public interface Block<T>
System.arraycopy(positions, positionCount, matchedPositions, positionCount, positionCount);
return positionCount;
}
default Object getValues()
{
throw new UnsupportedOperationException();
}
default int getBlockOffset()
{
throw new UnsupportedOperationException();
}
default boolean[] getValueNulls()
{
throw new UnsupportedOperationException();
}
default void close()
{
}
default boolean isExtensionBlock()
{
return false;
}
default void setClosable(boolean isClosable)
{
}
}

8
presto-spi/src/main/java/io/prestosql/spi/block/BlockBuilder.java
View File

@ -51,6 +51,14 @@ public interface BlockBuilder<T>
throw new UnsupportedOperationException(getClass().getName());
}
/**
* Write a double to the current entry;
*/
default BlockBuilder writeDouble(double value)
{
throw new UnsupportedOperationException(getClass().getName());
}
/**
* Write a byte sequences to the current entry;
*/

34
presto-spi/src/main/java/io/prestosql/spi/block/BlockUtil.java
View File

@ -22,7 +22,7 @@ import static java.lang.Math.ceil;
import static java.lang.String.format;
import static java.util.Objects.requireNonNull;
final class BlockUtil
public final class BlockUtil
{
private static final double BLOCK_RESET_SKEW = 1.25;
@ -34,7 +34,7 @@ final class BlockUtil
{
}
static void checkArrayRange(int[] array, int offset, int length)
public static void checkArrayRange(int[] array, int offset, int length)
{
requireNonNull(array, "array is null");
if (offset < 0 || length < 0 || offset + length > array.length) {
@ -42,28 +42,28 @@ final class BlockUtil
}
}
static void checkValidRegion(int positionCount, int positionOffset, int length)
public static void checkValidRegion(int positionCount, int positionOffset, int length)
{
if (positionOffset < 0 || length < 0 || positionOffset + length > positionCount) {
throw new IndexOutOfBoundsException(format("Invalid position %s and length %s in block with %s positions", positionOffset, length, positionCount));
}
}
static void checkValidPositions(boolean[] positions, int positionCount)
public static void checkValidPositions(boolean[] positions, int positionCount)
{
if (positions.length != positionCount) {
throw new IllegalArgumentException(format("Invalid positions array size %d, actual position count is %d", positions.length, positionCount));
}
}
static void checkValidPosition(int position, int positionCount)
public static void checkValidPosition(int position, int positionCount)
{
if (position < 0 || position >= positionCount) {
throw new IllegalArgumentException(format("Invalid position %s in block with %s positions", position, positionCount));
}
}
static int calculateNewArraySize(int currentSize)
public static int calculateNewArraySize(int currentSize)
{
// grow array by 50%
long newSize = (long) currentSize + (currentSize >> 1);
@ -81,7 +81,7 @@ final class BlockUtil
return (int) newSize;
}
static int calculateBlockResetSize(int currentSize)
public static int calculateBlockResetSize(int currentSize)
{
long newSize = (long) ceil(currentSize * BLOCK_RESET_SKEW);
@ -95,7 +95,7 @@ final class BlockUtil
return (int) newSize;
}
static int calculateBlockResetBytes(int currentBytes)
public static int calculateBlockResetBytes(int currentBytes)
{
long newBytes = (long) ceil(currentBytes * BLOCK_RESET_SKEW);
if (newBytes > MAX_ARRAY_SIZE) {
@ -110,7 +110,7 @@ final class BlockUtil
* with the first value set to 0.
* If the range matches the entire <code>offsets</code> array, the input array will be returned.
*/
static int[] compactOffsets(int[] offsets, int index, int length)
public static int[] compactOffsets(int[] offsets, int index, int length)
{
if (index == 0 && offsets.length == length + 1) {
return offsets;
@ -128,7 +128,7 @@ final class BlockUtil
* If the range matches the entire slice, the input slice will be returned.
* Otherwise, a copy will be returned.
*/
static Slice compactSlice(Slice slice, int index, int length)
public static Slice compactSlice(Slice slice, int index, int length)
{
if (slice.isCompact() && index == 0 && length == slice.length()) {
return slice;
@ -141,7 +141,7 @@ final class BlockUtil
* If the range matches the entire array, the input array will be returned.
* Otherwise, a copy will be returned.
*/
static boolean[] compactArray(boolean[] array, int index, int length)
public static boolean[] compactArray(boolean[] array, int index, int length)
{
if (index == 0 && length == array.length) {
return array;
@ -149,7 +149,7 @@ final class BlockUtil
return Arrays.copyOfRange(array, index, index + length);
}
static byte[] compactArray(byte[] array, int index, int length)
public static byte[] compactArray(byte[] array, int index, int length)
{
if (index == 0 && length == array.length) {
return array;
@ -157,7 +157,7 @@ final class BlockUtil
return Arrays.copyOfRange(array, index, index + length);
}
static short[] compactArray(short[] array, int index, int length)
public static short[] compactArray(short[] array, int index, int length)
{
if (index == 0 && length == array.length) {
return array;
@ -165,7 +165,7 @@ final class BlockUtil
return Arrays.copyOfRange(array, index, index + length);
}
static int[] compactArray(int[] array, int index, int length)
public static int[] compactArray(int[] array, int index, int length)
{
if (index == 0 && length == array.length) {
return array;
@ -173,7 +173,7 @@ final class BlockUtil
return Arrays.copyOfRange(array, index, index + length);
}
static long[] compactArray(long[] array, int index, int length)
public static long[] compactArray(long[] array, int index, int length)
{
if (index == 0 && length == array.length) {
return array;
@ -181,7 +181,7 @@ final class BlockUtil
return Arrays.copyOfRange(array, index, index + length);
}
static int countUsedPositions(boolean[] positions)
public static int countUsedPositions(boolean[] positions)
{
int used = 0;
for (boolean position : positions) {
@ -196,7 +196,7 @@ final class BlockUtil
* Returns <tt>true</tt> if the two specified arrays contain the same object in every position.
* Unlike the {@link Arrays#equals(Object[], Object[])} method, this method compares using reference equals.
*/
static boolean arraySame(Object[] array1, Object[] array2)
public static boolean arraySame(Object[] array1, Object[] array2)
{
if (array1 == null || array2 == null || array1.length != array2.length) {
throw new IllegalArgumentException("array1 and array2 cannot be null and should have same length");

18
presto-spi/src/main/java/io/prestosql/spi/block/ByteArrayBlock.java
View File

@ -120,6 +120,24 @@ public class ByteArrayBlock
return positionCount;
}
@Override
public byte[] getValues()
{
return values;
}
@Override
public int getBlockOffset()
{
return arrayOffset;
}
@Override
public boolean[] getValueNulls()
{
return valueIsNull;
}
@Override
public byte getByte(int position, int offset)
{

14
presto-spi/src/main/java/io/prestosql/spi/block/DictionaryBlock.java
View File

@ -397,11 +397,23 @@ public class DictionaryBlock<T>
return dictionary;
}
Slice getIds()
public Slice getIds()
{
return Slices.wrappedIntArray(ids, idsOffset, positionCount);
}
public int[] getIdsArray()
{
if (idsOffset == 0) {
return ids;
}
else {
int[] res = new int[positionCount];
System.arraycopy(ids, idsOffset, res, 0, positionCount);
return res;
}
}
public int getId(int position)
{
checkValidPosition(position, positionCount);

2
presto-spi/src/main/java/io/prestosql/spi/block/EncoderUtil.java
View File

@ -18,7 +18,7 @@ import io.airlift.slice.SliceOutput;
import java.util.Optional;
final class EncoderUtil
public final class EncoderUtil
{
private EncoderUtil()
{

18
presto-spi/src/main/java/io/prestosql/spi/block/Int128ArrayBlock.java
View File

@ -123,6 +123,24 @@ public class Int128ArrayBlock
return positionCount;
}
@Override
public long[] getValues()
{
return values;
}
@Override
public int getBlockOffset()
{
return positionOffset;
}
@Override
public boolean[] getValueNulls()
{
return valueIsNull;
}
@Override
public long getLong(int position, int offset)
{

18
presto-spi/src/main/java/io/prestosql/spi/block/IntArrayBlock.java
View File

@ -120,6 +120,24 @@ public class IntArrayBlock
return positionCount;
}
@Override
public int[] getValues()
{
return values;
}
@Override
public int getBlockOffset()
{
return arrayOffset;
}
@Override
public boolean[] getValueNulls()
{
return valueIsNull;
}
@Override
public int getInt(int position, int offset)
{

2
presto-spi/src/main/java/io/prestosql/spi/block/IntArrayList.java
View File

@ -21,7 +21,7 @@ import static java.lang.String.format;
/**
* A simplified version of fastutils IntArrayList for the purpose of positions copying.
*/
class IntArrayList
public class IntArrayList
{
private static final int DEFAULT_INITIAL_CAPACITY = 16;
private int[] array;

5
presto-spi/src/main/java/io/prestosql/spi/block/LazyBlock.java
View File

@ -98,6 +98,11 @@ public class LazyBlock<T>
return block.getObject(position, clazz);
}
public Block<T> getBlock()
{
return block;
}
@Override
public boolean bytesEqual(int position, int offset, Slice otherSlice, int otherOffset, int length)
{

18
presto-spi/src/main/java/io/prestosql/spi/block/LongArrayBlock.java
View File

@ -124,6 +124,24 @@ public class LongArrayBlock
return positionCount;
}
@Override
public long[] getValues()
{
return values;
}
@Override
public int getBlockOffset()
{
return arrayOffset;
}
@Override
public boolean[] getValueNulls()
{
return valueIsNull;
}
@Override
public long getLong(int position, int offset)
{

8
presto-spi/src/main/java/io/prestosql/spi/block/SingleRowBlockWriter.java
View File

@ -132,6 +132,14 @@ public class SingleRowBlockWriter<T>
return this;
}
@Override
public BlockBuilder writeDouble(double value)
{
checkFieldIndexToWrite();
fieldBlockBuilders[currentFieldIndexToWrite].writeDouble(value);
return this;
}
@Override
public BlockBuilder writeBytes(Slice source, int sourceIndex, int length)
{

19
presto-spi/src/main/java/io/prestosql/spi/block/VariableWidthBlock.java
View File

@ -99,6 +99,11 @@ public class VariableWidthBlock
this.isInitialized = true;
}
public int[] getOffsets()
{
return offsets;
}
@Override
protected final int getPositionOffset(int position)
{
@ -130,6 +135,18 @@ public class VariableWidthBlock
return positionCount;
}
@Override
public int getBlockOffset()
{
return arrayOffset;
}
@Override
public boolean[] getValueNulls()
{
return valueIsNull;
}
@Override
public long getSizeInBytes()
{
@ -203,7 +220,7 @@ public class VariableWidthBlock
}
@Override
protected Slice getRawSlice(int position)
public Slice getRawSlice(int position)
{
return slice;
}