fix: remove any rowtime or rowkey columns from query schema (MINOR) (Fixes 3039)

ksql-engine/src/main/java/io/confluent/ksql/planner/plan/KsqlBareOutputNode.java

@@ -39,7 +39,17 @@ public KsqlBareOutputNode(
       final OptionalInt limit,
       final TimestampExtractionPolicy extractionPolicy
   ) {
-    super(id, source, schema, limit, extractionPolicy);
+    super(
+        id,
+        source,
+        // KSQL internally copies the implicit and key fields into the value schema.
+        // This is done by DataSourceNode
+        // Hence, they must be removed again here if they are still in the sink schema.
+        // This leads to strange behaviour, but changing it is a breaking change.
+        schema.withoutMetaAndKeyFieldsInValue(),
+        limit,
+        extractionPolicy
+    );
     this.keyField = KeyField.of(source.getKeyField().name(), Optional.empty())
         .validateKeyExistsIn(schema);
   }

ksql-engine/src/main/java/io/confluent/ksql/util/QueryMetadata.java

@@ -79,6 +79,11 @@ protected QueryMetadata(
     this.closeCallback = Objects.requireNonNull(closeCallback, "closeCallback");
     this.sourceNames = Objects.requireNonNull(sourceNames, "sourceNames");
     this.logicalSchema = Objects.requireNonNull(logicalSchema, "logicalSchema");
+
+    if (logicalSchema.findValueField(SchemaUtil.ROWKEY_NAME).isPresent()
+        || logicalSchema.findValueField(SchemaUtil.ROWTIME_NAME).isPresent()) {
+      throw new IllegalArgumentException("Schema contains implicit columns in value schema");
+    }
   }
 
   protected QueryMetadata(final QueryMetadata other, final Consumer<QueryMetadata> closeCallback) {

ksql-engine/src/test/java/io/confluent/ksql/util/QueryMetadataTest.java

@@ -154,4 +154,52 @@ public void shouldReturnSources() {
   public void shouldReturnSchema() {
     assertThat(query.getLogicalSchema(), is(SOME_SCHEMA));
   }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void shouldThrowIfSchemaContainsRowTime() {
+    // Given:
+    final LogicalSchema invalidSchema = LogicalSchema.of(SchemaBuilder.struct()
+        .field("f0", SchemaBuilder.OPTIONAL_STRING_SCHEMA)
+        .field(SchemaUtil.ROWTIME_NAME, SchemaBuilder.OPTIONAL_INT64_SCHEMA)
+        .build());
+
+    // When:
+    new QueryMetadata(
+        "foo",
+        kafkaStreams,
+        invalidSchema,
+        SOME_SOURCES,
+        "bar",
+        DataSourceType.KSTREAM,
+        QUERY_APPLICATION_ID,
+        topoplogy,
+        Collections.emptyMap(),
+        Collections.emptyMap(),
+        closeCallback
+    );
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void shouldThrowIfSchemaContainsRowKey() {
+    // Given:
+    final LogicalSchema invalidSchema = LogicalSchema.of(SchemaBuilder.struct()
+        .field("f0", SchemaBuilder.OPTIONAL_STRING_SCHEMA)
+        .field(SchemaUtil.ROWKEY_NAME, SchemaBuilder.OPTIONAL_STRING_SCHEMA)
+        .build());
+
+    // When:
+    new QueryMetadata(
+        "foo",
+        kafkaStreams,
+        invalidSchema,
+        SOME_SOURCES,
+        "bar",
+        DataSourceType.KSTREAM,
+        QUERY_APPLICATION_ID,
+        topoplogy,
+        Collections.emptyMap(),
+        Collections.emptyMap(),
+        closeCallback
+    );
+  }
 }