@Test
public void getValue_ProperValuesGivenInConstructor_ShouldReturnWhatsSet() {
  // Arrange
  String expectedText = ANY_NAME;
  int expectedInt = ANY_INT;
  definitions.add(expectedText, DataType.cint());
  when(row.getInt(expectedText)).thenReturn(expectedInt);
  ResultImpl spy = spy(new ResultImpl(new ArrayList<>(), null));
  doReturn(definitions.get()).when(spy).getColumnDefinitions(row);
  spy.interpret(row);

  // Act
  Optional<Value> actual = spy.getValue(expectedText);

  // Assert
  assertThat(actual).isEqualTo(Optional.of(new IntValue(expectedText, expectedInt)));
}
 

@Override
public void init(CassandraSessionPool.Session session) {
    ImmutableMap<HugeKeys, DataType> pkeys = ImmutableMap.of(
            HugeKeys.ID, TYPE_SL
    );
    ImmutableMap<HugeKeys, DataType> ckeys = ImmutableMap.of();
    ImmutableMap<HugeKeys, DataType> columns = ImmutableMap
            .<HugeKeys, DataType>builder()
            .put(HugeKeys.NAME, DataType.text())
            .put(HugeKeys.ID_STRATEGY, DataType.tinyint())
            .put(HugeKeys.PRIMARY_KEYS, DataType.list(TYPE_PK))
            .put(HugeKeys.NULLABLE_KEYS, DataType.set(TYPE_PK))
            .put(HugeKeys.INDEX_LABELS, DataType.set(TYPE_IL))
            .put(HugeKeys.PROPERTIES, DataType.set(TYPE_PK))
            .put(HugeKeys.ENABLE_LABEL_INDEX, DataType.cboolean())
            .put(HugeKeys.USER_DATA, TYPE_UD)
            .put(HugeKeys.STATUS, DataType.tinyint())
            .put(HugeKeys.TTL, TYPE_TTL)
            .put(HugeKeys.TTL_START_TIME, TYPE_PK)
            .build();

    this.createTable(session, pkeys, ckeys, columns);
    this.createIndex(session, NAME_INDEX, HugeKeys.NAME);
}
 

@Override
public void init(CassandraSessionPool.Session session) {
    ImmutableMap<HugeKeys, DataType> pkeys = ImmutableMap.of(
            HugeKeys.ID, DataType.cint()
    );
    ImmutableMap<HugeKeys, DataType> ckeys = ImmutableMap.of();
    ImmutableMap<HugeKeys, DataType> columns = ImmutableMap
            .<HugeKeys, DataType>builder()
            .put(HugeKeys.NAME, DataType.text())
            .put(HugeKeys.DATA_TYPE, DataType.tinyint())
            .put(HugeKeys.CARDINALITY, DataType.tinyint())
            .put(HugeKeys.AGGREGATE_TYPE, DataType.tinyint())
            .put(HugeKeys.PROPERTIES, DataType.set(TYPE_PK))
            .put(HugeKeys.USER_DATA, TYPE_UD)
            .put(HugeKeys.STATUS, DataType.tinyint())
            .build();

    this.createTable(session, pkeys, ckeys, columns);
    this.createIndex(session, NAME_INDEX, HugeKeys.NAME);
}
 

@Override
public void init(CassandraSessionPool.Session session) {
    ImmutableMap<HugeKeys, DataType> pkeys = ImmutableMap.of(
            HugeKeys.ID, TYPE_IL
    );
    ImmutableMap<HugeKeys, DataType> ckeys = ImmutableMap.of();
    ImmutableMap<HugeKeys, DataType> columns = ImmutableMap
            .<HugeKeys, DataType>builder()
            .put(HugeKeys.NAME, DataType.text())
            .put(HugeKeys.BASE_TYPE, DataType.tinyint())
            .put(HugeKeys.BASE_VALUE, TYPE_SL)
            .put(HugeKeys.INDEX_TYPE, DataType.tinyint())
            .put(HugeKeys.FIELDS, DataType.list(TYPE_PK))
            .put(HugeKeys.USER_DATA, TYPE_UD)
            .put(HugeKeys.STATUS, DataType.tinyint())
            .build();

    this.createTable(session, pkeys, ckeys, columns);
    this.createIndex(session, NAME_INDEX, HugeKeys.NAME);
}
 

@Override
public void init(CassandraSessionPool.Session session) {
    ImmutableMap<HugeKeys, DataType> pkeys = ImmutableMap.of(
            HugeKeys.OWNER_VERTEX, TYPE_ID
    );
    ImmutableMap<HugeKeys, DataType> ckeys = ImmutableMap.of(
            HugeKeys.DIRECTION, DataType.tinyint(),
            HugeKeys.LABEL, TYPE_SL,
            HugeKeys.SORT_VALUES, DataType.text(),
            HugeKeys.OTHER_VERTEX, TYPE_ID
    );
    ImmutableMap<HugeKeys, DataType> columns = ImmutableMap.of(
            HugeKeys.PROPERTIES, DataType.map(TYPE_PK, TYPE_PROP),
            HugeKeys.EXPIRED_TIME, TYPE_EXPIRED_TIME
    );

    this.createTable(session, pkeys, ckeys, columns);

    /*
     * Only out-edges table needs label index because we query edges
     * by label from out-edges table
     */
    if (this.direction == Directions.OUT) {
        this.createIndex(session, LABEL_INDEX, HugeKeys.LABEL);
    }
}
 

private <T extends Serializable> void doSendMessage( T body, List<String> regions ) throws IOException {

        createQueueIfNecessary();

        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        ObjectOutputStream oos = new ObjectOutputStream(bos);
        oos.writeObject(body);
        oos.flush();
        oos.close();
        ByteBuffer byteBuffer = ByteBuffer.wrap( bos.toByteArray() );

        queueMessageManager.sendMessages(
            scope.getName(),
            regions,
            null, // delay millis
            null, // expiration seconds
            "application/octet-stream",
            DataType.serializeValue( byteBuffer, ProtocolVersion.NEWEST_SUPPORTED ));
    }
 

@Override
public UserType convert(DataType dataType) {
    com.datastax.driver.core.UserType userType = (com.datastax.driver.core.UserType) dataType;

    String typeNameString = userType.getTypeName();
    Collection<String> fieldNames = userType.getFieldNames();

    List<AbstractType<?>> innerAbstractTypes = new ArrayList<>(fieldNames.size());

    ByteBuffer typeNameBuffer = UTF8Type.instance.fromString(typeNameString);

    List<FieldIdentifier> fieldIdentifiers = new ArrayList<>(fieldNames.size());
    for (String fieldName : fieldNames) {
        fieldIdentifiers.add(FieldIdentifier.forInternalString(fieldName));
        innerAbstractTypes.add((CassandraTypeConverter.convert(userType.getFieldType(fieldName))));
    }

    return new UserType(userType.getKeyspace(),
            typeNameBuffer,
            fieldIdentifiers,
            innerAbstractTypes,
            !userType.isFrozen());
}
 

@Test
public void testGetPrimitiveAvroTypeFromCassandraType() throws Exception {
    assertEquals("string", AbstractCassandraProcessor.getPrimitiveAvroTypeFromCassandraType(DataType.ascii()));
    assertEquals("string", AbstractCassandraProcessor.getPrimitiveAvroTypeFromCassandraType(DataType.text()));
    assertEquals("string", AbstractCassandraProcessor.getPrimitiveAvroTypeFromCassandraType(DataType.varchar()));
    assertEquals("string", AbstractCassandraProcessor.getPrimitiveAvroTypeFromCassandraType(DataType.timestamp()));
    assertEquals("string", AbstractCassandraProcessor.getPrimitiveAvroTypeFromCassandraType(DataType.timeuuid()));
    assertEquals("string", AbstractCassandraProcessor.getPrimitiveAvroTypeFromCassandraType(DataType.uuid()));
    assertEquals("string", AbstractCassandraProcessor.getPrimitiveAvroTypeFromCassandraType(DataType.inet()));
    assertEquals("string", AbstractCassandraProcessor.getPrimitiveAvroTypeFromCassandraType(DataType.varint()));

    assertEquals("boolean", AbstractCassandraProcessor.getPrimitiveAvroTypeFromCassandraType(DataType.cboolean()));
    assertEquals("int", AbstractCassandraProcessor.getPrimitiveAvroTypeFromCassandraType(DataType.cint()));

    assertEquals("long", AbstractCassandraProcessor.getPrimitiveAvroTypeFromCassandraType(DataType.bigint()));
    assertEquals("long", AbstractCassandraProcessor.getPrimitiveAvroTypeFromCassandraType(DataType.counter()));

    assertEquals("float", AbstractCassandraProcessor.getPrimitiveAvroTypeFromCassandraType(DataType.cfloat()));
    assertEquals("double", AbstractCassandraProcessor.getPrimitiveAvroTypeFromCassandraType(DataType.cdouble()));

    assertEquals("bytes", AbstractCassandraProcessor.getPrimitiveAvroTypeFromCassandraType(DataType.blob()));
}
 

@Test
public void testMap() {
    // map from ASCII to Double
    // test non-frozen
    DataType mapType = DataType.map(DataType.ascii(), DataType.cdouble());
    AbstractType<?> convertedType = CassandraTypeConverter.convert(mapType);

    MapType<?, ?> expectedType = MapType.getInstance(AsciiType.instance, DoubleType.instance, true);
    Assert.assertEquals(expectedType, convertedType);

    // test frozen
    mapType = DataType.map(DataType.ascii(), DataType.cdouble(), true);
    convertedType = CassandraTypeConverter.convert(mapType);
    expectedType = MapType.getInstance(AsciiType.instance, DoubleType.instance, false);
    Assert.assertEquals(expectedType, convertedType);
    Assert.assertTrue("Expected convertType to be frozen", convertedType.isFrozenCollection());
}
 

/**
 * 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;
}
 

public int getColumnType(int column) throws SQLException
{        	
	DataType type = null;        	
	if(currentRow!=null){
		type = currentRow.getColumnDefinitions().getType(column-1);
	}else{
		type = driverResultSet.getColumnDefinitions().asList().get(column-1).getType();
	}
	return TypesMap.getTypeForComparator(type.toString()).getJdbcType();
    
}
 

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

private static String buildMapValue(Row row, int position)
{
    DataType type = row.getColumnDefinitions().getType(position);
    checkArgument(type.getTypeArguments().size() == 2, "Expected two type arguments, got: %s", type.getTypeArguments());
    DataType keyType = type.getTypeArguments().get(0);
    DataType valueType = type.getTypeArguments().get(1);
    return buildMapValue((Map<?, ?>) row.getObject(position), keyType, valueType);
}
 

/**
 * Initializes field info from property descriptor.
 *
 * @param accessor {@link PojoFieldAccessor} accessor.
 */
private void init(PojoFieldAccessor accessor) {
    DataType.Name cassandraType = PropertyMappingHelper.getCassandraType(accessor.getFieldType());
    cassandraType = cassandraType == null ? DataType.Name.BLOB : cassandraType;

    this.colDDL = "\"" + col + "\" " + cassandraType.toString();

    this.accessor = accessor;
}
 

public static boolean isFullySupported(DataType dataType)
{
    if (toCassandraType(dataType.getName()).isEmpty()) {
        return false;
    }

    return dataType.getTypeArguments().stream()
            .allMatch(CassandraType::isFullySupported);
}
 

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

private void mockTableMetadataWithIdAndNameColumns(TableMetadata tableMetadata) {
  ColumnMetadata colId = mock(ColumnMetadata.class);
  when(colId.getName()).thenReturn("id");
  when(colId.getType()).thenReturn(DataType.text());
  ColumnMetadata colName = mock(ColumnMetadata.class);
  when(colName.getName()).thenReturn("name");
  when(colName.getType()).thenReturn(DataType.text());

  List<ColumnMetadata> listOfColumns = ImmutableList.of(colId, colName);
  when(tableMetadata.getColumns()).thenReturn(listOfColumns);
}
 

/**
 * private constructor.
 */
private CellValidator(String validatorClassName, Kind validatorKind, Collection<String> validatorTypes,
                      DataType.Name cqlTypeName) {
    this.validatorClassName = validatorClassName != null ? validatorClassName : DEFAULT_VALIDATOR_CLASSNAME;
    this.validatorKind = validatorKind;
    this.validatorTypes = validatorTypes;
    this.cqlTypeName = cqlTypeName;
}
 

private String getCreateTaskLookupTableStatement() {
    return SchemaBuilder.createTable(config.getCassandraKeyspace(), TABLE_TASK_LOOKUP)
        .ifNotExists()
        .addPartitionKey(TASK_ID_KEY, DataType.uuid())
        .addColumn(WORKFLOW_ID_KEY, DataType.uuid())
        .getQueryString();
}
 

DataType.Name checkType(int i, DataType.Name name1, DataType.Name name2, DataType.Name name3) {
    DataType defined = getType(i);
    if (name1 != defined.getName() && name2 != defined.getName() && name3 != defined.getName())
        throw new InvalidTypeException(String.format("Column %s is of type %s", getName(i), defined));

    return defined.getName();
}
 

public static String spaceSeparatedKeyValue(Map<String, ?> columns){

        StringJoiner columnsSchema = new StringJoiner(",");
        columns.forEach( (key, value) -> {

            if( value == DataType.Name.CUSTOM ){
                columnsSchema.add(key+" "+COMPOSITE_TYPE);
            }else {
                columnsSchema.add(key + " " + String.valueOf(value));
            }
        });

        return columnsSchema.toString();

    }
 

private HashMap<String, String> getStreamFieldsAndTypes(List<ColumnNameTypeValue> columns) {

        HashMap<String, String> fields = new HashMap<String, String>();
        for (ColumnNameTypeValue column : columns) {
            switch (column.getType()) {
            case BOOLEAN:
                fields.put(column.getColumn(), DataType.Name.BOOLEAN.toString());
                break;
            case DOUBLE:
                fields.put(column.getColumn(), DataType.Name.DOUBLE.toString());
                break;
            case FLOAT:
                fields.put(column.getColumn(), DataType.Name.FLOAT.toString());
                break;
            case INTEGER:
                fields.put(column.getColumn(), DataType.Name.INT.toString());
                break;
            case LONG:
                fields.put(column.getColumn(), DataType.Name.DOUBLE.toString());
                break;
            case STRING:
                fields.put(column.getColumn(), DataType.Name.TEXT.toString());
                break;
            default:
                throw new RuntimeException("Unsupported Column type");
            }
        }

        return fields;
    }
 

@Override
public void init(CassandraSessionPool.Session session) {
    ImmutableMap<HugeKeys, DataType> pkeys = ImmutableMap.of(
            HugeKeys.SCHEMA_TYPE, DataType.text()
    );
    ImmutableMap<HugeKeys, DataType> ckeys = ImmutableMap.of();
    ImmutableMap<HugeKeys, DataType> columns = ImmutableMap.of(
            HugeKeys.ID, DataType.counter()
    );

    this.createTable(session, pkeys, ckeys, columns);
}
 

public void testDataTypeListInstantiation() {
    try {
        assertNull(cellValidator((DataType) null));
        fail();
    } catch (Exception e) {
        //ok
    }

    DataType type = DataType.list(DataType.timeuuid());

    CellValidator cv = cellValidator(type);
    assertNotNull(cv);
    assertEquals(cv.getValidatorClassName(), ListType.class.getName());
    assertNotNull(cv.getValidatorTypes());
    assertEquals(cv.validatorKind(), Kind.LIST);
    assertEquals(cv.getValidatorTypes().size(), 1);
    assertEquals(cv.getValidatorTypes().iterator().next(), "timeuuid");
    assertEquals(DataType.Name.LIST, cv.getCqlTypeName());

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

    //        assertNotNull(cv.getAbstractType());
    //        assertEquals(cv.getAbstractType(), ListType.getInstance(TimeUUIDType.instance));
}
 

private String getCreateTaskDefLimitTableStatement() {
    return SchemaBuilder.createTable(config.getCassandraKeyspace(), TABLE_TASK_DEF_LIMIT)
        .ifNotExists()
        .addPartitionKey(TASK_DEF_NAME_KEY, DataType.text())
        .addClusteringColumn(TASK_ID_KEY, DataType.uuid())
        .addColumn(WORKFLOW_ID_KEY, DataType.uuid())
        .getQueryString();
}
 

@Override
public void init(CassandraSessionPool.Session session) {
    ImmutableMap<HugeKeys, DataType> pkeys = ImmutableMap.of(
            HugeKeys.FIELD_VALUES, DataType.text()
    );
    ImmutableMap<HugeKeys, DataType> ckeys = ImmutableMap.of(
            HugeKeys.INDEX_LABEL_ID, TYPE_IL,
            HugeKeys.ELEMENT_IDS, TYPE_ID
    );
    ImmutableMap<HugeKeys, DataType> columns = ImmutableMap.of(
            HugeKeys.EXPIRED_TIME, TYPE_EXPIRED_TIME
    );

    this.createTable(session, pkeys, ckeys, columns);
}
 

/**
 * Returns Cassandra table columns DDL, corresponding to POJO fields which should be persisted.
 *
 * @param ignoreColumns Table columns to ignore (exclude) from DDL.
 * @return DDL statement for Cassandra table fields.
 */
public String getTableColumnsDDL(Set<String> ignoreColumns) {
    if (PersistenceStrategy.BLOB == stgy)
        return "  \"" + col + "\" " + DataType.Name.BLOB.toString();

    if (PersistenceStrategy.PRIMITIVE == stgy)
        return "  \"" + col + "\" " + PropertyMappingHelper.getCassandraType(javaCls);

    List<F> fields = getFields();

    if (fields == null || fields.isEmpty()) {
        throw new IllegalStateException("There are no POJO fields found for '" + javaCls.toString()
            + "' class to be presented as a Cassandra primary key");
    }

    // Accumulating already processed columns in the set, to prevent duplicating columns
    // shared by two different POJO fields.
    Set<String> processedColumns = new HashSet<>();

    StringBuilder builder = new StringBuilder();

    for (F field : fields) {
        if ((ignoreColumns != null && ignoreColumns.contains(field.getColumn())) ||
                processedColumns.contains(field.getColumn())) {
            continue;
        }

        if (builder.length() > 0)
            builder.append(",\n");

        builder.append("  ").append(field.getColumnDDL());

        processedColumns.add(field.getColumn());
    }

    return builder.toString();
}
 

public void testBlobDataType() {
    CellValidator cv = cellValidator(DataType.blob());
    assertNotNull(cv);
    assertEquals(cv.getValidatorClassName(), BytesType.class.getName());
    assertNull(cv.getValidatorTypes());
    assertNotNull(cv.getAbstractType());
    assertEquals(cv.getAbstractType(), BytesType.instance);
}
 

private DataType.Name validateName(DataType dataType) {

        final DataType.Name name = dataType.getName();
        if (DataTypeHelper.isNumber(name)) {
            return name;
        }

        throw new IllegalArgumentException("Datatype " + dataType + " not a number");
    }
 

@Override
public TupleType convert(DataType dataType) {
    com.datastax.driver.core.TupleType tupleDataType = (com.datastax.driver.core.TupleType) dataType;
    List<DataType> innerTypes = tupleDataType.getComponentTypes();
    List<AbstractType<?>> innerAbstractTypes = new ArrayList<>(innerTypes.size());
    for (DataType dt : innerTypes) {
        innerAbstractTypes.add(CassandraTypeConverter.convert(dt));
    }
    return new TupleType(innerAbstractTypes);
}