Java 类com.datastax.driver.core.TableMetadata 实例源码

/**
 * {@inheritDoc}
 */
@Override
public TableInfo get(@Nonnull @NonNull final ConnectorRequestContext context,
                     @Nonnull @NonNull final QualifiedName name) {
    final String keyspace = name.getDatabaseName();
    final String table = name.getTableName();
    log.debug("Attempting to get metadata for Cassandra table {}.{} for request {}", keyspace, table, context);
    try {
        final KeyspaceMetadata keyspaceMetadata = this.getCluster().getMetadata().getKeyspace(keyspace);
        if (keyspaceMetadata == null) {
            throw new DatabaseNotFoundException(name);
        }
        final TableMetadata tableMetadata = keyspaceMetadata.getTable(table);
        if (tableMetadata == null) {
            throw new TableNotFoundException(name);
        }

        final TableInfo tableInfo = this.getTableInfo(name, tableMetadata);
        log.debug("Successfully got metadata for Cassandra table {}.{} for request {}", keyspace, table, context);
        return tableInfo;
    } catch (final DriverException de) {
        log.error(de.getMessage(), de);
        throw this.getExceptionMapper().toConnectorException(de, name);
    }
}

private TableInfo getTableInfo(
    @Nonnull @NonNull final QualifiedName name,
    @Nonnull @NonNull final TableMetadata tableMetadata
) {
    final ImmutableList.Builder<FieldInfo> fieldInfoBuilder = ImmutableList.builder();
    // TODO: Ignores clustering, primary key, index, etc columns. We need to rework TableInfo to support
    for (final ColumnMetadata column : tableMetadata.getColumns()) {
        final String dataType = column.getType().toString();
        fieldInfoBuilder.add(
            FieldInfo.builder()
                .name(column.getName())
                .sourceType(dataType)
                .type(this.typeConverter.toMetacatType(dataType))
                .build()
        );
    }
    return TableInfo.builder()
        .name(QualifiedName.ofTable(name.getCatalogName(), name.getDatabaseName(), tableMetadata.getName()))
        .fields(fieldInfoBuilder.build())
        .build();
}

/**
 * Checks if the main table exists in the database.
 * @param appid name of the {@link com.erudika.para.core.App}
 * @return true if the table exists
 */
public static boolean existsTable(String appid) {
    if (StringUtils.isBlank(appid)) {
        return false;
    }
    if (cluster == null) {
        throw new IllegalStateException("Cassandra client not initialized.");
    }
    try {
        KeyspaceMetadata ks = cluster.getMetadata().getKeyspace(DBNAME);
        TableMetadata table = ks.getTable(getTableNameForAppid(appid));
        return table != null && table.getName() != null;
    } catch (Exception e) {
        return false;
    }
}

public boolean insertRow(String tablename, Map<String, Object> valuesMap, Map<String, String> consistencyInfo, JsonInsert insObj) throws Exception {
    // Note: https://docs.datastax.com/en/cql/3.0/cql/cql_reference/insert_r.html
    String[] parts = tablename.split("\\.");
    KeyspaceMetadata ks = cluster.getMetadata().getKeyspace(parts[0]);
    TableMetadata tableInfo =  ks.getTable(parts[1]);

    StringBuilder fields = new StringBuilder();
    StringBuilder values = new StringBuilder();
    String prefix = "";
    for (String key : valuesMap.keySet()) {
        fields.append(prefix).append(key);
        Object valueObj  = valuesMap.get(key);
        DataType colType = tableInfo.getColumn(key).getType();
        values.append(prefix).append(convertToSqlDataType(colType, valueObj));
        prefix = ", ";
    }

    String suffix = getTTLSuffix(insObj);
    String query = String.format("INSERT INTO %s (%s) VALUES (%s)%s;", tablename, fields.toString(), values.toString(), suffix);
    LOG.debug(query);

    String consistency = extractConsistencyInfo(tablename, consistencyInfo);
    executeCreateQuery(query, consistency);
    return false;
}

private RowIdentifier getRowIdentifier(String keyspace,String tablename, MultivaluedMap<String, String> rowParams){
    String rowIdString="";
    int counter =0;
    TableMetadata tableInfo = MusicCore.returnColumnMetadata(keyspace, tablename);
    String primaryKeyValue="";
    for (MultivaluedMap.Entry<String, List<String>> entry : rowParams.entrySet()){
        String keyName = entry.getKey();
        List<String> valueList = entry.getValue();
        String indValue = valueList.get(0);
        DataType colType = tableInfo.getColumn(entry.getKey()).getType();
        String formattedValue = MusicCore.convertToCQLDataType(colType,indValue);   
        if(counter ==0)
            primaryKeyValue = primaryKeyValue+indValue;
        rowIdString = rowIdString + keyName +"="+ formattedValue;
        if(counter!=rowParams.size()-1)
            rowIdString = rowIdString+" AND ";
        counter = counter +1;
    }
    return new RowIdentifier(primaryKeyValue, rowIdString); 
}

protected void createColumnFamily() {
    final String ks = getKeySpace();
    final String cf = getColumnFamily();

    final KeyspaceMetadata keySpaceMeta = this.cluster.getMetadata().getKeyspace(ks);
    final TableMetadata tableMetadata = keySpaceMeta.getTable(cf);

    // check if the table exists
    if (tableMetadata != null) {
        return;
    }

    final String stmt = String.format("CREATE TABLE %s (\n" +
            "  " + KEY_COLUMN + " text,\n" +
            "  " + COLL_COLUMN + " blob,\n" +
            "  PRIMARY KEY (" + KEY_COLUMN + ")\n" +
            ");", cf);

    getSession().execute(stmt);
}

public static Cluster loadTablesFromRemote(String host, int port, String cfidOverrides) throws IOException {
    Map<String, UUID> cfs = parseOverrides(cfidOverrides);
    Cluster.Builder builder = Cluster.builder().addContactPoints(host).withPort(port);
    Cluster cluster = builder.build();
    Metadata metadata = cluster.getMetadata();
    IPartitioner partitioner = FBUtilities.newPartitioner(metadata.getPartitioner());
    if (DatabaseDescriptor.getPartitioner() == null)
        DatabaseDescriptor.setPartitionerUnsafe(partitioner);
    for (com.datastax.driver.core.KeyspaceMetadata ksm : metadata.getKeyspaces()) {
        if (!ksm.getName().equals("system")) {
            for (TableMetadata tm : ksm.getTables()) {
                String name = ksm.getName()+"."+tm.getName();
                try {
                    CassandraUtils.tableFromCQL(
                            new ByteArrayInputStream(tm.asCQLQuery().getBytes()),
                            cfs.get(name) != null ? cfs.get(name) : tm.getId());
                } catch(SyntaxException e) {
                    // ignore tables that we cant parse (probably dse)
                    logger.debug("Ignoring table " + name + " due to syntax exception " + e.getMessage());
                }
            }
        }
    }
    return cluster;
}

private TableMetadata getTableMetadata(SchemaTableName schemaTableName)
{
    String schemaName = schemaTableName.getSchemaName();
    String tableName = schemaTableName.getTableName();

    KeyspaceMetadata keyspaceMetadata = getCheckedKeyspaceMetadata(schemaName);
    TableMetadata tableMetadata = keyspaceMetadata.getTable(tableName);
    if (tableMetadata != null) {
        return tableMetadata;
    }

    for (TableMetadata table : keyspaceMetadata.getTables()) {
        if (table.getName().equalsIgnoreCase(tableName)) {
            return table;
        }
    }
    throw new TableNotFoundException(schemaTableName);
}

/**
 * Kicks off table generation.
 *
 * @param tables the cassandra table meta data
 * @throws IOException if write to file fails
 */
public static void generate(Collection<TableMetadata> tables) throws IOException {
    String namespaceToUse = MetaData.instance.getTableNamespace();

    for (TableMetadata table : tables) {
        String rawName = table.getName();
        String name = CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.UPPER_CAMEL, rawName);

        TypeSpec.Builder tableClassBuilder = TypeSpec.classBuilder(name)
                .addModifiers(Modifier.PUBLIC)
                .addAnnotation(getTableAnnotation(table.getKeyspace().getName(), rawName));

        addFields(tableClassBuilder, table, name);

        tableClassBuilder.addJavadoc(GeneratorHelper.getJavaDocHeader("Table class for Cassandra - " + rawName, MetaData.instance.getUpdateTime()));

        JavaFile javaFile = JavaFile.builder(namespaceToUse, tableClassBuilder.build()).build();

        Disk.outputFile(javaFile);
    }
}

private static MethodSpec generateSpecificGet(TableMetadata table, ClassName entityTable, int desiredColumns) {
    String query = getBaseQuery(table) + " WHERE ";
    MethodSpec.Builder builder = MethodSpec.methodBuilder("getAll");

    List<ColumnMetadata> columns = table.getPrimaryKey();
    for(int i = 0; i < desiredColumns; i++) {
        ColumnMetadata column = columns.get(i);
        String name = column.getName();
        String newClause = name + "=:" + name;
        if(i != 0) {
            newClause = " AND " + newClause;
        }
        query += newClause;
        builder.addParameter(getSpec(column, true));
    }

    return builder.addModifiers(Modifier.ABSTRACT).addModifiers(Modifier.PUBLIC)
            .returns(ParameterizedTypeName.get(ClassName.get(ListenableFuture.class),
                    ParameterizedTypeName.get(ClassName.get(Result.class), entityTable)))
            .addAnnotation(AnnotationSpec.builder(Query.class).addMember("value", "$S", query).build())
            .build();

}

/**
 * Adds all columns of column family.
 * @param table         Teiid table
 * @param columnFamily  Column family
 */
private void addColumnsToTable(MetadataFactory factory, Table table, TableMetadata columnFamily) {
    for (ColumnMetadata column : columnFamily.getColumns()){

        Class<?> cqlTypeToJavaClass = column.getType().asJavaClass();
        Class<?> teiidRuntimeTypeFromJavaClass = TypeFacility.getRuntimeType(cqlTypeToJavaClass);
        String type = TypeFacility.getDataTypeName(teiidRuntimeTypeFromJavaClass);

        if (column.getType().getName().equals(com.datastax.driver.core.DataType.Name.TIMESTAMP)) {
            type = TypeFacility.RUNTIME_NAMES.TIMESTAMP;
        } else if (column.getType().getName().equals(com.datastax.driver.core.DataType.Name.CUSTOM)
                || column.getType().getName().equals(com.datastax.driver.core.DataType.Name.BLOB)) {
            type = TypeFacility.RUNTIME_NAMES.VARBINARY;
        }

        Column c = factory.addColumn(column.getName(), type, table);
        c.setUpdatable(true);
        if (column.getIndex() != null) {
            c.setSearchType(SearchType.Searchable);
        }
        else {
            c.setSearchType(SearchType.Unsearchable);
        }
    }
}

public List<Column> getSchema(String keySpace, String tableName) {
    Metadata m = session.getCluster().getMetadata();
    KeyspaceMetadata km = m.getKeyspace(keySpace);
    if (km == null)
        return null;
    TableMetadata tm = km.getTable(tableName);
    if (tm == null)
        return null;
    // build schema
    List<Column> columns = new LinkedList<Column>();
    for (ColumnMetadata cm : tm.getColumns()) {
        if (!meta.contains(cm.getName()))
            columns.add(Column.newBuilder().setName(cm.getName())
                    .setType(toSimbaType(cm.getType().toString())).build());
    }

    return columns;

}

private void fetchKeys()
{
    // get CF meta data
    TableMetadata tableMetadata = session.getCluster()
                                         .getMetadata()
                                         .getKeyspace(Metadata.quote(keyspace))
                                         .getTable(Metadata.quote(cfName));
    if (tableMetadata == null)
    {
        throw new RuntimeException("No table metadata found for " + keyspace + "." + cfName);
    }
    //Here we assume that tableMetadata.getPartitionKey() always
    //returns the list of columns in order of component_index
    for (ColumnMetadata partitionKey : tableMetadata.getPartitionKey())
    {
        partitionKeys.add(partitionKey.getName());
    }
}

private static <T, K> DatastaxCrud<T, K> createCrud(Type target, Type keyTarget,
                                                    TableMetadata tableMetadata,
                                                    Session session,
                                                    DatastaxMapperFactory mapperFactory) {
    DatastaxMapper<T> selectMapper = selectMapper(target, tableMetadata, mapperFactory);
    return new DatastaxCrud<T, K>(
            session.prepare(insertQuery(tableMetadata)),
            session.prepare(insertQuery(tableMetadata, "TTL", "TIMESTAMP")),
            session.prepare(insertQuery(tableMetadata, "TTL" )),
            session.prepare(insertQuery(tableMetadata, "TIMESTAMP")),
            session.prepare(readQuery(tableMetadata)),
            session.prepare(deleteQuery(tableMetadata)),
            session.prepare(deleteQueryWithTimestamp(tableMetadata)),
            DatastaxCrudFactory.<T>insertSetter(target, tableMetadata, mapperFactory, 0),
            DatastaxCrudFactory.<K>keySetter(keyTarget, tableMetadata, mapperFactory, 0),
            DatastaxCrudFactory.<K>keySetter(keyTarget, tableMetadata, mapperFactory, 1),
            selectMapper,
            tableMetadata.getColumns().size(), session);
}

private static String insertQuery(TableMetadata tableMetadata, String... options) {
    Insert insert = QueryBuilder.insertInto(tableMetadata);

    if (options != null) {
        Insert.Options using = insert.using();
        for (String option : options) {
            if ("TTL".equals(option)) {
                using.and(QueryBuilder.ttl(QueryBuilder.bindMarker()));
            } else {
                using.and(QueryBuilder.timestamp(QueryBuilder.bindMarker()));
            }
        }
    }

    List<ColumnMetadata> columns = tableMetadata.getColumns();

    for(ColumnMetadata column : columns) {
        insert.value(column.getName(), QueryBuilder.bindMarker());
    }

    return insert.toString();
}

private void createTableIfNecessary( RetentionTable table, KeyspaceMetadata metadata ) {
    for ( TableMetadata meta : metadata.getTables()) {
        log.debug( "Comparing " + meta.getName() + " with " + table.tableName() );
        if ( meta.getName().equalsIgnoreCase( table.tableName() )) {
            return;
        }
    }

    StringBuilder query = new StringBuilder();
    query.append( "CREATE TABLE " ).append( table.tableName() ).append( " (" );
    query.append( COL_NAME ).append( " text, " );
    query.append( COL_TIME ).append( " bigint, " );
    query.append( COL_VALUE ).append( " double, " );
    query.append( "PRIMARY KEY (" ).append( COL_NAME ).append( ", " ).append( COL_TIME ).append( ")");
    query.append( ");" );
    log.debug( "Creating table with query: <" + query.toString() + ">");
    try {
        session.execute( query.toString() );
    } catch( AlreadyExistsException e ) {
        // Some other gatherer might have already created the same table.
    }
}

public void createTableIfNecessary( RetentionTable table ) {
    if ( session == null ) {
        open();
    }
    Collection<TableMetadata> tables = cluster.getMetadata().getKeyspace( keyspace ).getTables();
    for ( TableMetadata meta : tables ) {
        if ( meta.getName().equalsIgnoreCase( table.tableName() ) ) {
            return;
        }
    }

    if ( dryRun ) {
        log.debug( "Creating table " + table );
        return;
    }

    StringBuilder query = new StringBuilder();
    query.append( "CREATE TABLE IF NOT EXISTS " ).append( table.tableName() ).append( " (" );
    query.append( COL_NAME ).append( " text, " );
    query.append( COL_TIME ).append( " bigint, " );
    query.append( COL_VALUE ).append( " double, " );
    query.append( "PRIMARY KEY (" ).append( COL_NAME ).append( ", " ).append( COL_TIME ).append( ")" );
    query.append( ");" );
    session.execute( query.toString() );
    EventBusManager.fire( new CreateTableEvent( System.currentTimeMillis(), table ) );
}

private static TableMetadata loadTableMetadata(Session session, Tablename tablename) {

    final String keyspacename = tablename.getKeyspacename();

    if (keyspacename == null) {
        throw new IllegalStateException("no keyspacename assigned for " + tablename);

    } else {
        final TableMetadata tableMetadata = session.getCluster().getMetadata().getKeyspace(tablename.getKeyspacename()).getTable(tablename.getTablename());
        if (tableMetadata == null) {
            throw new RuntimeException("table " + tablename + " is not defined");
        }

        return tableMetadata;
    }
}

private void updateTwcsDtcsGcSeconds() throws Exception {
    logger.info("updating gc_grace_seconds on TWCS/DTCS tables ...");
    for (TableMetadata table : keyspaceMetadata.getTables()) {
        String compaction = table.getOptions().getCompaction().get("class");
        if (compaction == null) {
            continue;
        }
        if (compaction.equals("org.apache.cassandra.db.compaction.TimeWindowCompactionStrategy")
                || compaction.equals(
                        "org.apache.cassandra.db.compaction.DateTieredCompactionStrategy")) {
            // see gc_grace_seconds related comments in Sessions.createTableWithTWCS()
            // for reasoning behind the value of 4 hours
            session.execute("alter table " + table.getName() + " with gc_grace_seconds = "
                    + HOURS.toSeconds(4));
        }
    }
    logger.info("updating gc_grace_seconds on TWCS/DTCS tables - complete");
}

private static @Nullable Integer getSchemaVersion(Session session, KeyspaceMetadata keyspace)
        throws Exception {
    ResultSet results =
            session.execute("select schema_version from schema_version where one = 1");
    Row row = results.one();
    if (row != null) {
        return row.getInt(0);
    }
    TableMetadata agentTable = keyspace.getTable("agent");
    if (agentTable != null && agentTable.getColumn("system_info") != null) {
        // special case, this is glowroot version 0.9.1, the only version supporting upgrades
        // prior to schema_version table
        return 1;
    }
    // new installation
    return null;
}

@BeforeMethod(alwaysRun = true)
public void initMethod() {
    session.execute("TRUNCATE tenants");
    session.execute("TRUNCATE data");
    session.execute(String.format("TRUNCATE %s", DataAccessImpl.OUT_OF_ORDER_TABLE_NAME));
    session.execute("TRUNCATE data_compressed");
    session.execute("TRUNCATE metrics_idx");
    session.execute("TRUNCATE retentions_idx");
    session.execute("TRUNCATE metrics_tags_idx");
    session.execute("TRUNCATE leases");

    // Need to truncate all the temp tables also..
    for (TableMetadata tableMetadata : session.getCluster().getMetadata().getKeyspace(session.getLoggedKeyspace())
            .getTables()) {
        if(tableMetadata.getName().startsWith(DataAccessImpl.TEMP_TABLE_NAME_PROTOTYPE)) {
            session.execute(String.format("TRUNCATE %s", tableMetadata.getName()));
        }
    }

    NumericDataPointCollector.createPercentile = defaultCreatePercentile;
}

/**
 * Creates the output column family if not exists. <br/>
 * We first check if the column family exists. <br/>
 * If not, we get the first element from <i>tupleRDD</i> and we use it as a template to get columns metadata.
 * <p>
 * This is a very heavy operation since to obtain the schema we need to get at least one element of the output RDD.
 * </p>
 *
 * @param first the pair RDD.
 */
public void createOutputTableIfNeeded(Tuple2<Cells, Cells> first) {

    TableMetadata metadata = getSession()
            .getCluster()
            .getMetadata()
            .getKeyspace(this.catalog)
            .getTable(quote(this.table));

    if (metadata == null && !createTableOnWrite) {
        throw new DeepIOException("Cannot write RDD, output table does not exists and configuration object has " +
                "'createTableOnWrite' = false");
    }

    if (metadata != null) {
        return;
    }

    if (first._1() == null || first._1().isEmpty()) {
        throw new DeepNoSuchFieldException("no key structure found on row metadata");
    }
    String createTableQuery = createTableQueryGenerator(first._1(), first._2(), this.catalog,
            quote(this.table));
    getSession().execute(createTableQuery);
    waitForNewTableMetadata();
}

/**
 * {@inheritDoc}
 */

@Override
public synchronized Map<String, Cell> columnDefinitions() {
    if (columnDefinitionMap != null) {
        return columnDefinitionMap;
    }

    TableMetadata tableMetadata = fetchTableMetadata();

    if (tableMetadata == null && !createTableOnWrite) {
        LOG.warn("Configuration not suitable for writing RDD: output table does not exists and configuration " +
                "object has 'createTableOnWrite' = false");

        return null;
    } else if (tableMetadata == null) {
        return null;
    }

    initColumnDefinitionMap(tableMetadata);

    return columnDefinitionMap;
}

private void validateTableMetadata(TableMetadata tableMetadata) {

        if (tableMetadata == null && !isWriteConfig) {
            throw new IllegalArgumentException(String.format("Column family {%s.%s} does not exist", catalog,
                    table));
        }

        if (tableMetadata == null && !createTableOnWrite) {
            throw new IllegalArgumentException(String.format("Column family {%s.%s} does not exist and " +
                    "createTableOnWrite = false", catalog, table));
        }

        if (!ArrayUtils.isEmpty(inputColumns)) {
            for (String column : inputColumns) {
                assert tableMetadata != null;
                ColumnMetadata columnMetadata = tableMetadata.getColumn(column);

                if (columnMetadata == null) {
                    throw new DeepNoSuchFieldException("No column with name " + column + " has been found on table "
                            + this.catalog + "." + this.table);
                }
            }
        }

    }

private void validateAdditionalFilters(TableMetadata tableMetadata) {
    for (Map.Entry<String, Serializable> entry : additionalFilters.entrySet()) {
        /* check if there's an index specified on the provided column */
        ColumnMetadata columnMetadata = tableMetadata.getColumn(entry.getKey());

        if (columnMetadata == null) {
            throw new DeepNoSuchFieldException("No column with name " + entry.getKey() + " has been found on " +
                    "table " + this.catalog + "." + this.table);
        }

        if (columnMetadata.getIndex() == null) {
            throw new DeepIndexNotFoundException("No index has been found on column " + columnMetadata.getName()
                    + " on table " + this.catalog + "." + this.table);
        }
    }
}

public CassandraTable(
    final Session session,
    final TableMetadata table,
    final ConsistencyLevel consistencyLevel,
    final String bodyColumn,
    final boolean ignoreCase) {
  this.session = session;
  this.table = table;
  this.consistencyLevel = consistencyLevel;
  this.bodyColumn = bodyColumn;

  this.columns = table.getColumns();
  this.totalColumns = this.columns.size();
  this.primaryKeys = new ArrayList<String>();
  for (final ColumnMetadata column : table.getPrimaryKey()) {
    primaryKeys.add(column.getName());
  }

  this.ignoreCase = ignoreCase;
}

private String getKeyColumn(String columnFamily) {
    KeyspaceMetadata keyspaceMetadata = session.getCluster().getMetadata().getKeyspace(keyspace);
    TableMetadata tableMetadata = keyspaceMetadata.getTable(columnFamily);

    if (tableMetadata == null) {
        return null;
    }

    for (String key : new String[] { "\"KEY\"", "key" }) {
        if (tableMetadata.getColumn(key) != null) {
            return key;
        }
    }

    return null;
}

@SuppressWarnings("unchecked")
public static JSONArray marshallKeyspaces(List<KeyspaceMetadata> keyspaces, boolean flatten)
        throws UnsupportedEncodingException {
    JSONArray keyspaceJson = new JSONArray();
    if (flatten) {
        for (KeyspaceMetadata keyspace : keyspaces) {
            for (TableMetadata table : keyspace.getTables()) {
                JSONObject json = new JSONObject();
                json.put("keyspace", keyspace.getName());
                json.put("columnFamily", table.getName());
                keyspaceJson.add(json);
            }
        }
    } 
    return keyspaceJson;
}

private void ensureTableExists() {
    KeyspaceMetadata keyspaceMetadata =
        session.getCluster().getMetadata().getKeyspace(session.getLoggedKeyspace());
    TableMetadata tableMetadata = keyspaceMetadata.getTable(TABLE_NAME);

    if (tableMetadata != null) {
        LOG.debug("Versioning column family already exists, skipping creation.");
        ensureTableSchema(tableMetadata);
        return;
    }

    LOG.info("Creating versioning column family.");
    session.execute(
        "CREATE TABLE " + TABLE_NAME + " ("
            + "key text PRIMARY KEY,"
            + "executed timestamp"
            + ");");

    LOG.debug("Versioning column family created.");
}

private static void ensureTableSchema(TableMetadata tableMetadata) throws IllegalStateException {
    ColumnMetadata primaryKey = tableMetadata.getPrimaryKey().get(0);

    if (!primaryKey.getName().equals("key")) {
        throw new IllegalStateException(String.format("The name of primary key in table [%s] should be 'key'", TABLE_NAME));
    }

    if (primaryKey.getType() != DataType.text()) {
        throw new IllegalStateException(String.format("Primary key in table [%s] should have type 'text'", TABLE_NAME));
    }

    ColumnMetadata executedColumn = tableMetadata.getColumn("executed");

    if (executedColumn == null) {
        throw new IllegalStateException(String.format("Cannot find column 'executed' in table [%s]", TABLE_NAME));
    }

    if (executedColumn.getType() != DataType.timestamp()) {
        throw new IllegalStateException(String.format("Column 'executed' in table [%s] should have type 'timestamp'", TABLE_NAME));
    }
}

/**
 * Performs an analysis of the given keyspace in a Cassandra cluster
 * {@link Cluster} instance and detects the cassandra types structure based
 * on the metadata provided by the datastax cassandra java client.
 *
 * @see #detectTable(TableMetadata)
 *
 * @param cluster
 *            the cluster to inspect
 * @param keyspaceName
 * @return a mutable schema instance, useful for further fine tuning by the
 *         user.
 */
public static SimpleTableDef[] detectSchema(Cluster cluster, String keyspaceName) {
    final Metadata metadata = cluster.getMetadata();
    final KeyspaceMetadata keyspace = metadata.getKeyspace(keyspaceName);
    if (keyspace == null) {
        throw new IllegalArgumentException("Keyspace '" + keyspaceName + "' does not exist in the database");
    }
    final Collection<TableMetadata> tables = keyspace.getTables();
    final SimpleTableDef[] result = new SimpleTableDef[tables.size()];
    int i = 0;
    for (final TableMetadata tableMetaData : tables) {
        final SimpleTableDef table = detectTable(tableMetaData);
        result[i] = table;
        i++;
    }
    return result;
}

@Override
protected Schema getMainSchema() throws MetaModelException {
    final MutableSchema theSchema = new MutableSchema(getMainSchemaName());
    for (final SimpleTableDef tableDef : tableDefs) {
        final MutableTable table = tableDef.toTable().setSchema(theSchema);

        final TableMetadata cassandraTable = cassandraCluster.getMetadata().getKeyspace(keySpaceName).getTable(table
                .getName());
        if (cassandraTable != null) {
            final List<ColumnMetadata> primaryKeys = cassandraTable.getPrimaryKey();
            for (ColumnMetadata primaryKey : primaryKeys) {
                final MutableColumn column = (MutableColumn) table.getColumnByName(primaryKey.getName());
                if (column != null) {
                    column.setPrimaryKey(true);
                }
                column.setNativeType(primaryKey.getType().getName().name());
            }
        }

        theSchema.addTable(table);
    }
    return theSchema;
}

public void clean() {
    log.info("Cleaning all tables");
    for (KeyspaceMetadata keyspaceMetadata : session.getCluster().getMetadata().getKeyspaces()) {
        String keyspace = keyspaceMetadata.getName();
        if (keyspace.startsWith("system")) {
            continue;
        }
        for (TableMetadata metadata : keyspaceMetadata.getTables()) {
            String statement = "TRUNCATE TABLE " + keyspace + "." + metadata.getName();
            session.execute(statement);
        }
    }
}

@Override
public List<String> getTableList(String keyspacename) {
  Collection<TableMetadata> tables =
      cassandraclusterMap.get(keyspacename).getMetadata().getKeyspace(keyspacename).getTables();

  // to convert to list of the names
  return tables.stream().map(tm -> tm.getName()).collect(Collectors.toList());
}

@Test
public void testZgetCluster() {
  Cluster cluster = connectionManager.getCluster(cassandraKeySpace);
  Collection<TableMetadata> tables = cluster.getMetadata().getKeyspace(cassandraKeySpace).getTables();
  List<String> tableList = tables.stream().map(tm -> tm.getName()).collect(Collectors.toList());
  assertTrue(tableList.contains(JsonKey.USER));
}

private void setupColumnFamiliesCombo() {
    String nodes = this.transMeta.environmentSubstitute(this.hostText.getText());
    String port_s = this.transMeta.environmentSubstitute(this.portText.getText());
    String username = this.transMeta.environmentSubstitute(this.userText.getText());
    String password = this.transMeta.environmentSubstitute(this.passText.getText());
    String keyspace = this.transMeta.environmentSubstitute(this.keyspaceText.getText());
    Boolean withSSL = this.sslenabledBut.getSelection();
    String truststorefile = this.transMeta.environmentSubstitute(this.truststorefileText.getText());
    String truststorepass = this.transMeta.environmentSubstitute(this.truststorepassText.getText());
    ConnectionCompression compression = ConnectionCompression.fromString(this.wCompression.getText());
    Cluster cluster;
    try {
        try {
            this.connection = Utils.connect(nodes, port_s, username, password, keyspace, withSSL, truststorefile, truststorepass, compression);
            cluster = this.connection.getSession().getCluster();
            Collection<TableMetadata> colFams = cluster.getMetadata().getKeyspace(this.transMeta.environmentSubstitute(this.keyspaceText.getText())).getTables();
            this.columnFamilyCombo.removeAll();
            for (TableMetadata row : colFams) {
                this.columnFamilyCombo.add(row.getName());
            }
        }
        catch (Exception ex) {
            this.logError(String.valueOf(BaseMessages.getString(PKG, "AutSoftCassandraOutputDialog.Error.ProblemGettingSchemaInfo.Message", (String[])new String[0])) + ":\n\n" + ex.getMessage(), ex);
            new org.pentaho.di.ui.core.dialog.ErrorDialog(this.shell, BaseMessages.getString(PKG, "AutSoftCassandraOutputDialog.Error.ProblemGettingSchemaInfo.Title", (String[])new String[0]), String.valueOf(BaseMessages.getString(PKG, "AutSoftCassandraOutputDialog.Error.ProblemGettingSchemaInfo.Message", (String[])new String[0])) + ":\n\n" + ex.getMessage(), ex);
            if (this.connection != null) {
                this.connection.release();
            }
        }
    }
    finally {
        if (this.connection != null) {
            this.connection.release();
        }
    }
}

private static void syncQuorum(String key){
    logger.info("Performing sync operation---");
    String[] splitString = key.split("\\.");
    String keyspaceName = splitString[0];
    String tableName = splitString[1];
    String primaryKeyValue = splitString[2];

    //get the primary key d
    TableMetadata tableInfo = returnColumnMetadata(keyspaceName, tableName);
    String primaryKeyName = tableInfo.getPrimaryKey().get(0).getName();//we only support single primary key
    DataType primaryKeyType = tableInfo.getPrimaryKey().get(0).getType();
    String cqlFormattedPrimaryKeyValue = convertToCQLDataType(primaryKeyType, primaryKeyValue);

    //get the row of data from a quorum
    String selectQuery =  "SELECT *  FROM "+keyspaceName+"."+tableName+ " WHERE "+primaryKeyName+"="+cqlFormattedPrimaryKeyValue+";"; 
    ResultSet results = getDSHandle().executeCriticalGet(selectQuery);

    //write it back to a quorum
    Row row = results.one();
    ColumnDefinitions colInfo = row.getColumnDefinitions();
    int totalColumns = colInfo.size();
    int counter =1;
    String fieldValueString="";
    for (Definition definition : colInfo){
        String colName = definition.getName();
        if(colName.equals(primaryKeyName))
            continue; 
        DataType colType = definition.getType();
        Object valueObj = getDSHandle().getColValue(row, colName, colType); 
        String valueString = convertToCQLDataType(colType,valueObj);    
        fieldValueString = fieldValueString+ colName+"="+valueString;
        if(counter!=(totalColumns-1))
            fieldValueString = fieldValueString+",";
        counter = counter +1;
    }

    String updateQuery =  "UPDATE "+keyspaceName+"."+tableName+" SET "+fieldValueString+" WHERE "+primaryKeyName+"="+cqlFormattedPrimaryKeyValue+";";
    getDSHandle().executePut(updateQuery, "critical");
}

/**
 * Because of the way databus tables were created historically using Astyanax and Cassandra 1.2 there may be
 * inconsistency in the names of the CQL columns in the subscription table.  To be safe read the table metadata
 * to get the column names.
 */
private void getColumnNames() {
    TableMetadata table = _keyspace.getKeyspaceMetadata().getTable(CF_NAME);
    _rowkeyColumn = table.getPrimaryKey().get(0).getName();
    _subscriptionNameColumn = table.getPrimaryKey().get(1).getName();
    _subscriptionColumn = table.getColumns().get(2).getName();
}

/**
 * All three placement tables -- delta, audit, and delta history -- follow the same DDL.
 */
private TableDDL createTableDDL(String tableName) {
    TableMetadata tableMetadata = _keyspace.getKeyspaceMetadata().getTable(tableName);
    String rowKeyColumnName = tableMetadata.getPrimaryKey().get(0).getName();
    String timeSeriesColumnName = tableMetadata.getPrimaryKey().get(1).getName();
    String valueColumnName = tableMetadata.getColumns().get(2).getName();

    return new TableDDL(tableMetadata, rowKeyColumnName, timeSeriesColumnName, valueColumnName);
}

private BlockedDeltaTableDDL createBlockedDeltaTableDDL(String tableName) {
    TableMetadata tableMetadata = _keyspace.getKeyspaceMetadata().getTable(tableName);
    String rowKeyColumnName = tableMetadata.getPrimaryKey().get(0).getName();
    String timeSeriesColumnName = tableMetadata.getPrimaryKey().get(1).getName();
    String blockColumnName = tableMetadata.getPrimaryKey().get(2).getName();
    String valueColumnName = tableMetadata.getColumns().get(3).getName();

    return new BlockedDeltaTableDDL(tableMetadata, rowKeyColumnName, timeSeriesColumnName, valueColumnName, blockColumnName);
}