diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/CalcCodeGenerator.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/CalcCodeGenerator.scala
index 9406de4..2fa275d 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/CalcCodeGenerator.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/CalcCodeGenerator.scala
@@ -19,8 +19,10 @@
 
 import org.apache.flink.api.common.functions.{FlatMapFunction, Function}
 import org.apache.flink.api.dag.Transformation
-import org.apache.flink.table.api.{TableConfig, TableException}
+import org.apache.flink.table.api.{TableConfig, TableException, ValidationException}
 import org.apache.flink.table.data.{BoxedWrapperRowData, RowData}
+import org.apache.flink.table.functions.FunctionKind
+import org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction
 import org.apache.flink.table.runtime.generated.GeneratedFunction
 import org.apache.flink.table.runtime.operators.CodeGenOperatorFactory
 import org.apache.flink.table.runtime.typeutils.InternalTypeInfo
@@ -117,6 +119,12 @@
       allowSplit: Boolean = false): String = {
 
     val projection = calcProgram.getProjectList.map(calcProgram.expandLocalRef)
+
+    // according to the SQL standard, every table function should also be a scalar function
+    // but we don't allow that for now
+    projection.foreach(_.accept(ScalarFunctionsValidator))
+    condition.foreach(_.accept(ScalarFunctionsValidator))
+
     val exprGenerator = new ExprCodeGenerator(ctx, false)
         .bindInput(inputType, inputTerm = inputTerm)
 
@@ -195,4 +203,17 @@
       }
     }
   }
+
+  private object ScalarFunctionsValidator extends RexVisitorImpl[Unit](true) {
+    override def visitCall(call: RexCall): Unit = {
+      super.visitCall(call)
+      call.getOperator match {
+        case bsf: BridgingSqlFunction if bsf.getDefinition.getKind != FunctionKind.SCALAR =>
+          throw new ValidationException(
+            s"Invalid use of function '$bsf'. " +
+              s"Currently, only scalar functions can be used in a projection or filter operation.")
+        case _ => // ok
+      }
+    }
+  }
 }
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/CorrelateCodeGenerator.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/CorrelateCodeGenerator.scala
index c09606f..3b1157b 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/CorrelateCodeGenerator.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/CorrelateCodeGenerator.scala
@@ -70,8 +70,10 @@
     rexCall.getOperator match {
       case func: BridgingSqlFunction if func.getDefinition.getKind == FunctionKind.TABLE => // ok
       case _: TableSqlFunction => // ok
-      case _ =>
-        throw new ValidationException("Currently, only table functions can emit rows.")
+      case f@_ =>
+        throw new ValidationException(
+          s"Invalid use of function '$f'. " +
+            s"Currently, only table functions can be used in a correlate operation.")
     }
 
     val swallowInputOnly = if (projectProgram.isDefined) {
diff --git a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java
index fdd2e4c..9b7281f 100644
--- a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java
+++ b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java
@@ -858,7 +858,7 @@
 				e,
 				hasMessage(
 					containsString(
-						"Currently, only table functions can emit rows.")));
+						"Currently, only table functions can be used in a correlate operation.")));
 		}
 	}
 
@@ -869,12 +869,19 @@
 		tEnv().executeSql("CREATE TABLE SinkTable(s ROW<s STRING, sa ARRAY<STRING> NOT NULL>) WITH ('connector' = 'COLLECTION')");
 
 		tEnv().createTemporarySystemFunction("RowTableFunction", RowTableFunction.class);
-		tEnv().executeSql(
-			"INSERT INTO SinkTable " +
-			"SELECT RowTableFunction('test')");
 
-		// currently, calling a table function like a scalar function produces no result
-		assertThat(TestCollectionTableFactory.getResult(), equalTo(Collections.emptyList()));
+		try {
+			tEnv().explainSql(
+				"INSERT INTO SinkTable " +
+				"SELECT RowTableFunction('test')");
+			fail();
+		} catch (ValidationException e) {
+			assertThat(
+				e,
+				hasMessage(
+					containsString(
+						"Currently, only scalar functions can be used in a projection or filter operation.")));
+		}
 	}
 
 	@Test
diff --git a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/stream/table/FunctionITCase.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/stream/table/FunctionITCase.java
index 2fb79da..35150b0 100644
--- a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/stream/table/FunctionITCase.java
+++ b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/stream/table/FunctionITCase.java
@@ -38,8 +38,10 @@
 
 import static org.apache.flink.table.api.Expressions.$;
 import static org.apache.flink.table.api.Expressions.call;
+import static org.hamcrest.CoreMatchers.containsString;
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.junit.Assert.assertThat;
+import static org.junit.internal.matchers.ThrowableMessageMatcher.hasMessage;
 
 /**
  * Tests for user defined functions in the Table API.
@@ -114,7 +116,10 @@
 	@Test
 	public void testLateralJoinWithScalarFunction() throws Exception {
 		thrown.expect(ValidationException.class);
-		thrown.expectMessage("Currently, only table functions can emit rows.");
+		thrown.expect(
+			hasMessage(
+				containsString(
+					"Currently, only table functions can be used in a correlate operation.")));
 
 		TestCollectionTableFactory.reset();
 		tEnv().executeSql("CREATE TABLE SourceTable(s STRING) WITH ('connector' = 'COLLECTION')");