Address some review comments

Author: xin-zhang2

presto-main-base/src/main/java/com/facebook/presto/metadata/Metadata.java

@@ -560,5 +560,10 @@ default boolean isPushdownSupportedForFilter(Session session, TableHandle tableH
 
     String normalizeIdentifier(Session session, String catalogName, String identifier);
 
+    /**
+     * Attempt to push down the table function invocation into the connector.
+     * @return {@link Optional#empty()} if the connector doesn't support table function invocation pushdown,
+     * or an {@code Optional<TableFunctionApplicationResult<TableHandle>>} containing the table handle that will be used in place of the table function invocation.
+     */
     Optional<TableFunctionApplicationResult<TableHandle>> applyTableFunction(Session session, TableFunctionHandle handle);
 }

presto-main-base/src/main/java/com/facebook/presto/operator/RegularTableFunctionPartition.java

@@ -315,7 +315,7 @@ private int findFirstNullPosition(int markerChannel)
         int end = partitionEnd;
         // value at start is not null, value at end is null
         while (end - start > 1) {
-            int mid = start + end >>> 1;
+            int mid = (start + end) >>> 1;
             if (pagesIndex.isNull(markerChannel, mid)) {
                 end = mid;
             }

presto-main-base/src/main/java/com/facebook/presto/sql/planner/PlanOptimizers.java

@@ -140,7 +140,6 @@
 import com.facebook.presto.sql.planner.iterative.rule.RewriteExcludeColumnsFunctionToProjection;
 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;
@@ -157,6 +156,7 @@
 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;
@@ -394,7 +394,7 @@ public PlanOptimizers(
                         ruleStats,
                         statsCalculator,
                         costCalculator,
-                        ImmutableSet.of(new RewriteTableFunctionToTableScan(metadata))));
+                        ImmutableSet.of(new TransformTableFunctionProcessorToTableScan(metadata))));
 
         builder.add(
                 new IterativeOptimizer(
@@ -894,7 +894,7 @@ public PlanOptimizers(
                         ruleStats,
                         statsCalculator,
                         costCalculator,
-                        ImmutableSet.of(new RewriteTableFunctionToTableScan(metadata), new RewriteExcludeColumnsFunctionToProjection())));
+                        ImmutableSet.of(new TransformTableFunctionProcessorToTableScan(metadata), new RewriteExcludeColumnsFunctionToProjection())));
 
         if (!noExchange) {
             builder.add(new ReplicateSemiJoinInDelete()); // Must run before AddExchanges

presto-main-base/src/main/java/com/facebook/presto/sql/planner/iterative/rule/RewriteTableFunctionToTableScan.java renamed to presto-main-base/src/main/java/com/facebook/presto/sql/planner/iterative/rule/TransformTableFunctionProcessorToTableScan.java

@@ -53,15 +53,15 @@
  *         outputVar2 -> my_function(arg1, arg2)_colHandle2
  *     }
  */
-public class RewriteTableFunctionToTableScan
+public class TransformTableFunctionProcessorToTableScan
         implements Rule<TableFunctionProcessorNode>
 {
     private static final Pattern<TableFunctionProcessorNode> PATTERN = tableFunctionProcessor()
             .with(empty(sources()));
 
     private final Metadata metadata;
 
-    public RewriteTableFunctionToTableScan(Metadata metadata)
+    public TransformTableFunctionProcessorToTableScan(Metadata metadata)
     {
         this.metadata = requireNonNull(metadata, "metadata is null");
     }
@@ -82,7 +82,9 @@ public Result apply(TableFunctionProcessorNode node, Captures captures, Context
         }
 
         List<ColumnHandle> columnHandles = result.get().getColumnHandles();
-        checkState(node.getOutputVariables().size() == columnHandles.size(), "returned table does not match the node's output");
+        checkState(node.getOutputVariables().size() == columnHandles.size(),
+                "Connector returned %s columns but TableFunctionProcessorNode expects %s outputs",
+                columnHandles.size(), node.getOutputVariables().size());
         ImmutableMap.Builder<VariableReferenceExpression, ColumnHandle> assignments = ImmutableMap.builder();
         for (int i = 0; i < columnHandles.size(); i++) {
             assignments.put(node.getOutputVariables().get(i), columnHandles.get(i));