@Override
  public SchemaKTable<K> into(
      final KsqlTopic topic,
      final QueryContext.Stacker contextStacker,
      final Optional<TimestampColumn> timestampColumn
  ) {
    if (!keyFormat.getWindowInfo().equals(topic.getKeyFormat().getWindowInfo())) {
      throw new IllegalArgumentException("Can't change windowing");
    }

    final TableSink<K> step = ExecutionStepFactory.tableSink(
        contextStacker,
        sourceTableStep,
        Formats.from(topic),
        topic.getKafkaTopicName(),
        timestampColumn
    );

    return new SchemaKTable<>(
        step,
        resolveSchema(step),
        keyFormat,
        ksqlConfig,
        functionRegistry
    );
  }
@Test
  public void shouldThrowOnIntoIfKeyFormatWindowInfoIsDifferent() {
    // Given:
    final SchemaKTable<?> table = buildSchemaKTable(ksqlTable, mockKTable);

    when(topic.getKeyFormat()).thenReturn(KeyFormat.windowed(
        keyFormat.getFormatInfo(),
        SerdeFeatures.of(),
        WindowInfo.of(WindowType.SESSION, Optional.empty(), Optional.empty())
    ));

    // When:
    assertThrows(
        IllegalArgumentException.class,
        () -> table.into(topic, childContextStacker, Optional.empty())
    );
  }