class KsqlQueryBuilder {

  public Serde<Struct> buildKeySerde(
      final FormatInfo format, final PhysicalSchema schema, final QueryContext queryContext
  ) {
    final String loggerNamePrefix = QueryLoggerUtil.queryLoggerName(queryId, queryContext);

    schemas.track(
        loggerNamePrefix,
        schema.logicalSchema(),
        KeyFormat.nonWindowed(format, schema.keySchema().features())
    );

    return keySerdeFactory.create(
        format,
        schema.keySchema(),
        ksqlConfig,
        serviceContext.getSchemaRegistryClientFactory(),
        loggerNamePrefix,
        processingLogContext
    );
  }

  @VisibleForTesting  KsqlQueryBuilder(
      final StreamsBuilder streamsBuilder,
      final KsqlConfig ksqlConfig,
      final ServiceContext serviceContext,
      final ProcessingLogContext processingLogContext,
      final FunctionRegistry functionRegistry,
      final QueryId queryId,
      final KeySerdeFactory keySerdeFactory,
      final ValueSerdeFactory valueSerdeFactory
  );

}