catalogFunctions = tableFunctions.get(connectorId);
+ if (catalogFunctions != null) {
+ String lowercasedSchemaName = name.getSchemaFunctionName().getSchemaName().toLowerCase(ENGLISH);
+ String lowercasedFunctionName = name.getSchemaFunctionName().getFunctionName().toLowerCase(ENGLISH);
+ TableFunctionMetadata function = catalogFunctions.get(new SchemaFunctionName(lowercasedSchemaName, lowercasedFunctionName));
+ if (function != null) {
+ return function;
+ }
+ }
+
+ throw new PrestoException(GENERIC_USER_ERROR, format("Table functions for catalog %s could not be resolved.", connectorId.getCatalogName()));
}
private static void validateTableFunction(ConnectorTableFunction tableFunction)
@@ -154,8 +186,8 @@ private static void validateTableFunction(ConnectorTableFunction tableFunction)
// Such a table argument is implicitly 'prune when empty'. The TableArgumentSpecification.Builder enforces the 'prune when empty' property
// for a table argument with row semantics.
- if (tableFunction.getReturnTypeSpecification() instanceof DescribedTable) {
- DescribedTable describedTable = (DescribedTable) tableFunction.getReturnTypeSpecification();
+ if (tableFunction.getReturnTypeSpecification() instanceof DescribedTableReturnTypeSpecification) {
+ DescribedTableReturnTypeSpecification describedTable = (DescribedTableReturnTypeSpecification) tableFunction.getReturnTypeSpecification();
checkArgument(describedTable.getDescriptor().isTyped(), "field types missing in returned type specification");
}
}
diff --git a/presto-main-base/src/main/java/com/facebook/presto/operator/EmptyTableFunctionPartition.java b/presto-main-base/src/main/java/com/facebook/presto/operator/EmptyTableFunctionPartition.java
new file mode 100644
index 0000000000000..bda83ae6319d4
--- /dev/null
+++ b/presto-main-base/src/main/java/com/facebook/presto/operator/EmptyTableFunctionPartition.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.facebook.presto.operator;
+
+import com.facebook.presto.common.Page;
+import com.facebook.presto.common.block.Block;
+import com.facebook.presto.common.block.RunLengthEncodedBlock;
+import com.facebook.presto.common.type.Type;
+import com.facebook.presto.spi.PrestoException;
+import com.facebook.presto.spi.function.table.TableFunctionDataProcessor;
+import com.facebook.presto.spi.function.table.TableFunctionProcessorState;
+
+import java.util.List;
+
+import static com.facebook.airlift.concurrent.MoreFutures.toListenableFuture;
+import static com.facebook.presto.spi.StandardErrorCode.FUNCTION_IMPLEMENTATION_ERROR;
+import static com.facebook.presto.spi.function.table.TableFunctionProcessorState.Finished.FINISHED;
+import static java.lang.String.format;
+import static java.util.Objects.requireNonNull;
+
+/**
+ * This is a class representing empty input to a table function. An EmptyTableFunctionPartition is created
+ * when the table function has KEEP WHEN EMPTY property, which means that the function should be executed
+ * even if the input is empty, and all the table arguments are empty relations.
+ *
+ * An EmptyTableFunctionPartition is created and processed once per node. To avoid duplicated execution,
+ * a table function having KEEP WHEN EMPTY property must have single distribution.
+ */
+public class EmptyTableFunctionPartition
+ implements TableFunctionPartition
+{
+ private final TableFunctionDataProcessor tableFunction;
+ private final int properChannelsCount;
+ private final int passThroughSourcesCount;
+ private final Type[] passThroughTypes;
+
+ public EmptyTableFunctionPartition(TableFunctionDataProcessor tableFunction, int properChannelsCount, int passThroughSourcesCount, List passThroughTypes)
+ {
+ this.tableFunction = requireNonNull(tableFunction, "tableFunction is null");
+ this.properChannelsCount = properChannelsCount;
+ this.passThroughSourcesCount = passThroughSourcesCount;
+ this.passThroughTypes = passThroughTypes.toArray(new Type[] {});
+ }
+
+ @Override
+ public WorkProcessor toOutputPages()
+ {
+ return WorkProcessor.create(() -> {
+ TableFunctionProcessorState state = tableFunction.process(null);
+ if (state == FINISHED) {
+ return WorkProcessor.ProcessState.finished();
+ }
+ if (state instanceof TableFunctionProcessorState.Blocked) {
+ return WorkProcessor.ProcessState.blocked(toListenableFuture(((TableFunctionProcessorState.Blocked) state).getFuture()));
+ }
+ TableFunctionProcessorState.Processed processed = (TableFunctionProcessorState.Processed) state;
+ if (processed.getResult() != null) {
+ return WorkProcessor.ProcessState.ofResult(appendNullsForPassThroughColumns(processed.getResult()));
+ }
+ throw new PrestoException(FUNCTION_IMPLEMENTATION_ERROR, "When function got no input, it should either produce output or return Blocked state");
+ });
+ }
+
+ private Page appendNullsForPassThroughColumns(Page page)
+ {
+ if (page.getChannelCount() != properChannelsCount + passThroughSourcesCount) {
+ throw new PrestoException(
+ FUNCTION_IMPLEMENTATION_ERROR,
+ format(
+ "Table function returned a page containing %s channels. Expected channel number: %s (%s proper columns, %s pass-through index columns)",
+ page.getChannelCount(),
+ properChannelsCount + passThroughSourcesCount,
+ properChannelsCount,
+ passThroughSourcesCount));
+ }
+
+ Block[] resultBlocks = new Block[properChannelsCount + passThroughTypes.length];
+
+ // proper outputs first
+ for (int channel = 0; channel < properChannelsCount; channel++) {
+ resultBlocks[channel] = page.getBlock(channel);
+ }
+
+ // pass-through columns next
+ // because no input was processed, all pass-through indexes in the result page must be null (there are no input rows they could refer to).
+ // for performance reasons this is not checked. All pass-through columns are filled with nulls.
+ int channel = properChannelsCount;
+ for (Type type : passThroughTypes) {
+ resultBlocks[channel] = RunLengthEncodedBlock.create(type, null, page.getPositionCount());
+ channel++;
+ }
+
+ // pass the position count so that the Page can be successfully created in the case when there are no output channels (resultBlocks is empty)
+ return new Page(page.getPositionCount(), resultBlocks);
+ }
+}
diff --git a/presto-main-base/src/main/java/com/facebook/presto/operator/LeafTableFunctionOperator.java b/presto-main-base/src/main/java/com/facebook/presto/operator/LeafTableFunctionOperator.java
new file mode 100644
index 0000000000000..3eb272cef09ec
--- /dev/null
+++ b/presto-main-base/src/main/java/com/facebook/presto/operator/LeafTableFunctionOperator.java
@@ -0,0 +1,205 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.facebook.presto.operator;
+
+import com.facebook.presto.common.Page;
+import com.facebook.presto.execution.ScheduledSplit;
+import com.facebook.presto.metadata.Split;
+import com.facebook.presto.spi.ConnectorSplit;
+import com.facebook.presto.spi.UpdatablePageSource;
+import com.facebook.presto.spi.function.table.ConnectorTableFunctionHandle;
+import com.facebook.presto.spi.function.table.TableFunctionProcessorProvider;
+import com.facebook.presto.spi.function.table.TableFunctionProcessorState;
+import com.facebook.presto.spi.function.table.TableFunctionProcessorState.Blocked;
+import com.facebook.presto.spi.function.table.TableFunctionProcessorState.Processed;
+import com.facebook.presto.spi.function.table.TableFunctionSplitProcessor;
+import com.facebook.presto.spi.plan.PlanNodeId;
+import com.google.common.util.concurrent.ListenableFuture;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Supplier;
+
+import static com.facebook.airlift.concurrent.MoreFutures.toListenableFuture;
+import static com.facebook.presto.spi.function.table.TableFunctionProcessorState.Finished.FINISHED;
+import static com.google.common.base.Preconditions.checkState;
+import static java.util.Objects.requireNonNull;
+
+public class LeafTableFunctionOperator
+ implements SourceOperator
+{
+ public static class LeafTableFunctionOperatorFactory
+ implements SourceOperatorFactory
+ {
+ private final int operatorId;
+ private final PlanNodeId sourceId;
+ private final TableFunctionProcessorProvider tableFunctionProvider;
+ private final ConnectorTableFunctionHandle functionHandle;
+ private boolean closed;
+
+ public LeafTableFunctionOperatorFactory(int operatorId, PlanNodeId sourceId, TableFunctionProcessorProvider tableFunctionProvider, ConnectorTableFunctionHandle functionHandle)
+ {
+ this.operatorId = operatorId;
+ this.sourceId = requireNonNull(sourceId, "sourceId is null");
+ this.tableFunctionProvider = requireNonNull(tableFunctionProvider, "tableFunctionProvider is null");
+ this.functionHandle = requireNonNull(functionHandle, "functionHandle is null");
+ }
+
+ @Override
+ public PlanNodeId getSourceId()
+ {
+ return sourceId;
+ }
+
+ @Override
+ public SourceOperator createOperator(DriverContext driverContext)
+ {
+ checkState(!closed, "Factory is already closed");
+ OperatorContext operatorContext = driverContext.addOperatorContext(operatorId, sourceId, LeafTableFunctionOperator.class.getSimpleName());
+ return new LeafTableFunctionOperator(operatorContext, sourceId, tableFunctionProvider, functionHandle);
+ }
+
+ @Override
+ public void noMoreOperators()
+ {
+ closed = true;
+ }
+ }
+
+ private final OperatorContext operatorContext;
+ private final PlanNodeId sourceId;
+ private final TableFunctionProcessorProvider tableFunctionProvider;
+ private final ConnectorTableFunctionHandle functionHandle;
+
+ private ConnectorSplit currentSplit;
+ private final List pendingSplits = new ArrayList<>();
+ private boolean noMoreSplits;
+
+ private TableFunctionSplitProcessor processor;
+ private boolean processorUsedData;
+ private boolean processorFinishedSplit = true;
+ private ListenableFuture processorBlocked = NOT_BLOCKED;
+
+ public LeafTableFunctionOperator(OperatorContext operatorContext, PlanNodeId sourceId, TableFunctionProcessorProvider tableFunctionProvider, ConnectorTableFunctionHandle functionHandle)
+ {
+ this.operatorContext = requireNonNull(operatorContext, "operatorContext is null");
+ this.sourceId = requireNonNull(sourceId, "sourceId is null");
+ this.tableFunctionProvider = requireNonNull(tableFunctionProvider, "tableFunctionProvider is null");
+ this.functionHandle = requireNonNull(functionHandle, "functionHandle is null");
+ }
+
+ private void resetProcessor()
+ {
+ this.processor = tableFunctionProvider.getSplitProcessor(functionHandle);
+ this.processorUsedData = false;
+ this.processorFinishedSplit = false;
+ this.processorBlocked = NOT_BLOCKED;
+ }
+
+ @Override
+ public OperatorContext getOperatorContext()
+ {
+ return operatorContext;
+ }
+
+ @Override
+ public PlanNodeId getSourceId()
+ {
+ return sourceId;
+ }
+
+ @Override
+ public boolean needsInput()
+ {
+ return false;
+ }
+
+ @Override
+ public void addInput(Page page)
+ {
+ throw new UnsupportedOperationException(getClass().getName() + " does not take input");
+ }
+
+ @Override
+ public Supplier> addSplit(ScheduledSplit split)
+ {
+ Split curSplit = requireNonNull(split, "split is null").getSplit();
+ checkState(!noMoreSplits, "no more splits expected");
+ ConnectorSplit curConnectorSplit = curSplit.getConnectorSplit();
+ pendingSplits.add(curConnectorSplit);
+ return Optional::empty;
+ }
+
+ @Override
+ public void noMoreSplits()
+ {
+ noMoreSplits = true;
+ }
+
+ @Override
+ public Page getOutput()
+ {
+ if (processorFinishedSplit) {
+ // start processing a new split
+ if (pendingSplits.isEmpty()) {
+ // no more splits to process at the moment
+ return null;
+ }
+ currentSplit = pendingSplits.remove(0);
+ resetProcessor();
+ }
+ else {
+ // a split is being processed
+ requireNonNull(currentSplit, "currentSplit is null");
+ }
+
+ TableFunctionProcessorState state = processor.process(processorUsedData ? null : currentSplit);
+ if (state == FINISHED) {
+ processorFinishedSplit = true;
+ }
+ if (state instanceof Blocked) {
+ Blocked blocked = (Blocked) state;
+ processorBlocked = toListenableFuture(blocked.getFuture());
+ }
+ if (state instanceof Processed) {
+ Processed processed = (Processed) state;
+ if (processed.isUsedInput()) {
+ processorUsedData = true;
+ }
+ if (processed.getResult() != null) {
+ return processed.getResult();
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public ListenableFuture isBlocked()
+ {
+ return processorBlocked;
+ }
+
+ @Override
+ public void finish()
+ {
+ // this method is redundant. the operator takes no input at all. noMoreSplits() should be called instead.
+ }
+
+ @Override
+ public boolean isFinished()
+ {
+ return processorFinishedSplit && pendingSplits.isEmpty() && noMoreSplits;
+ }
+}
diff --git a/presto-main-base/src/main/java/com/facebook/presto/operator/PageBuffer.java b/presto-main-base/src/main/java/com/facebook/presto/operator/PageBuffer.java
new file mode 100644
index 0000000000000..cb14500597a1b
--- /dev/null
+++ b/presto-main-base/src/main/java/com/facebook/presto/operator/PageBuffer.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.facebook.presto.operator;
+
+import com.facebook.presto.common.Page;
+import com.facebook.presto.operator.WorkProcessor.ProcessState;
+import jakarta.annotation.Nullable;
+
+import static com.facebook.presto.operator.WorkProcessor.ProcessState.finished;
+import static com.facebook.presto.operator.WorkProcessor.ProcessState.ofResult;
+import static com.google.common.base.Preconditions.checkState;
+import static java.util.Objects.requireNonNull;
+
+public class PageBuffer
+{
+ @Nullable
+ private Page page;
+ private boolean finished;
+
+ public WorkProcessor pages()
+ {
+ return WorkProcessor.create(() -> {
+ if (isFinished() && isEmpty()) {
+ return finished();
+ }
+
+ if (!isEmpty()) {
+ Page result = page;
+ page = null;
+ return ofResult(result);
+ }
+
+ return ProcessState.yield();
+ });
+ }
+
+ public boolean isEmpty()
+ {
+ return page == null;
+ }
+
+ public boolean isFinished()
+ {
+ return finished;
+ }
+
+ public void add(Page page)
+ {
+ checkState(isEmpty(), "page buffer is not empty");
+ checkState(!isFinished(), "page buffer is finished");
+ requireNonNull(page, "page is null");
+
+ if (page.getPositionCount() == 0) {
+ return;
+ }
+
+ this.page = page;
+ }
+
+ public void finish()
+ {
+ finished = true;
+ }
+}
diff --git a/presto-main-base/src/main/java/com/facebook/presto/operator/PagesIndex.java b/presto-main-base/src/main/java/com/facebook/presto/operator/PagesIndex.java
index 640ae9919ca90..f4da55ac314e0 100644
--- a/presto-main-base/src/main/java/com/facebook/presto/operator/PagesIndex.java
+++ b/presto-main-base/src/main/java/com/facebook/presto/operator/PagesIndex.java
@@ -270,9 +270,9 @@ public void swap(int a, int b)
valueAddresses.swap(a, b);
}
- public int buildPage(int position, int[] outputChannels, PageBuilder pageBuilder)
+ public int buildPage(int position, int endPosition, int[] outputChannels, PageBuilder pageBuilder)
{
- while (!pageBuilder.isFull() && position < positionCount) {
+ while (!pageBuilder.isFull() && position < endPosition) {
long pageAddress = valueAddresses.get(position);
int blockIndex = decodeSliceIndex(pageAddress);
int blockPosition = decodePosition(pageAddress);
@@ -562,10 +562,29 @@ protected Page computeNext()
}
public Iterator getSortedPages()
+ {
+ return getSortedPagesFromRange(0, positionCount);
+ }
+
+ /**
+ * Get sorted pages from the specified section of the PagesIndex.
+ *
+ * @param start start position of the section, inclusive
+ * @param end end position of the section, exclusive
+ * @return iterator of pages
+ */
+ public Iterator getSortedPages(int start, int end)
+ {
+ checkArgument(start >= 0 && end <= positionCount, "position range out of bounds");
+ checkArgument(start <= end, "invalid position range");
+ return getSortedPagesFromRange(start, end);
+ }
+
+ private Iterator getSortedPagesFromRange(int start, int end)
{
return new AbstractIterator()
{
- private int currentPosition;
+ private int currentPosition = start;
private final PageBuilder pageBuilder = new PageBuilder(types);
private final int[] outputChannels = new int[types.size()];
@@ -576,7 +595,7 @@ public Iterator getSortedPages()
@Override
public Page computeNext()
{
- currentPosition = buildPage(currentPosition, outputChannels, pageBuilder);
+ currentPosition = buildPage(currentPosition, end, outputChannels, pageBuilder);
if (pageBuilder.isEmpty()) {
return endOfData();
}
diff --git a/presto-main-base/src/main/java/com/facebook/presto/operator/RegularTableFunctionPartition.java b/presto-main-base/src/main/java/com/facebook/presto/operator/RegularTableFunctionPartition.java
new file mode 100644
index 0000000000000..5d0376f3be7ee
--- /dev/null
+++ b/presto-main-base/src/main/java/com/facebook/presto/operator/RegularTableFunctionPartition.java
@@ -0,0 +1,438 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.facebook.presto.operator;
+
+import com.facebook.presto.common.Page;
+import com.facebook.presto.common.block.Block;
+import com.facebook.presto.common.block.BlockBuilder;
+import com.facebook.presto.common.block.RunLengthEncodedBlock;
+import com.facebook.presto.common.type.Type;
+import com.facebook.presto.spi.PrestoException;
+import com.facebook.presto.spi.function.table.TableFunctionDataProcessor;
+import com.facebook.presto.spi.function.table.TableFunctionProcessorState;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.primitives.Ints;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.Set;
+
+import static com.facebook.airlift.concurrent.MoreFutures.toListenableFuture;
+import static com.facebook.presto.common.Utils.checkState;
+import static com.facebook.presto.common.type.BigintType.BIGINT;
+import static com.facebook.presto.spi.StandardErrorCode.FUNCTION_IMPLEMENTATION_ERROR;
+import static com.facebook.presto.spi.function.table.TableFunctionProcessorState.Finished.FINISHED;
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.collect.ImmutableList.toImmutableList;
+import static com.google.common.util.concurrent.Futures.immediateFuture;
+import static java.lang.Math.min;
+import static java.lang.Math.toIntExact;
+import static java.lang.String.format;
+import static java.util.Objects.requireNonNull;
+
+public class RegularTableFunctionPartition
+ implements TableFunctionPartition
+{
+ private final PagesIndex pagesIndex;
+ private final int partitionStart;
+ private final int partitionEnd;
+ private final Iterator sortedPages;
+
+ private final TableFunctionDataProcessor tableFunction;
+ private final int properChannelsCount;
+ private final int passThroughSourcesCount;
+
+ // channels required by the table function, listed by source in order of argument declarations
+ private final int[][] requiredChannels;
+
+ // for each input channel, the end position of actual data in that channel (exclusive) relative to partition. The remaining rows are "filler" rows, and should not be passed to table function or passed-through
+ private final int[] endOfData;
+
+ // a builder for each pass-through column, in order of argument declarations
+ private final PassThroughColumnProvider[] passThroughProviders;
+
+ // number of processed input positions from partition start. all sources have been processed up to this position, except the sources whose partitions ended earlier.
+ private int processedPositions;
+
+ public RegularTableFunctionPartition(
+ PagesIndex pagesIndex,
+ int partitionStart,
+ int partitionEnd,
+ TableFunctionDataProcessor tableFunction,
+ int properChannelsCount,
+ int passThroughSourcesCount,
+ List> requiredChannels,
+ Optional> markerChannels,
+ List passThroughSpecifications)
+
+ {
+ checkArgument(pagesIndex.getPositionCount() != 0, "PagesIndex is empty for regular table function partition");
+ this.pagesIndex = pagesIndex;
+ this.partitionStart = partitionStart;
+ this.partitionEnd = partitionEnd;
+ this.sortedPages = pagesIndex.getSortedPages(partitionStart, partitionEnd);
+ this.tableFunction = requireNonNull(tableFunction, "tableFunction is null");
+ this.properChannelsCount = properChannelsCount;
+ this.passThroughSourcesCount = passThroughSourcesCount;
+ this.requiredChannels = requiredChannels.stream()
+ .map(Ints::toArray)
+ .toArray(int[][]::new);
+ this.endOfData = findEndOfData(markerChannels, requiredChannels, passThroughSpecifications);
+ for (List channels : requiredChannels) {
+ checkState(
+ channels.stream()
+ .mapToInt(channel -> endOfData[channel])
+ .distinct()
+ .count() <= 1,
+ "end-of-data position is inconsistent within a table function source");
+ }
+ this.passThroughProviders = new PassThroughColumnProvider[passThroughSpecifications.size()];
+ for (int i = 0; i < passThroughSpecifications.size(); i++) {
+ passThroughProviders[i] = createColumnProvider(passThroughSpecifications.get(i));
+ }
+ }
+
+ @Override
+ public WorkProcessor toOutputPages()
+ {
+ return WorkProcessor.create(new WorkProcessor.Process()
+ {
+ List> inputPages = prepareInputPages();
+
+ @Override
+ public WorkProcessor.ProcessState process()
+ {
+ TableFunctionProcessorState state = tableFunction.process(inputPages);
+ boolean functionGotNoData = inputPages == null;
+ if (state == FINISHED) {
+ return WorkProcessor.ProcessState.finished();
+ }
+ if (state instanceof TableFunctionProcessorState.Blocked) {
+ return WorkProcessor.ProcessState.blocked(toListenableFuture(((TableFunctionProcessorState.Blocked) state).getFuture()));
+ }
+ TableFunctionProcessorState.Processed processed = (TableFunctionProcessorState.Processed) state;
+ if (processed.isUsedInput()) {
+ inputPages = prepareInputPages();
+ }
+ if (processed.getResult() != null) {
+ return WorkProcessor.ProcessState.ofResult(appendPassThroughColumns(processed.getResult()));
+ }
+ if (functionGotNoData) {
+ throw new PrestoException(FUNCTION_IMPLEMENTATION_ERROR, "When function got no input, it should either produce output or return Blocked state");
+ }
+ return WorkProcessor.ProcessState.blocked(immediateFuture(null));
+ }
+ });
+ }
+
+ /**
+ * Iterate over the partition by page and extract pages for each table function source from the input page.
+ * For each source, project the columns required by the table function.
+ * If for some source all data in the partition has been consumed, Optional.empty() is returned for that source.
+ * It happens when the partition of this source is shorter than the partition of some other source.
+ * The overall length of the table function partition is equal to the length of the longest source partition.
+ * When all sources are fully consumed, this method returns null.
+ *
+ * NOTE: There are two types of table function's source semantics: set and row. The two types of sources should be handled
+ * by the TableFunctionDataProcessor in different ways. For a source with set semantics, the whole partition can be used for computations,
+ * while for a source with row semantics, each row should be processed independently from all other rows.
+ * To enforce that behavior, we could pass to the TableFunctionDataProcessor only one row from a table with row semantics.
+ * However, for performance reasons, we handle sources with row and set semantics in the same way: the TableFunctionDataProcessor
+ * gets a page of data from each source. The TableFunctionDataProcessor is responsible for using the provided data accordingly
+ * to the declared source semantics (set or rows).
+ *
+ * @return A List containing:
+ * - Optional Page for every source that is not fully consumed
+ * - Optional.empty() for every source that is fully consumed
+ * or null if all sources are fully consumed.
+ */
+ private List> prepareInputPages()
+ {
+ if (!sortedPages.hasNext()) {
+ return null;
+ }
+
+ Page inputPage = sortedPages.next();
+ ImmutableList.Builder> sourcePages = ImmutableList.builder();
+
+ for (int[] channelsForSource : requiredChannels) {
+ if (channelsForSource.length == 0) {
+ sourcePages.add(Optional.of(new Page(inputPage.getPositionCount())));
+ }
+ else {
+ int endOfDataForSource = endOfData[channelsForSource[0]]; // end-of-data position is validated to be consistent for all channels from source
+ if (endOfDataForSource <= processedPositions) {
+ // all data for this source was already processed
+ sourcePages.add(Optional.empty());
+ }
+ else {
+ Block[] sourceBlocks = new Block[channelsForSource.length];
+ if (endOfDataForSource < processedPositions + inputPage.getPositionCount()) {
+ // data for this source ends within the current page
+ for (int i = 0; i < channelsForSource.length; i++) {
+ int inputChannel = channelsForSource[i];
+ sourceBlocks[i] = inputPage.getBlock(inputChannel).getRegion(0, endOfDataForSource - processedPositions);
+ }
+ }
+ else {
+ // data for this source does not end within the current page
+ for (int i = 0; i < channelsForSource.length; i++) {
+ int inputChannel = channelsForSource[i];
+ sourceBlocks[i] = inputPage.getBlock(inputChannel);
+ }
+ }
+ sourcePages.add(Optional.of(new Page(sourceBlocks)));
+ }
+ }
+ }
+
+ processedPositions += inputPage.getPositionCount();
+
+ return sourcePages.build();
+ }
+
+ /**
+ * There are two types of table function's source semantics: set and row.
+ *
+ * For a source with row semantics, the table function result depends on the whole partition,
+ * so it is not always possible to associate an output row with a specific input row.
+ * The TableFunctionDataProcessor can return null as the pass-through index to indicate that
+ * the output row is not associated with any row from the given source.
+ *
+ * For a source with row semantics, the output is determined on a row-by-row basis, so every
+ * output row is associated with a specific input row. In such case, the pass-through index
+ * should never be null.
+ *
+ * In our implementation, we handle sources with row and set semantics in the same way.
+ * For performance reasons, we do not validate the null pass-through indexes.
+ * The TableFunctionDataProcessor is responsible for using the pass-through capability
+ * accordingly to the declared source semantics (set or rows).
+ */
+ private Page appendPassThroughColumns(Page page)
+ {
+ if (page.getChannelCount() != properChannelsCount + passThroughSourcesCount) {
+ throw new PrestoException(
+ FUNCTION_IMPLEMENTATION_ERROR,
+ format(
+ "Table function returned a page containing %s channels. Expected channel number: %s (%s proper columns, %s pass-through index columns)",
+ page.getChannelCount(),
+ properChannelsCount + passThroughSourcesCount,
+ properChannelsCount,
+ passThroughSourcesCount));
+ }
+ // TODO is it possible to verify types of columns returned by TF?
+
+ Block[] resultBlocks = new Block[properChannelsCount + passThroughProviders.length];
+
+ // proper outputs first
+ for (int channel = 0; channel < properChannelsCount; channel++) {
+ resultBlocks[channel] = page.getBlock(channel);
+ }
+
+ // pass-through columns next
+ int channel = properChannelsCount;
+ for (PassThroughColumnProvider provider : passThroughProviders) {
+ resultBlocks[channel] = provider.getPassThroughColumn(page);
+ channel++;
+ }
+
+ // pass the position count so that the Page can be successfully created in the case when there are no output channels (resultBlocks is empty)
+ return new Page(page.getPositionCount(), resultBlocks);
+ }
+
+ private int[] findEndOfData(Optional> markerChannels, List> requiredChannels, List passThroughSpecifications)
+ {
+ Set referencedChannels = ImmutableSet.builder()
+ .addAll(requiredChannels.stream()
+ .flatMap(Collection::stream)
+ .collect(toImmutableList()))
+ .addAll(passThroughSpecifications.stream()
+ .map(PassThroughColumnSpecification::getInputChannel)
+ .collect(toImmutableList()))
+ .build();
+
+ if (referencedChannels.isEmpty()) {
+ // no required or pass-through channels
+ return null;
+ }
+
+ int maxInputChannel = referencedChannels.stream()
+ .mapToInt(Integer::intValue)
+ .max()
+ .orElseThrow(NoSuchElementException::new);
+
+ int[] result = new int[maxInputChannel + 1];
+ Arrays.fill(result, -1);
+
+ // if table function had one source, adding a marker channel was not necessary.
+ // end-of-data position is equal to partition end for each input channel
+ if (!markerChannels.isPresent()) {
+ referencedChannels.stream()
+ .forEach(channel -> result[channel] = partitionEnd - partitionStart);
+ return result;
+ }
+
+ // if table function had more than one source, the markers map shall be present, and it shall contain mapping for each input channel
+ ImmutableMap.Builder endOfDataPerMarkerBuilder = ImmutableMap.builder();
+ for (int markerChannel : ImmutableSet.copyOf(markerChannels.orElseThrow(NoSuchElementException::new).values())) {
+ endOfDataPerMarkerBuilder.put(markerChannel, findFirstNullPosition(markerChannel));
+ }
+ Map endOfDataPerMarker = endOfDataPerMarkerBuilder.buildOrThrow();
+ referencedChannels.stream()
+ .forEach(channel -> result[channel] = endOfDataPerMarker.get(markerChannels.orElseThrow(NoSuchElementException::new).get(channel)) - partitionStart);
+
+ return result;
+ }
+
+ private int findFirstNullPosition(int markerChannel)
+ {
+ if (pagesIndex.isNull(markerChannel, partitionStart)) {
+ return partitionStart;
+ }
+ if (!pagesIndex.isNull(markerChannel, partitionEnd - 1)) {
+ return partitionEnd;
+ }
+
+ int start = partitionStart;
+ int end = partitionEnd;
+ // value at start is not null, value at end is null
+ while (end - start > 1) {
+ int mid = (start + end) >>> 1;
+ if (pagesIndex.isNull(markerChannel, mid)) {
+ end = mid;
+ }
+ else {
+ start = mid;
+ }
+ }
+ return end;
+ }
+
+ public static class PassThroughColumnSpecification
+ {
+ private final boolean isPartitioningColumn;
+ private final int inputChannel;
+ private final int indexChannel;
+
+ public PassThroughColumnSpecification(boolean isPartitioningColumn, int inputChannel, int indexChannel)
+ {
+ this.isPartitioningColumn = isPartitioningColumn;
+ this.inputChannel = inputChannel;
+ this.indexChannel = indexChannel;
+ }
+
+ public boolean isPartitioningColumn()
+ {
+ return isPartitioningColumn;
+ }
+
+ public int getInputChannel()
+ {
+ return inputChannel;
+ }
+
+ public int getIndexChannel()
+ {
+ return indexChannel;
+ }
+ }
+
+ private PassThroughColumnProvider createColumnProvider(PassThroughColumnSpecification specification)
+ {
+ if (specification.isPartitioningColumn()) {
+ return new PartitioningColumnProvider(pagesIndex.getSingleValueBlock(specification.getInputChannel(), partitionStart));
+ }
+ return new NonPartitioningColumnProvider(specification.getInputChannel(), specification.getIndexChannel());
+ }
+
+ private interface PassThroughColumnProvider
+ {
+ Block getPassThroughColumn(Page page);
+ }
+
+ private static class PartitioningColumnProvider
+ implements PassThroughColumnProvider
+ {
+ private final Block partitioningValue;
+
+ private PartitioningColumnProvider(Block partitioningValue)
+ {
+ this.partitioningValue = requireNonNull(partitioningValue, "partitioningValue is null");
+ }
+
+ @Override
+ public Block getPassThroughColumn(Page page)
+ {
+ return new RunLengthEncodedBlock(partitioningValue, page.getPositionCount());
+ }
+
+ public Block getPartitioningValue()
+ {
+ return partitioningValue;
+ }
+ }
+
+ private final class NonPartitioningColumnProvider
+ implements PassThroughColumnProvider
+ {
+ private final int inputChannel;
+ private final Type type;
+ private final int indexChannel;
+
+ public NonPartitioningColumnProvider(int inputChannel, int indexChannel)
+ {
+ this.inputChannel = inputChannel;
+ this.type = pagesIndex.getType(inputChannel);
+ this.indexChannel = indexChannel;
+ }
+
+ @Override
+ public Block getPassThroughColumn(Page page)
+ {
+ Block indexes = page.getBlock(indexChannel);
+ BlockBuilder builder = type.createBlockBuilder(null, page.getPositionCount());
+ for (int position = 0; position < page.getPositionCount(); position++) {
+ if (indexes.isNull(position)) {
+ builder.appendNull();
+ }
+ else {
+ // table function returns index from partition start
+ long index = BIGINT.getLong(indexes, position);
+ // validate index
+ if (index < 0 || index >= endOfData[inputChannel] || index >= processedPositions) {
+ int end = min(endOfData[inputChannel], processedPositions) - 1;
+ if (end >= 0) {
+ throw new PrestoException(FUNCTION_IMPLEMENTATION_ERROR, format("Index of a pass-through row: %s out of processed portion of partition [0, %s]", index, end));
+ }
+ else {
+ throw new PrestoException(FUNCTION_IMPLEMENTATION_ERROR, "Index of a pass-through row must be null when no input data from the partition was processed. Actual: " + index);
+ }
+ }
+ // index in PagesIndex
+ long absoluteIndex = partitionStart + index;
+ pagesIndex.appendTo(inputChannel, toIntExact(absoluteIndex), builder);
+ }
+ }
+
+ return builder.build();
+ }
+ }
+}
diff --git a/presto-main-base/src/main/java/com/facebook/presto/operator/TableFunctionOperator.java b/presto-main-base/src/main/java/com/facebook/presto/operator/TableFunctionOperator.java
new file mode 100644
index 0000000000000..97d71899d5b9a
--- /dev/null
+++ b/presto-main-base/src/main/java/com/facebook/presto/operator/TableFunctionOperator.java
@@ -0,0 +1,642 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.facebook.presto.operator;
+
+import com.facebook.presto.common.Page;
+import com.facebook.presto.common.block.SortOrder;
+import com.facebook.presto.common.type.Type;
+import com.facebook.presto.memory.context.LocalMemoryContext;
+import com.facebook.presto.spi.function.table.ConnectorTableFunctionHandle;
+import com.facebook.presto.spi.function.table.TableFunctionProcessorProvider;
+import com.facebook.presto.spi.plan.PlanNodeId;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Iterables;
+import com.google.common.primitives.Ints;
+import com.google.common.util.concurrent.ListenableFuture;
+import jakarta.annotation.Nullable;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.OptionalInt;
+
+import static com.facebook.presto.common.block.SortOrder.ASC_NULLS_LAST;
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkPositionIndex;
+import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.collect.ImmutableList.toImmutableList;
+import static com.google.common.collect.Iterables.concat;
+import static java.util.Collections.nCopies;
+import static java.util.Objects.requireNonNull;
+
+public class TableFunctionOperator
+ implements Operator
+{
+ public static class TableFunctionOperatorFactory
+ implements OperatorFactory
+ {
+ private final int operatorId;
+ private final PlanNodeId planNodeId;
+
+ // a provider of table function processor to be called once per partition
+ private final TableFunctionProcessorProvider tableFunctionProvider;
+
+ // all information necessary to execute the table function collected during analysis
+ private final ConnectorTableFunctionHandle functionHandle;
+
+ // number of proper columns produced by the table function
+ private final int properChannelsCount;
+
+ // number of input tables declared as pass-through
+ private final int passThroughSourcesCount;
+
+ // columns required by the table function, in order of input tables
+ private final List> requiredChannels;
+
+ // map from input channel to marker channel
+ // for each input table, there is a channel that marks which rows contain original data, and which are "filler" rows.
+ // the "filler" rows are part of the algorithm, and they should not be processed by the table function, or passed-through.
+ // In this map, every original column from the input table is associated with the appropriate marker.
+ private final Optional> markerChannels;
+
+ // necessary information to build a pass-through column, for all pass-through columns, ordered as expected on the output
+ // it includes columns from sources declared as pass-through as well as partitioning columns from other sources
+ private final List passThroughSpecifications;
+
+ // specifies whether the function should be pruned or executed when the input is empty
+ // pruneWhenEmpty is false if and only if all original input tables are KEEP WHEN EMPTY
+ private final boolean pruneWhenEmpty;
+
+ // partitioning channels from all sources
+ private final List partitionChannels;
+
+ // subset of partition channels that are already grouped
+ private final List prePartitionedChannels;
+
+ // channels necessary to sort all sources:
+ // - for a single source, these are the source's sort channels
+ // - for multiple sources, this is a single synthesized row number channel
+ private final List sortChannels;
+ private final List sortOrders;
+
+ // number of leading sort channels that are already sorted
+ private final int preSortedPrefix;
+
+ private final List sourceTypes;
+ private final int expectedPositions;
+ private final PagesIndex.Factory pagesIndexFactory;
+
+ private boolean closed;
+
+ public TableFunctionOperatorFactory(
+ int operatorId,
+ PlanNodeId planNodeId,
+ TableFunctionProcessorProvider tableFunctionProvider,
+ ConnectorTableFunctionHandle functionHandle,
+ int properChannelsCount,
+ int passThroughSourcesCount,
+ List> requiredChannels,
+ Optional> markerChannels,
+ List passThroughSpecifications,
+ boolean pruneWhenEmpty,
+ List partitionChannels,
+ List prePartitionedChannels,
+ List sortChannels,
+ List sortOrders,
+ int preSortedPrefix,
+ List sourceTypes,
+ int expectedPositions,
+ PagesIndex.Factory pagesIndexFactory)
+ {
+ requireNonNull(planNodeId, "planNodeId is null");
+ requireNonNull(tableFunctionProvider, "tableFunctionProvider is null");
+ requireNonNull(functionHandle, "functionHandle is null");
+ requireNonNull(requiredChannels, "requiredChannels is null");
+ requireNonNull(markerChannels, "markerChannels is null");
+ requireNonNull(passThroughSpecifications, "passThroughSpecifications is null");
+ requireNonNull(partitionChannels, "partitionChannels is null");
+ requireNonNull(prePartitionedChannels, "prePartitionedChannels is null");
+ checkArgument(partitionChannels.containsAll(prePartitionedChannels), "prePartitionedChannels must be a subset of partitionChannels");
+ requireNonNull(sortChannels, "sortChannels is null");
+ requireNonNull(sortOrders, "sortOrders is null");
+ checkArgument(sortChannels.size() == sortOrders.size(), "The number of sort channels must be equal to the number of sort orders");
+ checkArgument(preSortedPrefix <= sortChannels.size(), "The number of pre-sorted channels must be lower or equal to the number of sort channels");
+ checkArgument(preSortedPrefix == 0 || ImmutableSet.copyOf(prePartitionedChannels).equals(ImmutableSet.copyOf(partitionChannels)), "preSortedPrefix can only be greater than zero if all partition channels are pre-grouped");
+ requireNonNull(sourceTypes, "sourceTypes is null");
+ requireNonNull(pagesIndexFactory, "pagesIndexFactory is null");
+
+ this.operatorId = operatorId;
+ this.planNodeId = planNodeId;
+ this.tableFunctionProvider = tableFunctionProvider;
+ this.functionHandle = functionHandle;
+ this.properChannelsCount = properChannelsCount;
+ this.passThroughSourcesCount = passThroughSourcesCount;
+ this.requiredChannels = requiredChannels.stream()
+ .map(ImmutableList::copyOf)
+ .collect(toImmutableList());
+ this.markerChannels = markerChannels.map(ImmutableMap::copyOf);
+ this.passThroughSpecifications = ImmutableList.copyOf(passThroughSpecifications);
+ this.pruneWhenEmpty = pruneWhenEmpty;
+ this.partitionChannels = ImmutableList.copyOf(partitionChannels);
+ this.prePartitionedChannels = ImmutableList.copyOf(prePartitionedChannels);
+ this.sortChannels = ImmutableList.copyOf(sortChannels);
+ this.sortOrders = ImmutableList.copyOf(sortOrders);
+ this.preSortedPrefix = preSortedPrefix;
+ this.sourceTypes = ImmutableList.copyOf(sourceTypes);
+ this.expectedPositions = expectedPositions;
+ this.pagesIndexFactory = pagesIndexFactory;
+ }
+
+ @Override
+ public Operator createOperator(DriverContext driverContext)
+ {
+ checkState(!closed, "Factory is already closed");
+
+ OperatorContext operatorContext = driverContext.addOperatorContext(operatorId, planNodeId, TableFunctionOperator.class.getSimpleName());
+ return new TableFunctionOperator(
+ operatorContext,
+ tableFunctionProvider,
+ functionHandle,
+ properChannelsCount,
+ passThroughSourcesCount,
+ requiredChannels,
+ markerChannels,
+ passThroughSpecifications,
+ pruneWhenEmpty,
+ partitionChannels,
+ prePartitionedChannels,
+ sortChannels,
+ sortOrders,
+ preSortedPrefix,
+ sourceTypes,
+ expectedPositions,
+ pagesIndexFactory);
+ }
+
+ @Override
+ public void noMoreOperators()
+ {
+ closed = true;
+ }
+
+ @Override
+ public OperatorFactory duplicate()
+ {
+ return new TableFunctionOperatorFactory(
+ operatorId,
+ planNodeId,
+ tableFunctionProvider,
+ functionHandle,
+ properChannelsCount,
+ passThroughSourcesCount,
+ requiredChannels,
+ markerChannels,
+ passThroughSpecifications,
+ pruneWhenEmpty,
+ partitionChannels,
+ prePartitionedChannels,
+ sortChannels,
+ sortOrders,
+ preSortedPrefix,
+ sourceTypes,
+ expectedPositions,
+ pagesIndexFactory);
+ }
+ }
+
+ private final OperatorContext operatorContext;
+
+ private final PageBuffer pageBuffer = new PageBuffer();
+ private final WorkProcessor outputPages;
+ private final boolean processEmptyInput;
+
+ @Nullable
+ private Page pendingInput;
+ private boolean operatorFinishing;
+
+ public TableFunctionOperator(
+ OperatorContext operatorContext,
+ TableFunctionProcessorProvider tableFunctionProvider,
+ ConnectorTableFunctionHandle functionHandle,
+ int properChannelsCount,
+ int passThroughSourcesCount,
+ List> requiredChannels,
+ Optional> markerChannels,
+ List passThroughSpecifications,
+ boolean pruneWhenEmpty,
+ List partitionChannels,
+ List prePartitionedChannels,
+ List sortChannels,
+ List sortOrders,
+ int preSortedPrefix,
+ List sourceTypes,
+ int expectedPositions,
+ PagesIndex.Factory pagesIndexFactory)
+ {
+ requireNonNull(operatorContext, "operatorContext is null");
+ requireNonNull(tableFunctionProvider, "tableFunctionProvider is null");
+ requireNonNull(functionHandle, "functionHandle is null");
+ requireNonNull(requiredChannels, "requiredChannels is null");
+ requireNonNull(markerChannels, "markerChannels is null");
+ requireNonNull(passThroughSpecifications, "passThroughSpecifications is null");
+ requireNonNull(partitionChannels, "partitionChannels is null");
+ requireNonNull(prePartitionedChannels, "prePartitionedChannels is null");
+ checkArgument(partitionChannels.containsAll(prePartitionedChannels), "prePartitionedChannels must be a subset of partitionChannels");
+ requireNonNull(sortChannels, "sortChannels is null");
+ requireNonNull(sortOrders, "sortOrders is null");
+ checkArgument(sortChannels.size() == sortOrders.size(), "The number of sort channels must be equal to the number of sort orders");
+ checkArgument(preSortedPrefix <= sortChannels.size(), "The number of pre-sorted channels must be lower or equal to the number of sort channels");
+ checkArgument(preSortedPrefix == 0 || ImmutableSet.copyOf(prePartitionedChannels).equals(ImmutableSet.copyOf(partitionChannels)), "preSortedPrefix can only be greater than zero if all partition channels are pre-grouped");
+ requireNonNull(sourceTypes, "sourceTypes is null");
+ requireNonNull(pagesIndexFactory, "pagesIndexFactory is null");
+
+ this.operatorContext = operatorContext;
+
+ this.processEmptyInput = !pruneWhenEmpty;
+
+ PagesIndex pagesIndex = pagesIndexFactory.newPagesIndex(sourceTypes, expectedPositions);
+ HashStrategies hashStrategies = new HashStrategies(pagesIndex, partitionChannels, prePartitionedChannels, sortChannels, sortOrders, preSortedPrefix);
+
+ this.outputPages = WorkProcessor.create(new PagesSource())
+ .transform(new PartitionAndSort(pagesIndex, hashStrategies, processEmptyInput))
+ .flatMap(groupPagesIndex -> pagesIndexToTableFunctionPartitions(
+ groupPagesIndex,
+ hashStrategies,
+ tableFunctionProvider,
+ functionHandle,
+ properChannelsCount,
+ passThroughSourcesCount,
+ requiredChannels,
+ markerChannels,
+ passThroughSpecifications,
+ processEmptyInput))
+ .flatMap(TableFunctionPartition::toOutputPages);
+ }
+
+ @Override
+ public OperatorContext getOperatorContext()
+ {
+ return operatorContext;
+ }
+
+ @Override
+ public void finish()
+ {
+ operatorFinishing = true;
+ }
+
+ @Override
+ public boolean isFinished()
+ {
+ return outputPages.isFinished();
+ }
+
+ @Override
+ public ListenableFuture isBlocked()
+ {
+ if (outputPages.isBlocked()) {
+ return outputPages.getBlockedFuture();
+ }
+
+ return NOT_BLOCKED;
+ }
+
+ @Override
+ public boolean needsInput()
+ {
+ return pendingInput == null && !operatorFinishing;
+ }
+
+ @Override
+ public void addInput(Page page)
+ {
+ requireNonNull(page, "page is null");
+ checkState(pendingInput == null, "Operator already has pending input");
+
+ if (page.getPositionCount() == 0) {
+ return;
+ }
+
+ pendingInput = page;
+ }
+
+ @Override
+ public Page getOutput()
+ {
+ if (!outputPages.process()) {
+ return null;
+ }
+
+ if (outputPages.isFinished()) {
+ return null;
+ }
+
+ return outputPages.getResult();
+ }
+
+ private static class HashStrategies
+ {
+ final PagesHashStrategy prePartitionedStrategy;
+ final PagesHashStrategy remainingPartitionStrategy;
+ final PagesHashStrategy preSortedStrategy;
+ final List remainingPartitionAndSortChannels;
+ final List remainingSortOrders;
+ final int[] prePartitionedChannelsArray;
+
+ public HashStrategies(
+ PagesIndex pagesIndex,
+ List partitionChannels,
+ List prePartitionedChannels,
+ List sortChannels,
+ List sortOrders,
+ int preSortedPrefix)
+ {
+ this.prePartitionedStrategy = pagesIndex.createPagesHashStrategy(prePartitionedChannels, OptionalInt.empty());
+
+ List remainingPartitionChannels = partitionChannels.stream()
+ .filter(channel -> !prePartitionedChannels.contains(channel))
+ .collect(toImmutableList());
+ this.remainingPartitionStrategy = pagesIndex.createPagesHashStrategy(remainingPartitionChannels, OptionalInt.empty());
+
+ List preSortedChannels = sortChannels.stream()
+ .limit(preSortedPrefix)
+ .collect(toImmutableList());
+ this.preSortedStrategy = pagesIndex.createPagesHashStrategy(preSortedChannels, OptionalInt.empty());
+
+ if (preSortedPrefix > 0) {
+ // preSortedPrefix > 0 implies that all partition channels are already pre-partitioned (enforced by check in the constructor), so we only need to do the remaining sort
+ this.remainingPartitionAndSortChannels = ImmutableList.copyOf(Iterables.skip(sortChannels, preSortedPrefix));
+ this.remainingSortOrders = ImmutableList.copyOf(Iterables.skip(sortOrders, preSortedPrefix));
+ }
+ else {
+ // we need to sort by the remaining partition channels so that the input is fully partitioned,
+ // and then need to we sort by all the sort channels so that the input is fully sorted
+ this.remainingPartitionAndSortChannels = ImmutableList.copyOf(concat(remainingPartitionChannels, sortChannels));
+ this.remainingSortOrders = ImmutableList.copyOf(concat(nCopies(remainingPartitionChannels.size(), ASC_NULLS_LAST), sortOrders));
+ }
+
+ this.prePartitionedChannelsArray = Ints.toArray(prePartitionedChannels);
+ }
+ }
+
+ private class PartitionAndSort
+ implements WorkProcessor.Transformation
+ {
+ private final PagesIndex pagesIndex;
+ private final HashStrategies hashStrategies;
+ private final LocalMemoryContext memoryContext;
+
+ private boolean resetPagesIndex;
+ private int inputPosition;
+ private boolean processEmptyInput;
+
+ public PartitionAndSort(PagesIndex pagesIndex, HashStrategies hashStrategies, boolean processEmptyInput)
+ {
+ this.pagesIndex = pagesIndex;
+ this.hashStrategies = hashStrategies;
+ this.memoryContext = operatorContext.aggregateUserMemoryContext().newLocalMemoryContext(PartitionAndSort.class.getSimpleName());
+ this.processEmptyInput = processEmptyInput;
+ }
+
+ @Override
+ public WorkProcessor.TransformationState process(Optional input)
+ {
+ if (resetPagesIndex) {
+ pagesIndex.clear();
+ updateMemoryUsage();
+ resetPagesIndex = false;
+ }
+
+ if (!input.isPresent() && pagesIndex.getPositionCount() == 0) {
+ if (processEmptyInput) {
+ // it can only happen at the first call to process(), which implies that there is no input. Empty PagesIndex can be passed on only once.
+ processEmptyInput = false;
+ return WorkProcessor.TransformationState.ofResult(pagesIndex, false);
+ }
+ else {
+ memoryContext.close();
+ return WorkProcessor.TransformationState.finished();
+ }
+ }
+
+ // there is input, so we are not interested in processing empty input
+ processEmptyInput = false;
+
+ if (input.isPresent()) {
+ // append rows from input which belong to the current group wrt pre-partitioned columns
+ // it might be one or more partitions
+ inputPosition = appendCurrentGroup(pagesIndex, hashStrategies, input.get(), inputPosition);
+ updateMemoryUsage();
+
+ if (inputPosition >= input.get().getPositionCount()) {
+ inputPosition = 0;
+ return WorkProcessor.TransformationState.needsMoreData();
+ }
+ }
+
+ // we have unused input or the input is finished. we have buffered a full group
+ // the group contains one or more partitions, as it was determined by the pre-partitioned columns
+ // sorting serves two purposes:
+ // - sort by the remaining partition channels so that the input is fully partitioned,
+ // - sort by all the sort channels so that the input is fully sorted
+ sortCurrentGroup(pagesIndex, hashStrategies);
+ resetPagesIndex = true;
+ return WorkProcessor.TransformationState.ofResult(pagesIndex, false);
+ }
+
+ void updateMemoryUsage()
+ {
+ memoryContext.setBytes(pagesIndex.getEstimatedSize().toBytes());
+ }
+ }
+
+ private static int appendCurrentGroup(PagesIndex pagesIndex, HashStrategies hashStrategies, Page page, int startPosition)
+ {
+ checkArgument(page.getPositionCount() > startPosition);
+
+ PagesHashStrategy prePartitionedStrategy = hashStrategies.prePartitionedStrategy;
+ Page prePartitionedPage = page.extractChannels(hashStrategies.prePartitionedChannelsArray);
+
+ if (pagesIndex.getPositionCount() == 0 || pagesIndex.positionEqualsRow(prePartitionedStrategy, 0, startPosition, prePartitionedPage)) {
+ // we are within the current group. find the position where the pre-grouped columns change
+ int groupEnd = findGroupEnd(prePartitionedPage, prePartitionedStrategy, startPosition);
+
+ // add the section of the page that contains values for the current group
+ pagesIndex.addPage(page.getRegion(startPosition, groupEnd - startPosition));
+
+ if (page.getPositionCount() - groupEnd > 0) {
+ // the remaining prt of the page contains the next group
+ return groupEnd;
+ }
+ // page fully consumed: it contains the current group only
+ return page.getPositionCount();
+ }
+
+ // we had previous results buffered, but the remaining page starts with new group values
+ return startPosition;
+ }
+
+ private static void sortCurrentGroup(PagesIndex pagesIndex, HashStrategies hashStrategies)
+ {
+ PagesHashStrategy preSortedStrategy = hashStrategies.preSortedStrategy;
+ List remainingPartitionAndSortChannels = hashStrategies.remainingPartitionAndSortChannels;
+ List remainingSortOrders = hashStrategies.remainingSortOrders;
+
+ if (pagesIndex.getPositionCount() > 1 && !remainingPartitionAndSortChannels.isEmpty()) {
+ int startPosition = 0;
+ while (startPosition < pagesIndex.getPositionCount()) {
+ int endPosition = findGroupEnd(pagesIndex, preSortedStrategy, startPosition);
+ pagesIndex.sort(remainingPartitionAndSortChannels, remainingSortOrders, startPosition, endPosition);
+ startPosition = endPosition;
+ }
+ }
+ }
+
+ // Assumes input grouped on relevant pagesHashStrategy columns
+ private static int findGroupEnd(Page page, PagesHashStrategy pagesHashStrategy, int startPosition)
+ {
+ checkArgument(page.getPositionCount() > 0, "Must have at least one position");
+ checkPositionIndex(startPosition, page.getPositionCount(), "startPosition out of bounds");
+
+ return findEndPosition(startPosition, page.getPositionCount(), (firstPosition, secondPosition) -> pagesHashStrategy.rowEqualsRow(firstPosition, page, secondPosition, page));
+ }
+
+ // Assumes input grouped on relevant pagesHashStrategy columns
+ private static int findGroupEnd(PagesIndex pagesIndex, PagesHashStrategy pagesHashStrategy, int startPosition)
+ {
+ checkArgument(pagesIndex.getPositionCount() > 0, "Must have at least one position");
+ checkPositionIndex(startPosition, pagesIndex.getPositionCount(), "startPosition out of bounds");
+
+ return findEndPosition(startPosition, pagesIndex.getPositionCount(), (firstPosition, secondPosition) -> pagesIndex.positionEqualsPosition(pagesHashStrategy, firstPosition, secondPosition));
+ }
+
+ /**
+ * @param startPosition - inclusive
+ * @param endPosition - exclusive
+ * @param comparator - returns true if positions given as parameters are equal
+ * @return the end of the group position exclusive (the position the very next group starts)
+ */
+ @VisibleForTesting
+ static int findEndPosition(int startPosition, int endPosition, PositionComparator comparator)
+ {
+ checkArgument(startPosition >= 0, "startPosition must be greater or equal than zero: %s", startPosition);
+ checkArgument(startPosition < endPosition, "startPosition (%s) must be less than endPosition (%s)", startPosition, endPosition);
+
+ int left = startPosition;
+ int right = endPosition;
+
+ while (right - left > 1) {
+ int middle = (left + right) >>> 1;
+
+ if (comparator.test(startPosition, middle)) {
+ left = middle;
+ }
+ else {
+ right = middle;
+ }
+ }
+
+ return right;
+ }
+
+ private interface PositionComparator
+ {
+ boolean test(int first, int second);
+ }
+
+ private WorkProcessor pagesIndexToTableFunctionPartitions(
+ PagesIndex pagesIndex,
+ HashStrategies hashStrategies,
+ TableFunctionProcessorProvider tableFunctionProvider,
+ ConnectorTableFunctionHandle functionHandle,
+ int properChannelsCount,
+ int passThroughSourcesCount,
+ List> requiredChannels,
+ Optional> markerChannels,
+ List passThroughSpecifications,
+ boolean processEmptyInput)
+ {
+ // pagesIndex contains the full grouped and sorted data for one or more partitions
+
+ PagesHashStrategy remainingPartitionStrategy = hashStrategies.remainingPartitionStrategy;
+
+ return WorkProcessor.create(new WorkProcessor.Process()
+ {
+ private int partitionStart;
+ private boolean processEmpty = processEmptyInput;
+
+ @Override
+ public WorkProcessor.ProcessState process()
+ {
+ if (partitionStart == pagesIndex.getPositionCount()) {
+ if (processEmpty && pagesIndex.getPositionCount() == 0) {
+ // empty PagesIndex can only be passed once as the result of PartitionAndSort. Neither this nor any future instance of Process will ever get an empty PagesIndex again.
+ processEmpty = false;
+ return WorkProcessor.ProcessState.ofResult(new EmptyTableFunctionPartition(
+ tableFunctionProvider.getDataProcessor(functionHandle),
+ properChannelsCount,
+ passThroughSourcesCount,
+ passThroughSpecifications.stream()
+ .map(RegularTableFunctionPartition.PassThroughColumnSpecification::getInputChannel)
+ .map(pagesIndex::getType)
+ .collect(toImmutableList())));
+ }
+ return WorkProcessor.ProcessState.finished();
+ }
+
+ // there is input, so we are not interested in processing empty input
+ processEmpty = false;
+
+ int partitionEnd = findGroupEnd(pagesIndex, remainingPartitionStrategy, partitionStart);
+
+ RegularTableFunctionPartition partition = new RegularTableFunctionPartition(
+ pagesIndex,
+ partitionStart,
+ partitionEnd,
+ tableFunctionProvider.getDataProcessor(functionHandle),
+ properChannelsCount,
+ passThroughSourcesCount,
+ requiredChannels,
+ markerChannels,
+ passThroughSpecifications);
+
+ partitionStart = partitionEnd;
+ return WorkProcessor.ProcessState.ofResult(partition);
+ }
+ });
+ }
+
+ private class PagesSource
+ implements WorkProcessor.Process
+ {
+ @Override
+ public WorkProcessor.ProcessState process()
+ {
+ if (operatorFinishing && pendingInput == null) {
+ return WorkProcessor.ProcessState.finished();
+ }
+
+ if (pendingInput != null) {
+ Page result = pendingInput;
+ pendingInput = null;
+ return WorkProcessor.ProcessState.ofResult(result);
+ }
+
+ return WorkProcessor.ProcessState.yield();
+ }
+ }
+}
diff --git a/presto-main-base/src/main/java/com/facebook/presto/operator/TableFunctionPartition.java b/presto-main-base/src/main/java/com/facebook/presto/operator/TableFunctionPartition.java
new file mode 100644
index 0000000000000..1876b352bd251
--- /dev/null
+++ b/presto-main-base/src/main/java/com/facebook/presto/operator/TableFunctionPartition.java
@@ -0,0 +1,21 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.facebook.presto.operator;
+
+import com.facebook.presto.common.Page;
+
+public interface TableFunctionPartition
+{
+ WorkProcessor toOutputPages();
+}
diff --git a/presto-main-base/src/main/java/com/facebook/presto/operator/project/PageProcessor.java b/presto-main-base/src/main/java/com/facebook/presto/operator/project/PageProcessor.java
index 736662d471eaf..744b487e670d4 100644
--- a/presto-main-base/src/main/java/com/facebook/presto/operator/project/PageProcessor.java
+++ b/presto-main-base/src/main/java/com/facebook/presto/operator/project/PageProcessor.java
@@ -43,7 +43,6 @@
import static com.facebook.presto.common.block.DictionaryId.randomDictionaryId;
import static com.facebook.presto.operator.WorkProcessor.ProcessState.finished;
import static com.facebook.presto.operator.WorkProcessor.ProcessState.ofResult;
-import static com.facebook.presto.operator.WorkProcessor.ProcessState.yield;
import static com.facebook.presto.operator.project.SelectedPositions.positionsRange;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Verify.verify;
@@ -204,7 +203,7 @@ public ProcessState process()
lastComputeYielded = true;
lastComputeBatchSize = batchSize;
updateRetainedSize();
- return yield();
+ return ProcessState.yield();
}
if (result.isPageTooLarge()) {
diff --git a/presto-main-base/src/main/java/com/facebook/presto/operator/table/ExcludeColumns.java b/presto-main-base/src/main/java/com/facebook/presto/operator/table/ExcludeColumns.java
new file mode 100644
index 0000000000000..d7881b63c0bc3
--- /dev/null
+++ b/presto-main-base/src/main/java/com/facebook/presto/operator/table/ExcludeColumns.java
@@ -0,0 +1,169 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.facebook.presto.operator.table;
+
+import com.facebook.presto.common.type.RowType;
+import com.facebook.presto.spi.ConnectorSession;
+import com.facebook.presto.spi.PrestoException;
+import com.facebook.presto.spi.connector.ConnectorTransactionHandle;
+import com.facebook.presto.spi.function.table.AbstractConnectorTableFunction;
+import com.facebook.presto.spi.function.table.Argument;
+import com.facebook.presto.spi.function.table.ConnectorTableFunction;
+import com.facebook.presto.spi.function.table.ConnectorTableFunctionHandle;
+import com.facebook.presto.spi.function.table.Descriptor;
+import com.facebook.presto.spi.function.table.DescriptorArgument;
+import com.facebook.presto.spi.function.table.DescriptorArgumentSpecification;
+import com.facebook.presto.spi.function.table.TableArgument;
+import com.facebook.presto.spi.function.table.TableArgumentSpecification;
+import com.facebook.presto.spi.function.table.TableFunctionAnalysis;
+import com.facebook.presto.spi.function.table.TableFunctionDataProcessor;
+import com.facebook.presto.spi.function.table.TableFunctionProcessorProvider;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Sets;
+
+import javax.inject.Provider;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import static com.facebook.presto.spi.StandardErrorCode.INVALID_ARGUMENTS;
+import static com.facebook.presto.spi.StandardErrorCode.INVALID_FUNCTION_ARGUMENT;
+import static com.facebook.presto.spi.function.table.DescriptorArgument.NULL_DESCRIPTOR;
+import static com.facebook.presto.spi.function.table.GenericTableReturnTypeSpecification.GENERIC_TABLE;
+import static com.facebook.presto.spi.function.table.TableFunctionProcessorState.Finished.FINISHED;
+import static com.facebook.presto.spi.function.table.TableFunctionProcessorState.Processed.usedInputAndProduced;
+import static com.google.common.collect.ImmutableSet.toImmutableSet;
+import static com.google.common.collect.Iterables.getOnlyElement;
+import static java.lang.String.format;
+import static java.util.Locale.ENGLISH;
+import static java.util.stream.Collectors.joining;
+
+public class ExcludeColumns
+ implements Provider
+{
+ public static final String NAME = "exclude_columns";
+
+ @Override
+ public ConnectorTableFunction get()
+ {
+ return new ExcludeColumnsFunction();
+ }
+
+ public static class ExcludeColumnsFunction
+ extends AbstractConnectorTableFunction
+ {
+ private static final String TABLE_ARGUMENT_NAME = "INPUT";
+ private static final String DESCRIPTOR_ARGUMENT_NAME = "COLUMNS";
+
+ public ExcludeColumnsFunction()
+ {
+ super(
+ "builtin",
+ NAME,
+ ImmutableList.of(
+ TableArgumentSpecification.builder()
+ .name(TABLE_ARGUMENT_NAME)
+ .rowSemantics()
+ .build(),
+ DescriptorArgumentSpecification.builder()
+ .name(DESCRIPTOR_ARGUMENT_NAME)
+ .build()),
+ GENERIC_TABLE);
+ }
+
+ @Override
+ public TableFunctionAnalysis analyze(ConnectorSession session, ConnectorTransactionHandle transaction, Map arguments)
+ {
+ DescriptorArgument excludedColumns = (DescriptorArgument) arguments.get(DESCRIPTOR_ARGUMENT_NAME);
+ if (excludedColumns.equals(NULL_DESCRIPTOR)) {
+ throw new PrestoException(INVALID_FUNCTION_ARGUMENT, "COLUMNS descriptor is null");
+ }
+ Descriptor excludedColumnsDescriptor = excludedColumns.getDescriptor().orElseThrow(() -> new PrestoException(INVALID_ARGUMENTS, "Missing exclude columns descriptor"));
+ if (excludedColumnsDescriptor.getFields().stream().anyMatch(field -> field.getType().isPresent())) {
+ throw new PrestoException(INVALID_FUNCTION_ARGUMENT, "COLUMNS descriptor contains types");
+ }
+
+ // column names in DescriptorArgument are canonical wrt SQL identifier semantics.
+ // column names in TableArgument are not canonical wrt SQL identifier semantics, as they are taken from the corresponding RelationType.
+ // because of that, we match the excluded columns names case-insensitive
+ // TODO: apply proper identifier semantics
+ Set excludedNames = excludedColumnsDescriptor.getFields().stream()
+ .map(Descriptor.Field::getName)
+ .map(name -> name.orElseThrow(() -> new PrestoException(INVALID_ARGUMENTS, "Missing Descriptor field name")).toLowerCase(ENGLISH))
+ .collect(toImmutableSet());
+
+ List inputSchema = ((TableArgument) arguments.get(TABLE_ARGUMENT_NAME)).getRowType().getFields();
+ Set inputNames = inputSchema.stream()
+ .map(RowType.Field::getName)
+ .filter(Optional::isPresent)
+ .map(Optional::get)
+ .map(name -> name.toLowerCase(ENGLISH))
+ .collect(toImmutableSet());
+
+ if (!inputNames.containsAll(excludedNames)) {
+ String missingColumns = Sets.difference(excludedNames, inputNames).stream()
+ .collect(joining(", ", "[", "]"));
+ throw new PrestoException(INVALID_FUNCTION_ARGUMENT, format("Excluded columns: %s not present in the table", missingColumns));
+ }
+
+ ImmutableList.Builder requiredColumns = ImmutableList.builder();
+ ImmutableList.Builder returnedColumns = ImmutableList.builder();
+
+ for (int i = 0; i < inputSchema.size(); i++) {
+ Optional name = inputSchema.get(i).getName();
+ if (!name.isPresent() || !excludedNames.contains(name.orElseThrow(() -> new PrestoException(INVALID_FUNCTION_ARGUMENT, "Missing schema name")).toLowerCase(ENGLISH))) {
+ requiredColumns.add(i);
+ // per SQL standard, all columns produced by a table function must be named. We allow anonymous columns.
+ returnedColumns.add(new Descriptor.Field(name, Optional.of(inputSchema.get(i).getType())));
+ }
+ }
+
+ List returnedType = returnedColumns.build();
+ if (returnedType.isEmpty()) {
+ throw new PrestoException(INVALID_FUNCTION_ARGUMENT, "All columns are excluded");
+ }
+
+ return TableFunctionAnalysis.builder()
+ .requiredColumns(TABLE_ARGUMENT_NAME, requiredColumns.build())
+ .returnedType(new Descriptor(returnedType))
+ .handle(new ExcludeColumnsFunctionHandle())
+ .build();
+ }
+ }
+
+ public static TableFunctionProcessorProvider getExcludeColumnsFunctionProcessorProvider()
+ {
+ return new TableFunctionProcessorProvider()
+ {
+ @Override
+ public TableFunctionDataProcessor getDataProcessor(ConnectorTableFunctionHandle handle)
+ {
+ return input -> {
+ if (input == null) {
+ return FINISHED;
+ }
+ return usedInputAndProduced(getOnlyElement(input).orElseThrow(() -> new PrestoException(INVALID_ARGUMENTS, "Missing data processor input")));
+ };
+ }
+ };
+ }
+
+ public static class ExcludeColumnsFunctionHandle
+ implements ConnectorTableFunctionHandle
+ {
+ // there's no information to remember. All logic is effectively delegated to the engine via `requiredColumns`.
+ }
+}
diff --git a/presto-main-base/src/main/java/com/facebook/presto/operator/table/Sequence.java b/presto-main-base/src/main/java/com/facebook/presto/operator/table/Sequence.java
new file mode 100644
index 0000000000000..d992ff9993316
--- /dev/null
+++ b/presto-main-base/src/main/java/com/facebook/presto/operator/table/Sequence.java
@@ -0,0 +1,332 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.facebook.presto.operator.table;
+
+import com.facebook.presto.common.PageBuilder;
+import com.facebook.presto.common.block.BlockBuilder;
+import com.facebook.presto.spi.ConnectorSession;
+import com.facebook.presto.spi.ConnectorSplit;
+import com.facebook.presto.spi.ConnectorSplitSource;
+import com.facebook.presto.spi.FixedSplitSource;
+import com.facebook.presto.spi.HostAddress;
+import com.facebook.presto.spi.NodeManager;
+import com.facebook.presto.spi.NodeProvider;
+import com.facebook.presto.spi.PrestoException;
+import com.facebook.presto.spi.connector.ConnectorTransactionHandle;
+import com.facebook.presto.spi.function.table.AbstractConnectorTableFunction;
+import com.facebook.presto.spi.function.table.Argument;
+import com.facebook.presto.spi.function.table.ConnectorTableFunction;
+import com.facebook.presto.spi.function.table.ConnectorTableFunctionHandle;
+import com.facebook.presto.spi.function.table.DescribedTableReturnTypeSpecification;
+import com.facebook.presto.spi.function.table.ScalarArgument;
+import com.facebook.presto.spi.function.table.ScalarArgumentSpecification;
+import com.facebook.presto.spi.function.table.TableFunctionAnalysis;
+import com.facebook.presto.spi.function.table.TableFunctionProcessorProvider;
+import com.facebook.presto.spi.function.table.TableFunctionProcessorState;
+import com.facebook.presto.spi.function.table.TableFunctionSplitProcessor;
+import com.facebook.presto.spi.schedule.NodeSelectionStrategy;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+
+import javax.inject.Provider;
+
+import java.math.BigInteger;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static com.facebook.presto.common.type.BigintType.BIGINT;
+import static com.facebook.presto.operator.table.Sequence.SequenceFunctionSplit.MAX_SPLIT_SIZE;
+import static com.facebook.presto.spi.StandardErrorCode.INVALID_FUNCTION_ARGUMENT;
+import static com.facebook.presto.spi.function.table.Descriptor.descriptor;
+import static com.facebook.presto.spi.function.table.TableFunctionProcessorState.Finished.FINISHED;
+import static com.facebook.presto.spi.function.table.TableFunctionProcessorState.Processed.produced;
+import static com.facebook.presto.spi.function.table.TableFunctionProcessorState.Processed.usedInputAndProduced;
+import static com.facebook.presto.spi.schedule.NodeSelectionStrategy.NO_PREFERENCE;
+import static com.google.common.base.Preconditions.checkArgument;
+import static java.lang.String.format;
+
+public class Sequence
+ implements Provider
+{
+ public static final String NAME = "sequence";
+
+ @Override
+ public ConnectorTableFunction get()
+ {
+ return new SequenceFunction();
+ }
+
+ public static class SequenceFunction
+ extends AbstractConnectorTableFunction
+ {
+ private static final String START_ARGUMENT_NAME = "START";
+ private static final String STOP_ARGUMENT_NAME = "STOP";
+ private static final String STEP_ARGUMENT_NAME = "STEP";
+
+ public SequenceFunction()
+ {
+ super(
+ "builtin",
+ NAME,
+ ImmutableList.of(
+ ScalarArgumentSpecification.builder()
+ .name(START_ARGUMENT_NAME)
+ .type(BIGINT)
+ .defaultValue(0L)
+ .build(),
+ ScalarArgumentSpecification.builder()
+ .name(STOP_ARGUMENT_NAME)
+ .type(BIGINT)
+ .build(),
+ ScalarArgumentSpecification.builder()
+ .name(STEP_ARGUMENT_NAME)
+ .type(BIGINT)
+ .defaultValue(1L)
+ .build()),
+ new DescribedTableReturnTypeSpecification(descriptor(ImmutableList.of("sequential_number"), ImmutableList.of(BIGINT))));
+ }
+
+ @Override
+ public TableFunctionAnalysis analyze(ConnectorSession session, ConnectorTransactionHandle transaction, Map arguments)
+ {
+ Object startValue = ((ScalarArgument) arguments.get(START_ARGUMENT_NAME)).getValue();
+ if (startValue == null) {
+ throw new PrestoException(INVALID_FUNCTION_ARGUMENT, "Start is null");
+ }
+
+ Object stopValue = ((ScalarArgument) arguments.get(STOP_ARGUMENT_NAME)).getValue();
+ if (stopValue == null) {
+ throw new PrestoException(INVALID_FUNCTION_ARGUMENT, "Stop is null");
+ }
+
+ Object stepValue = ((ScalarArgument) arguments.get(STEP_ARGUMENT_NAME)).getValue();
+ if (stepValue == null) {
+ throw new PrestoException(INVALID_FUNCTION_ARGUMENT, "Step is null");
+ }
+
+ long start = (long) startValue;
+ long stop = (long) stopValue;
+ long step = (long) stepValue;
+
+ if (start < stop && step <= 0) {
+ throw new PrestoException(INVALID_FUNCTION_ARGUMENT, format("Step must be positive for sequence [%s, %s]", start, stop));
+ }
+
+ if (start > stop && step >= 0) {
+ throw new PrestoException(INVALID_FUNCTION_ARGUMENT, format("Step must be negative for sequence [%s, %s]", start, stop));
+ }
+
+ return TableFunctionAnalysis.builder()
+ .handle(new SequenceFunctionHandle(start, stop, start == stop ? 0 : step))
+ .build();
+ }
+ }
+
+ public static class SequenceFunctionHandle
+ implements ConnectorTableFunctionHandle
+ {
+ private final long start;
+ private final long stop;
+ private final long step;
+
+ @JsonCreator
+ public SequenceFunctionHandle(@JsonProperty("start") long start, @JsonProperty("stop") long stop, @JsonProperty("step") long step)
+ {
+ this.start = start;
+ this.stop = stop;
+ this.step = step;
+ }
+
+ @JsonProperty
+ public long start()
+ {
+ return start;
+ }
+
+ @JsonProperty
+ public long stop()
+ {
+ return stop;
+ }
+
+ @JsonProperty
+ public long step()
+ {
+ return step;
+ }
+
+ @Override
+ public ConnectorSplitSource getSplits(ConnectorTransactionHandle transaction, ConnectorSession session, NodeManager nodeManager, Object functionAndTypeManager)
+ {
+ return getSequenceFunctionSplitSource(this);
+ }
+ }
+
+ public static ConnectorSplitSource getSequenceFunctionSplitSource(SequenceFunctionHandle handle)
+ {
+ // using BigInteger to avoid long overflow since it's not in the main data processing loop
+ BigInteger start = BigInteger.valueOf(handle.start());
+ BigInteger stop = BigInteger.valueOf(handle.stop());
+ BigInteger step = BigInteger.valueOf(handle.step());
+
+ if (step.equals(BigInteger.ZERO)) {
+ checkArgument(start.equals(stop), "start is not equal to stop for step = 0");
+ return new FixedSplitSource(ImmutableList.of(new SequenceFunctionSplit(start.longValueExact(), stop.longValueExact())));
+ }
+
+ ImmutableList.Builder splits = ImmutableList.builder();
+
+ BigInteger totalSteps = stop.subtract(start).divide(step).add(BigInteger.ONE);
+ BigInteger totalSplits = totalSteps.divide(BigInteger.valueOf(MAX_SPLIT_SIZE)).add(BigInteger.ONE);
+ BigInteger[] stepsPerSplit = totalSteps.divideAndRemainder(totalSplits);
+ BigInteger splitJump = stepsPerSplit[0].subtract(BigInteger.ONE).multiply(step);
+
+ BigInteger splitStart = start;
+ for (BigInteger i = BigInteger.ZERO; i.compareTo(totalSplits) < 0; i = i.add(BigInteger.ONE)) {
+ BigInteger splitStop = splitStart.add(splitJump);
+ // distribute the remaining steps between the initial splits, one step per split
+ if (i.compareTo(stepsPerSplit[1]) < 0) {
+ splitStop = splitStop.add(step);
+ }
+ splits.add(new SequenceFunctionSplit(splitStart.longValueExact(), splitStop.longValueExact()));
+ splitStart = splitStop.add(step);
+ }
+
+ return new FixedSplitSource(splits.build());
+ }
+
+ public static class SequenceFunctionSplit
+ implements ConnectorSplit
+ {
+ public static final int DEFAULT_SPLIT_SIZE = 1000000;
+ public static final int MAX_SPLIT_SIZE = 1000000;
+
+ // the first value of sub-sequence
+ private final long start;
+
+ // the last value of sub-sequence. this value is aligned so that it belongs to the sequence.
+ private final long stop;
+
+ @JsonCreator
+ public SequenceFunctionSplit(@JsonProperty("start") long start, @JsonProperty("stop") long stop)
+ {
+ this.start = start;
+ this.stop = stop;
+ }
+
+ @JsonProperty
+ public long getStart()
+ {
+ return start;
+ }
+
+ @JsonProperty
+ public long getStop()
+ {
+ return stop;
+ }
+
+ @Override
+ public NodeSelectionStrategy getNodeSelectionStrategy()
+ {
+ return NO_PREFERENCE;
+ }
+
+ @Override
+ public List getPreferredNodes(NodeProvider nodeProvider)
+ {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public Object getInfo()
+ {
+ return ImmutableMap.builder()
+ .put("start", start)
+ .put("stop", stop)
+ .buildOrThrow();
+ }
+ }
+
+ public static TableFunctionProcessorProvider getSequenceFunctionProcessorProvider()
+ {
+ return new TableFunctionProcessorProvider() {
+ @Override
+ public TableFunctionSplitProcessor getSplitProcessor(ConnectorTableFunctionHandle handle)
+ {
+ return new SequenceFunctionProcessor(((SequenceFunctionHandle) handle).step());
+ }
+ };
+ }
+
+ public static class SequenceFunctionProcessor
+ implements TableFunctionSplitProcessor
+ {
+ private final PageBuilder page = new PageBuilder(ImmutableList.of(BIGINT));
+ private final long step;
+ private long start;
+ private long stop;
+ private boolean finished;
+
+ public SequenceFunctionProcessor(long step)
+ {
+ this.step = step;
+ }
+
+ @Override
+ public TableFunctionProcessorState process(ConnectorSplit split)
+ {
+ if (split != null) {
+ SequenceFunctionSplit sequenceSplit = (SequenceFunctionSplit) split;
+ start = sequenceSplit.getStart();
+ stop = sequenceSplit.getStop();
+ BlockBuilder block = page.getBlockBuilder(0);
+ while (start != stop && !page.isFull()) {
+ page.declarePosition();
+ BIGINT.writeLong(block, start);
+ start += step;
+ }
+ if (!page.isFull()) {
+ page.declarePosition();
+ BIGINT.writeLong(block, start);
+ finished = true;
+ return usedInputAndProduced(page.build());
+ }
+ return usedInputAndProduced(page.build());
+ }
+
+ if (finished) {
+ return FINISHED;
+ }
+
+ page.reset();
+ BlockBuilder block = page.getBlockBuilder(0);
+ while (start != stop && !page.isFull()) {
+ page.declarePosition();
+ BIGINT.writeLong(block, start);
+ start += step;
+ }
+ if (!page.isFull()) {
+ page.declarePosition();
+ BIGINT.writeLong(block, start);
+ finished = true;
+ return produced(page.build());
+ }
+ return produced(page.build());
+ }
+ }
+}
diff --git a/presto-main-base/src/main/java/com/facebook/presto/server/PluginManager.java b/presto-main-base/src/main/java/com/facebook/presto/server/PluginManager.java
index d072762d14dbc..34cb2d7821b13 100644
--- a/presto-main-base/src/main/java/com/facebook/presto/server/PluginManager.java
+++ b/presto-main-base/src/main/java/com/facebook/presto/server/PluginManager.java
@@ -52,6 +52,7 @@
import com.facebook.presto.spi.tracing.TracerProvider;
import com.facebook.presto.spi.ttl.ClusterTtlProviderFactory;
import com.facebook.presto.spi.ttl.NodeTtlFetcherFactory;
+import com.facebook.presto.spi.tvf.TVFProviderFactory;
import com.facebook.presto.spi.type.TypeManagerFactory;
import com.facebook.presto.sql.analyzer.AnalyzerProviderManager;
import com.facebook.presto.sql.analyzer.QueryPreparerProviderManager;
@@ -340,6 +341,11 @@ public void installCoordinatorPlugin(CoordinatorPlugin plugin)
log.info("Registering type manager factory %s", typeManagerFactory.getName());
metadata.getFunctionAndTypeManager().addTypeManagerFactory(typeManagerFactory);
}
+
+ for (TVFProviderFactory tvfProviderFactory : plugin.getTVFProviderFactories()) {
+ log.info("Registering table functions provider factory %s", tvfProviderFactory.getName());
+ metadata.getFunctionAndTypeManager().addTVFProviderFactory(tvfProviderFactory);
+ }
}
private class MainPluginInstaller
diff --git a/presto-main-base/src/main/java/com/facebook/presto/split/CloseableSplitSourceProvider.java b/presto-main-base/src/main/java/com/facebook/presto/split/CloseableSplitSourceProvider.java
index 8bcd62f6006ef..f7f6afc493848 100644
--- a/presto-main-base/src/main/java/com/facebook/presto/split/CloseableSplitSourceProvider.java
+++ b/presto-main-base/src/main/java/com/facebook/presto/split/CloseableSplitSourceProvider.java
@@ -15,6 +15,7 @@
import com.facebook.airlift.log.Logger;
import com.facebook.presto.Session;
+import com.facebook.presto.metadata.TableFunctionHandle;
import com.facebook.presto.spi.TableHandle;
import com.facebook.presto.spi.WarningCollector;
import com.facebook.presto.spi.connector.ConnectorSplitManager.SplitSchedulingStrategy;
@@ -32,14 +33,14 @@ public class CloseableSplitSourceProvider
{
private static final Logger log = Logger.get(CloseableSplitSourceProvider.class);
- private final SplitSourceProvider delegate;
+ private final SplitManager delegate;
@GuardedBy("this")
private List splitSources = new ArrayList<>();
@GuardedBy("this")
private boolean closed;
- public CloseableSplitSourceProvider(SplitSourceProvider delegate)
+ public CloseableSplitSourceProvider(SplitManager delegate)
{
this.delegate = requireNonNull(delegate, "delegate is null");
}
@@ -53,6 +54,15 @@ public synchronized SplitSource getSplits(Session session, TableHandle tableHand
return splitSource;
}
+ @Override
+ public synchronized SplitSource getSplits(Session session, TableFunctionHandle tableFunctionHandle)
+ {
+ checkState(!closed, "split source provider is closed");
+ SplitSource splitSource = delegate.getSplitsForTableFunction(session, tableFunctionHandle);
+ splitSources.add(splitSource);
+ return splitSource;
+ }
+
@Override
public synchronized void close()
{
diff --git a/presto-main-base/src/main/java/com/facebook/presto/split/SplitManager.java b/presto-main-base/src/main/java/com/facebook/presto/split/SplitManager.java
index 8cd7efc5e2f9b..d334e5a5e6a70 100644
--- a/presto-main-base/src/main/java/com/facebook/presto/split/SplitManager.java
+++ b/presto-main-base/src/main/java/com/facebook/presto/split/SplitManager.java
@@ -18,6 +18,7 @@
import com.facebook.presto.execution.scheduler.NodeSchedulerConfig;
import com.facebook.presto.metadata.Metadata;
import com.facebook.presto.metadata.MetadataManager;
+import com.facebook.presto.metadata.TableFunctionHandle;
import com.facebook.presto.metadata.TableLayoutResult;
import com.facebook.presto.spi.ConnectorId;
import com.facebook.presto.spi.ConnectorSession;
@@ -104,4 +105,17 @@ private ConnectorSplitManager getConnectorSplitManager(ConnectorId connectorId)
checkArgument(result != null, "No split manager for connector '%s'", connectorId);
return result;
}
+
+ public SplitSource getSplitsForTableFunction(Session session, TableFunctionHandle function)
+ {
+ ConnectorId connectorId = function.getConnectorId();
+ ConnectorSplitManager splitManager = splitManagers.get(connectorId);
+
+ ConnectorSplitSource source = splitManager.getSplits(
+ function.getTransactionHandle(),
+ session.toConnectorSession(connectorId),
+ function.getFunctionHandle());
+
+ return new ConnectorAwareSplitSource(connectorId, function.getTransactionHandle(), source);
+ }
}
diff --git a/presto-main-base/src/main/java/com/facebook/presto/split/SplitSourceProvider.java b/presto-main-base/src/main/java/com/facebook/presto/split/SplitSourceProvider.java
index 617fba7093613..30b54174c27b6 100644
--- a/presto-main-base/src/main/java/com/facebook/presto/split/SplitSourceProvider.java
+++ b/presto-main-base/src/main/java/com/facebook/presto/split/SplitSourceProvider.java
@@ -14,6 +14,7 @@
package com.facebook.presto.split;
import com.facebook.presto.Session;
+import com.facebook.presto.metadata.TableFunctionHandle;
import com.facebook.presto.spi.TableHandle;
import com.facebook.presto.spi.WarningCollector;
import com.facebook.presto.spi.connector.ConnectorSplitManager.SplitSchedulingStrategy;
@@ -21,4 +22,5 @@
public interface SplitSourceProvider
{
SplitSource getSplits(Session session, TableHandle tableHandle, SplitSchedulingStrategy splitSchedulingStrategy, WarningCollector warningCollector);
+ SplitSource getSplits(Session session, TableFunctionHandle tableFunctionHandle);
}
diff --git a/presto-main-base/src/main/java/com/facebook/presto/sql/analyzer/StatementAnalyzer.java b/presto-main-base/src/main/java/com/facebook/presto/sql/analyzer/StatementAnalyzer.java
index fc383fc928d23..2deb6dc592da2 100644
--- a/presto-main-base/src/main/java/com/facebook/presto/sql/analyzer/StatementAnalyzer.java
+++ b/presto-main-base/src/main/java/com/facebook/presto/sql/analyzer/StatementAnalyzer.java
@@ -36,7 +36,6 @@
import com.facebook.presto.metadata.FunctionAndTypeManager;
import com.facebook.presto.metadata.Metadata;
import com.facebook.presto.metadata.OperatorNotFoundException;
-import com.facebook.presto.metadata.TableFunctionMetadata;
import com.facebook.presto.spi.ColumnHandle;
import com.facebook.presto.spi.ColumnMetadata;
import com.facebook.presto.spi.ConnectorId;
@@ -62,6 +61,7 @@
import com.facebook.presto.spi.function.table.Argument;
import com.facebook.presto.spi.function.table.ArgumentSpecification;
import com.facebook.presto.spi.function.table.ConnectorTableFunction;
+import com.facebook.presto.spi.function.table.DescribedTableReturnTypeSpecification;
import com.facebook.presto.spi.function.table.Descriptor;
import com.facebook.presto.spi.function.table.DescriptorArgument;
import com.facebook.presto.spi.function.table.DescriptorArgumentSpecification;
@@ -71,6 +71,7 @@
import com.facebook.presto.spi.function.table.TableArgument;
import com.facebook.presto.spi.function.table.TableArgumentSpecification;
import com.facebook.presto.spi.function.table.TableFunctionAnalysis;
+import com.facebook.presto.spi.function.table.TableFunctionMetadata;
import com.facebook.presto.spi.procedure.DistributedProcedure;
import com.facebook.presto.spi.procedure.TableDataRewriteDistributedProcedure;
import com.facebook.presto.spi.relation.DomainTranslator;
@@ -265,7 +266,7 @@
import static com.facebook.presto.spi.function.FunctionKind.AGGREGATE;
import static com.facebook.presto.spi.function.FunctionKind.WINDOW;
import static com.facebook.presto.spi.function.table.DescriptorArgument.NULL_DESCRIPTOR;
-import static com.facebook.presto.spi.function.table.ReturnTypeSpecification.GenericTable.GENERIC_TABLE;
+import static com.facebook.presto.spi.function.table.GenericTableReturnTypeSpecification.GENERIC_TABLE;
import static com.facebook.presto.spi.security.ViewSecurity.DEFINER;
import static com.facebook.presto.spi.security.ViewSecurity.INVOKER;
import static com.facebook.presto.sql.MaterializedViewUtils.buildOwnerSession;
@@ -1507,14 +1508,10 @@ protected Scope visitLateral(Lateral node, Optional scope)
@Override
protected Scope visitTableFunctionInvocation(TableFunctionInvocation node, Optional scope)
{
- TableFunctionMetadata tableFunctionMetadata = metadata.getFunctionAndTypeManager()
- .getTableFunctionRegistry()
- .resolve(session, node.getName())
- .orElseThrow(() -> new SemanticException(
- FUNCTION_NOT_FOUND,
- node,
- "Table function %s not registered",
- node.getName()));
+ TableFunctionMetadata tableFunctionMetadata = metadata.getFunctionAndTypeManager().resolveTableFunction(session, node.getName());
+ if (tableFunctionMetadata == null) {
+ throw new SemanticException(FUNCTION_NOT_FOUND, node, "Table function %s not registered", node.getName());
+ }
ConnectorTableFunction function = tableFunctionMetadata.getFunction();
ConnectorId connectorId = tableFunctionMetadata.getConnectorId();
@@ -1608,7 +1605,7 @@ private void verifyRequiredColumns(TableFunctionInvocation node, Map column < 0 || column >= inputScope.getRelationType().getAllFieldCount()) // hidden columns can be required as well as visible columns
+ .filter(column -> column < 0 || column >= inputScope.getRelationType().getVisibleFieldCount())
.findFirst()
.ifPresent(column -> {
throw new SemanticException(TABLE_FUNCTION_IMPLEMENTATION_ERROR, "Invalid index: %s of required column from table argument %s", column, name);
@@ -1626,7 +1623,7 @@ private void verifyRequiredColumns(TableFunctionInvocation node, Map analyzedProperColumnsDescriptor)
{
switch (returnTypeSpecification.getReturnType()) {
- case ReturnTypeSpecification.OnlyPassThrough.returnType:
+ case "PASSTHROUGH":
if (analysis.isAliased(node)) {
// According to SQL standard ISO/IEC 9075-2, 7.6 , p. 409,
// table alias is prohibited for a table function with ONLY PASS THROUGH returned type.
@@ -1648,7 +1645,7 @@ private Descriptor verifyProperColumnsDescriptor(TableFunctionInvocation node, C
throw new SemanticException(TABLE_FUNCTION_IMPLEMENTATION_ERROR, "A table function with ONLY_PASS_THROUGH return type must have a table argument with pass-through columns.");
}
return null;
- case ReturnTypeSpecification.GenericTable.returnType:
+ case "GENERIC":
// According to SQL standard ISO/IEC 9075-2, 7.6 , p. 409,
// table alias is mandatory for a polymorphic table function invocation which produces proper columns.
// We don't enforce this requirement.
@@ -1664,7 +1661,7 @@ private Descriptor verifyProperColumnsDescriptor(TableFunctionInvocation node, C
// so the function's analyze() method should not return the proper columns descriptor.
throw new SemanticException(TABLE_FUNCTION_AMBIGUOUS_RETURN_TYPE, node, "Returned relation type for table function %s is ambiguous", node.getName());
}
- return ((ReturnTypeSpecification.DescribedTable) returnTypeSpecification).getDescriptor();
+ return ((DescribedTableReturnTypeSpecification) returnTypeSpecification).getDescriptor();
}
}
diff --git a/presto-main-base/src/main/java/com/facebook/presto/sql/planner/BasePlanFragmenter.java b/presto-main-base/src/main/java/com/facebook/presto/sql/planner/BasePlanFragmenter.java
index 7683f51b60973..2326c62b56dfd 100644
--- a/presto-main-base/src/main/java/com/facebook/presto/sql/planner/BasePlanFragmenter.java
+++ b/presto-main-base/src/main/java/com/facebook/presto/sql/planner/BasePlanFragmenter.java
@@ -51,6 +51,8 @@
import com.facebook.presto.sql.planner.plan.SequenceNode;
import com.facebook.presto.sql.planner.plan.SimplePlanRewriter;
import com.facebook.presto.sql.planner.plan.StatisticsWriterNode;
+import com.facebook.presto.sql.planner.plan.TableFunctionNode;
+import com.facebook.presto.sql.planner.plan.TableFunctionProcessorNode;
import com.facebook.presto.sql.planner.sanity.PlanChecker;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
@@ -296,6 +298,22 @@ public PlanNode visitValues(ValuesNode node, RewriteContext
return context.defaultRewrite(node, context.get());
}
+ @Override
+ public PlanNode visitTableFunction(TableFunctionNode node, RewriteContext context)
+ {
+ throw new IllegalStateException(format("Unexpected node: TableFunctionNode (%s)", node.getName()));
+ }
+
+ @Override
+ public PlanNode visitTableFunctionProcessor(TableFunctionProcessorNode node, RewriteContext context)
+ {
+ if (!node.getSource().isPresent()) {
+ // context is mutable. The leaf node should set the PartitioningHandle.
+ context.get().addSourceDistribution(node.getId(), SOURCE_DISTRIBUTION, metadata, session);
+ }
+ return context.defaultRewrite(node, context.get());
+ }
+
@Override
public PlanNode visitExchange(ExchangeNode exchange, RewriteContext context)
{
diff --git a/presto-main-base/src/main/java/com/facebook/presto/sql/planner/LocalExecutionPlanner.java b/presto-main-base/src/main/java/com/facebook/presto/sql/planner/LocalExecutionPlanner.java
index 878f46dc04fcb..13cc38caef3fe 100644
--- a/presto-main-base/src/main/java/com/facebook/presto/sql/planner/LocalExecutionPlanner.java
+++ b/presto-main-base/src/main/java/com/facebook/presto/sql/planner/LocalExecutionPlanner.java
@@ -69,6 +69,7 @@
import com.facebook.presto.operator.JoinBridgeManager;
import com.facebook.presto.operator.JoinOperatorFactory;
import com.facebook.presto.operator.JoinOperatorFactory.OuterOperatorFactoryResult;
+import com.facebook.presto.operator.LeafTableFunctionOperator;
import com.facebook.presto.operator.LimitOperator.LimitOperatorFactory;
import com.facebook.presto.operator.LocalPlannerAware;
import com.facebook.presto.operator.LookupJoinOperators;
@@ -89,6 +90,7 @@
import com.facebook.presto.operator.PartitionFunction;
import com.facebook.presto.operator.PartitionedLookupSourceFactory;
import com.facebook.presto.operator.PipelineExecutionStrategy;
+import com.facebook.presto.operator.RegularTableFunctionPartition;
import com.facebook.presto.operator.RemoteProjectOperator.RemoteProjectOperatorFactory;
import com.facebook.presto.operator.RowNumberOperator;
import com.facebook.presto.operator.ScanFilterAndProjectOperator.ScanFilterAndProjectOperatorFactory;
@@ -102,6 +104,7 @@
import com.facebook.presto.operator.StreamingAggregationOperator.StreamingAggregationOperatorFactory;
import com.facebook.presto.operator.TableCommitContext;
import com.facebook.presto.operator.TableFinishOperator.PageSinkCommitter;
+import com.facebook.presto.operator.TableFunctionOperator;
import com.facebook.presto.operator.TableScanOperator.TableScanOperatorFactory;
import com.facebook.presto.operator.TableWriterMergeOperator.TableWriterMergeOperatorFactory;
import com.facebook.presto.operator.TaskContext;
@@ -145,11 +148,13 @@
import com.facebook.presto.spi.function.SqlFunctionId;
import com.facebook.presto.spi.function.SqlInvokedFunction;
import com.facebook.presto.spi.function.aggregation.LambdaProvider;
+import com.facebook.presto.spi.function.table.TableFunctionProcessorProvider;
import com.facebook.presto.spi.plan.AbstractJoinNode;
import com.facebook.presto.spi.plan.AggregationNode;
import com.facebook.presto.spi.plan.AggregationNode.Aggregation;
import com.facebook.presto.spi.plan.AggregationNode.Step;
import com.facebook.presto.spi.plan.Assignments;
+import com.facebook.presto.spi.plan.DataOrganizationSpecification;
import com.facebook.presto.spi.plan.DeleteNode;
import com.facebook.presto.spi.plan.DistinctLimitNode;
import com.facebook.presto.spi.plan.EquiJoinClause;
@@ -218,6 +223,7 @@
import com.facebook.presto.sql.planner.plan.SampleNode;
import com.facebook.presto.sql.planner.plan.StatisticsWriterNode;
import com.facebook.presto.sql.planner.plan.TableFunctionNode;
+import com.facebook.presto.sql.planner.plan.TableFunctionProcessorNode;
import com.facebook.presto.sql.planner.plan.TableWriterMergeNode;
import com.facebook.presto.sql.planner.plan.TopNRowNumberNode;
import com.facebook.presto.sql.planner.plan.UpdateNode;
@@ -250,6 +256,7 @@
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.NoSuchElementException;
import java.util.Optional;
import java.util.OptionalInt;
import java.util.Set;
@@ -358,9 +365,11 @@
import static com.google.common.base.Verify.verify;
import static com.google.common.collect.DiscreteDomain.integers;
import static com.google.common.collect.ImmutableList.toImmutableList;
+import static com.google.common.collect.ImmutableMap.toImmutableMap;
import static com.google.common.collect.ImmutableSet.toImmutableSet;
import static com.google.common.collect.Iterables.getOnlyElement;
import static com.google.common.collect.Range.closedOpen;
+import static java.lang.Math.toIntExact;
import static java.lang.String.format;
import static java.util.Objects.requireNonNull;
import static java.util.stream.IntStream.range;
@@ -1218,6 +1227,92 @@ public PhysicalOperation visitTableFunction(TableFunctionNode node, LocalExecuti
throw new UnsupportedOperationException("execution by operator is not yet implemented for table function " + node.getName());
}
+ @Override
+ public PhysicalOperation visitTableFunctionProcessor(TableFunctionProcessorNode node, LocalExecutionPlanContext context)
+ {
+ TableFunctionProcessorProvider processorProvider = metadata.getFunctionAndTypeManager().getTableFunctionProcessorProvider(node.getHandle());
+
+ if (!node.getSource().isPresent()) {
+ OperatorFactory operatorFactory = new LeafTableFunctionOperator.LeafTableFunctionOperatorFactory(context.getNextOperatorId(), node.getId(), processorProvider, node.getHandle().getFunctionHandle());
+ return new PhysicalOperation(operatorFactory, makeLayout(node), context, Optional.empty(), UNGROUPED_EXECUTION);
+ }
+
+ PhysicalOperation source = node.getSource().orElseThrow(NoSuchElementException::new).accept(this, context);
+
+ int properChannelsCount = node.getProperOutputs().size();
+
+ long passThroughSourcesCount = node.getPassThroughSpecifications().stream()
+ .filter(TableFunctionNode.PassThroughSpecification::isDeclaredAsPassThrough)
+ .count();
+
+ List> requiredChannels = node.getRequiredVariables().stream()
+ .map(list -> getChannelsForVariables(list, source.getLayout()))
+ .collect(toImmutableList());
+
+ Optional> markerChannels = node.getMarkerVariables()
+ .map(map -> map.entrySet().stream()
+ .collect(toImmutableMap(entry -> source.getLayout().get(entry.getKey()), entry -> source.getLayout().get(entry.getValue()))));
+
+ int channel = properChannelsCount;
+ ImmutableList.Builder passThroughColumnSpecifications = ImmutableList.builder();
+ for (TableFunctionNode.PassThroughSpecification specification : node.getPassThroughSpecifications()) {
+ // the table function produces one index channel for each source declared as pass-through. They are laid out after the proper channels.
+ int indexChannel = specification.isDeclaredAsPassThrough() ? channel++ : -1;
+ for (TableFunctionNode.PassThroughColumn column : specification.getColumns()) {
+ passThroughColumnSpecifications.add(new RegularTableFunctionPartition.PassThroughColumnSpecification(column.isPartitioningColumn(), source.getLayout().get(column.getOutputVariables()), indexChannel));
+ }
+ }
+
+ List partitionChannels = node.getSpecification()
+ .map(DataOrganizationSpecification::getPartitionBy)
+ .map(list -> getChannelsForVariables(list, source.getLayout()))
+ .orElse(ImmutableList.of());
+
+ List sortChannels = ImmutableList.of();
+ List sortOrders = ImmutableList.of();
+ if (node.getSpecification().flatMap(DataOrganizationSpecification::getOrderingScheme).isPresent()) {
+ OrderingScheme orderingScheme = node.getSpecification().flatMap(DataOrganizationSpecification::getOrderingScheme).orElseThrow(NoSuchElementException::new);
+ sortChannels = getChannelsForVariables(orderingScheme.getOrderByVariables(), source.getLayout());
+ sortOrders = orderingScheme.getOrderingsMap().values().stream().collect(toImmutableList());
+ }
+
+ OperatorFactory operator = new TableFunctionOperator.TableFunctionOperatorFactory(
+ context.getNextOperatorId(),
+ node.getId(),
+ processorProvider,
+ node.getHandle().getFunctionHandle(),
+ properChannelsCount,
+ toIntExact(passThroughSourcesCount),
+ requiredChannels,
+ markerChannels,
+ passThroughColumnSpecifications.build(),
+ node.isPruneWhenEmpty(),
+ partitionChannels,
+ getChannelsForVariables(ImmutableList.copyOf(node.getPrePartitioned()), source.getLayout()),
+ sortChannels,
+ sortOrders,
+ node.getPreSorted(),
+ source.getTypes(),
+ 10_000,
+ pagesIndexFactory);
+
+ ImmutableMap.Builder outputMappings = ImmutableMap.builder();
+ for (int i = 0; i < node.getProperOutputs().size(); i++) {
+ outputMappings.put(node.getProperOutputs().get(i), i);
+ }
+ List passThroughVariables = node.getPassThroughSpecifications().stream()
+ .map(TableFunctionNode.PassThroughSpecification::getColumns)
+ .flatMap(Collection::stream)
+ .map(TableFunctionNode.PassThroughColumn::getOutputVariables)
+ .collect(toImmutableList());
+ int outputChannel = properChannelsCount;
+ for (VariableReferenceExpression passThroughVariable : passThroughVariables) {
+ outputMappings.put(passThroughVariable, outputChannel++);
+ }
+
+ return new PhysicalOperation(operator, outputMappings.buildOrThrow(), context, source);
+ }
+
@Override
public PhysicalOperation visitTopN(TopNNode node, LocalExecutionPlanContext context)
{
@@ -2942,7 +3037,7 @@ public PhysicalOperation visitTableFinish(TableFinishNode node, LocalExecutionPl
Map aggregationMap =
aggregation.getAggregations().entrySet()
.stream().collect(
- ImmutableMap.toImmutableMap(
+ toImmutableMap(
Map.Entry::getKey,
entry -> createAggregation(entry.getValue())));
if (groupingVariables.isEmpty()) {
diff --git a/presto-main-base/src/main/java/com/facebook/presto/sql/planner/PlanOptimizers.java b/presto-main-base/src/main/java/com/facebook/presto/sql/planner/PlanOptimizers.java
index a12b350dcdd1e..d56bdb9b82048 100644
--- a/presto-main-base/src/main/java/com/facebook/presto/sql/planner/PlanOptimizers.java
+++ b/presto-main-base/src/main/java/com/facebook/presto/sql/planner/PlanOptimizers.java
@@ -84,6 +84,8 @@
import com.facebook.presto.sql.planner.iterative.rule.PruneRedundantProjectionAssignments;
import com.facebook.presto.sql.planner.iterative.rule.PruneSemiJoinColumns;
import com.facebook.presto.sql.planner.iterative.rule.PruneSemiJoinFilteringSourceColumns;
+import com.facebook.presto.sql.planner.iterative.rule.PruneTableFunctionProcessorColumns;
+import com.facebook.presto.sql.planner.iterative.rule.PruneTableFunctionProcessorSourceColumns;
import com.facebook.presto.sql.planner.iterative.rule.PruneTableScanColumns;
import com.facebook.presto.sql.planner.iterative.rule.PruneTopNColumns;
import com.facebook.presto.sql.planner.iterative.rule.PruneUpdateSourceColumns;
@@ -123,6 +125,7 @@
import com.facebook.presto.sql.planner.iterative.rule.RemoveRedundantLimit;
import com.facebook.presto.sql.planner.iterative.rule.RemoveRedundantSort;
import com.facebook.presto.sql.planner.iterative.rule.RemoveRedundantSortColumns;
+import com.facebook.presto.sql.planner.iterative.rule.RemoveRedundantTableFunctionProcessor;
import com.facebook.presto.sql.planner.iterative.rule.RemoveRedundantTopN;
import com.facebook.presto.sql.planner.iterative.rule.RemoveRedundantTopNColumns;
import com.facebook.presto.sql.planner.iterative.rule.RemoveTrivialFilters;
@@ -137,7 +140,6 @@
import com.facebook.presto.sql.planner.iterative.rule.RewriteConstantArrayContainsToInExpression;
import com.facebook.presto.sql.planner.iterative.rule.RewriteFilterWithExternalFunctionToProject;
import com.facebook.presto.sql.planner.iterative.rule.RewriteSpatialPartitioningAggregation;
-import com.facebook.presto.sql.planner.iterative.rule.RewriteTableFunctionToTableScan;
import com.facebook.presto.sql.planner.iterative.rule.RuntimeReorderJoinSides;
import com.facebook.presto.sql.planner.iterative.rule.ScaledWriterRule;
import com.facebook.presto.sql.planner.iterative.rule.SimplifyCardinalityMap;
@@ -154,6 +156,8 @@
import com.facebook.presto.sql.planner.iterative.rule.TransformDistinctInnerJoinToLeftEarlyOutJoin;
import com.facebook.presto.sql.planner.iterative.rule.TransformDistinctInnerJoinToRightEarlyOutJoin;
import com.facebook.presto.sql.planner.iterative.rule.TransformExistsApplyToLateralNode;
+import com.facebook.presto.sql.planner.iterative.rule.TransformTableFunctionProcessorToTableScan;
+import com.facebook.presto.sql.planner.iterative.rule.TransformTableFunctionToTableFunctionProcessor;
import com.facebook.presto.sql.planner.iterative.rule.TransformUncorrelatedInPredicateSubqueryToDistinctInnerJoin;
import com.facebook.presto.sql.planner.iterative.rule.TransformUncorrelatedInPredicateSubqueryToSemiJoin;
import com.facebook.presto.sql.planner.iterative.rule.TransformUncorrelatedLateralToJoin;
@@ -319,6 +323,8 @@ public PlanOptimizers(
new PruneValuesColumns(),
new PruneWindowColumns(),
new PruneLimitColumns(),
+ new PruneTableFunctionProcessorColumns(),
+ new PruneTableFunctionProcessorSourceColumns(),
new PruneTableScanColumns());
builder.add(new LogicalCteOptimizer(metadata));
@@ -424,6 +430,7 @@ public PlanOptimizers(
.addAll(predicatePushDownRules)
.addAll(columnPruningRules)
.addAll(ImmutableSet.of(
+ new TransformTableFunctionToTableFunctionProcessor(metadata),
new MergeDuplicateAggregation(metadata.getFunctionAndTypeManager()),
new RemoveRedundantIdentityProjections(),
new RemoveFullSample(),
@@ -446,6 +453,8 @@ public PlanOptimizers(
new MergeLimitWithDistinct(),
new PruneCountAggregationOverScalar(metadata.getFunctionAndTypeManager()),
new PruneOrderByInAggregation(metadata.getFunctionAndTypeManager()),
+ new RemoveRedundantTableFunctionProcessor(), // must run after TransformTableFunctionToTableFunctionProcessor
+ new TransformTableFunctionProcessorToTableScan(metadata), // must run after TransformTableFunctionToTableFunctionProcessor
new RewriteSpatialPartitioningAggregation(metadata)))
.build()),
new IterativeOptimizer(
@@ -786,7 +795,11 @@ public PlanOptimizers(
ruleStats,
statsCalculator,
estimatedExchangesCostCalculator,
- ImmutableSet.of(new RemoveRedundantIdentityProjections(), new PruneRedundantProjectionAssignments())),
+ ImmutableSet.of(
+ new RemoveRedundantIdentityProjections(),
+ new PruneRedundantProjectionAssignments(),
+ // Re-run RemoveRedundantTableFunctionProcessor after SimplifyPlanWithEmptyInput to optimize empty input tables to empty ValueNode
+ new RemoveRedundantTableFunctionProcessor())),
new PushdownSubfields(metadata, expressionOptimizerManager));
builder.add(predicatePushDown); // Run predicate push down one more time in case we can leverage new information from layouts' effective predicate
@@ -881,14 +894,6 @@ public PlanOptimizers(
costCalculator,
ImmutableSet.of(new ScaledWriterRule())));
- builder.add(
- new IterativeOptimizer(
- metadata,
- ruleStats,
- statsCalculator,
- costCalculator,
- ImmutableSet.of(new RewriteTableFunctionToTableScan(metadata))));
-
if (!noExchange) {
builder.add(new ReplicateSemiJoinInDelete()); // Must run before AddExchanges
diff --git a/presto-main-base/src/main/java/com/facebook/presto/sql/planner/PlannerUtils.java b/presto-main-base/src/main/java/com/facebook/presto/sql/planner/PlannerUtils.java
index 826eaea10044b..4e32a6e6918ed 100644
--- a/presto-main-base/src/main/java/com/facebook/presto/sql/planner/PlannerUtils.java
+++ b/presto-main-base/src/main/java/com/facebook/presto/sql/planner/PlannerUtils.java
@@ -205,6 +205,9 @@ public static PlanNode addOverrideProjection(PlanNode source, PlanNodeIdAllocato
|| source.getOutputVariables().stream().distinct().count() != source.getOutputVariables().size()) {
return source;
}
+ if (source instanceof ProjectNode && ((ProjectNode) source).getAssignments().getMap().equals(variableMap)) {
+ return source;
+ }
Assignments.Builder assignmentsBuilder = Assignments.builder();
assignmentsBuilder.putAll(source.getOutputVariables().stream().collect(toImmutableMap(identity(), x -> variableMap.containsKey(x) ? variableMap.get(x) : x)));
return new ProjectNode(source.getSourceLocation(), planNodeIdAllocator.getNextId(), source, assignmentsBuilder.build(), LOCAL);
diff --git a/presto-main-base/src/main/java/com/facebook/presto/sql/planner/QueryPlanner.java b/presto-main-base/src/main/java/com/facebook/presto/sql/planner/QueryPlanner.java
index f2c102667e06f..c7deb6a31cdec 100644
--- a/presto-main-base/src/main/java/com/facebook/presto/sql/planner/QueryPlanner.java
+++ b/presto-main-base/src/main/java/com/facebook/presto/sql/planner/QueryPlanner.java
@@ -175,7 +175,7 @@
import static java.lang.String.format;
import static java.util.Objects.requireNonNull;
-class QueryPlanner
+public class QueryPlanner
{
private final Analysis analysis;
private final VariableAllocator variableAllocator;
@@ -891,7 +891,7 @@ private PlanBuilder project(PlanBuilder subPlan, Iterable expression
*
* @return the new subplan and a mapping of each expression to the symbol representing the coercion or an existing symbol if a coercion wasn't needed
*/
- private PlanAndMappings coerce(PlanBuilder subPlan, List expressions, Analysis analysis, PlanNodeIdAllocator idAllocator, VariableAllocator variableAllocator, Metadata metadata)
+ public PlanAndMappings coerce(PlanBuilder subPlan, List expressions, Analysis analysis, PlanNodeIdAllocator idAllocator, VariableAllocator variableAllocator, Metadata metadata)
{
Assignments.Builder assignments = Assignments.builder();
assignments.putAll(subPlan.getRoot().getOutputVariables().stream().collect(toImmutableMap(Function.identity(), Function.identity())));
@@ -1713,15 +1713,18 @@ private RowExpression rowExpression(Expression expression, SqlPlannerContext con
context.getTranslatorContext());
}
- private static List toSymbolReferences(List variables)
+ public static List toSymbolReferences(List variables)
{
return variables.stream()
- .map(variable -> new SymbolReference(
- variable.getSourceLocation().map(location -> new NodeLocation(location.getLine(), location.getColumn())),
- variable.getName()))
+ .map(QueryPlanner::toSymbolReference)
.collect(toImmutableList());
}
+ public static SymbolReference toSymbolReference(VariableReferenceExpression variable)
+ {
+ return new SymbolReference(variable.getSourceLocation().map(location -> new NodeLocation(location.getLine(), location.getColumn())), variable.getName());
+ }
+
public static class PlanAndMappings
{
private final PlanBuilder subPlan;
diff --git a/presto-main-base/src/main/java/com/facebook/presto/sql/planner/RelationPlanner.java b/presto-main-base/src/main/java/com/facebook/presto/sql/planner/RelationPlanner.java
index be6b7851ec99b..e07fadcbf4666 100644
--- a/presto-main-base/src/main/java/com/facebook/presto/sql/planner/RelationPlanner.java
+++ b/presto-main-base/src/main/java/com/facebook/presto/sql/planner/RelationPlanner.java
@@ -16,6 +16,7 @@
import com.facebook.presto.Session;
import com.facebook.presto.SystemSessionProperties;
import com.facebook.presto.common.QualifiedObjectName;
+import com.facebook.presto.common.block.SortOrder;
import com.facebook.presto.common.predicate.TupleDomain;
import com.facebook.presto.common.type.ArrayType;
import com.facebook.presto.common.type.MapType;
@@ -31,12 +32,14 @@
import com.facebook.presto.spi.plan.AggregationNode;
import com.facebook.presto.spi.plan.Assignments;
import com.facebook.presto.spi.plan.CteReferenceNode;
+import com.facebook.presto.spi.plan.DataOrganizationSpecification;
import com.facebook.presto.spi.plan.EquiJoinClause;
import com.facebook.presto.spi.plan.ExceptNode;
import com.facebook.presto.spi.plan.FilterNode;
import com.facebook.presto.spi.plan.IntersectNode;
import com.facebook.presto.spi.plan.JoinNode;
import com.facebook.presto.spi.plan.MaterializedViewScanNode;
+import com.facebook.presto.spi.plan.OrderingScheme;
import com.facebook.presto.spi.plan.PlanNode;
import com.facebook.presto.spi.plan.PlanNodeIdAllocator;
import com.facebook.presto.spi.plan.ProjectNode;
@@ -52,14 +55,17 @@
import com.facebook.presto.sql.analyzer.Field;
import com.facebook.presto.sql.analyzer.RelationId;
import com.facebook.presto.sql.analyzer.RelationType;
+import com.facebook.presto.sql.analyzer.ResolvedField;
import com.facebook.presto.sql.analyzer.Scope;
-import com.facebook.presto.sql.analyzer.SemanticException;
import com.facebook.presto.sql.parser.SqlParser;
import com.facebook.presto.sql.planner.optimizations.JoinNodeUtils;
import com.facebook.presto.sql.planner.optimizations.SampleNodeUtil;
import com.facebook.presto.sql.planner.plan.LateralJoinNode;
import com.facebook.presto.sql.planner.plan.SampleNode;
import com.facebook.presto.sql.planner.plan.TableFunctionNode;
+import com.facebook.presto.sql.planner.plan.TableFunctionNode.PassThroughColumn;
+import com.facebook.presto.sql.planner.plan.TableFunctionNode.PassThroughSpecification;
+import com.facebook.presto.sql.planner.plan.TableFunctionNode.TableArgumentProperties;
import com.facebook.presto.sql.tree.AliasedRelation;
import com.facebook.presto.sql.tree.Cast;
import com.facebook.presto.sql.tree.CoalesceExpression;
@@ -87,11 +93,10 @@
import com.facebook.presto.sql.tree.Row;
import com.facebook.presto.sql.tree.SampledRelation;
import com.facebook.presto.sql.tree.SetOperation;
+import com.facebook.presto.sql.tree.SortItem;
import com.facebook.presto.sql.tree.SymbolReference;
import com.facebook.presto.sql.tree.Table;
-import com.facebook.presto.sql.tree.TableFunctionDescriptorArgument;
import com.facebook.presto.sql.tree.TableFunctionInvocation;
-import com.facebook.presto.sql.tree.TableFunctionTableArgument;
import com.facebook.presto.sql.tree.TableSubquery;
import com.facebook.presto.sql.tree.Union;
import com.facebook.presto.sql.tree.Unnest;
@@ -99,6 +104,7 @@
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableListMultimap;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.ListMultimap;
import com.google.common.collect.UnmodifiableIterator;
@@ -119,6 +125,7 @@
import static com.facebook.presto.SystemSessionProperties.getQueryAnalyzerTimeout;
import static com.facebook.presto.common.type.TypeUtils.isEnumType;
import static com.facebook.presto.metadata.MetadataUtil.createQualifiedObjectName;
+import static com.facebook.presto.spi.StandardErrorCode.INVALID_PLAN_ERROR;
import static com.facebook.presto.spi.StandardErrorCode.QUERY_PLANNING_TIMEOUT;
import static com.facebook.presto.spi.plan.AggregationNode.singleGroupingSet;
import static com.facebook.presto.spi.plan.ProjectNode.Locality.LOCAL;
@@ -127,7 +134,6 @@
import static com.facebook.presto.sql.analyzer.ExpressionTreeUtils.isEqualComparisonExpression;
import static com.facebook.presto.sql.analyzer.ExpressionTreeUtils.resolveEnumLiteral;
import static com.facebook.presto.sql.analyzer.FeaturesConfig.CteMaterializationStrategy.NONE;
-import static com.facebook.presto.sql.analyzer.SemanticErrorCode.NOT_SUPPORTED;
import static com.facebook.presto.sql.analyzer.SemanticExceptions.notSupportedException;
import static com.facebook.presto.sql.planner.PlannerUtils.newVariable;
import static com.facebook.presto.sql.planner.TranslateExpressionsUtil.toRowExpression;
@@ -364,51 +370,185 @@ private RelationPlan planMaterializedView(Table node, Analysis.MaterializedViewI
return new RelationPlan(mvScanNode, scope, outputVariables);
}
+ /**
+ * Processes a {@code TableFunctionInvocation} node to construct and return a {@link RelationPlan}.
+ * This involves preparing the necessary plan nodes, variable mappings, and associated properties
+ * to represent the execution plan for the invoked table function.
+ *
+ * @param node The {@code TableFunctionInvocation} syntax tree node to be processed.
+ * @param context The SQL planner context used for planning and analysis tasks.
+ * @return A {@link RelationPlan} encapsulating the execution plan for the table function invocation.
+ */
@Override
protected RelationPlan visitTableFunctionInvocation(TableFunctionInvocation node, SqlPlannerContext context)
{
- node.getArguments().stream()
- .forEach(argument -> {
- if (argument.getValue() instanceof TableFunctionTableArgument) {
- throw new SemanticException(NOT_SUPPORTED, argument, "Table arguments are not yet supported for table functions");
- }
- if (argument.getValue() instanceof TableFunctionDescriptorArgument) {
- throw new SemanticException(NOT_SUPPORTED, argument, "Descriptor arguments are not yet supported for table functions");
- }
- });
Analysis.TableFunctionInvocationAnalysis functionAnalysis = analysis.getTableFunctionAnalysis(node);
+ ImmutableList.Builder sources = ImmutableList.builder();
+ ImmutableList.Builder sourceProperties = ImmutableList.builder();
+ ImmutableList.Builder outputVariables = ImmutableList.builder();
+
+ // create new symbols for table function's proper columns
+ RelationType relationType = analysis.getScope(node).getRelationType();
+ List properOutputs = IntStream.range(0, functionAnalysis.getProperColumnsCount())
+ .mapToObj(relationType::getFieldByIndex)
+ .map(field -> variableAllocator.newVariable(getSourceLocation(node), field.getName().orElse("field"), field.getType()))
+ .collect(toImmutableList());
- // TODO handle input relations:
- // 1. extract the input relations from node.getArguments() and plan them. Apply relation coercions if requested.
- // 2. for each input relation, prepare the TableArgumentProperties record, consisting of:
- // - row or set semantics (from the actualArgument)
- // - prune when empty property (from the actualArgument)
- // - pass through columns property (from the actualArgument)
- // - optional Specification: ordering scheme and partitioning (from the node's argument) <- planned upon the source's RelationPlan (or combined RelationPlan from all sources)
- // TODO add - argument name
- // TODO add - mapping column name => Symbol // TODO mind the fields without names and duplicate field names in RelationType
- List sources = ImmutableList.of();
- List inputRelationsProperties = ImmutableList.of();
-
- Scope scope = analysis.getScope(node);
+ outputVariables.addAll(properOutputs);
- ImmutableList.Builder outputVariablesBuilder = ImmutableList.builder();
- for (Field field : scope.getRelationType().getAllFields()) {
- VariableReferenceExpression variable = variableAllocator.newVariable(getSourceLocation(node), field.getName().get(), field.getType());
- outputVariablesBuilder.add(variable);
- }
+ processTableArguments(context, functionAnalysis, outputVariables, sources, sourceProperties);
- List outputVariables = outputVariablesBuilder.build();
PlanNode root = new TableFunctionNode(
idAllocator.getNextId(),
functionAnalysis.getFunctionName(),
functionAnalysis.getArguments(),
- outputVariablesBuilder.build(),
- sources.stream().map(RelationPlan::getRoot).collect(toImmutableList()),
- inputRelationsProperties,
- new TableFunctionHandle(functionAnalysis.getConnectorId(), functionAnalysis.getConnectorTableFunctionHandle(), functionAnalysis.getTransactionHandle()));
+ properOutputs,
+ sources.build(),
+ sourceProperties.build(),
+ functionAnalysis.getCopartitioningLists(),
+ new TableFunctionHandle(
+ functionAnalysis.getConnectorId(),
+ functionAnalysis.getConnectorTableFunctionHandle(),
+ functionAnalysis.getTransactionHandle()));
+
+ return new RelationPlan(root, analysis.getScope(node), outputVariables.build());
+ }
+
+ private void processTableArguments(SqlPlannerContext context,
+ Analysis.TableFunctionInvocationAnalysis functionAnalysis,
+ ImmutableList.Builder outputVariables,
+ ImmutableList.Builder sources,
+ ImmutableList.Builder sourceProperties)
+ {
+ QueryPlanner partitionQueryPlanner = new QueryPlanner(analysis, variableAllocator, idAllocator, lambdaDeclarationToVariableMap, metadata, session, context, sqlParser);
+ // process sources in order of argument declarations
+ for (Analysis.TableArgumentAnalysis tableArgument : functionAnalysis.getTableArgumentAnalyses()) {
+ RelationPlan sourcePlan = process(tableArgument.getRelation(), context);
+ PlanBuilder sourcePlanBuilder = initializePlanBuilder(sourcePlan);
+
+ int[] fieldIndexForVisibleColumn = getFieldIndexesForVisibleColumns(sourcePlan);
+
+ List requiredColumns = functionAnalysis.getRequiredColumns().get(tableArgument.getArgumentName()).stream()
+ .map(column -> fieldIndexForVisibleColumn[column])
+ .map(sourcePlan::getVariable)
+ .collect(toImmutableList());
+
+ Optional specification = Optional.empty();
+
+ // if the table argument has set semantics, create Specification
+ if (!tableArgument.isRowSemantics()) {
+ // partition by
+ List partitionBy = ImmutableList.of();
+ // if there are partitioning columns, they might have to be coerced for copartitioning
+ if (tableArgument.getPartitionBy().isPresent() && !tableArgument.getPartitionBy().get().isEmpty()) {
+ List partitioningColumns = tableArgument.getPartitionBy().get();
+ for (Expression partitionColumn : partitioningColumns) {
+ if (!sourcePlanBuilder.canTranslate(partitionColumn)) {
+ ResolvedField partition = sourcePlan.getScope().tryResolveField(partitionColumn).orElseThrow(() -> new PrestoException(INVALID_PLAN_ERROR, "Missing equivalent alias"));
+ sourcePlanBuilder.getTranslations().put(partitionColumn, sourcePlan.getVariable(partition.getRelationFieldIndex()));
+ }
+ }
+ QueryPlanner.PlanAndMappings copartitionCoercions = partitionQueryPlanner.coerce(sourcePlanBuilder, partitioningColumns, analysis, idAllocator, variableAllocator, metadata);
+ sourcePlanBuilder = copartitionCoercions.getSubPlan();
+ partitionBy = partitioningColumns.stream()
+ .map(copartitionCoercions::get)
+ .collect(toImmutableList());
+ }
- return new RelationPlan(root, scope, outputVariables);
+ // order by
+ Optional orderBy = getOrderingScheme(tableArgument, sourcePlanBuilder, sourcePlan);
+ specification = Optional.of(new DataOrganizationSpecification(partitionBy, orderBy));
+ }
+
+ // add output symbols passed from the table argument
+ ImmutableList.Builder passThroughColumns = ImmutableList.builder();
+ addPassthroughColumns(outputVariables, tableArgument, sourcePlan, specification, passThroughColumns, sourcePlanBuilder);
+ sources.add(sourcePlanBuilder.getRoot());
+
+ sourceProperties.add(new TableArgumentProperties(
+ tableArgument.getArgumentName(),
+ tableArgument.isRowSemantics(),
+ tableArgument.isPruneWhenEmpty(),
+ new PassThroughSpecification(tableArgument.isPassThroughColumns(), passThroughColumns.build()),
+ requiredColumns,
+ specification));
+ }
+ }
+
+ private static int[] getFieldIndexesForVisibleColumns(RelationPlan sourcePlan)
+ {
+ // required columns are a subset of visible columns of the source. remap required column indexes to field indexes in source relation type.
+ RelationType sourceRelationType = sourcePlan.getScope().getRelationType();
+ int[] fieldIndexForVisibleColumn = new int[sourceRelationType.getVisibleFieldCount()];
+ int visibleColumn = 0;
+ for (int i = 0; i < sourceRelationType.getAllFieldCount(); i++) {
+ if (!sourceRelationType.getFieldByIndex(i).isHidden()) {
+ fieldIndexForVisibleColumn[visibleColumn] = i;
+ visibleColumn++;
+ }
+ }
+ return fieldIndexForVisibleColumn;
+ }
+
+ private static Optional getOrderingScheme(Analysis.TableArgumentAnalysis tableArgument, PlanBuilder sourcePlanBuilder, RelationPlan sourcePlan)
+ {
+ Optional orderBy = Optional.empty();
+ if (tableArgument.getOrderBy().isPresent()) {
+ List sortItems = tableArgument.getOrderBy().get().getSortItems();
+
+ // Ensure all ORDER BY columns can be translated (populate missing translations if needed)
+ for (SortItem sortItem : sortItems) {
+ Expression sortKey = sortItem.getSortKey();
+ if (!sourcePlanBuilder.canTranslate(sortKey)) {
+ Optional resolvedField = sourcePlan.getScope().tryResolveField(sortKey);
+ resolvedField.ifPresent(field -> sourcePlanBuilder.getTranslations().put(
+ sortKey,
+ sourcePlan.getVariable(field.getRelationFieldIndex())));
+ }
+ }
+
+ // The ordering symbols are coerced
+ List coerced = sortItems.stream()
+ .map(SortItem::getSortKey)
+ .map(sourcePlanBuilder::translate)
+ .collect(toImmutableList());
+
+ List sortOrders = sortItems.stream()
+ .map(PlannerUtils::toSortOrder)
+ .collect(toImmutableList());
+
+ orderBy = Optional.of(PlannerUtils.toOrderingScheme(coerced, sortOrders));
+ }
+ return orderBy;
+ }
+
+ private static void addPassthroughColumns(ImmutableList.Builder outputVariables,
+ Analysis.TableArgumentAnalysis tableArgument, RelationPlan sourcePlan,
+ Optional specification,
+ ImmutableList.Builder passThroughColumns,
+ PlanBuilder sourcePlanBuilder)
+ {
+ if (tableArgument.isPassThroughColumns()) {
+ // the original output symbols from the source node, not coerced
+ // note: hidden columns are included. They are present in sourcePlan.fieldMappings
+ outputVariables.addAll(sourcePlan.getFieldMappings());
+ Set partitionBy = specification
+ .map(DataOrganizationSpecification::getPartitionBy)
+ .map(ImmutableSet::copyOf)
+ .orElse(ImmutableSet.of());
+ sourcePlan.getFieldMappings().stream()
+ .map(variable -> new PassThroughColumn(variable, partitionBy.contains(variable)))
+ .forEach(passThroughColumns::add);
+ }
+ else if (tableArgument.getPartitionBy().isPresent()) {
+ tableArgument.getPartitionBy().get().stream()
+ .map(sourcePlanBuilder::translate)
+ // the original symbols for partitioning columns, not coerced
+ .forEach(variable -> {
+ outputVariables.add(variable);
+ passThroughColumns.add(new PassThroughColumn(variable, true));
+ });
+ }
}
@Override
diff --git a/presto-main-base/src/main/java/com/facebook/presto/sql/planner/SchedulingOrderVisitor.java b/presto-main-base/src/main/java/com/facebook/presto/sql/planner/SchedulingOrderVisitor.java
index abb784cdaa298..471c797c426a8 100644
--- a/presto-main-base/src/main/java/com/facebook/presto/sql/planner/SchedulingOrderVisitor.java
+++ b/presto-main-base/src/main/java/com/facebook/presto/sql/planner/SchedulingOrderVisitor.java
@@ -22,9 +22,11 @@
import com.facebook.presto.spi.plan.SpatialJoinNode;
import com.facebook.presto.spi.plan.TableScanNode;
import com.facebook.presto.sql.planner.plan.InternalPlanVisitor;
+import com.facebook.presto.sql.planner.plan.TableFunctionProcessorNode;
import com.google.common.collect.ImmutableList;
import java.util.List;
+import java.util.NoSuchElementException;
import java.util.function.Consumer;
public class SchedulingOrderVisitor
@@ -88,5 +90,17 @@ public Void visitTableScan(TableScanNode node, Consumer