diff --git a/ksql-parser/src/main/java/io/confluent/ksql/schema/ksql/DefaultSqlValueCoercer.java b/ksql-parser/src/main/java/io/confluent/ksql/schema/ksql/DefaultSqlValueCoercer.java
index e066a2bd7a42..eb0b15f7713b 100644
--- a/ksql-parser/src/main/java/io/confluent/ksql/schema/ksql/DefaultSqlValueCoercer.java
+++ b/ksql-parser/src/main/java/io/confluent/ksql/schema/ksql/DefaultSqlValueCoercer.java
@@ -58,6 +58,10 @@ public <T> Optional<T> coerce(final Object value, final SqlType targetType) {
       return Optional.empty();
     }
 
+    if (targetType.baseType() == SqlBaseType.STRING) {
+      return optional(String.valueOf(value));
+    }
+
     final Number result = UPCASTER.get(targetType.baseType()).apply((Number) value);
     return optional(result);
   }
diff --git a/ksql-parser/src/test/java/io/confluent/ksql/schema/ksql/DefaultSqlValueCoercerTest.java b/ksql-parser/src/test/java/io/confluent/ksql/schema/ksql/DefaultSqlValueCoercerTest.java
index a91c956737b7..703e35bfb13d 100644
--- a/ksql-parser/src/test/java/io/confluent/ksql/schema/ksql/DefaultSqlValueCoercerTest.java
+++ b/ksql-parser/src/test/java/io/confluent/ksql/schema/ksql/DefaultSqlValueCoercerTest.java
@@ -20,6 +20,7 @@
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
 import io.confluent.ksql.schema.ksql.types.SqlTypes;
 import io.confluent.ksql.util.KsqlException;
 import java.math.BigDecimal;
@@ -140,6 +141,17 @@ public void shouldCoerceStringToDecimal() {
     assertThat(coercer.coerce(val, SqlTypes.decimal(2, 1)), is(Optional.of(new BigDecimal("1.0"))));
   }
 
+  @Test
+  public void shouldCoerceNumbersToString() {
+    ImmutableSet.of(
+        1,
+        2L,
+        3.3
+    ).forEach(number -> {
+      assertThat(coercer.coerce(number, SqlTypes.STRING), is(Optional.of(String.valueOf(number))));
+    });
+  }
+
   @Test
   public void shouldThrowIfInvalidCoercionString() {
     // Given: