@Override
public DynamoDBResponse deleteTable(DeleteTableRequest deleteTableRequest) {
    logger.info("deleting JSON table");

    String keyspace = keyspaceName;
    String table = deleteTableRequest.getTableName();
    String statement = String.format("DROP TABLE %s.\"%s\";\n", keyspace, table);
    ResultSet result = session().execute(statement);
    if (result.wasApplied()) {

        logger.info("deleted table " + table);
        cassandraManager.refreshSchema();

        TableDescription newTableDesc = this.getTableDescription(table, null,null);
        DeleteTableResult createResult = (new DeleteTableResult()).withTableDescription(newTableDesc);
        return new DynamoDBResponse(createResult, 200);
    }
    return null;
}
 

@Override
public DynamoDBResponse createTable(CreateTableRequest createTableRequest) throws IOException {

    logger.info("creating JSON table");

    String columnPairs = createTableRequest.getAttributeDefinitions().stream().map(this::attributeToPairs).collect(Collectors.joining(", "));
    columnPairs += ",json_blob text";
    String keyspace = keyspaceName;
    String table = createTableRequest.getTableName();
    String primaryKey = getPrimaryKey(createTableRequest.getKeySchema());
    String statement = String.format("CREATE TABLE IF NOT EXISTS %s.\"%s\" ( %s, PRIMARY KEY %s);\n", keyspace, table, columnPairs, primaryKey);
    ResultSet result = session().execute(statement);
    if (result.wasApplied()) {

        logger.info("created {} as {}", table, statement);

        cassandraManager.refreshSchema();

        TableDescription newTableDesc = this.getTableDescription(table, createTableRequest.getAttributeDefinitions(), createTableRequest.getKeySchema());
        CreateTableResult createResult = (new CreateTableResult()).withTableDescription(newTableDesc);
        return new DynamoDBResponse(createResult, 200);
    }
    return null;
}
 

/**
 * Find existing row by primary key lock.name
 *
 * @param name lock name
 * @return optional lock row or empty
 */
Optional<Lock> find(String name) {
    SimpleStatement selectStatement = QueryBuilder.selectFrom(table)
        .column(LOCK_UNTIL)
        .column(LOCKED_AT)
        .column(LOCKED_BY)
        .whereColumn(LOCK_NAME).isEqualTo(literal(name))
        .build()
        .setConsistencyLevel(consistencyLevel);

    ResultSet resultSet = cqlSession.execute(selectStatement);
    Row row = resultSet.one();
    if (row != null) {
        return Optional.of(new Lock(row.getInstant(LOCK_UNTIL), row.getInstant(LOCKED_AT), row.getString(LOCKED_BY)));
    } else {
        return Optional.empty();
    }
}
 

private static void createTable(CqlSession session, String keyspace, String tableName) {
    ResultSet execute = session.execute("select table_name from system_schema.tables where keyspace_name = '"+keyspace+"';");
    List<Row> all = execute.all();
    boolean found = false;
    for(Row row : all) {
        String table = row.getString("table_name");
        if (table.equalsIgnoreCase(tableName)) {
            found = true;
            break;
        }
    }
    if (!found) {
        session.execute(String.format(createTableTemplate, tableName));
        LOGGER.debug("Table {} has been created", tableName);
    }
}
 

@Override
public List<ActorSystemEventListener> getAll(ShardKey shardKey, ActorSystemEvent event) {
    ResultSet resultSet = executeWithRetry(cassandraSession, selectStatement.bind(clusterName, shardKey.toString(), event.name()).setPageSize(Integer.MAX_VALUE), logger);
    List<ActorSystemEventListener> resultList = new LinkedList<>();
    for (Row resultRow : resultSet) {
        for (int i = 0; i < resultRow.getColumnDefinitions().size(); i++) {
            ByteBuffer resultBuffer = resultRow.getByteBuffer(i);
            try {
                byte[] resultBytes = new byte[requireNonNull(resultBuffer).remaining()];
                resultBuffer.get(resultBytes);
                resultList.add(ActorSystemEventListenerDeserializer.get().deserialize(resultBytes));
            } catch(NullPointerException | IOException e)  {
                logger.error("IOException while deserializing ActorSystemEventListener",e);
            }
        }
    }
    return resultList;
}
 

@Override
public List<ScheduledMessage> getAll(ShardKey shardKey) {
    ResultSet resultSet = executeWithRetry(cassandraSession, selectStatement.bind(clusterName, shardKey.toString()).setPageSize(Integer.MAX_VALUE), logger);
    List<ScheduledMessage> resultList = new LinkedList<>();
    for (Row resultRow : resultSet) {
        for (int i = 0; i < resultRow.getColumnDefinitions().size(); i++) {
            ByteBuffer resultBuffer = resultRow.getByteBuffer(i);
            try {
                byte[] resultBytes = new byte[requireNonNull(resultBuffer).remaining()];
                resultBuffer.get(resultBytes);
                resultList.add(scheduledMessageDeserializer.deserialize(resultBytes));
            } catch (NullPointerException | IOException e) {
                logger.error("IOException while deserializing ScheduledMessage", e);
            }
        }
    }
    return resultList;
}
 

private Map<TokenRange, Long> describeSplits(String keyspace, String table, TokenRange tokenRange, int splitSize, int splitSizeMb, CqlSession session) {
    String query = String.format("SELECT mean_partition_size, partitions_count " +
                    "FROM %s.%s " +
                    "WHERE keyspace_name = ? AND table_name = ? AND range_start = ? AND range_end = ?",
            SchemaConstants.SYSTEM_KEYSPACE_NAME,
            SystemKeyspace.SIZE_ESTIMATES);

    ResultSet resultSet = session.execute(session.prepare(query).bind(keyspace, table, tokenRange.getStart().toString(), tokenRange.getEnd().toString()));

    Row row = resultSet.one();

    long meanPartitionSize = 0;
    long partitionCount = 0;
    int splitCount = 0;

    if (row != null) {
        meanPartitionSize = row.getLong("mean_partition_size");
        partitionCount = row.getLong("partitions_count");

        splitCount = splitSizeMb > 0
                ? (int) (meanPartitionSize * partitionCount / splitSizeMb / 1024 / 1024)
                : (int) (partitionCount / splitSize);
    }

    // If we have no data on this split or the size estimate is 0,
    // return the full split i.e., do not sub-split
    // Assume smallest granularity of partition count available from CASSANDRA-7688
    if (splitCount == 0) {
        Map<TokenRange, Long> wrappedTokenRange = new HashMap<>();
        wrappedTokenRange.put(tokenRange, (long) 128);
        return wrappedTokenRange;
    }

    List<TokenRange> splitRanges = tokenRange.splitEvenly(splitCount);
    Map<TokenRange, Long> rangesWithLength = new HashMap<>();
    for (TokenRange range : splitRanges) {
        rangesWithLength.put(range, partitionCount / splitCount);
    }
    return rangesWithLength;
}
 

public RowIterator() {
    AbstractType type = partitioner.getTokenValidator();
    ResultSet rs = session.execute(session.prepare(cqlQuery).bind(type.compose(type.fromString(split.getStartToken())), type.compose(type.fromString(split.getEndToken()))));
    for (ColumnMetadata meta : session.getMetadata().getKeyspace(quote(keyspace)).get().getTable(quote(cfName)).get().getPartitionKey()) {
        partitionBoundColumns.put(meta.getName().toString(), Boolean.TRUE);
    }
    rows = rs.iterator();
}
 

ResultSet executeOnSession(Statement statement) {
    try {
        this.semaphore.acquire();
        return this.session.execute(statement);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new JanusGraphException("Interrupted while acquiring resource to execute query on Session.");
    } finally {
        this.semaphore.release();
    }
}
 

CQLResultSetKeyIterator(SliceQuery sliceQuery, CQLColValGetter getter, ResultSet resultSet) {
    this.sliceQuery = sliceQuery;
    this.getter = getter;
    this.iterator = Iterator.ofAll(resultSet.iterator())
            .peek(row -> {
                this.currentRow = row;
                this.currentKey = StaticArrayBuffer.of(row.getByteBuffer(CQLKeyColumnValueStore.KEY_COLUMN_NAME));
            });
}
 

@Override
public EntryList getSlice(KeySliceQuery query, StoreTransaction txh) {
    ResultSet result = this.storeManager.executeOnSession(this.getSlice.bind()
            .setByteBuffer(KEY_BINDING, query.getKey().asByteBuffer())
            .setByteBuffer(SLICE_START_BINDING, query.getSliceStart().asByteBuffer())
            .setByteBuffer(SLICE_END_BINDING, query.getSliceEnd().asByteBuffer())
            .setInt(LIMIT_BINDING, query.getLimit())
            .setConsistencyLevel(getTransaction(txh).getReadConsistencyLevel()));

    return fromResultSet(result, this.getter);
}
 

private static EntryList fromResultSet(ResultSet resultSet, StaticArrayEntry.GetColVal<Tuple3<StaticBuffer, StaticBuffer, Row>, StaticBuffer> getter) {
    Lazy<ArrayList<Row>> lazyList = Lazy.of(() -> Lists.newArrayList(resultSet));

    // Use the Iterable overload of of ByteBuffer as it's able to allocate
    // the byte array up front.
    // To ensure that the Iterator instance is recreated, it is created
    // within the closure otherwise
    // the same iterator would be reused and would be exhausted.
    return StaticArrayEntryList.ofStaticBuffer(() -> Iterator.ofAll(lazyList.get()).map(row -> Tuple.of(
            StaticArrayBuffer.of(row.getByteBuffer(COLUMN_COLUMN_NAME)),
            StaticArrayBuffer.of(row.getByteBuffer(VALUE_COLUMN_NAME)),
            row)),
            getter);
}
 

@Test
public void test() {
    Marshaller marshaller = new JsonMarshaller();
    CassandraAppender appender = new CassandraAppender();
    Dictionary<String, Object> config = new Hashtable<>();
    config.put(CassandraAppender.CASSANDRA_PORT_PROPERTY, CASSANDRA_HOST);
    config.put(CassandraAppender.CASSANDRA_PORT_PROPERTY, CASSANDRA_PORT);
    config.put(CassandraAppender.KEYSPACE_PROPERTY, KEYSPACE);
    config.put(CassandraAppender.TABLE_PROPERTY, TABLE_NAME);
    appender.marshaller = marshaller;
    appender.activate(config);
    
    Map<String, Object> properties = new HashMap<>();
    properties.put(EventConstants.TIMESTAMP, TIMESTAMP);
    Event event = new Event(TOPIC, properties);
    
    appender.handleEvent(event);
    appender.deactivate();

    CqlSession session = getSession();

    ResultSet execute = session.execute("SELECT * FROM "+ KEYSPACE+"."+TABLE_NAME+";");
    List<Row> all = execute.all();
    Assert.assertEquals(1, all.size());
    assertThat(all, not(nullValue()));
    
    assertThat(all.get(0).getInstant("timeStamp").toEpochMilli(), is(TIMESTAMP));
    
    session.close();
}
 

@Override
public void activate() {
  super.activate();
  if (isRemembering() || isHashing()) {
    CqlSession session = getCassandra().getSession();
    List<DocKey> strandedDocs = new ArrayList<>();
    PreparedStatement preparedQuery = getCassandra().getPreparedQuery(RESET_PROCESSING_Q);
    BoundStatement statement =  preparedQuery.bind(getName());
    ResultSet procRs = session.execute(statement);
    strandedDocs.addAll(procRs.all().stream()
        .map((row) -> new DocKey(row.getString(0), row.getString(1))).collect(Collectors.toList()));
    preparedQuery = getCassandra().getPreparedQuery(RESET_ERROR_Q);
    statement = preparedQuery.bind(getName());
    ResultSet errorRs = session.execute(statement);
    strandedDocs.addAll(errorRs.all().stream()
        .map((row) -> new DocKey(row.getString(0), row.getString(1))).collect(Collectors.toList()));
    preparedQuery = getCassandra().getPreparedQuery(RESET_BATCHED_Q);
    statement = preparedQuery.bind(getName());
    ResultSet batchedRs = session.execute(statement);
    strandedDocs.addAll(batchedRs.all().stream()
        .map((row) -> new DocKey(row.getString(0), row.getString(1))).collect(Collectors.toList()));

    preparedQuery = getCassandra().getPreparedQuery(RESET_DOCS_U);
    // todo: batch
    for (DocKey docId : strandedDocs) {
      statement = preparedQuery.bind(docId.docid, docId.scanner);
      session.execute(statement);
    }

  }
}
 

private ResultSet executeStatement(SimpleStatement statement, String keyspace) {
    if (keyspace != null) {
        statement.setKeyspace(CqlIdentifier.fromCql(keyspace));
    }

    return session.execute(statement);
}
 

private Row internalGet(final ShardKey shard,final String actorId) {
    // log a warning when we exceed the readExecutionThreshold
    final long startTime = currentTimeMillis();
    try {
        ResultSet resultSet = executeWithRetry(cassandraSession, selectStatement.bind(clusterName, shard.toString(), actorId), logger);
        return resultSet.one();
    } finally {
        final long endTime = currentTimeMillis();
        if((endTime - startTime) > readExecutionThresholdMillis) {
            logger.warn("Cassandra read operation took {} msecs for actorId [{}] on shard [{}]", (endTime - startTime), actorId, shard);
        }
    }
}
 

@Override
public DynamoDBResponse query(QueryRequest payload)
{
    logger.debug("query against JSON table");
    ResultSet resultSet;

    if (payload.getKeyConditionExpression() != null)
        resultSet = queryByKeyExpression(payload);
    else if (payload.getKeyConditions() != null)
        resultSet = queryByKeyCondition(payload);
    else
        throw new UnsupportedOperationException("un-supported query type");

   try
    {
        Collection<Map<String, AttributeValue>> items = new HashSet<Map<String, AttributeValue>>();
        for (Row row : resultSet)
        {
            AttributeValue item;
            ColumnDefinitions colDefs = row.getColumnDefinitions();
            Map<String, AttributeValue> keysSet = new HashMap<>();
            for (ColumnDefinition colDef : colDefs)
            {
                if (colDef.getName().asInternal().equals("json_blob"))
                    continue;

                item = rowToAV(colDef, row);
                keysSet.put(colDef.getName().asInternal(), item);
            }

            Map<String, AttributeValue> itemSet = blobToItemSet(row.getString("json_blob"));
            itemSet.putAll(keysSet);

            if (payload.getFilterExpression() != null){
                if(!matchesFilterExpression(itemSet, payload)){
                    continue;
                }
            }
            items.add(itemSet);
        }

        QueryResult queryResult = new QueryResult();
        queryResult.setItems(items);

        return new DynamoDBResponse(queryResult, 200);
    } catch (Throwable e) {
        logger.warn("Query error", e);

        DynamoDBResponse ddbResponse = new DynamoDBResponse(null, 500);
        String msg= String.format("query failed with error: %s", e.getMessage());
        ddbResponse.setError(msg);
        return ddbResponse;
    }
}
 

@Override
public DynamoDBResponse deleteItem(DeleteItemRequest dir) {
    logger.debug("delete item into JSON table");
    String tableName = dir.getTableName();
    TableDef tableDef = cassandraManager.getTableDef(tableName);

    PreparedStatement deleteStatement = tableDef.getDeleteStatement();

    AttributeDefinition partitionKeyAttr = tableDef.getPartitionKey();
    Optional<AttributeDefinition> maybeCusteringKeyAttr = tableDef.getClusteringKey();

    Map<String, AttributeValue> keys = dir.getKey();

    Object partitionKeyValue = getAttributeObject(
            ScalarAttributeType.fromValue(partitionKeyAttr.getAttributeType()),
            keys.get(partitionKeyAttr.getAttributeName())
    );

    BoundStatement boundStatement;

    if (maybeCusteringKeyAttr.isPresent())
    {
        Object clusteringKeyValue = getAttributeObject(
                ScalarAttributeType.fromValue(maybeCusteringKeyAttr.get().getAttributeType()),
                keys.get(maybeCusteringKeyAttr.get().getAttributeName())
        );

        boundStatement = deleteStatement.bind(partitionKeyValue, clusteringKeyValue);
    }
    else
    {
        boundStatement = deleteStatement.bind(partitionKeyValue);
    }

    ResultSet result = session().execute(boundStatement);

    if (result.wasApplied()){
        DeleteItemResult dres = new DeleteItemResult();
        return new DynamoDBResponse(dres, 200);
    }
    else return null;

}
 

@Override
public DynamoDBResponse getItem(GetItemRequest getItemRequest) {
    logger.debug("get item from JSON table");

    String tableName = getItemRequest.getTableName();
    TableDef tableDef = cassandraManager.getTableDef(tableName);
    PreparedStatement selectStatement = tableDef.getQueryRowStatement();

    AttributeDefinition partitionKeyDef = tableDef.getPartitionKey();
    Optional<AttributeDefinition> clusteringKeyDef = tableDef.getClusteringKey();

    Map<String, AttributeValue> keys = getItemRequest.getKey();

    AttributeValue partitionKey = keys.get(partitionKeyDef.getAttributeName());
    AttributeValue clusteringKey = clusteringKeyDef.isPresent() ?
            keys.get(clusteringKeyDef.get().getAttributeName()) : null;

    ScalarAttributeType partitionKeyType = ScalarAttributeType.valueOf(partitionKeyDef.getAttributeType());
    ScalarAttributeType clusteringKeyType = clusteringKeyDef.isPresent() ?
            ScalarAttributeType.valueOf(clusteringKeyDef.get().getAttributeType()) : null;

    BoundStatement boundStatement = clusteringKey == null ?
            selectStatement.bind(getAttributeObject(partitionKeyType, partitionKey)) :
            selectStatement.bind(getAttributeObject(partitionKeyType, partitionKey),
                    getAttributeObject(clusteringKeyType, clusteringKey));

    ResultSet result = session().execute(boundStatement);

    GetItemResult gir = new GetItemResult();
    Map<String, AttributeValue> item = new HashMap<>();
    ColumnDefinitions colDefs = result.getColumnDefinitions();

    Row row = result.one();

    //Case that nothing is found
    if (row == null)
        return new DynamoDBResponse(null, 200);

    Map<String, AttributeValue> keysSet = new HashMap<>();
    for (ColumnDefinition colDef : colDefs)
    {
        if (colDef.getName().asInternal().equals("json_blob"))
            continue;

        keysSet.put(colDef.getName().asInternal(), rowToAV(colDef, row));
    }

    try
    {
        item = blobToItemSet(row.getString("json_blob"));
        item.putAll(keysSet);

        gir.withItem(item);
        return new DynamoDBResponse(gir, 200);
    } catch (IOException e) {
        DynamoDBResponse ddbResponse = new DynamoDBResponse(gir, 500);
        String msg = String.format("GetItem failed", getItemRequest.getTableName());
        ddbResponse.setError(msg);
        return ddbResponse;
    }
}
 

private TableResponseContainer executeCql(String finalCql, String datacenter, String cassandraHost, int port, String keyspace, String username, String password) {
	TableResponseContainer response = new TableResponseContainer();


	RowSetResponseAspect rowSetAspect = new RowSetResponseAspect();
	var builder = CqlSession.builder()
			.withLocalDatacenter(datacenter)
			.addContactPoint(InetSocketAddress.createUnresolved(cassandraHost, port));

	if (StringUtils.isNotBlank(keyspace)) {
		builder = builder.withKeyspace("\"" + keyspace + "\"");
	}

	if (StringUtils.isNotBlank(username)) {
		builder = builder.withAuthCredentials(username, password == null ? "" : password);
	}

	long requestTimeInMs = 0;
	try (CqlSession session = builder.build()) {
		long startTime = System.currentTimeMillis();
		ResultSet rs = session.execute(finalCql);
		requestTimeInMs = System.currentTimeMillis() - startTime;

		var columnNames = Streams.stream(rs.getColumnDefinitions())
				.map(cd -> cd.getName().asCql(true))
				.collect(Collectors.toList());

		var rows = new LinkedList<List<String>>();
		for (Row row : rs) {
			List<String> rowValues = new LinkedList<>();
			for(int c = 0; c < columnNames.size(); ++c){
				var value = row.getObject(c);
				rowValues.add(value != null ? value.toString() : "NULL");
			}
			rows.add(rowValues);
		}

		rowSetAspect.setColumnNames(columnNames);
		rowSetAspect.setRows(rows);
	}

	response.getAspects().add(rowSetAspect);
	response.getStatusInformations().complete(Map.of(
			"Rows", ""+ rowSetAspect.getRows().size(),
			"Time", requestTimeInMs + "ms"));


	return response;
}
 

@Override
public ResultSet execute(String query) {
	Objects.requireNonNull(query, "'query' must not be null");
	return this.session.execute(query);
}
 

@Override
public ResultSet execute(String query, Object... values) {
	Objects.requireNonNull(query, "'query' must not be null");
	Objects.requireNonNull(values, "'values' must not be null");
	return this.session.execute(SimpleStatement.newInstance(query, values));
}
 

/**
 * What to do when a document has been recognized as required for indexing.
 *
 * @param doc The document to be processed
 */
public void docFound(Document doc) {
  log.trace("{} found doc: {}", getName(), doc.getId());
  String id = doc.getId();
  Function<String, String> idFunction = getIdFunction();
  String result = idFunction.apply(id);
  String idField = doc.getIdField();
  doc.removeAll(idField);
  doc.put(idField, result);

  id = doc.getId();
  String status = null;
  String md5 = null;
  if (isRemembering()) {
    PreparedStatement preparedQuery = getCassandra().getPreparedQuery(FTI_CHECK_Q);
    CqlSession session = getCassandra().getSession();
    ResultSet statusRs = session.execute(preparedQuery.bind(id, getName()));
    if (statusRs.getAvailableWithoutFetching() > 0) {
      if (statusRs.getAvailableWithoutFetching() > 1 || !statusRs.isFullyFetched()) {
        log.error("FATAL: duplicate primary keys in cassandra table??");
        throw new RuntimeException("VERY BAD: duplicate primary keys in FTI table?");
      } else {
        Row next = statusRs.all().iterator().next();
        status = next.getString(0);
        log.trace("Found '{}' with status {}", id, status);
        if (isHashing()) {
          md5 = next.getString(1);
        }
      }
    }
  }
  // written with negated and's so I can defer doc.getHash() until we are sure we
  // need to check the hash.
  if (isRemembering() &&                                         // easier to read, let jvm optimize this check out
      status != null &&                                          // A status was found so we have seen this before
      Status.valueOf(status) != Status.DIRTY &&                  // not marked dirty
      !heuristicDirty(doc)                                       // not dirty by subclass logic
      ) {
    if (!isHashing()) {
      log.trace("{} ignoring previously seen document {}", getName(), id);
      return;
    }
    if (md5 != null) {
      String hash = doc.getHash();
      if (md5.equals(hash)) {
        log.trace("{} ignoring document with previously seen content {}", getName(), id);
        return;
      }
    }
  }
  sendToNext(doc);
}
 

/**
 * Retrieve the Cassandra release version.
 *
 * @param session must not be {@literal null}.
 * @return the release {@link Version}.
 */
public static Version getReleaseVersion(CqlSession session) {

	Assert.notNull(session, "Session must not be null");

	ResultSet resultSet = session.execute("SELECT release_version FROM system.local;");
	Row row = resultSet.one();

	return Version.parse(row.getString(0));
}
 

public List<Video> selectAll(String keyspace) {
    Select select = QueryBuilder.selectFrom(TABLE_NAME).all();

    ResultSet resultSet = executeStatement(select.build(), keyspace);

    List<Video> result = new ArrayList<>();

    resultSet.forEach(x -> result.add(
        new Video(x.getUuid("video_id"), x.getString("title"), x.getInstant("creation_date"))
    ));

    return result;
}