<commit_before>package com.thinkaurelius.faunus.formats.titan.cassandra;

import com.thinkaurelius.faunus.FaunusVertex;
import com.thinkaurelius.faunus.mapreduce.FaunusCompiler;
import org.apache.cassandra.hadoop.ColumnFamilyInputFormat;
import org.apache.cassandra.hadoop.ColumnFamilyRecordReader;
import org.apache.cassandra.hadoop.ConfigHelper;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.commons.configuration.BaseConfiguration;
import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;
import java.util.List;

/**
 * @author <NAME> (http://markorodriguez.com)
 */
public class TitanCassandraInputFormat extends InputFormat implements Configurable {

    private final ColumnFamilyInputFormat columnFamilyInputFormat;
    private FaunusTitanCassandraGraph graph;
    private boolean pathEnabled;
    private Configuration config;


    public TitanCassandraInputFormat() {
        this.columnFamilyInputFormat = new ColumnFamilyInputFormat();
    }

    @Override
    public List<InputSplit> getSplits(final JobContext jobContext) throws IOException, InterruptedException {
        return this.columnFamilyInputFormat.getSplits(jobContext);
    }

    @Override
    public RecordReader<NullWritable, FaunusVertex> createRecordReader(final InputSplit inputSplit, final TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        return new TitanCassandraRecordReader(this.graph, this.pathEnabled, (ColumnFamilyRecordReader) this.columnFamilyInputFormat.createRecordReader(inputSplit, taskAttemptContext));
    }

    @Override
    public void setConf(final Configuration config) {
        ConfigHelper.setInputColumnFamily(config, ConfigHelper.getInputKeyspace(config), "edgestore", false);
        final SlicePredicate predicate = new SlicePredicate();
        final SliceRange sliceRange = new SliceRange();
        sliceRange.setStart(new byte[0]);
        sliceRange.setFinish(new byte[0]);
        predicate.setSlice_range(sliceRange);
        ConfigHelper.setInputSlicePredicate(config, predicate);

        final BaseConfiguration titanconfig = new BaseConfiguration();
        //General Titan configuration for read-only
        titanconfig.setProperty("storage.read-only", "true");
        titanconfig.setProperty("autotype", "none");
        //Cassandra specific configuration
        titanconfig.setProperty("storage.backend", "cassandra");   // todo: astyanax
        titanconfig.setProperty("storage.hostname", ConfigHelper.getInputInitialAddress(config));
        titanconfig.setProperty("storage.keyspace", ConfigHelper.getInputKeyspace(config));
        titanconfig.setProperty("storage.port", ConfigHelper.getInputRpcPort(config));<commit_msg>added Integer.MAX_VALUE count to Cassandra splits to ensure all columns are retrieved. Bam.<commit_after>package com.thinkaurelius.faunus.formats.titan.cassandra;

import com.thinkaurelius.faunus.FaunusVertex;
import com.thinkaurelius.faunus.mapreduce.FaunusCompiler;
import com.thinkaurelius.titan.graphdb.configuration.GraphDatabaseConfiguration;
import org.apache.cassandra.hadoop.ColumnFamilyInputFormat;
import org.apache.cassandra.hadoop.ColumnFamilyRecordReader;
import org.apache.cassandra.hadoop.ConfigHelper;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.commons.configuration.BaseConfiguration;
import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;
import java.util.List;

/**
 * @author <NAME> (http://markorodriguez.com)
 */
public class TitanCassandraInputFormat extends InputFormat implements Configurable {

    private final ColumnFamilyInputFormat columnFamilyInputFormat;
    private FaunusTitanCassandraGraph graph;
    private boolean pathEnabled;
    private Configuration config;


    public TitanCassandraInputFormat() {
        this.columnFamilyInputFormat = new ColumnFamilyInputFormat();
    }

    @Override
    public List<InputSplit> getSplits(final JobContext jobContext) throws IOException, InterruptedException {
        return this.columnFamilyInputFormat.getSplits(jobContext);
    }

    @Override
    public RecordReader<NullWritable, FaunusVertex> createRecordReader(final InputSplit inputSplit, final TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        return new TitanCassandraRecordReader(this.graph, this.pathEnabled, (ColumnFamilyRecordReader) this.columnFamilyInputFormat.createRecordReader(inputSplit, taskAttemptContext));
    }

    @Override
    public void setConf(final Configuration config) {
        ConfigHelper.setInputColumnFamily(config, ConfigHelper.getInputKeyspace(config), GraphDatabaseConfiguration.STORAGE_EDGESTORE_NAME);
        final SlicePredicate predicate = new SlicePredicate();
        final SliceRange sliceRange = new SliceRange();
        sliceRange.setStart(new byte[0]);
        sliceRange.setFinish(new byte[0]);
        sliceRange.setCount(Integer.MAX_VALUE);
        predicate.setSlice_range(sliceRange);
        ConfigHelper.setInputSlicePredicate(config, predicate);

        final BaseConfiguration titanconfig = new BaseConfiguration();
        //General Titan configuration for read-only
        titanconfig.setProperty("storage.read-only", "true");
        titanconfig.setProperty("autotype", "none");
        //Cassandra specific configuration
        titanconfig.setProperty("storage.backend", "cassandra");   // todo: astyanax
        titanconfig.setProperty("storage.hostname", ConfigHelper.getInputInitialAddress(config));
        titanconfig.setProperty("storage.keyspace", ConfigHelper.getInputKeyspace(config));
        titanconfig.setProperty("storage.port", ConfigHelper.getInputRpcPort(config));
{'fraction_non_alphanumeric': 0.08849009900990099, 'fraction_numerical': 0.0006188118811881188, 'mean_word_length': 5.252417794970986, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, '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': '3321310', 'n_tokens_mistral': 1770, 'n_tokens_neox': 1717, 'n_words': 315}