This is a dummy implementation and this feature is not implemented
for Cassandra backed Conductor
164332
@Override
    public List<Workflow> getWorkflowsByType(String workflowName, Long startTime, Long endTime) {
        throw new UnsupportedOperationException("This method is not implemented in CassandraExecutionDAO. Please use ExecutionDAOFacade instead.");
    }
1313486
@Override
    public void mutateMany(Map<String, Map<StaticBuffer, KCVMutation>> mutations, StoreTransaction txh) throws BackendException {
        Preconditions.checkNotNull(mutations);

        final MaskedTimestamp commitTime = new MaskedTimestamp(txh);

        int size = 0;
        for (Map<StaticBuffer, KCVMutation> mutation : mutations.values()) size += mutation.size();
        Map<StaticBuffer, org.apache.cassandra.db.Mutation> rowMutations = new HashMap<>(size);

        for (Map.Entry<String, Map<StaticBuffer, KCVMutation>> mutEntry : mutations.entrySet()) {
            String columnFamily = mutEntry.getKey();
            for (Map.Entry<StaticBuffer, KCVMutation> titanMutation : mutEntry.getValue().entrySet()) {
                StaticBuffer key = titanMutation.getKey();
                KCVMutation mut = titanMutation.getValue();

                org.apache.cassandra.db.Mutation rm = rowMutations.get(key);
                if (rm == null) {
                    rm = new org.apache.cassandra.db.Mutation(keySpaceName, key.asByteBuffer());
                    rowMutations.put(key, rm);
                }

                if (mut.hasAdditions()) {
                    for (Entry e : mut.getAdditions()) {
                        Integer ttl = (Integer) e.getMetaData().get(EntryMetaData.TTL);

                        if (null != ttl && ttl > 0) {
                            rm.add(columnFamily, CellNames.simpleDense(e.getColumnAs(StaticBuffer.BB_FACTORY)),
                                    e.getValueAs(StaticBuffer.BB_FACTORY), commitTime.getAdditionTime(times), ttl);
                        } else {
                            rm.add(columnFamily, CellNames.simpleDense(e.getColumnAs(StaticBuffer.BB_FACTORY)),
                                    e.getValueAs(StaticBuffer.BB_FACTORY), commitTime.getAdditionTime(times));
                        }
                    }
                }

                if (mut.hasDeletions()) {
                    for (StaticBuffer col : mut.getDeletions()) {
                        rm.delete(columnFamily, CellNames.simpleDense(col.as(StaticBuffer.BB_FACTORY)),
                                commitTime.getDeletionTime(times));
                    }
                }

            }
        }

        mutate(new ArrayList<org.apache.cassandra.db.Mutation>(rowMutations.values()), getTx(txh).getWriteConsistencyLevel().getDB());

        sleepAfterWrite(txh, commitTime);
    }
1009947
private void dropKeyspaceOrCFs() throws InvalidRequestException, SchemaDisagreementException, TException, Exception
    {
        if (createdKeyspaces.contains(databaseName))// drop if created during
                                                    // create-drop call.
        {
            cassandra_client.system_drop_keyspace(databaseName);
        }
        else
        {
            cassandra_client.set_keyspace(databaseName);
            for (TableInfo tableInfo : tableInfos)
            {
                dropColumnFamily(tableInfo);
            }
        }
    }
780426
protected boolean initiateClient()
    {
        Throwable message = null;

        for (String host : hosts)
        {
            if (host == null || !StringUtils.isNumeric(port) || port.isEmpty())
            {
                log.error("Host or port should not be null, Port should be numeric.");
                throw new IllegalArgumentException("Host or port should not be null, Port should be numeric.");
            }
            int thriftPort = externalProperties.get(CassandraConstants.THRIFT_PORT) != null ? Integer
                    .parseInt((String) externalProperties.get(CassandraConstants.THRIFT_PORT)) : Integer.parseInt(port);
            TSocket socket = new TSocket(host, thriftPort);
            TTransport transport = new TFramedTransport(socket);
            TProtocol protocol = new TBinaryProtocol(transport, true, true);
            cassandra_client = new Cassandra.Client(protocol);
            try
            {
                if (!socket.isOpen())
                {
                    socket.open();
                    if (userName != null)
                    {
                        Map<String, String> credentials = new HashMap<String, String>();
                        credentials.put("username", userName);
                        credentials.put("password", password);
                        AuthenticationRequest auth_request = new AuthenticationRequest(credentials);
                        cassandra_client.login(auth_request);
                    }
                }
                return true;
            }
            catch (TTransportException e)
            {
                message = e;
                log.warn("Error while opening socket for host {}, skipping for next available node ", host);
            }
            catch (Exception e)
            {
                log.error("Error during creating schema in cassandra, Caused by: .", e);
                throw new SchemaGenerationException(e, "Cassandra");
            }
        }
        throw new SchemaGenerationException("Error while opening socket, Caused by: .", message, "Cassandra");
    }
548665
private <C> Iterator<Column<C>> columnScan(final ByteBuffer rowKey,
                                               final DeltaPlacement placement,
                                               final ColumnFamily<ByteBuffer, C> columnFamily,
                                               final C start,
                                               final C end,
                                               final boolean reversed,
                                               final ColumnInc<C> columnInc,
                                               final long limit,
                                               final long page,
                                               final ReadConsistency consistency) {
        return Iterators.concat(new AbstractIterator<Iterator<Column<C>>>() {
            private C _from = start;
            private long _remaining = limit;
            private long _page = page;

            @Override
            protected Iterator<Column<C>> computeNext() {
                if (_remaining <= 0) {
                    return endOfData();
                }

                // For page N+1, treat "_from" as exclusive.  Since Cassandra doesn't support exclusive column ranges
                // bump the from value up to the next possible time UUID (assumes from != null when page != 0).
                if (_page > 0) {
                    if (_from.equals(end)) {
                        return endOfData();
                    }
                    _from = reversed ? columnInc.previous(_from) : columnInc.next(_from);
                    if (_from == null) {
                        return endOfData();
                    }
                }

                // Execute the query
                int batchSize = (int) Math.min(_remaining, MAX_COLUMN_SCAN_BATCH);
                ColumnList<C> columns = execute(placement.getKeyspace()
                                .prepareQuery(columnFamily, SorConsistencies.toAstyanax(consistency))
                                .getKey(rowKey)
                                .withColumnRange(_from, end, reversed, batchSize),
                        "scan columns in placement %s, column family %s, row %s, from %s to %s",
                        placement.getName(), columnFamily.getName(), rowKey, start, end);

                // Update state for the next iteration.
                if (columns.size() >= batchSize) {
                    // Save the last column key so we can use it as the start (exclusive) if we must query to get more data.
                    _from = columns.getColumnByIndex(columns.size() - 1).getName();
                    _remaining = _remaining - columns.size();
                    _page++;
                } else {
                    // If we got fewer columns than we asked for, another query won't find more columns.
                    _remaining = 0;
                }

                // Track metrics.  For rows w/more than 50 columns, count subsequent reads w/_largeRowReadMeter.
                (_page == 0 ? _randomReadMeter : _largeRowReadMeter).mark();

                return columns.iterator();
            }
        });
    }
917632
private void doDeliverHintsToEndpoint(InetAddress endpoint)
    {
        // find the hints for the node using its token.
        UUID hostId = Gossiper.instance.getHostId(endpoint);
        logger.info("Started hinted handoff for host: {} with IP: {}", hostId, endpoint);
        final ByteBuffer hostIdBytes = ByteBuffer.wrap(UUIDGen.decompose(hostId));
        DecoratedKey epkey =  StorageService.getPartitioner().decorateKey(hostIdBytes);

        final AtomicInteger rowsReplayed = new AtomicInteger(0);
        Composite startColumn = Composites.EMPTY;

        int pageSize = calculatePageSize();
        logger.debug("Using pageSize of {}", pageSize);

        // rate limit is in bytes per second. Uses Double.MAX_VALUE if disabled (set to 0 in cassandra.yaml).
        // max rate is scaled by the number of nodes in the cluster (CASSANDRA-5272).
        int throttleInKB = DatabaseDescriptor.getHintedHandoffThrottleInKB()
                           / (StorageService.instance.getTokenMetadata().getAllEndpoints().size() - 1);
        RateLimiter rateLimiter = RateLimiter.create(throttleInKB == 0 ? Double.MAX_VALUE : throttleInKB * 1024);

        delivery:
        while (true)
        {
            long now = System.currentTimeMillis();
            QueryFilter filter = QueryFilter.getSliceFilter(epkey,
                                                            SystemKeyspace.HINTS_CF,
                                                            startColumn,
                                                            Composites.EMPTY,
                                                            false,
                                                            pageSize,
                                                            now);

            ColumnFamily hintsPage = ColumnFamilyStore.removeDeleted(hintStore.getColumnFamily(filter), (int) (now / 1000));

            if (pagingFinished(hintsPage, startColumn))
            {
                logger.info("Finished hinted handoff of {} rows to endpoint {}", rowsReplayed, endpoint);
                break;
            }

            // check if node is still alive and we should continue delivery process
            if (!FailureDetector.instance.isAlive(endpoint))
            {
                logger.info("Endpoint {} died during hint delivery; aborting ({} delivered)", endpoint, rowsReplayed);
                break;
            }

            List<WriteResponseHandler> responseHandlers = Lists.newArrayList();
            for (final Cell hint : hintsPage)
            {
                // check if hints delivery has been paused during the process
                if (hintedHandOffPaused)
                {
                    logger.debug("Hints delivery process is paused, aborting");
                    break delivery;
                }

                // Skip tombstones:
                // if we iterate quickly enough, it's possible that we could request a new page in the same millisecond
                // in which the local deletion timestamp was generated on the last column in the old page, in which
                // case the hint will have no columns (since it's deleted) but will still be included in the resultset
                // since (even with gcgs=0) it's still a "relevant" tombstone.
                if (!hint.isLive())
                    continue;

                startColumn = hint.name();

                int version = Int32Type.instance.compose(hint.name().get(1));
                DataInputStream in = new DataInputStream(ByteBufferUtil.inputStream(hint.value()));
                Mutation mutation;
                try
                {
                    mutation = Mutation.serializer.deserialize(in, version);
                }
                catch (UnknownColumnFamilyException e)
                {
                    logger.debug("Skipping delivery of hint for deleted columnfamily", e);
                    deleteHint(hostIdBytes, hint.name(), hint.timestamp());
                    continue;
                }
                catch (IOException e)
                {
                    throw new AssertionError(e);
                }

                for (UUID cfId : mutation.getColumnFamilyIds())
                {
                    if (hint.timestamp() <= SystemKeyspace.getTruncatedAt(cfId))
                    {
                        logger.debug("Skipping delivery of hint for truncated columnfamily {}", cfId);
                        mutation = mutation.without(cfId);
                    }
                }

                if (mutation.isEmpty())
                {
                    deleteHint(hostIdBytes, hint.name(), hint.timestamp());
                    continue;
                }

                MessageOut<Mutation> message = mutation.createMessage();
                rateLimiter.acquire(message.serializedSize(MessagingService.current_version));
                Runnable callback = new Runnable()
                {
                    public void run()
                    {
                        rowsReplayed.incrementAndGet();
                        deleteHint(hostIdBytes, hint.name(), hint.timestamp());
                    }
                };
                WriteResponseHandler responseHandler = new WriteResponseHandler(endpoint, WriteType.SIMPLE, callback);
                MessagingService.instance().sendRR(message, endpoint, responseHandler, false);
                responseHandlers.add(responseHandler);
            }

            for (WriteResponseHandler handler : responseHandlers)
            {
                try
                {
                    handler.get();
                }
                catch (WriteTimeoutException e)
                {
                    logger.info("Timed out replaying hints to {}; aborting ({} delivered)", endpoint, rowsReplayed);
                    break delivery;
                }
            }
        }

        // Flush all the tombstones to disk
        hintStore.forceBlockingFlush();
    }
440483
@Override
    protected int onExecuteUpdate() {
        EntityMetadata m = getEntityMetadata();
        Client client = m != null ? persistenceDelegeator.getClient(m)
            : persistenceDelegeator.getClient(kunderaQuery.getPersistenceUnit());
        externalProperties = ((CassandraClientBase) client).getExternalProperties();
        ApplicationMetadata appMetadata = kunderaMetadata.getApplicationMetadata();

        String query = appMetadata.getQuery(getJPAQuery());

        boolean isNative = kunderaQuery.isNative();

        if (isNative) {
            ((CassandraClientBase) client).executeQuery(m == null ? null : m.getEntityClazz(), null, isNative,
                query != null ? query : getJPAQuery());
        } else if (kunderaQuery.isDeleteUpdate()) {
            // If query is not convertible to CQL, fetch and merge records usual
            // way, otherwise
            // convert to CQL and execute
            if (!isQueryConvertibleToCQL(kunderaQuery)) {
                return onUpdateDeleteEvent();
            } else {
                query = null;
                if (kunderaQuery.isUpdateClause()) {
                    query = createUpdateQuery(kunderaQuery);
                } else {
                    query = createDeleteQuery(kunderaQuery);
                }
                return ((CassandraClientBase) client).executeUpdateDeleteQuery(query);
            }
        }
        return 0;
    }
441269
public CassandraJavaRDD<R> withDescOrder() {
        CassandraRDD<R> newRDD = rdd().withDescOrder();
        return wrap(newRDD);
    }
926965
public static CqlBulkRecordWriter newWriter(Configuration conf,
                                              String host,
                                              int port,
                                              String username,
                                              String password,
                                              String keyspace,
                                              String table,
                                              String partitioner,
                                              String tableSchema,
                                              String insertStatement)
      throws IOException {
    Config.setClientMode(true);
    ConfigHelper.setOutputInitialAddress(conf, host);
    if (port >= 0) {
      ConfigHelper.setOutputRpcPort(conf, String.valueOf(port));
    }
    if (username != null && password != null) {
      ConfigHelper.setOutputKeyspaceUserNameAndPassword(conf, username, password);
    }
    ConfigHelper.setOutputKeyspace(conf, keyspace);
    ConfigHelper.setOutputColumnFamily(conf, table);
    ConfigHelper.setOutputPartitioner(conf, partitioner);
    CqlBulkOutputFormat.setTableSchema(conf, table, tableSchema);
    CqlBulkOutputFormat.setTableInsertStatement(conf, table, insertStatement);
    conf.set(CqlBulkRecordWriter.OUTPUT_LOCATION, Files.createTempDirectory("scio-cassandra-").toString());

    // workaround for Hadoop static initialization
    if (!System.getProperties().containsKey("hadoop.home.dir") &&
        !System.getenv().containsKey("HADOOP_HOME")) {
      System.setProperty("hadoop.home.dir", "/");
    }

    return new CqlBulkRecordWriter(conf);
  }
49212
protected void createIndexesOnColumns(EntityMetadata m, String tableName, List<Column> columns, Class columnType) {
        Object pooledConnection = null;
        try {
            Cassandra.Client api = null;
            pooledConnection = getConnection();
            api = (org.apache.cassandra.thrift.Cassandra.Client) getConnection(pooledConnection);
            KsDef ksDef = api.describe_keyspace(m.getSchema());
            List<CfDef> cfDefs = ksDef.getCf_defs();

            // Column family definition on which secondary index creation is
            // required
            CfDef columnFamilyDefToUpdate = null;
            boolean isUpdatable = false;
            for (CfDef cfDef : cfDefs) {
                if (cfDef.getName().equals(tableName)) {
                    columnFamilyDefToUpdate = cfDef;
                    break;
                }
            }

            if (columnFamilyDefToUpdate == null) {
                log.error("Join table {} not available.", tableName);
                throw new PersistenceException("table" + tableName + " not found!");
            }
            // create a column family, in case it is not already available.

            // Get list of indexes already created
            List<ColumnDef> columnMetadataList = columnFamilyDefToUpdate.getColumn_metadata();
            List<String> indexList = new ArrayList<String>();

            if (columnMetadataList != null) {
                for (ColumnDef columnDef : columnMetadataList) {
                    indexList.add(new StringAccessor().fromBytes(String.class, columnDef.getName()));
                }
                // need to set them to null else it is giving problem on update
                // column family and trying to add again existing indexes.
                // columnFamilyDefToUpdate.column_metadata = null;
            }

            // Iterate over all columns for creating secondary index on them
            for (Column column : columns) {

                ColumnDef columnDef = new ColumnDef();

                columnDef.setName(column.getName());
                columnDef.setValidation_class(CassandraValidationClassMapper.getValidationClass(columnType, false));
                columnDef.setIndex_type(IndexType.KEYS);

                // Add secondary index only if it's not already created
                // (if already created, it would be there in column family
                // definition)
                if (!indexList.contains(new StringAccessor().fromBytes(String.class, column.getName()))) {
                    isUpdatable = true;
                    columnFamilyDefToUpdate.addToColumn_metadata(columnDef);
                }
            }

            // Finally, update column family with modified column family
            // definition
            if (isUpdatable) {
                columnFamilyDefToUpdate.setKey_validation_class(CassandraValidationClassMapper
                    .getValidationClass(m.getIdAttribute().getJavaType(), isCql3Enabled(m)));
                api.system_update_column_family(columnFamilyDefToUpdate);
            }

        } catch (Exception e) {
            log.warn("Could not create secondary index on column family {}, Caused by: . ", tableName, e);

        } finally {
            releaseConnection(pooledConnection);
        }
    }
757207
@SuppressWarnings("unchecked")
    public CassandraJavaPairRDD<K, V> withReadConf(ReadConf config) {
        CassandraRDD<Tuple2<K, V>> newRDD = rdd().withReadConf(config);
        return wrap(newRDD);
    }
1271331
public CassandraJavaRDD<R> withConnector(CassandraConnector connector) {
        CassandraRDD<R> newRDD = rdd().withConnector(connector);
        return wrap(newRDD);
    }
1089992
@Given("^I create a Cassandra index named '(.+?)' in table '(.+?)' using magic_column '(.+?)' using keyspace '(.+?)'$")
    public void createBasicMapping(String index_name, String table, String column, String keyspace) throws Exception {
        String query = "CREATE INDEX " + index_name + " ON " + table + " (" + column + ");";
        commonspec.getCassandraClient().executeQuery(query);
    }
1356657
@SuppressWarnings({ "unchecked", "rawtypes" })
    @Override
    public State makeState(Map conf, IMetricsContext metrics, int partitionIndex, int numPartitions) {
        LOG.info("Making new CassandraState object for cluster " + cassandraClusterId + ": partition [" + partitionIndex + "] of [" + numPartitions + "]");
        AstyanaxClient client = AstyanaxClientFactory.getInstance(cassandraClusterId, (Map)conf.get(cassandraClusterId));
        int batchMaxSize = Utils.getInt(Utils.get(conf, StormCassandraConstants.CASSANDRA_BATCH_MAX_SIZE,
                CassandraState.DEFAULT_MAX_BATCH_SIZE));
        return new CassandraState(client, batchMaxSize, this.exceptionHandler);
    }
548500
@Override
    public List<EventExecution> getEventExecutions(String eventHandlerName, String eventName, String messageId, int max) {
        throw new UnsupportedOperationException("This method is not implemented in CassandraExecutionDAO. Please use ExecutionDAOFacade instead.");
    }
1879
public static Collection marshalCollection(Class cassandraTypeClazz, Collection result, Class clazz, Class resultTypeClass)
    {
        Collection mappedCollection = result;

        if (cassandraTypeClazz.isAssignableFrom(BytesType.class))
        {
            mappedCollection = (Collection) PropertyAccessorHelper.getObject(resultTypeClass);
            for (Object value : result)
            {
                byte[] valueAsBytes = new byte[((ByteBuffer) value).remaining()];
                ((ByteBuffer) value).get(valueAsBytes);
                 mappedCollection.add(PropertyAccessorHelper.getObject(clazz, valueAsBytes));
            }
        }
        return mappedCollection;
    }
544416
static <V> EntityMetadata buildEntityMetadata(@NotNull Class<V> clazz, @NotNull CassandraClient cassandraClient) throws MetadataException {
        Table tableAnnotation = clazz.getAnnotation(Table.class);

        if (tableAnnotation == null) {
            throw new MetadataException(String.format("Table annotation is missing for %s", clazz.getName()));
        }

        String tableKeyspace = StringUtils.isEmpty(tableAnnotation.keyspace()) ? cassandraClient.getDefaultKeyspace() : tableAnnotation.keyspace();

        String tableName = tableAnnotation.name();

        TableMetadata tableMetadata = cassandraClient.getTableMetadata(tableKeyspace, tableName);

        EntityMetadata entityMetadata = new EntityMetadata(
                tableName,
                tableKeyspace,
                tableAnnotation.compactStorage(),
                tableAnnotation.fetchSize(),
                tableAnnotation.readConsistencyLevel(),
                tableAnnotation.writeConsistencyLevel(),
                tableMetadata
        );

        try {
            BeanInfo beanInfo = Introspector.getBeanInfo(clazz);

            for (PropertyDescriptor propertyDescriptor : beanInfo.getPropertyDescriptors()) {

                ColumnMetadata columnMetadata = buildColumnMetadata(propertyDescriptor, clazz, cassandraClient, tableMetadata);

                if (columnMetadata != null) {
                    entityMetadata.addColumnMetadata(columnMetadata);
                }

                LOGGER.debug("Property descriptor: {} {}", propertyDescriptor.getName(), propertyDescriptor.getDisplayName());
            }
        } catch (IntrospectionException e) {
            LOGGER.warn("Can't build column metadata", e);
        }

        EntityMetadata.validate(entityMetadata);

        return entityMetadata;
    }