fix test issues

ksqldb-execution/src/main/java/io/confluent/ksql/execution/ddl/commands/CreateTableCommand.java

@@ -15,6 +15,7 @@
 
 package io.confluent.ksql.execution.ddl.commands;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.errorprone.annotations.Immutable;
@@ -23,6 +24,7 @@
 import io.confluent.ksql.name.SourceName;
 import io.confluent.ksql.schema.ksql.LogicalSchema;
 import io.confluent.ksql.serde.WindowInfo;
+import java.util.Objects;
 import java.util.Optional;
 
 @JsonIgnoreProperties({"keyField"}) // Removed at version 0.10
@@ -60,10 +62,42 @@ public CreateTableCommand(
   // This can be in CreateSourceCommand, but it fails deserializing the JSON property when
   // loading a CreateStreamCommand because it is not supported there yet. We should move this
   // source variable and method to the CreateSourceCommand after supporting source streams.
-  public Boolean isSource() {
+  // Also, this getIsSource() is required so that serialized QTT historic plans do not fail
+  // that the 'isSource' field is not present.
+  public Optional<Boolean> getIsSource() {
+    return isSource;
+  }
+
+  @JsonIgnore
+  public boolean isSource() {
     return isSource.orElse(false);
   }
 
+  @Override
+  public boolean equals(final Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    final CreateTableCommand that = (CreateTableCommand) o;
+    return super.equals(that)
+        && isSource() == that.isSource();
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(
+        getSourceName(),
+        getSchema(),
+        getTimestampColumn(),
+        getTopicName(),
+        getFormats(),
+        getWindowInfo(),
+        getIsSource());
+  }
+
   @Override
   public DdlCommandResult execute(final Executor executor) {
     return executor.executeCreateTable(this);

ksqldb-parser/src/main/java/io/confluent/ksql/parser/SqlFormatter.java

@@ -259,11 +259,7 @@ protected Void visitCreateStream(final CreateStream node, final Integer indent)
 
     @Override
     protected Void visitCreateTable(final CreateTable node, final Integer indent) {
-      formatCreate(node,
-          node.isSource()
-              ? "SOURCE TABLE"
-              : "TABLE"
-      );
+      formatCreate(node, "TABLE");
       return null;
     }
 
@@ -611,6 +607,10 @@ private void formatCreate(final CreateSource node, final String type) {
         builder.append("OR REPLACE ");
       }
 
+      if (node.isSource()) {
+        builder.append("SOURCE ");
+      }
+
       builder.append(type);
       builder.append(" ");