From 5de63b2caf582e9af56012bbe5822fc3b5d268a2 Mon Sep 17 00:00:00 2001
From: Tim Fox <tfox@confluent.io>
Date: Mon, 9 Dec 2019 14:59:06 +0000
Subject: [PATCH] fix: Explicitly disallow table functions with table sources,
 fixes #4033

---
 .../io/confluent/ksql/analyzer/QueryAnalyzer.java     |  9 +++++++++
 .../query-validation-tests/table-functions.json       | 11 +++++++++++
 2 files changed, 20 insertions(+)

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
index f1f21b69aa2c..1b1e87c5fea1 100644
--- 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
index d602f14ccae6..495cbe9cef6a 100644
--- 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"
+      }
     }
   ]
 }
\ No newline at end of file