feat: add basic support for key syntax

ksql-benchmark/src/main/java/io/confluent/ksql/benchmark/SerdeBenchmark.java

@@ -33,6 +33,7 @@
 import io.confluent.ksql.serde.json.KsqlJsonSerdeFactory;
 import io.confluent.ksql.util.KsqlConfig;
 import io.confluent.ksql.util.Pair;
+import io.confluent.ksql.util.SchemaUtil;
 import java.io.InputStream;
 import java.nio.file.Path;
 import java.nio.file.Paths;
@@ -117,6 +118,11 @@ private InputStream getSchemaStream() {
 
   @State(Scope.Thread)
   public static class SerdeState {
+
+    private static final org.apache.kafka.connect.data.Schema KEY_SCHEMA = SchemaBuilder.struct()
+        .field(SchemaUtil.ROWKEY_NAME, org.apache.kafka.connect.data.Schema.OPTIONAL_STRING_SCHEMA)
+        .build();
+
     Serializer<GenericRow> serializer;
     Deserializer<GenericRow> deserializer;
     GenericRow row;
@@ -165,9 +171,14 @@ private static org.apache.kafka.connect.data.Schema convertFieldNamesToUppercase
     private static Serde<GenericRow> getJsonSerdeHelper(
         final org.apache.kafka.connect.data.Schema schema
     ) {
+      final PhysicalSchema physicalSchema = PhysicalSchema.from(
+          LogicalSchema.of(KEY_SCHEMA, schema),
+          SerdeOption.none()
+      );
+
       return GenericRowSerDe.from(
           new KsqlJsonSerdeFactory(),
-          PhysicalSchema.from(LogicalSchema.of(schema), SerdeOption.none()),
+          physicalSchema,
           new KsqlConfig(Collections.emptyMap()),
           () -> null,
           "benchmark",
@@ -178,9 +189,15 @@ private static Serde<GenericRow> getAvroSerde(
         final org.apache.kafka.connect.data.Schema schema
     ) {
       final SchemaRegistryClient schemaRegistryClient = new MockSchemaRegistryClient();
+
+      final PhysicalSchema physicalSchema = PhysicalSchema.from(
+          LogicalSchema.of(KEY_SCHEMA, schema),
+          SerdeOption.none()
+      );
+
       return GenericRowSerDe.from(
           new KsqlAvroSerdeFactory("benchmarkSchema"),
-          PhysicalSchema.from(LogicalSchema.of(schema), SerdeOption.none()),
+          physicalSchema,
           new KsqlConfig(Collections.emptyMap()),
           () -> schemaRegistryClient,
           "benchmark",

ksql-common/src/main/java/io/confluent/ksql/schema/ksql/LogicalSchema.java

@@ -63,7 +63,7 @@ public final class LogicalSchema {
       .field(SchemaUtil.ROWTIME_NAME, Schema.OPTIONAL_INT64_SCHEMA)
       .build();
 
-  private static final Schema KEY_SCHEMA = SchemaBuilder
+  private static final Schema IMPLICIT_KEY_SCHEMA = SchemaBuilder
       .struct()
       .field(SchemaUtil.ROWKEY_NAME, Schema.OPTIONAL_STRING_SCHEMA)
       .build();
@@ -90,7 +90,14 @@ public final class LogicalSchema {
   private final ConnectSchema valueSchema;
 
   public static LogicalSchema of(final Schema valueSchema) {
-    return new LogicalSchema(METADATA_SCHEMA, KEY_SCHEMA, valueSchema, Optional.empty());
+    return LogicalSchema.of(IMPLICIT_KEY_SCHEMA, valueSchema);
+  }
+
+  public static LogicalSchema of(
+      final Schema keySchema,
+      final Schema valueSchema
+  ) {
+    return new LogicalSchema(METADATA_SCHEMA, keySchema, valueSchema, Optional.empty());
   }
 
   private LogicalSchema(
@@ -105,6 +112,10 @@ private LogicalSchema(
     this.alias = requireNonNull(alias, "alias");
   }
 
+  public ConnectSchema keySchema() {
+    return keySchema;
+  }
+
   public ConnectSchema valueSchema() {
     return valueSchema;
   }
@@ -237,6 +248,13 @@ public LogicalSchema withoutAlias() {
     );
   }
 
+  /**
+   * @return {@code true} is aliased, {@code false} otherwise.
+   */
+  public boolean isAliased() {
+    return alias.isPresent();
+  }
+
   /**
    * Copies metadata and key fields to the value schema.
    *
@@ -310,12 +328,13 @@ public boolean equals(final Object o) {
       return false;
     }
     final LogicalSchema that = (LogicalSchema) o;
-    return schemasAreEqual(valueSchema, that.valueSchema);
+    return Objects.equals(keySchema, that.keySchema)
+        && Objects.equals(valueSchema, that.valueSchema);
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(valueSchema);
+    return Objects.hash(keySchema, valueSchema);
   }
 
   @Override

ksql-common/src/main/java/io/confluent/ksql/util/SchemaUtil.java

@@ -50,7 +50,6 @@ public final class SchemaUtil {
   public static final String ROWKEY_NAME = "ROWKEY";
   public static final String ROWTIME_NAME = "ROWTIME";
 
-  public static final int ROWTIME_INDEX = 0;
   public static final int ROWKEY_INDEX = 1;
   private static final Map<Type, Supplier<SchemaBuilder>> typeToSchema
       = ImmutableMap.<Type, Supplier<SchemaBuilder>>builder()