static void validateNoUserColumnsWithSameNameAsPseudoColumns(final Analysis analysis) {

    final String disallowedNames = analysis.getAllDataSources()
        .stream()
        .map(AliasedDataSource::getDataSource)
        .map(DataSource::getSchema)
        .map(LogicalSchema::value)
        .flatMap(Collection::stream)
        .map(Column::name)
        .filter(name -> SystemColumns.isPseudoColumn(name))
        .map(ColumnName::toString)
        .collect(Collectors.joining(", "));

    if (disallowedNames.length() > 0) {
      throw new KsqlException(
          "Your stream/table has columns with the same name as newly introduced pseudocolumns in"
              + " ksqlDB, and cannot be queried as a result. The conflicting names are: "
              + disallowedNames + ".\n"
      );
    }
  }
@Test
  public void shouldThrowOnUserColumnsWithSameNameAsPseudoColumn() {
    // Given:
    givenAnalysisOfQueryWithUserColumnsWithSameNameAsPseudoColumn();

    // When:
    final Exception e = assertThrows(
        KsqlException.class,
        () -> QueryValidatorUtil.validateNoUserColumnsWithSameNameAsPseudoColumns(analysis)
    );

    // Then:
    assertThat(e.getMessage(), containsString("Your stream/table has columns with the "
        + "same name as newly introduced pseudocolumns in "
        + "ksqlDB, and cannot be queried as a result. The conflicting names are: `ROWPARTITION`."));

  }