feat: Implement EXPLODE(ARRAY) for single table function in SELECT

ksql-common/src/main/java/io/confluent/ksql/function/AggregateFunctionFactory.java

@@ -18,6 +18,7 @@
 import com.google.common.collect.ImmutableList;
 import io.confluent.ksql.function.udf.UdfMetadata;
 import io.confluent.ksql.util.DecimalUtil;
+import io.confluent.ksql.util.KsqlConstants;
 import java.util.List;
 import java.util.Objects;
 import java.util.function.Consumer;
@@ -38,7 +39,14 @@ public abstract class AggregateFunctionFactory {
       .build();
 
   public AggregateFunctionFactory(final String functionName) {
-    this(new UdfMetadata(functionName, "", "confluent", "", KsqlFunction.INTERNAL_PATH, false));
+    this(new UdfMetadata(
+        functionName,
+        "",
+        KsqlConstants.CONFLUENT_AUTHOR,
+        "",
+        KsqlFunction.INTERNAL_PATH,
+        false
+    ));
   }
 
   public AggregateFunctionFactory(final UdfMetadata metadata) {

ksql-common/src/main/java/io/confluent/ksql/function/FunctionRegistry.java

@@ -33,6 +33,16 @@ public interface FunctionRegistry {
    */
   boolean isAggregate(String functionName);
 
+  /**
+   * Test if the supplied {@code functionName} is a table function.
+   *
+   * <p>Note: unknown functions result in {@code false} return value.
+   *
+   * @param functionName the name of the function to test
+   * @return {@code true} if it is a table function, {@code false} otherwise.
+   */
+  boolean isTableFunction(String functionName);
+
   /**
    * Get the factory for a UDF.
    *
@@ -72,6 +82,8 @@ public interface FunctionRegistry {
   KsqlAggregateFunction<?, ?, ?> getAggregateFunction(String functionName, Schema argumentType,
       AggregateFunctionInitArguments initArgs);
 
+  KsqlTableFunction<?, ?> getTableFunction(String functionName, Schema argumentType);
+
   /**
    * @return all UDF factories.
    */

ksql-common/src/main/java/io/confluent/ksql/function/FunctionSignature.java

@@ -38,7 +38,9 @@ public interface FunctionSignature {
    * @return whether or not to consider the final argument in
    *         {@link #getArguments()} as variadic
    */
-  boolean isVariadic();
+  default boolean isVariadic() {
+    return false;
+  }
 
 
 }

ksql-common/src/main/java/io/confluent/ksql/function/KsqlAggregateFunction.java

@@ -51,9 +51,4 @@ public interface KsqlAggregateFunction<I, A, O> extends FunctionSignature {
   Function<A, O> getResultMapper();
 
   String getDescription();
-
-  @Override
-  default boolean isVariadic() {
-    return false;
-  }
 }

ksql-common/src/main/java/io/confluent/ksql/function/KsqlTableFunction.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2018 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.function;
+
+import io.confluent.ksql.schema.ksql.types.SqlType;
+import java.util.List;
+import org.apache.kafka.connect.data.Schema;
+
+public interface KsqlTableFunction<I, O> extends FunctionSignature {
+
+  Schema getReturnType();
+
+  SqlType returnType();
+
+  List<O> flatMap(List<I> input);
+
+  String getDescription();
+}

ksql-common/src/main/java/io/confluent/ksql/function/MutableFunctionRegistry.java

@@ -51,4 +51,6 @@ public interface MutableFunctionRegistry extends FunctionRegistry {
    * @throws KsqlException if a function, (of any type), with the same name exists.
    */
   void addAggregateFunctionFactory(AggregateFunctionFactory aggregateFunctionFactory);
+
+  void addTableFunctionFactory(TableFunctionFactory tableFunctionFactory);
 }

ksql-common/src/main/java/io/confluent/ksql/function/TableFunctionFactory.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2018 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.function;
+
+import io.confluent.ksql.function.udf.UdfMetadata;
+import java.util.List;
+import java.util.Objects;
+import org.apache.kafka.connect.data.Schema;
+
+
+public abstract class TableFunctionFactory {
+
+  private final UdfMetadata metadata;
+
+  public TableFunctionFactory(final UdfMetadata metadata) {
+    this.metadata = Objects.requireNonNull(metadata, "metadata can't be null");
+  }
+
+  public abstract KsqlTableFunction<?, ?> createTableFunction(List<Schema> argTypeList);
+
+  protected abstract List<List<Schema>> supportedArgs();
+
+  public String getName() {
+    return metadata.getName();
+  }
+
+  public String getDescription() {
+    return metadata.getDescription();
+  }
+
+  public String getPath() {
+    return metadata.getPath();
+  }
+
+  public String getAuthor() {
+    return metadata.getAuthor();
+  }
+
+  public String getVersion() {
+    return metadata.getVersion();
+  }
+
+  public boolean isInternal() {
+    return metadata.isInternal();
+  }
+}

ksql-common/src/main/java/io/confluent/ksql/name/ColumnName.java

@@ -24,11 +24,16 @@
 public final class ColumnName extends Name<ColumnName> {
 
   private static final String AGGREGATE_COLUMN_PREFIX = "KSQL_AGG_VARIABLE_";
+  private static final String GENERATED_COLUMN_PREFIX = "KSQL_COL_";
 
-  public static ColumnName aggregate(final int idx) {
+  public static ColumnName aggregateColumn(final int idx) {
     return of(AGGREGATE_COLUMN_PREFIX + idx);
   }
 
+  public static ColumnName generatedColumnName(final int idx) {
+    return ColumnName.of(GENERATED_COLUMN_PREFIX + idx);
+  }
+
   public static ColumnName of(final String name) {
     return new ColumnName(name);
   }

ksql-engine/src/main/java/io/confluent/ksql/util/AggregateExpressionRewriter.java → ksql-engine/src/main/java/io/confluent/ksql/analyzer/AggregateExpressionRewriter.java

@@ -13,7 +13,7 @@
  * specific language governing permissions and limitations under the License.
  */
 
-package io.confluent.ksql.util;
+package io.confluent.ksql.analyzer;
 
 import io.confluent.ksql.engine.rewrite.ExpressionTreeRewriter;
 import io.confluent.ksql.engine.rewrite.ExpressionTreeRewriter.Context;
@@ -45,7 +45,7 @@ public Optional<Expression> visitFunctionCall(
       final ExpressionTreeRewriter.Context<Void> context) {
     final String functionName = node.getName().name();
     if (functionRegistry.isAggregate(functionName)) {
-      final ColumnName aggVarName = ColumnName.aggregate(aggVariableIndex);
+      final ColumnName aggVarName = ColumnName.aggregateColumn(aggVariableIndex);
       aggVariableIndex++;
       return Optional.of(
           new ColumnReferenceExp(node.getLocation(), ColumnRef.withoutSource(aggVarName)));

ksql-engine/src/main/java/io/confluent/ksql/analyzer/MutableTableFunctionAnalysis.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2018 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License; you may not use this file
+ * except in compliance with the License.  You may obtain a copy of the License at
+ *
+ * http://www.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.analyzer;
+
+import io.confluent.ksql.execution.expression.tree.FunctionCall;
+import io.confluent.ksql.execution.plan.SelectExpression;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+public class MutableTableFunctionAnalysis implements TableFunctionAnalysis {
+
+  private final List<FunctionCall> tableFunctions = new ArrayList<>();
+  private final List<SelectExpression> selectExpressions = new ArrayList<>();
+
+  @Override
+  public List<FunctionCall> getTableFunctions() {
+    return Collections.unmodifiableList(tableFunctions);
+  }
+
+  @Override
+  public List<SelectExpression> getFinalSelectExpressions() {
+    return selectExpressions;
+  }
+
+  void addTableFunction(final FunctionCall functionCall) {
+    tableFunctions.add(Objects.requireNonNull(functionCall));
+  }
+
+  void addFinalSelectExpression(final SelectExpression selectExpression) {
+    this.selectExpressions.add(Objects.requireNonNull(selectExpression));
+  }
+}

ksql-engine/src/main/java/io/confluent/ksql/analyzer/QueryAnalyzer.java

@@ -31,15 +31,16 @@
 import io.confluent.ksql.parser.tree.Query;
 import io.confluent.ksql.parser.tree.Sink;
 import io.confluent.ksql.serde.SerdeOption;
-import io.confluent.ksql.util.AggregateExpressionRewriter;
 import io.confluent.ksql.util.KsqlException;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 
+// CHECKSTYLE_RULES.OFF: ClassDataAbstractionCoupling
 public class QueryAnalyzer {
+  // CHECKSTYLE_RULES.ON: ClassDataAbstractionCoupling
 
   private final Analyzer analyzer;
   private final MetaStore metaStore;
@@ -130,6 +131,27 @@ public AggregateAnalysis analyzeAggregate(final Query query, final Analysis anal
     return aggregateAnalysis;
   }
 
+  public TableFunctionAnalysis analyzeTableFunctions(final Analysis analysis) {
+    final MutableTableFunctionAnalysis tableFunctionAnalysis = new MutableTableFunctionAnalysis();
+    final TableFunctionAnalyzer tableFunctionAnalyzer =
+        new TableFunctionAnalyzer(tableFunctionAnalysis, metaStore);
+    final TableFunctionExpressionRewriter tableFunctionExpressionRewriter =
+        new TableFunctionExpressionRewriter(metaStore);
+
+    processSelectExpressionsForTableFunctionAnalysis(
+        analysis,
+        tableFunctionAnalysis,
+        tableFunctionAnalyzer,
+        tableFunctionExpressionRewriter
+    );
+
+    if (tableFunctionAnalysis.getTableFunctions().size() > 1) {
+      throw new KsqlException("Only one table function per query currently is supported");
+    }
+
+    return tableFunctionAnalysis;
+  }
+
   private static void processHavingExpression(
       final Expression having,
       final MutableAggregateAnalysis aggregateAnalysis,
@@ -166,6 +188,24 @@ private static void processSelectExpressions(
     }
   }
 
+  private static void processSelectExpressionsForTableFunctionAnalysis(
+      final Analysis analysis,
+      final MutableTableFunctionAnalysis tableFunctionAnalysis,
+      final TableFunctionAnalyzer tableFunctionAnalyzer,
+      final TableFunctionExpressionRewriter tableFunctionExpressionRewriter
+  ) {
+    for (final SelectExpression select : analysis.getSelectExpressions()) {
+      final Expression exp = select.getExpression();
+      tableFunctionAnalyzer.processSelect(exp);
+
+      tableFunctionAnalysis.addFinalSelectExpression(
+          SelectExpression.of(
+              select.getAlias(),
+              ExpressionTreeRewriter.rewriteWith(
+                  tableFunctionExpressionRewriter::process, exp)));
+    }
+  }
+
   private static void enforceAggregateRules(
       final Query query,
       final Analysis analysis,

ksql-engine/src/main/java/io/confluent/ksql/analyzer/TableFunctionAnalysis.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2018 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License; you may not use this file
+ * except in compliance with the License.  You may obtain a copy of the License at
+ *
+ * http://www.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.analyzer;
+
+import io.confluent.ksql.execution.expression.tree.FunctionCall;
+import io.confluent.ksql.execution.plan.SelectExpression;
+import java.util.List;
+
+/**
+ * Encapsulates data that's been extracted from a query related to table functions.
+ */
+public interface TableFunctionAnalysis {
+
+  List<FunctionCall> getTableFunctions();
+
+  List<SelectExpression> getFinalSelectExpressions();
+}