package org.pingles.cascading.cassandra;

import cascading.tap.SinkMode;
import cascading.tap.Tap;
import cascading.tap.hadoop.TapCollector;
import cascading.tap.hadoop.TapIterator;
import cascading.tuple.TupleEntryCollector;
import cascading.tuple.TupleEntryIterator;
import org.apache.cassandra.hadoop.ConfigHelper;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.JobConf;
import org.pingles.cascading.cassandra.hadoop.ColumnFamilyInputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;

public class CassandraTap extends Tap {
    private static final Logger LOGGER = LoggerFactory.getLogger(CassandraScheme.class);
    private static final String SCHEME = "cassandra";

    private final String initialAddress;
    private final Integer rpcPort;
    private String columnFamilyName;
    private String keyspace;

    public CassandraTap(String initialAddress, Integer rpcPort, String keyspace, String columnFamilyName, CassandraScheme scheme) {
        super(scheme, SinkMode.APPEND);
        this.initialAddress = initialAddress;
        this.rpcPort = rpcPort;
        this.columnFamilyName = columnFamilyName;
        this.keyspace = keyspace;
    }

    @Override
    public void sinkInit(JobConf conf) throws IOException {
        LOGGER.info("Created Cassandra tap {}", getPath());
        LOGGER.info("Sinking to column family: {}", columnFamilyName);

        super.sinkInit(conf);

        ConfigHelper.setOutputColumnFamily(conf, keyspace, columnFamilyName);
        ConfigHelper.setPartitioner(conf, "org.apache.cassandra.dht.RandomPartitioner");
        ConfigHelper.setInitialAddress(conf, this.initialAddress);
        ConfigHelper.setRpcPort(conf, this.rpcPort.toString());
    }

    @Override
    public void sourceInit(JobConf conf) throws IOException {
        LOGGER.info("Sourcing from column family: {}", columnFamilyName);

        FileInputFormat.addInputPaths(conf, getPath().toString());
        conf.setInputFormat(ColumnFamilyInputFormat.class);
        ConfigHelper.setInputColumnFamily(conf, keyspace, columnFamilyName);
        ConfigHelper.setInitialAddress(conf, initialAddress);
        ConfigHelper.setRpcPort(conf, rpcPort.toString());

        super.sourceInit(conf);
    }

    @Override
    public Path getPath() {
        return new Path(String.format("%s://%s:%d/%s/%s", SCHEME, initialAddress, rpcPort, keyspace, columnFamilyName));
    }

    @Override
    public TupleEntryIterator openForRead(JobConf jobConf) throws IOException {
        return new TupleEntryIterator(getSourceFields(), new TapIterator(this, jobConf));
    }

    @Override
    public TupleEntryCollector openForWrite(JobConf jobConf) throws IOException {
        return new TapCollector(this, jobConf);
    }

    @Override
    public boolean makeDirs(JobConf jobConf) throws IOException {
        throw new UnsupportedOperationException("makeDirs unsupported with Cassandra.");
    }

    @Override
    public boolean deletePath(JobConf jobConf) throws IOException {
        throw new UnsupportedOperationException("deletePath unsupported with Cassandra.");
    }

    @Override
    public boolean pathExists(JobConf jobConf) throws IOException {
        throw new UnsupportedOperationException("pathExists unsupported with Cassandra.");
    }

    @Override
    public long getPathModified(JobConf jobConf) throws IOException {
        throw new UnsupportedOperationException("getPathModified unsupported with Cassandra.");
    }
}

@@ -258,24 +258,52 @@
 ryIterator;%0A
+import cascading.util.Util;%0A
 import org.a
@@ -700,16 +700,24 @@
 LOGGER =
+%0A       
  LoggerF
@@ -789,16 +789,20 @@
  String 
+URI_
 SCHEME =
@@ -983,16 +983,29 @@
 ndraTap(
+%0A            
 String i
@@ -1052,16 +1052,28 @@
 eyspace,
+%0A           
  String 
@@ -1681,16 +1681,28 @@
 er(conf,
+%0A           
  %22org.ap
@@ -2403,37 +2403,34 @@
     
-@Override%0A    public Path get
+protected String getString
 Path
@@ -2453,17 +2453,8 @@
 urn 
-new Path(
 Stri
@@ -2482,17 +2482,33 @@
 /%25s/%25s%22,
- 
+%0A            URI_
 SCHEME, 
@@ -2559,17 +2559,865 @@
 ilyName)
-)
+;%0A    %7D%0A%0A    @Override%0A    public Path getPath() %7B%0A        return new Path(getStringPath());%0A    %7D%0A%0A    @Override%0A    public String toString() %7B%0A        return getClass().getSimpleName()%0A            + %22%5B%5C%22%22 + URI_SCHEME + %22%5C%22%5D%22%0A            + %22%5B%5C%22%22 + Util.sanitizeUrl(getStringPath()) + %22%5C%22%5D%22;%0A    %7D%0A%0A    @Override%0A    public boolean equals(Object obj) %7B%0A        if (this == obj) return true;%0A        if (obj == null %7C%7C getClass() != obj.getClass()) return false;%0A        if (!super.equals(obj)) return false;%0A%0A        CassandraTap tap = (CassandraTap) obj;%0A        if (!getScheme().equals(tap.getScheme())) return false;%0A        return getStringPath().equals(tap.getStringPath());%0A    %7D%0A%0A    @Override%0A    public int hashCode() %7B%0A        int result = super.hashCode();%0A        result = 31 * result + getStringPath().hashCode();%0A        return result
 ;%0A    %7D%0A
@@ -3549,16 +3549,29 @@
 terator(
+%0A            
 getSourc
@@ -3900,16 +3900,29 @@
 ception(
+%0A            
 %22makeDir
@@ -4091,16 +4091,29 @@
 ception(
+%0A            
 %22deleteP
@@ -4284,16 +4284,29 @@
 ception(
+%0A            
 %22pathExi
@@ -4479,16 +4479,29 @@
 ception(
+%0A            
 %22getPath

Fix bug causing Cascading to think all Cassandra taps are identical.