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 shouldThrowOnV1TableSourceWithPseudoColumnVersionGreaterThanZero() {
    // Given:
    givenNonWindowedTable();
    givenExistingQueryWithOldPseudoColumnVersion(tableSourceV1);
    when(tableSourceV1.getPseudoColumnVersion()).thenReturn(CURRENT_PSEUDOCOLUMN_VERSION_NUMBER);

    // When:
    final Exception e = assertThrows(
        IllegalStateException.class,
        () -> SchemaKSourceFactory.buildSource(
            buildContext,
            dataSource,
            contextStacker
        )
    );

    // Then:
    assertThat(
        e.getMessage(),
        containsString("TableSourceV2 was released in conjunction with pseudocolumn version 1. Something has gone very wrong")
    );
  }