public static SchemaKStream<?> buildSource(
      final PlanBuildContext buildContext,
      final DataSource dataSource,
      final QueryContext.Stacker contextStacker
  ) {
    final boolean windowed = dataSource.getKsqlTopic().getKeyFormat().isWindowed();
    switch (dataSource.getDataSourceType()) {
      case KSTREAM:
        return windowed
            ? buildWindowedStream(
            buildContext,
            dataSource,
            contextStacker
        ) : buildStream(
            buildContext,
            dataSource,
            contextStacker
        );

      case KTABLE:
        return windowed
            ? buildWindowedTable(
            buildContext,
            dataSource,
            contextStacker
        ) : buildTable(
            buildContext,
            dataSource,
            contextStacker
        );

      default:
        throw new UnsupportedOperationException("Source type:" + dataSource.getDataSourceType());
    }
  }
@Test
  public void shouldReplaceWindowedStreamSourceWithMatchingPseudoColumnVersion() {
    // Given:
    givenWindowedStream();
    givenExistingQueryWithOldPseudoColumnVersion(windowedStreamSource);

    // When:
    final SchemaKStream<?> result = SchemaKSourceFactory.buildSource(
        buildContext,
        dataSource,
        contextStacker
    );

    // Then:
    assertThat(((WindowedStreamSource) result.getSourceStep()).getPseudoColumnVersion(), equalTo(LEGACY_PSEUDOCOLUMN_VERSION_NUMBER));
    assertValidSchema(result);
  }