cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From brandonwilli...@apache.org
Subject [3/3] git commit: Merge branch 'cassandra-1.2' into cassandra-2.0
Date Fri, 11 Oct 2013 20:33:20 GMT
Merge branch 'cassandra-1.2' into cassandra-2.0


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/a3ad2e82
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/a3ad2e82
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/a3ad2e82

Branch: refs/heads/cassandra-2.0
Commit: a3ad2e82249b88d4a05f24140948cdbc809d14f3
Parents: 8a506e6 639c01a
Author: Brandon Williams <brandonwilliams@apache.org>
Authored: Fri Oct 11 15:30:27 2013 -0500
Committer: Brandon Williams <brandonwilliams@apache.org>
Committed: Fri Oct 11 15:30:27 2013 -0500

----------------------------------------------------------------------
 .../hadoop/pig/AbstractCassandraStorage.java    | 97 ++++++++++++++------
 .../cassandra/hadoop/pig/CassandraStorage.java  | 55 +++++++----
 .../apache/cassandra/hadoop/pig/CqlStorage.java |  8 +-
 3 files changed, 109 insertions(+), 51 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/a3ad2e82/src/java/org/apache/cassandra/hadoop/pig/AbstractCassandraStorage.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/hadoop/pig/AbstractCassandraStorage.java
index c881734,dbebfb5..486c781
--- a/src/java/org/apache/cassandra/hadoop/pig/AbstractCassandraStorage.java
+++ b/src/java/org/apache/cassandra/hadoop/pig/AbstractCassandraStorage.java
@@@ -127,14 -128,36 +128,37 @@@ public abstract class AbstractCassandra
              setTupleValue(pair, 0, cassandraToObj(comparator, col.name()));
  
          // value
 -        if (col instanceof Column)
 +        Map<ByteBuffer,AbstractType> validators = getValidatorMap(cfDef);
 +        if (validators.get(col.name()) == null)
          {
-             Map<MarshallerType, AbstractType> marshallers = getDefaultMarshallers(cfDef);
-             setTupleValue(pair, 1, cassandraToObj(marshallers.get(MarshallerType.DEFAULT_VALIDATOR),
col.value()));
+             // standard
+             Map<ByteBuffer,AbstractType> validators = getValidatorMap(cfDef);
+             ByteBuffer colName;
+             if (cfInfo.cql3Table && !cfInfo.compactCqlTable)
+             {
+                 ByteBuffer[] names = ((AbstractCompositeType) parseType(cfDef.comparator_type)).split(col.name());
+                 colName = names[names.length-1];
+             }
+             else
+                 colName = col.name();
+             if (validators.get(colName) == null)
+             {
+                 Map<MarshallerType, AbstractType> marshallers = getDefaultMarshallers(cfDef);
+                 setTupleValue(pair, 1, cassandraToObj(marshallers.get(MarshallerType.DEFAULT_VALIDATOR),
col.value()));
+             }
+             else
+                 setTupleValue(pair, 1, cassandraToObj(validators.get(colName), col.value()));
+             return pair;
          }
          else
-             setTupleValue(pair, 1, cassandraToObj(validators.get(col.name()), col.value()));
+         {
+             // super
+             ArrayList<Tuple> subcols = new ArrayList<Tuple>();
+             for (IColumn subcol : col.getSubColumns())
+                 subcols.add(columnToTuple(subcol, cfInfo, parseType(cfDef.getSubcomparator_type())));
+ 
+             pair.set(1, new DefaultDataBag(subcols));
+         }
          return pair;
      }
  

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a3ad2e82/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
index 4083236,a7cc1ad..d9c55a1
--- a/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
+++ b/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
@@@ -124,9 -123,9 +125,9 @@@ public class CassandraStorage extends A
                              key = (ByteBuffer)reader.getCurrentKey();
                              tuple = keyToTuple(key, cfDef, parseType(cfDef.getKey_validation_class()));
                          }
 -                        for (Map.Entry<ByteBuffer, IColumn> entry : lastRow.entrySet())
 +                        for (Map.Entry<ByteBuffer, Column> entry : lastRow.entrySet())
                          {
-                             bag.add(columnToTuple(entry.getValue(), cfDef, parseType(cfDef.getComparator_type())));
+                             bag.add(columnToTuple(entry.getValue(), cfInfo, parseType(cfDef.getComparator_type())));
                          }
                          lastKey = null;
                          lastRow = null;
@@@ -162,9 -161,9 +163,9 @@@
                              tuple = keyToTuple(lastKey, cfDef, parseType(cfDef.getKey_validation_class()));
                          else
                              addKeyToTuple(tuple, lastKey, cfDef, parseType(cfDef.getKey_validation_class()));
 -                        for (Map.Entry<ByteBuffer, IColumn> entry : lastRow.entrySet())
 +                        for (Map.Entry<ByteBuffer, Column> entry : lastRow.entrySet())
                          {
-                             bag.add(columnToTuple(entry.getValue(), cfDef, parseType(cfDef.getComparator_type())));
+                             bag.add(columnToTuple(entry.getValue(), cfInfo, parseType(cfDef.getComparator_type())));
                          }
                          tuple.append(bag);
                          lastKey = key;
@@@ -176,19 -175,19 +177,19 @@@
                      else
                          addKeyToTuple(tuple, lastKey, cfDef, parseType(cfDef.getKey_validation_class()));
                  }
 -                SortedMap<ByteBuffer,IColumn> row = (SortedMap<ByteBuffer,IColumn>)reader.getCurrentValue();
 +                SortedMap<ByteBuffer, Column> row = (SortedMap<ByteBuffer, Column>)reader.getCurrentValue();
                  if (lastRow != null) // prepend what was read last time
                  {
 -                    for (Map.Entry<ByteBuffer, IColumn> entry : lastRow.entrySet())
 +                    for (Map.Entry<ByteBuffer, Column> entry : lastRow.entrySet())
                      {
-                         bag.add(columnToTuple(entry.getValue(), cfDef, parseType(cfDef.getComparator_type())));
+                         bag.add(columnToTuple(entry.getValue(), cfInfo, parseType(cfDef.getComparator_type())));
                      }
                      lastKey = null;
                      lastRow = null;
                  }
 -                for (Map.Entry<ByteBuffer, IColumn> entry : row.entrySet())
 +                for (Map.Entry<ByteBuffer, Column> entry : row.entrySet())
                  {
-                     bag.add(columnToTuple(entry.getValue(), cfDef, parseType(cfDef.getComparator_type())));
+                     bag.add(columnToTuple(entry.getValue(), cfInfo, parseType(cfDef.getComparator_type())));
                  }
              }
          }
@@@ -210,9 -209,10 +211,10 @@@
              if (!reader.nextKeyValue())
                  return null;
  
-             CfDef cfDef = getCfDef(loadSignature);
+             CfInfo cfInfo = getCfInfo(loadSignature);
+             CfDef cfDef = cfInfo.cfDef;
              ByteBuffer key = reader.getCurrentKey();
 -            Map<ByteBuffer, IColumn> cf = reader.getCurrentValue();
 +            Map<ByteBuffer, Column> cf = reader.getCurrentValue();
              assert key != null && cf != null;
  
              // output tuple, will hold the key, each indexed column in a tuple, then a bag
of the rest
@@@ -237,10 -247,10 +249,10 @@@
                  added.put(cdef.name, true);
              }
              // now add all the other columns
 -            for (Map.Entry<ByteBuffer, IColumn> entry : cf.entrySet())
 +            for (Map.Entry<ByteBuffer, Column> entry : cf.entrySet())
              {
                  if (!added.containsKey(entry.getKey()))
-                     bag.add(columnToTuple(entry.getValue(), cfDef, parseType(cfDef.getComparator_type())));
+                     bag.add(columnToTuple(entry.getValue(), cfInfo, parseType(cfDef.getComparator_type())));
              }
              tuple.append(bag);
              // finally, special top-level indexes if needed

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a3ad2e82/src/java/org/apache/cassandra/hadoop/pig/CqlStorage.java
----------------------------------------------------------------------


Mime
View raw message