public AlterSourceCommand create(final AlterSource statement) {
    final DataSource dataSource = metaStore.getSource(statement.getName());
    final String dataSourceType = statement.getDataSourceType().getKsqlType();

    if (dataSource != null && dataSource.isSource()) {
      throw new KsqlException(
          String.format("Cannot alter %s '%s': ALTER operations are not supported on source %s.",
              dataSourceType.toLowerCase(),
              statement.getName().text(),
              dataSourceType.toLowerCase() + "s"));
    }

    final List<Column> newColumns = statement
        .getAlterOptions()
        .stream()
        .map(
            alterOption -> Column.of(
                ColumnName.of(alterOption.getColumnName()),
                alterOption.getType().getSqlType(),
                Namespace.VALUE,
                0))
        .collect(Collectors.toList());

    return new AlterSourceCommand(
        statement.getName(),
        dataSourceType,
        newColumns
    );
  }
@Test
  public void shouldCreateCommandForAlterTable() {
    // Given:
    final AlterSource alterSource = new AlterSource(TABLE_NAME, DataSourceType.KTABLE, NEW_COLUMNS);

    // When:
    final AlterSourceCommand result = alterSourceFactory.create(alterSource);

    // Then:
    assertEquals(result.getKsqlType(), DataSourceType.KTABLE.getKsqlType());
    assertEquals(result.getSourceName(), TABLE_NAME);
    assertEquals(result.getNewColumns().size(), 1);
  }