fix: Explicitly disallow table functions with table sources, fixes #4033

(#4085)
confluentinc · Dec 10, 2019 · 60e20ef · 60e20ef
1 parent 9a2bdec
commit 60e20ef
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 0 deletions.
diff --git a/ksql-engine/src/main/java/io/confluent/ksql/analyzer/QueryAnalyzer.java b/ksql-engine/src/main/java/io/confluent/ksql/analyzer/QueryAnalyzer.java
@@ -21,12 +21,14 @@
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Sets;
 import com.google.common.collect.Sets.SetView;
+import io.confluent.ksql.analyzer.Analysis.AliasedDataSource;
 import io.confluent.ksql.engine.rewrite.ExpressionTreeRewriter;
 import io.confluent.ksql.execution.expression.tree.ColumnReferenceExp;
 import io.confluent.ksql.execution.expression.tree.Expression;
 import io.confluent.ksql.execution.expression.tree.FunctionCall;
 import io.confluent.ksql.execution.plan.SelectExpression;
 import io.confluent.ksql.metastore.MetaStore;
+import io.confluent.ksql.metastore.model.DataSource.DataSourceType;
 import io.confluent.ksql.name.FunctionName;
 import io.confluent.ksql.parser.tree.Query;
 import io.confluent.ksql.parser.tree.Sink;
@@ -83,6 +85,13 @@ public Analysis analyze(
       pullQueryValidator.validate(analysis);
     }
 
+    if (!analysis.getTableFunctions().isEmpty()) {
+      AliasedDataSource ds = analysis.getFromDataSources().get(0);
+      if (ds.getDataSource().getDataSourceType() == DataSourceType.KTABLE) {
+        throw new KsqlException("Table source is not supported with table functions");
+      }
+    }
+
     return analysis;
   }
 

diff --git a/ksql-functional-tests/src/test/resources/query-validation-tests/table-functions.json b/ksql-functional-tests/src/test/resources/query-validation-tests/table-functions.json
@@ -237,6 +237,17 @@
         {"topic": "OUTPUT", "key": "0", "value": {"F0": 2, "VAL": 8}},
         {"topic": "OUTPUT", "key": "0", "value": {"F0": 2, "VAL": 9}}
       ]
+    },
+    {
+      "name": "table functions don't support table sources",
+      "statements": [
+        "CREATE TABLE TEST (ID BIGINT, MY_ARR ARRAY<BIGINT>) WITH (kafka_topic='test_topic', KEY='ID', value_format='JSON');",
+        "CREATE TABLE OUTPUT AS SELECT ID, EXPLODE(MY_ARR) VAL FROM TEST;"
+      ],
+      "expectedException": {
+        "type": "io.confluent.ksql.util.KsqlException",
+        "message": "Table source is not supported with table functions"
+      }
     }
   ]
 }