@Override
  public SchemaKGroupedTable groupBy(
      final FormatInfo valueFormat,
      final List<Expression> groupByExpressions,
      final Stacker contextStacker
  ) {
    // Since tables must have a key, we know that the keyFormat is both
    // not NONE and has at least one column; this allows us to inherit
    // the key format directly (as opposed to the logic in SchemaKStream)
    final KeyFormat groupedKeyFormat = SerdeFeaturesFactory.sanitizeKeyFormat(
        KeyFormat.nonWindowed(keyFormat.getFormatInfo(), keyFormat.getFeatures()),
        toSqlTypes(groupByExpressions),
        true
    );

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

    return new SchemaKGroupedTable(
        step,
        resolveSchema(step),
        groupedKeyFormat,
        ksqlConfig,
        functionRegistry);
  }
@Test
  public void testGroupBy() {
    // Given:
    final String selectQuery = "SELECT col0, col1, col2 FROM test2 EMIT CHANGES;";
    final PlanNode logicalPlan = buildLogicalPlan(selectQuery);
    initialSchemaKTable = buildSchemaKTableFromPlan(logicalPlan);
    final List<Expression> groupByExpressions = Arrays.asList(TEST_2_COL_2, TEST_2_COL_1);

    // When:
    final SchemaKGroupedTable groupedSchemaKTable = initialSchemaKTable.groupBy(
        valueFormat.getFormatInfo(),
        groupByExpressions,
        childContextStacker
    );

    // Then:
    assertThat(groupedSchemaKTable, instanceOf(SchemaKGroupedTable.class));
  }