<commit_before>package com.hmsonline.trident.cql.incremental;

import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.Statement;
import com.hmsonline.trident.cql.CqlClientFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import storm.trident.operation.CombinerAggregator;
import storm.trident.state.State;
import storm.trident.tuple.TridentTuple;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class CassandraCqlIncrementalState<K, V> implements State {
    private static final Logger LOG = LoggerFactory.getLogger(CassandraCqlIncrementalState.class);
    private CqlClientFactory clientFactory;
    private CombinerAggregator<V> aggregator;
    private CqlIncrementMapper<K, V> mapper;
    private Map<K, V> aggregateValues;
    public static int MAX_ATTEMPTS = 10;
    private int partitionIndex;

    public CassandraCqlIncrementalState(CqlClientFactory clientFactory, CombinerAggregator<V> aggregator, CqlIncrementMapper<K, V> mapper, int partitionIndex) {
        this.clientFactory = clientFactory;
        this.aggregator = aggregator;
        this.mapper = mapper;
        this.partitionIndex = partitionIndex;
    }

    @Override
    public void beginCommit(Long txid) {
        aggregateValues = new HashMap<K, V>();
    }

    private boolean applyUpdate(Statement updateStatement) {
        LOG.debug("APPLYING [{}]", updateStatement.toString());
        boolean applied = false;
        int attempts = 0;
        while (!applied && attempts < MAX_ATTEMPTS) {
            ResultSet results = clientFactory.getSession().execute(updateStatement);
            Row row = results.one();
            if (row != null)
                applied = row.getBool("[applied]");
            attempts++;
        }
        return applied;
    }

    @Override
    public void commit(Long txid) {
        // Read current value.
        for (Map.Entry<K, V> entry : aggregateValues.entrySet()) {

            //Try to check pre-condition first; this is optional
            //The logic of condition is simple. The return statement checks a table for existing
            //key and insert if not exist. Following code checks the execution of condition statement
            //and skip the loop when receiving false value => Insert is failed, key is existed.
            final Statement condition = mapper.condition(entry.getKey(), txid, partitionIndex);
            if (condition != null && !applyUpdate(condition)) {
                continue;
            }

            Statement readStatement = mapper.read(entry.getKey());
            LOG.debug("EXECUTING [{}]", readStatement.toString());

            ResultSet results = clientFactory.getSession().execute(readStatement);

            V persistedValue = null;

            if (results != null) {
                List<Row> rows = results.all();
                if (rows.size() > 1) {
                    LOG.error("Found non-unique value for key [{}]", entry.getKey());
                } else if (rows.size() == 1) {
                    persistedValue = mapper.currentValue(rows.get(0));
                    LOG.debug("Persisted value = [{}]", persistedValue);
                }

                V combinedValue;
                // TODO: more elegant solution to this issue
                // Must be careful here as the first persisted value might not exist yet!
                if (persistedValue != null)<commit_msg>- change to use new feature if_not_exist of CQL 2.0<commit_after>package com.hmsonline.trident.cql.incremental;

import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.Statement;
import com.hmsonline.trident.cql.CqlClientFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import storm.trident.operation.CombinerAggregator;
import storm.trident.state.State;
import storm.trident.tuple.TridentTuple;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class CassandraCqlIncrementalState<K, V> implements State {
    private static final Logger LOG = LoggerFactory.getLogger(CassandraCqlIncrementalState.class);
    private CqlClientFactory clientFactory;
    private CombinerAggregator<V> aggregator;
    private CqlIncrementMapper<K, V> mapper;
    private Map<K, V> aggregateValues;
    public static int MAX_ATTEMPTS = 10;
    private int partitionIndex;

    public CassandraCqlIncrementalState(CqlClientFactory clientFactory, CombinerAggregator<V> aggregator, CqlIncrementMapper<K, V> mapper, int partitionIndex) {
        this.clientFactory = clientFactory;
        this.aggregator = aggregator;
        this.mapper = mapper;
        this.partitionIndex = partitionIndex;
    }

    @Override
    public void beginCommit(Long txid) {
        aggregateValues = new HashMap<K, V>();
    }

    private boolean applyUpdate(Statement updateStatement) {
        LOG.debug("APPLYING [{}]", updateStatement.toString());
        boolean applied = false;
        int attempts = 0;
        while (!applied && attempts < MAX_ATTEMPTS) {
            ResultSet results = clientFactory.getSession().execute(updateStatement);
            Row row = results.one();
            if (row != null)
                applied = row.getBool("[applied]");
            attempts++;
        }
        return applied;
    }

    @Override
    public void commit(Long txid) {
        // Read current value.
        for (Map.Entry<K, V> entry : aggregateValues.entrySet()) {

            //Try to check pre-condition first; this is optional
            //The logic of condition is simple. The return statement checks a table for existing
            //key and insert if not exist. Following code checks the execution of condition statement
            //and skip the loop when receiving false value => Insert is failed, key is existed.
            final Statement condition = mapper.condition(entry.getKey(), txid, partitionIndex);
            final boolean notSatisfiedPrecondition = condition != null && !applyUpdate(condition);

            if (notSatisfiedPrecondition) {
                continue;
            }

            Statement readStatement = mapper.read(entry.getKey());
            LOG.debug("EXECUTING [{}]", readStatement.toString());

            ResultSet results = clientFactory.getSession().execute(readStatement);

            V persistedValue = null;

            if (results != null) {
                List<Row> rows = results.all();
                if (rows.size() > 1) {
                    LOG.error("Found non-unique value for key [{}]", entry.getKey());
                } else if (rows.size() == 1) {
                    persistedValue = mapper.currentValue(rows.get(0));
                    LOG.debug("Persisted value = [{}]", persistedValue);
                }

                V combinedValue;
                // TODO: more elegant solution to this issue
                // Must be careful here as the first persisted value might not exist yet!
                if (persistedValue != null)
{'fraction_non_alphanumeric': 0.09657320872274143, 'fraction_numerical': 0.002548853016142736, 'mean_word_length': 2.5244510978043913, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3755142', 'n_tokens_mistral': 1889, 'n_tokens_neox': 1832, 'n_words': 514}