class CreateSourceFactory {

  public CreateTableCommand createTableCommand(
      final CreateTable statement,
      final KsqlConfig ksqlConfig
  ) {
    final SourceName sourceName = statement.getName();
    final CreateSourceProperties props = statement.getProperties();
    final String topicName = ensureTopicExists(props, serviceContext);
    final LogicalSchema schema = buildSchema(statement.getElements());
    if (schema.key().isEmpty()) {
      final boolean usingSchemaInference = props.getSchemaId().isPresent();

      final String additional = usingSchemaInference
          ? System.lineSeparator()
          + "Use a partial schema to define the primary key and still load the value columns from "
          + "the Schema Registry, for example:"
          + System.lineSeparator()
          + "\tCREATE TABLE " + sourceName.text() + " (ID INT PRIMARY KEY) WITH (...);"
          : "";

      throw new KsqlException(
          "Tables require a PRIMARY KEY. Please define the PRIMARY KEY." + additional
      );
    }

    final Optional<TimestampColumn> timestampColumn =
        buildTimestampColumn(ksqlConfig, props, schema);

    return new CreateTableCommand(
        sourceName,
        schema,
        timestampColumn,
        topicName,
        buildFormats(schema, props, ksqlConfig),
        getWindowInfo(props),
        Optional.of(statement.isOrReplace())
    );
  }

  public  CreateSourceFactory(final ServiceContext serviceContext);
  @VisibleForTesting  CreateSourceFactory(
      final ServiceContext serviceContext,
      final SerdeFeaturessSupplier keySerdeFeaturesSupplier,
      final SerdeFeaturessSupplier valueSerdeFeaturesSupplier,
      final KeySerdeFactory keySerdeFactory,
      final ValueSerdeFactory valueSerdeFactory
  );

}