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

/**
 * Returns a map with column name as key and column date type as value.
 *
 * The value might be as simple as "Boolean" or more complex like
 *  - "Set|Boolean"
 *  - "List|String"
 *  - "Map|String|Integer"
 *  these are cases when the data type is a container of primitive data types.
 *
 * @param tableName
 * @return
 * @throws DbException
 */
public Map<String, String> getColumnInfo(
                                          String tableName ) throws DbException {

    connect();

    ResultSet results = session.execute("SELECT * FROM " + this.dbName + "." + tableName + " LIMIT 1");

    Map<String, String> columnInfo = new HashMap<String, String>();
    for (Definition columnDefinition : results.getColumnDefinitions()) {
        DataType dataType = columnDefinition.getType();
        String dataTypeName = dataType.getName().name();
        if ("Set".equalsIgnoreCase(dataTypeName)) {
            dataTypeName = dataTypeName + "|" + dataType.getTypeArguments().get(0);
        } else if ("List".equalsIgnoreCase(dataTypeName)) {
            dataTypeName = dataTypeName + "|" + dataType.getTypeArguments().get(0);
        } else if ("Map".equalsIgnoreCase(dataTypeName)) {
            dataTypeName = dataTypeName + "|" + dataType.getTypeArguments().get(0) + "|"
                           + dataType.getTypeArguments().get(1);
        }
        columnInfo.put(columnDefinition.getName(), dataTypeName);
    }

    return columnInfo;
}

public static String convertToCQLDataType(DataType type,Object valueObj){
    String value ="";
    switch (type.getName()) {
    case UUID:
        value = valueObj+"";
        break;
    case TEXT: case VARCHAR:
        String valueString = valueObj+"";
        valueString = valueString.replace("'", "''");
        value = "'"+valueString+"'";
        break;
    case MAP:{
        Map<String,Object> otMap = (Map<String,Object>)valueObj;
        value = "{"+jsonMaptoSqlString(otMap, ",")+"}";
        break;
    }   
    default:
        value = valueObj+"";
        break;
    }
    return value;
}

public Object getColValue(Row row, String colName, DataType colType){   
    switch(colType.getName()){
    case VARCHAR: 
        return row.getString(colName);
    case UUID: 
        return row.getUUID(colName);
    case VARINT: 
        return row.getVarint(colName);
    case BIGINT: 
        return row.getLong(colName);
    case INT: 
        return row.getInt(colName);
    case FLOAT: 
        return row.getFloat(colName);   
    case DOUBLE: 
        return row.getDouble(colName);
    case BOOLEAN: 
        return row.getBool(colName);
    case MAP: 
        return row.getMap(colName, String.class, String.class);
    default: 
        return null;
    }
}

public static Object convertToActualDataType(DataType colType,Object valueObj){
    String valueObjString = valueObj+"";
    switch(colType.getName()){
    case UUID: 
        return UUID.fromString(valueObjString);
    case VARINT: 
        return BigInteger.valueOf(Long.parseLong(valueObjString));
    case BIGINT: 
        return Long.parseLong(valueObjString);
    case INT: 
        return Integer.parseInt(valueObjString);
    case FLOAT: 
        return Float.parseFloat(valueObjString);    
    case DOUBLE: 
        return Double.parseDouble(valueObjString);
    case BOOLEAN: 
        return Boolean.parseBoolean(valueObjString);
    case MAP: 
        return (Map<String,Object>)valueObj;
    default:
        return valueObjString;
    }
}

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); 
}

/**
 * Build schema programmatically
 * <p>
 * DDL equivalent:
 * 
 * <pre>
 * CREATE TABLE messages (
 *   sessionId uuid,
 *   seqNo bigint,
 *   message blob,
 *   PRIMARY KEY  (sessionId, seqNo ) );
 * </pre>
 * 
 * @throws StoreException if the store is not open
 *
 */
public void buildSchema() throws StoreException {
  if (session != null) {
    // Appropriate for a local test only
    session.execute(new SimpleStatement("CREATE KEYSPACE IF NOT EXISTS " + KEYSPACE_NAME
        + " WITH REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 }"));
    System.out.format("Keyspace %s available\n", KEYSPACE_NAME);

    Create create = SchemaBuilder.createTable(KEYSPACE_NAME, TABLE_NAME).ifNotExists()
        .addPartitionKey(SESSION_ID_COLNAME, DataType.uuid())
        .addClusteringColumn(SEQ_NO_COLNAME, DataType.bigint())
        .addColumn(MESSAGE_COLNAME, DataType.blob());

    ResultSet resultSet = session.execute(create);
    System.out.format("Table %s available\n", TABLE_NAME);
  } else {
    throw new StoreException("Schema not created; store not open");
  }
}

private static DataType cqlType(FieldInfo info) {
    if (String.class.equals(info.type())) {
        return DataType.text();
    } else if (Boolean.class.equals(info.type()) || Boolean.TYPE.equals(info.type())) {
        return DataType.cboolean();
    } else if (Long.class.equals(info.type()) || Long.TYPE.equals(info.type())) {
        return DataType.cint();
    } else if (Double.class.equals(info.type()) || Double.TYPE.equals(info.type())) {
        return DataType.cint();
    } else if (Float.class.equals(info.type()) || Float.TYPE.equals(info.type())) {
        return DataType.cint();
    } else if (Integer.class.equals(info.type()) || Integer.TYPE.equals(info.type())) {
        return DataType.cint();
    } else if (LocalDate.class.equals(info.type())) {
        return DataType.date();
    } else if (Enum.class.isAssignableFrom(info.type())) {
        return DataType.set(DataType.text());
    } else if (Collection.class.isAssignableFrom(info.type())) {
        return DataType.set(DataType.text());
    }
    throw new IllegalArgumentException("unknown type " + info.type() + " for " + info.id());
}

private static DataType cqlType(FieldInfo info) {
    if (String.class.equals(info.type())) {
        return text();
    } else if (Boolean.class.equals(info.type()) || Boolean.TYPE.equals(info.type())) {
        return DataType.cboolean();
    } else if (Long.class.equals(info.type()) || Long.TYPE.equals(info.type())) {
        return DataType.cint();
    } else if (LocalDate.class.equals(info.type())) {
        return DataType.date();
    } else if (Enum.class.isAssignableFrom(info.type())) {
        return DataType.set(text());
    } else if (Collection.class.isAssignableFrom(info.type())) {
        return DataType.set(text());
    }
    throw new IllegalArgumentException("unknown type " + info.type() + " for " + info.id());
}

private CassandraColumnHandle buildColumnHandle(ColumnMetadata columnMeta, boolean partitionKey, boolean clusteringKey, int ordinalPosition, boolean hidden)
{
    CassandraType cassandraType = CassandraType.getCassandraType(columnMeta.getType().getName());
    List<CassandraType> typeArguments = null;
    if (cassandraType != null && cassandraType.getTypeArgumentSize() > 0) {
        List<DataType> typeArgs = columnMeta.getType().getTypeArguments();
        switch (cassandraType.getTypeArgumentSize()) {
            case 1:
                typeArguments = ImmutableList.of(CassandraType.getCassandraType(typeArgs.get(0).getName()));
                break;
            case 2:
                typeArguments = ImmutableList.of(CassandraType.getCassandraType(typeArgs.get(0).getName()), CassandraType.getCassandraType(typeArgs.get(1).getName()));
                break;
            default:
                throw new IllegalArgumentException("Invalid type arguments: " + typeArgs);
        }
    }
    boolean indexed = columnMeta.getIndex() != null;
    return new CassandraColumnHandle(connectorId, columnMeta.getName(), ordinalPosition, cassandraType, typeArguments, partitionKey, clusteringKey, indexed, hidden);
}

/**
 * Handle getting the class names for parameterized types.
 *
 * @param type the cassandra data type to extract from
 * @return the parameterized type result
 */
public static TypeResult getClassWithTypes(DataType type) {
    ClassName outer = getRawType(type);

    List<TypeName> generics = new ArrayList<>();
    boolean hasFrozenType = false;
    for(DataType genericType : type.getTypeArguments()) {
        if(Udt.instance.isUdt(genericType)) {
            generics.add(MetaData.getClassNameForUdt((UserType) genericType));
            if(genericType.isFrozen()) {
                hasFrozenType = true;
            }
        } else {
            generics.add(getRawType(genericType).box());
        }
    }
    return new TypeResult(ParameterizedTypeName.get(outer, generics.toArray(new TypeName[generics.size()])), hasFrozenType);
}

/**
 * Get a setter spec for a entity field.
 *
 * @param field the field name
 * @param type the cassandra field type
 * @return the setter method spec
 */
public static MethodSpec getSetter(String field, DataType type) {
    String methodRoot = CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.UPPER_CAMEL, field);
    String paramName = CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL, field);
    MethodSpec.Builder spec;

    if (type.getTypeArguments().size() == 0) {
        if(Udt.instance.isUdt(type)) {
            spec = MethodSpec.methodBuilder("set" + methodRoot).addParameter(MetaData.getClassNameForUdt((UserType) type), paramName);
        } else {
            spec = MethodSpec.methodBuilder("set" + methodRoot).addParameter(getRawType(type), paramName);
        }
    } else {
        TypeResult result = getClassWithTypes(type);
        spec = MethodSpec.methodBuilder("set" + methodRoot).addParameter(result.type, paramName);
    }
    spec.addModifiers(Modifier.PUBLIC).addStatement("this.$L = $L", paramName, paramName);

    return spec.build();
}

/**
 * Get a getter spec for a entity field.
 *
 * @param field the field name
 * @param type the cassandra field type
 * @return the getter method spec
 */
public static MethodSpec getGetter(String field, DataType type) {
    String methodRoot = CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.UPPER_CAMEL, field);
    String paramName = CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL, field);
    MethodSpec.Builder spec;

    if (type.getTypeArguments().size() == 0) {
        if(Udt.instance.isUdt(type)) {
            spec = MethodSpec.methodBuilder("get" + methodRoot).returns(MetaData.getClassNameForUdt((UserType) type));
        } else {
            spec = MethodSpec.methodBuilder("get" + methodRoot).returns(getRawType(type));
        }
    } else {
        TypeResult result = getClassWithTypes(type);
        spec = MethodSpec.methodBuilder("get" + methodRoot).returns(result.type);
    }
    spec.addModifiers(Modifier.PUBLIC).addStatement("return $L", paramName);

    return spec.build();
}

/**
 * Gets field value as an object having Cassandra compatible type.
 * This it could be stored directly into Cassandra without any conversions.
 *
 * @param obj Object instance.
 * @param serializer {@link org.apache.ignite.cache.store.cassandra.serializer.Serializer} to use.
 * @return Object to store in Cassandra table column.
 */
public Object getValueFromObject(Object obj, Serializer serializer) {
    Object val = accessor.getValue(obj);

    if (val == null)
        return null;

    DataType.Name cassandraType = PropertyMappingHelper.getCassandraType(val.getClass());

    if (cassandraType != null)
        return val;

    if (serializer == null) {
        throw new IllegalStateException("Can't serialize value from object '" +
            val.getClass().getName() + "' field '" + name + "', cause there is no BLOB serializer specified");
    }

    return serializer.serialize(val);
}

/**
      * Spec says "database specific type name"; for Cassandra this means the AbstractType.
      */
     public String getColumnTypeName(int column) throws SQLException
     {

         //checkIndex(column);
         DataType type = null;
         try {
    if(currentRow!=null){
        type = currentRow.getColumnDefinitions().getType(column-1);
    }else{
        type = driverResultSet.getColumnDefinitions().getType(column-1); 
    }

    return type.toString();
} catch (Exception e) {
    return DataType.varchar().toString();
}            
     }

protected UDFunction(FunctionName name,
                     List<ColumnIdentifier> argNames,
                     List<AbstractType<?>> argTypes,
                     DataType[] argDataTypes,
                     AbstractType<?> returnType,
                     DataType returnDataType,
                     boolean calledOnNullInput,
                     String language,
                     String body)
{
    super(name, argTypes, returnType);
    assert new HashSet<>(argNames).size() == argNames.size() : "duplicate argument names";
    this.argNames = argNames;
    this.language = language;
    this.body = body;
    this.argCodecs = UDHelper.codecsFor(argDataTypes);
    this.returnCodec = UDHelper.codecFor(returnDataType);
    this.calledOnNullInput = calledOnNullInput;
}

BoundStatement bindRow(Row row)
{
    for (int i = 0 ; i < argumentIndex.length ; i++)
    {
        Object value = row.get(argumentIndex[i]);
        if (definitions.getType(i).getName().equals(DataType.date().getName()))
        {
            // the java driver only accepts com.datastax.driver.core.LocalDate for CQL type "DATE"
            value= LocalDate.fromDaysSinceEpoch((Integer) value);
        }
        bindBuffer[i] = value;
        if (bindBuffer[i] == null && !spec.partitionGenerator.permitNulls(argumentIndex[i]))
            throw new IllegalStateException();
    }
    return statement.bind(bindBuffer);
}

public void testDataTypeMapInstantiation() {
    DataType type = DataType.map(DataType.text(), DataType.bigint());

    CellValidator cv = cellValidator(type);
    assertNotNull(cv);
    assertEquals(cv.getValidatorClassName(), MapType.class.getName());
    assertNotNull(cv.getValidatorTypes());
    assertEquals(cv.validatorKind(), Kind.MAP);
    assertEquals(cv.getValidatorTypes().size(), 2);
    Iterator<String> types = cv.getValidatorTypes().iterator();
    assertEquals(types.next(), "text");
    assertEquals(types.next(), "bigint");
    assertEquals(DataType.Name.MAP, cv.getCqlTypeName());

    try {
        Collection<String> ctypes = cv.getValidatorTypes();
        ctypes.add("test");
        fail("Validator types collection must be inmutable");
    } catch (Exception ex) {
        // ok
    }

    //        assertNotNull(cv.getAbstractType());
    //        assertEquals(cv.getAbstractType(), MapType.getInstance(UTF8Type.instance, LongType.instance));
}

public static Object parseValue(final DataType type, final String value) {
  if (value == null) {
    return null;
  }
  switch (type.getName()) {
  case TEXT:
  case VARCHAR:
  case ASCII:
    return value;
  case INET:
    return type.parse("'" + value + "'");
  case INT:
  case VARINT:
  case BIGINT:
  case FLOAT:
  case DOUBLE:
  case DECIMAL:
  case BOOLEAN:
    return type.parse(value.replaceAll("\\s+", ""));
  default:
    return type.parse(value);
  }
}

/**
 * Create a parameterized token
 * 
 * @param sqlType
 * @param key
 * @param mode
 * @param position
 * @throws IllegalArgumentException
 */
public CqlToken(String cqlType, String key, int position)
        throws IllegalArgumentException {

    // convert the given cqlType to an enumerated value
    this.cqlType = DataType.Name.valueOf(cqlType.toUpperCase());

    // a tuple is a special case of a collection in that arbitrary types can
    // be added to a tuple
    if (this.cqlType == DataType.Name.TUPLE) {
        this.collectionType = this.cqlType;
    }

    // the key, which always equals the value property, is used to id this
    // token as a parameter field, as opposed to a token that represents a
    // CQL keyword. A keyword token will have its key property set to null
    this.key = key.toLowerCase();
    this.value = this.key;
    addPosition(position);
}

public HecubaClientManager<Long> getHecubaClientManagerWithLongKeys(CassandraParamsBean parameters,
        HecubaConstants.CassandraClientImplementation cassandraManagerType) {
    switch (cassandraManagerType) {
    case ASTYANAX:
        return new AstyanaxBasedHecubaClientManager<>(parameters,
                com.netflix.astyanax.serializers.LongSerializer.get());
    case HECTOR:
        return new HectorBasedHecubaClientManager<>(parameters,
                me.prettyprint.cassandra.serializers.LongSerializer.get());
    case DATASTAX:
        return new DataStaxBasedHecubaClientManager<>(parameters, DataType.bigint());
    case DATASTAX_SHARED:
        return new DataStaxBasedSharedHecubaClientManager<>(parameters, DataType.bigint());
    default:
        throw new RuntimeException("Unhandled CassandraManagerType: " + cassandraManagerType);
    }
}

@SuppressWarnings("unchecked")
@Test
public void testLocalDateSetter() throws Exception {
    Setter<SettableByIndexData, Object> setter = DataTypeTest.getSetter((Class<Object>)localDateClass, (DataType) DataType.class.getMethod("date").invoke(null));

    RecorderInvocationHandler recorder = new RecorderInvocationHandler();
    SettableByIndexData settableByDataInstance = (SettableByIndexData) Proxy.newProxyInstance(getClass().getClassLoader(), new Class[] { SettableByIndexData.class }, recorder);

    Object localDateInstance = localDateClass.getDeclaredMethod("fromMillisSinceEpoch", long.class).invoke(null, System.currentTimeMillis());
    setter.set(settableByDataInstance, localDateInstance);

    recorder.invokedOnce("setDate", 1, localDateInstance);

    recorder.reset();

    setter.set(settableByDataInstance, null);
    recorder.invokedOnce("setToNull", 1);
}

/**
 * Binds an object, which is provided by the calling application, to a
 * BoundStatement.
 */
@SuppressWarnings({ "unchecked", "rawtypes" })
public void bindObject(Session session, BoundStatement bs, Object obj)
        throws Exception {
    LOG.trace("bindObject: entered with this type {}", obj.getClass()
            .getName());        
    if (obj instanceof String) {
        bindString(bs, (String) obj);
    }
    else if (obj instanceof Map) {
        bindMap(bs, (Map) obj);
    } else if (obj instanceof List) {
        if (getCollectionType() == DataType.Name.TUPLE) {
            bindTuple(session, bs, (List) obj);
        } else {
            bindList(bs, (List) obj);
        }
    } else if (obj instanceof Set) {
        bindSet(bs, (Set) obj);
    } else {
        Class classZ = obj.getClass();
        for (Integer pos : getPositions()) {
            bs.set(pos, obj, classZ);
        }
    }
}

private static void createTableIfNotExists(final com.datastax.driver.core.Session session,
    final String table, final Logger log) {
  Create createTable = SchemaBuilder.createTable(table)
      .addPartitionKey(ID, DataType.varchar())
      .addColumn(CREATED_AT, DataType.timestamp())
      .addColumn(ACCESSED_AT, DataType.timestamp())
      .addColumn(SAVED_AT, DataType.timestamp())
      .addColumn(ATTRIBUTES, DataType.map(DataType.varchar(), DataType.varchar()))
      .ifNotExists();

  Futures.addCallback(session.executeAsync(createTable), new FutureCallback<ResultSet>() {
    @Override
    public void onSuccess(final ResultSet result) {
      log.debug("Session table successfully created");
    }

    @Override
    public void onFailure(final Throwable x) {
      log.error("Create session table resulted in exception", x);
    }
  });
}

@SuppressWarnings("unchecked")
@Override
public List<K> retrieveKeysBySecondaryIndex(String columnName, String columnValue) {
    final String query = "select * from " + secondaryIndexColumnFamily + " where "
            + secondaryIndexKeyColumn + " = ?";
    CassandraResultSet<K, String> keysResultSet = read(query, DataType.ascii(), keyType,
            ImmutableMap.of("*", keyType), getSecondaryIndexKey(columnName, columnValue));
    List<K> keys = new ArrayList<>();
    if (keysResultSet.hasResults()) {
        for (String key : keysResultSet.getColumnNames()) {
            if (keyType == DataType.bigint()) {
                keys.add((K) NumberUtils.createLong(key));
            } else {
                keys.add((K) key);
            }
        }

        return keys;
    }

    return null;
}

private void mockTableMetadata() {
  final ColumnMetadata idColumn = mock(ColumnMetadata.class);
  when(idColumn.getName()).thenReturn("id");
  when(idColumn.getType()).thenReturn(DataType.cint());

  final ColumnMetadata textColumn = mock(ColumnMetadata.class);
  when(textColumn.getName()).thenReturn("text_col");
  when(textColumn.getType()).thenReturn(DataType.text());

  final KeyspaceMetadata keyspaceMetadata = mock(KeyspaceMetadata.class);
  when(keyspaceMetadata.getName()).thenReturn("my_keyspace");

  when(tableMetadata.getName()).thenReturn("my_table");
  when(tableMetadata.getColumns()).thenReturn(ImmutableList.of(idColumn, textColumn));
  when(tableMetadata.getKeyspace()).thenReturn(keyspaceMetadata);
  when(tableMetadata.getPrimaryKey()).thenReturn(ImmutableList.of(idColumn));
}

@Override
public <K, V> ImmutableMap<K, V> getMap(String name, Class<K> keysClass, Class<V> valuesClass) {
    if (isNull(name)) {
        return ImmutableMap.of();
    }

    final DataType datatype = ctx.getCatalog().getColumnMetadata(tablename, name).getType();
    if (UDTValueMapper.isBuildInType(datatype)) {
        return ImmutableMap.copyOf(getRow().getMap(name, keysClass, valuesClass));

    } else {
        if (UDTValueMapper.isBuildInType(datatype.getTypeArguments().get(0))) {
            return ctx.getUDTValueMapper().fromUdtValues(datatype.getTypeArguments().get(0), datatype.getTypeArguments().get(1), ImmutableMap.copyOf(getRow().getMap(name, keysClass, UDTValue.class)), keysClass, valuesClass);

        } else if (UDTValueMapper.isBuildInType(datatype.getTypeArguments().get(1))) {
            return ctx.getUDTValueMapper().fromUdtValues(datatype.getTypeArguments().get(0), datatype.getTypeArguments().get(1), ImmutableMap.copyOf(getRow().getMap(name, UDTValue.class, valuesClass)), keysClass, valuesClass);

        } else {
            return ctx.getUDTValueMapper().fromUdtValues(datatype.getTypeArguments().get(0), datatype.getTypeArguments().get(1), ImmutableMap.copyOf(getRow().getMap(name, UDTValue.class, UDTValue.class)), keysClass, valuesClass);
        }
    }
}

public HecubaClientManager<String> getHecubaClientManagerWithStringKeys(CassandraParamsBean parameters,
        HecubaConstants.CassandraClientImplementation cassandraManagerType) {
    switch (cassandraManagerType) {
    case ASTYANAX:
        return new AstyanaxBasedHecubaClientManager<String>(parameters,
                com.netflix.astyanax.serializers.StringSerializer.get());
    case HECTOR:
        return new HectorBasedHecubaClientManager<String>(parameters,
                me.prettyprint.cassandra.serializers.StringSerializer.get());
    case DATASTAX:
        return new DataStaxBasedHecubaClientManager<>(parameters, DataType.text());
    case DATASTAX_SHARED:
        return new DataStaxBasedSharedHecubaClientManager<>(parameters, DataType.text());
    default:
        throw new RuntimeException("Unhandled CassandraManagerType: " + cassandraManagerType);
    }
}

/** Get a field for the current row from the underlying object.
 *
 * @param index Index of the field within the Row object
 * @param typeName Type of the field in this row
 */
private Object currentRowField(int index, SqlTypeName typeName) {
  DataType type = current.getColumnDefinitions().getType(index);
  if (type == DataType.ascii() || type == DataType.text() || type == DataType.varchar()) {
    return current.getString(index);
  } else if (type == DataType.cint() || type == DataType.varint()) {
    return current.getInt(index);
  } else if (type == DataType.bigint()) {
    return current.getLong(index);
  } else if (type == DataType.cdouble()) {
    return current.getDouble(index);
  } else if (type == DataType.cfloat()) {
    return current.getFloat(index);
  } else if (type == DataType.uuid() || type == DataType.timeuuid()) {
    return current.getUUID(index).toString();
  } else {
    return null;
  }
}

/**
 * Returns the partition key related to a given {@link Cells}.
 *
 * @param cells        {@link Cells} from Cassandra to extract the partition key.
 * @param keyValidator Cassandra key type.
 * @param numberOfKeys Number of keys.
 * @return Partition key.
 */
public static ByteBuffer getPartitionKey(Cells cells, AbstractType<?> keyValidator, int numberOfKeys) {
    ByteBuffer partitionKey;
    if (keyValidator instanceof CompositeType) {
        ByteBuffer[] keys = new ByteBuffer[numberOfKeys];

        for (int i = 0; i < cells.size(); i++) {
            Cell c = cells.getCellByIdx(i);

            if (c.isKey()) {
                keys[i] = DataType.serializeValue(c.getValue(), CassandraDeepJobConfig.PROTOCOL_VERSION);
            }
        }

        partitionKey = CompositeType.build(keys);
    } else {
        Cell cell = cells.getCellByIdx(0);
        partitionKey = DataType.serializeValue(cell.getValue(), CassandraDeepJobConfig.PROTOCOL_VERSION);
    }
    return partitionKey;
}

private Cluster doCreateCluster(CassandraProperties properties) {
    Cluster cluster = Cluster.builder()
            .withClusterName(properties.getCluster())
            .withPort(properties.getPort())
            .addContactPoints(properties.getContactPoints())
            .withTimestampGenerator(getTimestampGenerator())
            .withPoolingOptions(
                    //TODO some default options - move to config
                    new PoolingOptions()
                            .setConnectionsPerHost(HostDistance.LOCAL, 4, 4)
                            .setConnectionsPerHost(HostDistance.REMOTE, 2, 2)
                            .setMaxRequestsPerConnection(HostDistance.LOCAL, 1024)
                            .setMaxRequestsPerConnection(HostDistance.REMOTE, 256)
            )
            .build();
    //almost all queries are idempotent except counter updates, so it's easier to mark them as idempotent
    cluster.getConfiguration().getQueryOptions().setDefaultIdempotence(true);

    CodecRegistry codecRegistry = cluster.getConfiguration().getCodecRegistry();

    TupleType tupleType = cluster.getMetadata()
            .newTupleType(DataType.timestamp(), DataType.varchar());
    codecRegistry.register(new ZonedDateTimeCodec(tupleType));

    QueryLogger queryLogger = QueryLogger.builder()
            .withConstantThreshold(100)
            .withMaxQueryStringLength(200)
            .build();
    cluster.register(queryLogger);

    return cluster;
}

public static int convertDataType(DataType type) {
    switch (type.getName()) {
        case SMALLINT: 
        case INT:
        case BIGINT:
        case COUNTER:
        case TIME:
            return ValueMetaInterface.TYPE_INTEGER; // 5 > java.lang.Long
        case ASCII:
        case TEXT:
        case VARCHAR:
        case UUID:
        case TIMEUUID:              
            return ValueMetaInterface.TYPE_STRING; // 2 > java.lang.String
        case INET:
            return ValueMetaInterface.TYPE_INET; // 10 > 
        case BOOLEAN:
            return ValueMetaInterface.TYPE_BOOLEAN; // 4 > java.lang.Boolean
        case DECIMAL:
        case FLOAT:
        case DOUBLE:
            return ValueMetaInterface.TYPE_NUMBER; // 1 > java.lang.Double
        case VARINT:
            return ValueMetaInterface.TYPE_BIGNUMBER; // 6 > java.math.BigDecimal
        case TIMESTAMP:
            return ValueMetaInterface.TYPE_DATE; // 3 > java.util.Date
        case BLOB:
            return ValueMetaInterface.TYPE_BINARY; // 8 > java.lang.byte[]
        case LIST:
        case MAP:
        case SET:
            return ValueMetaInterface.TYPE_SERIALIZABLE; // 0
        default:
            return ValueMetaInterface.TYPE_STRING;
    }
}

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");
}

public boolean doesRowSatisfyCondition(Row row, Map<String, Object> condition){
    ColumnDefinitions colInfo = row.getColumnDefinitions();

    for (Map.Entry<String, Object> entry : condition.entrySet()){
        String colName = entry.getKey();
        DataType colType = colInfo.getType(colName);
        Object columnValue = getColValue(row, colName, colType);
        Object conditionValue = convertToActualDataType(colType, entry.getValue());
        if(columnValue.equals(conditionValue) == false)
            return false;       
    }
    return true;    
}

private Object readRow(final Row row, final String name, final DataType colType) {
        switch (colType.getName()) {
        case BIGINT:
            return row.getLong(name);
        case BOOLEAN:
            return row.getBool(name);
        case DOUBLE:
            return row.getDouble(name);
        case FLOAT:
            return row.getFloat(name);
        case INT:
            return row.getInt(name);
        case MAP:
            return row.getMap(name, String.class, String.class);
        case UUID:
            return row.getUUID(name);
        case TEXT:
        case VARCHAR:
            return row.getString(name);
        case VARINT:
            return row.getVarint(name);
// These are not supported right now....
// ASCII
// BLOB
// COUNTER
// CUSTOM
// DECIMAL
// INET
// LIST
// SET
// TIMESTAMP
// TIMEUUID
// TUPLE
// UDT
        default:
            return null;
        }
    }

private String convertToSqlDataType(DataType type, Object valueObj) {
    switch (type.getName()) {
    case TEXT:
        String t = valueObj.toString();
        t = t.replaceAll("'", "''");
        return "'" + t + "'";
    case MAP:
        @SuppressWarnings("unchecked")
        Map<String,Object> otMap = (Map<String,Object>) valueObj;
        return "{" + jsonMaptoSqlString(otMap, ",") + "}";
    default:
    case UUID:
        return valueObj.toString();
    }
}

private void ensureStashTokenRangeTableExists() {
    if (!_verifiedStashTokenRangeTableExists) {
        synchronized(this) {
            if (!_verifiedStashTokenRangeTableExists) {
                // Primary key is ((stash_id, data_center), placement, range_token, is_start_token).
                // Note that Cassandra performs unsigned byte comparison for "range_token" and sorts False before
                // True for "is_start_token".  The latter is necessary because it sorts two tables with
                // adjacent UUIDs correctly, returning the exclusive "to" token for the previous table before the
                // inclusive "from" token for the next table.
                _placementCache.get(_systemTablePlacement)
                        .getKeyspace()
                        .getCqlSession()
                        .execute(SchemaBuilder.createTable(STASH_TOKEN_RANGE_TABLE)
                                .ifNotExists()
                                .addPartitionKey(STASH_ID_COLUMN, DataType.text())
                                .addPartitionKey(DATA_CENTER_COLUMN, DataType.text())
                                .addClusteringColumn(PLACEMENT_COLUMN, DataType.text())
                                .addClusteringColumn(RANGE_TOKEN_COLUMN, DataType.blob())
                                .addClusteringColumn(IS_START_TOKEN_COLUMN, DataType.cboolean())
                                .addColumn(TABLE_JSON_COLUMN, DataType.text())
                                .withOptions()
                                // The following cluster orders should be the defaults but for clarity let's be explicit
                                .clusteringOrder(PLACEMENT_COLUMN, SchemaBuilder.Direction.ASC)
                                .clusteringOrder(RANGE_TOKEN_COLUMN, SchemaBuilder.Direction.ASC)
                                .clusteringOrder(IS_START_TOKEN_COLUMN, SchemaBuilder.Direction.ASC)
                                .compactStorage()
                                .defaultTimeToLive(TTL));

                _verifiedStashTokenRangeTableExists = true;
            }
        }
    }
}

public CassandraPOJOInputOperator()
{
  super();
  columnDataTypes = new ArrayList<DataType>();
  setters = new ArrayList<Object>();
  this.store = new CassandraStore();
}

/**
 * Generates a Boundstatement that can be executed for the given incoming tuple. This boundstatement is then
 * executed as a command
 * @param ps The prepared statement that was shortlisted to execute the given tuple
 * @param tuple The tuple that represents the current execution context
 * @param setNulls This represents the value whether the columns in the prepared statement need to be ignored or
 *                 considered
 * @return The boundstatement appropriately built
 */
@SuppressWarnings(value = "unchecked")
private BoundStatement processPayloadForExecution(final PreparedStatement ps, final UpsertExecutionContext tuple,
    final boolean setNulls)
{
  BoundStatement boundStatement = ps.bind();
  Object pojoPayload = tuple.getPayload();
  for (String cassandraColName : getters.keySet()) {
    DataType dataType = columnDefinitions.get(cassandraColName);
    CassandraPojoUtils.populateBoundStatementWithValue(boundStatement,getters,dataType,cassandraColName,
        pojoPayload,setNulls,codecsForCassandraColumnNames);
  }
  return boundStatement;
}

public CassandraPreparedStatementGenerator(Set<String> pkColumnNames, Set<String> counterColumns,
    Set<String> listColumns, Set<String> mapColumns, Set<String> setColumns,
    Map<String, DataType> columnDefinitions)
{
  this.pkColumnNames = pkColumnNames;
  this.counterColumns = counterColumns;
  this.listColumns = listColumns;
  this.mapColumns = mapColumns;
  this.setColumns = setColumns;
  this.columnDefinitions = columnDefinitions;
}