class SchemaKTable extends SchemaKStream<K> {

  @Override
  public SchemaKGroupedTable groupBy(
      final ValueFormat valueFormat,
      final List<Expression> groupByExpressions,
      final Stacker contextStacker
  ) {
    final KeyFormat groupedKeyFormat = KeyFormat
        .nonWindowed(keyFormat.getFormatInfo(), SerdeFeatures.of());

    final TableGroupBy<K> step = ExecutionStepFactory.tableGroupBy(
        contextStacker,
        sourceTableStep,
        InternalFormats.of(groupedKeyFormat, valueFormat),
        groupByExpressions
    );

    return new SchemaKGroupedTable(
        step,
        resolveSchema(step),
        groupedKeyFormat,
        ksqlConfig,
        functionRegistry);
  }

  public  SchemaKTable(
      final ExecutionStep<KTableHolder<K>> sourceTableStep,
      final LogicalSchema schema,
      final KeyFormat keyFormat,
      final KsqlConfig ksqlConfig,
      final FunctionRegistry functionRegistry
  );

}