class SaveToCassandraOperationsService {

    public void alterTable(String streamName, Set<String> oldColumnNames, List<ColumnNameTypeValue> columns) {

        StringBuilder sb = new StringBuilder();

        for (Entry<String, String> entry : getStreamFieldsAndTypes(columns).entrySet()) {
            if (!oldColumnNames.contains(entry.getKey())) {
                sb.append(String.format("ALTER TABLE %s.%s ADD %s %s;", STREAMING.STREAMING_KEYSPACE_NAME,
                        addQuotes(streamName), addQuotes(entry.getKey()), entry.getValue()));
            }
        }
        if (!"".equals(sb.toString())) {
            session.execute(sb.toString());
        }
    }

    public  SaveToCassandraOperationsService(Session session);

}