@Override
  public SchemaKStream<?> buildStream(final PlanBuildContext buildContext) {
    final Stacker contextStacker = buildContext.buildNodeContext(getId().toString());
    return schemaKStreamFactory.create(
        buildContext,
        dataSource,
        contextStacker.push(SOURCE_OP_NAME)
    );
  }
@Test
  public void shouldBuildSchemaKTableWhenKTableSource() {
    // Given:
    final KsqlTable<String> table = new KsqlTable<>("sqlExpression",
        SourceName.of("datasource"),
        REAL_SCHEMA,
        Optional.of(TIMESTAMP_COLUMN),
        false,
        new KsqlTopic(
            "topic2",
            KeyFormat.nonWindowed(FormatInfo.of(FormatFactory.KAFKA.name()), SerdeFeatures.of()),
            ValueFormat.of(FormatInfo.of(FormatFactory.JSON.name()), SerdeFeatures.of())
        ),
        false
    );

    node = new DataSourceNode(
        PLAN_NODE_ID,
        table,
        table.getName(),
        false
    );

    // When:
    final SchemaKStream<?> result = buildStream(node);

    // Then:
    assertThat(result.getClass(), equalTo(SchemaKTable.class));
  }