Mailing-List: contact commits-help@cassandra.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@cassandra.apache.org
Content-Type: text/plain; charset="us-ascii"
MIME-Version: 1.0
Content-Transfer-Encoding: 8bit
From: tylerhobbs@apache.org
To: commits@cassandra.apache.org
Date: Wed, 01 Apr 2015 17:42:46 -0000
Message-Id: <c2bb2e3dd00b4b7185b663a87d6192e5@git.apache.org>
In-Reply-To: <c11ffa503bbb44dba71f3fe845c0e733@git.apache.org>
References: <c11ffa503bbb44dba71f3fe845c0e733@git.apache.org>
Subject: [3/3] cassandra git commit: Add SELECT/INSERT JSON support, toJson(),
 fromJson()

Add SELECT/INSERT JSON support, toJson(), fromJson()

Patch by Tyler Hobbs; reviewed by Sylvain Lebresne and Robert Stupp for
CASSANDRA-7970


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

Branch: refs/heads/trunk
Commit: c7b02d1a6d11dc4436340a5cbeb0f377e8ac8605
Parents: 61e063b
Author: Tyler Hobbs <tyler@datastax.com>
Authored: Wed Apr 1 12:42:01 2015 -0500
Committer: Tyler Hobbs <tyler@datastax.com>
Committed: Wed Apr 1 12:42:01 2015 -0500

----------------------------------------------------------------------
 CHANGES.txt                                     |   2 +
 doc/cql3/CQL.textile                            |  77 +-
 pylib/cqlshlib/cql3handling.py                  |   9 +-
 .../apache/cassandra/cql3/AbstractMarker.java   |   2 +-
 .../apache/cassandra/cql3/ColumnCondition.java  |  31 +-
 .../org/apache/cassandra/cql3/Constants.java    |  34 +-
 src/java/org/apache/cassandra/cql3/Cql.g        |  64 +-
 src/java/org/apache/cassandra/cql3/Json.java    | 329 +++++++
 src/java/org/apache/cassandra/cql3/Lists.java   |  28 +-
 src/java/org/apache/cassandra/cql3/Maps.java    |  28 +-
 .../org/apache/cassandra/cql3/QueryOptions.java |   5 +
 .../org/apache/cassandra/cql3/ResultSet.java    |   1 -
 src/java/org/apache/cassandra/cql3/Sets.java    |  24 +-
 src/java/org/apache/cassandra/cql3/Term.java    |  13 +-
 src/java/org/apache/cassandra/cql3/Tuples.java  |   4 +-
 .../apache/cassandra/cql3/UntypedResultSet.java |   4 +-
 .../cassandra/cql3/functions/FromJsonFct.java   |  78 ++
 .../cassandra/cql3/functions/FunctionCall.java  |  28 +-
 .../cassandra/cql3/functions/FunctionName.java  |  11 +-
 .../cassandra/cql3/functions/Functions.java     |  33 +-
 .../cassandra/cql3/functions/ToJsonFct.java     |  67 ++
 .../cassandra/cql3/selection/Selectable.java    |  11 +-
 .../cassandra/cql3/selection/Selection.java     | 113 ++-
 .../cql3/selection/SelectorFactories.java       |  17 +
 .../cql3/statements/ModificationStatement.java  |   2 +-
 .../cql3/statements/SelectStatement.java        |  18 +-
 .../cql3/statements/UpdateStatement.java        |  76 +-
 .../db/marshal/AbstractCompositeType.java       |  13 +
 .../cassandra/db/marshal/AbstractType.java      |  13 +
 .../apache/cassandra/db/marshal/AsciiType.java  |  31 +
 .../cassandra/db/marshal/BooleanType.java       |  22 +-
 .../apache/cassandra/db/marshal/BytesType.java  |  25 +
 .../db/marshal/ColumnToCollectionType.java      |  13 +
 .../cassandra/db/marshal/CounterColumnType.java |  13 +
 .../apache/cassandra/db/marshal/DateType.java   |  26 +
 .../cassandra/db/marshal/DecimalType.java       |  21 +
 .../apache/cassandra/db/marshal/DoubleType.java |  27 +-
 .../db/marshal/DynamicCompositeType.java        |  13 +
 .../apache/cassandra/db/marshal/EmptyType.java  |  13 +
 .../apache/cassandra/db/marshal/FloatType.java  |  27 +-
 .../apache/cassandra/db/marshal/FrozenType.java |  11 +
 .../cassandra/db/marshal/InetAddressType.java   |  24 +-
 .../apache/cassandra/db/marshal/Int32Type.java  |  31 +-
 .../cassandra/db/marshal/IntegerType.java       |  22 +
 .../cassandra/db/marshal/LexicalUUIDType.java   |  16 +
 .../apache/cassandra/db/marshal/ListType.java   |  43 +
 .../db/marshal/LocalByPartionerType.java        |  13 +
 .../apache/cassandra/db/marshal/LongType.java   |  31 +-
 .../apache/cassandra/db/marshal/MapType.java    |  42 +
 .../cassandra/db/marshal/ReversedType.java      |  14 +
 .../apache/cassandra/db/marshal/SetType.java    |  29 +
 .../cassandra/db/marshal/SimpleDateType.java    |  23 +
 .../apache/cassandra/db/marshal/TimeType.java   |  22 +
 .../cassandra/db/marshal/TimeUUIDType.java      |  15 +
 .../cassandra/db/marshal/TimestampType.java     |  26 +
 .../apache/cassandra/db/marshal/TupleType.java  |  56 ++
 .../apache/cassandra/db/marshal/UTF8Type.java   |  35 +
 .../apache/cassandra/db/marshal/UUIDType.java   |  21 +-
 .../apache/cassandra/db/marshal/UserType.java   |  93 +-
 .../hadoop/pig/AbstractCassandraStorage.java    |   5 +-
 .../serializers/CollectionSerializer.java       |   4 +-
 .../cassandra/serializers/ListSerializer.java   |  14 +
 .../cassandra/serializers/MapSerializer.java    |  15 +
 .../cassandra/serializers/SetSerializer.java    |  13 +
 .../serializers/TimestampSerializer.java        |  10 +-
 .../org/apache/cassandra/cql3/CQLTester.java    |  22 +-
 .../org/apache/cassandra/cql3/JsonTest.java     | 860 +++++++++++++++++++
 67 files changed, 2634 insertions(+), 242 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b02d1a/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 9178fb3..3f7dc9e 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,6 @@
 3.0
+ * Add support for SELECT JSON, INSERT JSON syntax and new toJson(), fromJson()
+   functions (CASSANDRA-7970)
  * Optimise max purgeable timestamp calculation in compaction (CASSANDRA-8920)
  * Constrain internode message buffer sizes, and improve IO class hierarchy (CASSANDRA-8670) 
  * New tool added to validate all sstables in a node (CASSANDRA-5791)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b02d1a/doc/cql3/CQL.textile
----------------------------------------------------------------------
diff --git a/doc/cql3/CQL.textile b/doc/cql3/CQL.textile
index 7ebda61..8a227fb 100644
--- a/doc/cql3/CQL.textile
+++ b/doc/cql3/CQL.textile
@@ -120,7 +120,7 @@ p. A @<variable>@ can be either anonymous (a question mark (@?@)) or named (an i
 
 p. The @<properties>@ production is use by statement that create and alter keyspaces and tables. Each @<property>@ is either a _simple_ one, in which case it just has a value, or a _map_ one, in which case it's value is a map grouping sub-options. The following will refer to one or the other as the _kind_ (_simple_ or _map_) of the property.
 
-p. A @<tablename>@ will be used to identify a table. This is an identifier representing the table name that can be preceded by a keyspace name. The keyspace name, if provided, allow to identify a table in another keyspace than the currently active one (the currently active keyspace is set through the <a href="#useStmt"><tt>USE</tt></a> statement).
+p. A @<tablename>@ will be used to identify a table. This is an identifier representing the table name that can be preceded by a keyspace name. The keyspace name, if provided, allow to identify a table in another keyspace than the currently active one (the currently active keyspace is set through the <a href=":#useStmt"><tt>USE</tt></a> statement).
 
 p. For supported @<function>@, see the section on "functions":#functions.
 
@@ -743,11 +743,15 @@ __Syntax:__
 
 bc(syntax).. 
 <insertStatement> ::= INSERT INTO <tablename>
-                             '(' <identifier> ( ',' <identifier> )* ')'
-                      VALUES '(' <term-or-literal> ( ',' <term-or-literal> )* ')'
+                      ( ( <name-list> VALUES <value-list> )
+                      | ( JSON <string> ))
                       ( IF NOT EXISTS )?
                       ( USING <option> ( AND <option> )* )?
 
+<names-list> ::= '(' <identifier> ( ',' <identifier> )* ')'
+
+<value-list> ::= '(' <term-or-literal> ( ',' <term-or-literal> )* ')'
+
 <term-or-literal> ::= <term>
                     | <collection-literal>
 
@@ -756,12 +760,14 @@ bc(syntax)..
 p. 
 __Sample:__
 
-bc(sample). 
+bc(sample).. 
 INSERT INTO NerdMovies (movie, director, main_actor, year)
                 VALUES ('Serenity', 'Joss Whedon', 'Nathan Fillion', 2005)
 USING TTL 86400;
 
-The @INSERT@ statement writes one or more columns for a given row in a table. Note that since a row is identified by its @PRIMARY KEY@, at least the columns composing it must be specified.
+INSERT INTO NerdMovies JSON '{"movie": "Serenity", "director": "Joss Whedon", "year": 2005}'
+p. 
+The @INSERT@ statement writes one or more columns for a given row in a table. Note that since a row is identified by its @PRIMARY KEY@, at least the columns composing it must be specified.  The list of columns to insert to must be supplied when using the @VALUES@ syntax.  When using the @JSON@ syntax, they are optional.  See the section on "@INSERT JSON@":#insertJson for more details.
 
 Note that unlike in SQL, @INSERT@ does not check the prior existence of the row by default: the row is created if none existed before, and updated otherwise. Furthermore, there is no mean to know which of creation or update happened.
 
@@ -926,7 +932,7 @@ h3(#selectStmt). SELECT
 __Syntax:__
 
 bc(syntax).. 
-<select-stmt> ::= SELECT <select-clause>
+<select-stmt> ::= SELECT ( JSON )? <select-clause>
                   FROM <tablename>
                   ( WHERE <where-clause> )?
                   ( ORDER BY <order-by> )?
@@ -962,6 +968,8 @@ __Sample:__
 bc(sample).. 
 SELECT name, occupation FROM users WHERE userid IN (199, 200, 207);
 
+SELECT JSON name, occupation FROM users WHERE userid = 199;
+
 SELECT name AS user_name, occupation AS user_occupation FROM users;
 
 SELECT time, value
@@ -975,7 +983,7 @@ SELECT COUNT(*) FROM users;
 SELECT COUNT(*) AS user_count FROM users;
 
 p. 
-The @SELECT@ statements reads one or more columns for one or more rows in a table. It returns a result-set of rows, where each row contains the collection of columns corresponding to the query.
+The @SELECT@ statements reads one or more columns for one or more rows in a table. It returns a result-set of rows, where each row contains the collection of columns corresponding to the query.  If the @JSON@ keyword is used, the results for each row will contain only a single column named "json".  See the section on "@SELECT JSON@":#selectJson for more details.
 
 h4(#selectSelection). @<select-clause>@
 
@@ -1481,6 +1489,58 @@ p.
 
 See "@CREATE AGGREGATE@":#createAggregateStmt and "@DROP AGGREGATE@":#dropAggregateStmt.
 
+h2(#json). JSON Support
+
+Cassandra 3.0 introduces JSON support to "@SELECT@":#selectStmt and "@INSERT@":#insertStmt statements.  This support does not fundamentally alter the CQL API (for example, the schema is still enforced), it simply provides a convenient way to work with JSON documents.
+
+h3(#selectJson). SELECT JSON
+
+With @SELECT@ statements, the new @JSON@ keyword can be used to return each row as a single @JSON@ encoded map.  The remainder of the @SELECT@ statment behavior is the same.
+
+The result map keys are the same as the column names in a normal result set.  For example, a statement like "@SELECT JSON a, ttl(b) FROM ...@" would result in a map with keys @"a"@ and @"ttl(b)"@.  However, this is one notable exception: for symmetry with @INSERT JSON@ behavior, case-sensitive column names with upper-case letters will be surrounded with double quotes.  For example, "@SELECT JSON myColumn FROM ...@" would result in a map key @"\"myColumn\""@ (note the escaped quotes).
+
+The map values will @JSON@-encoded representations (as described below) of the result set values.
+
+h3(#insertJson). INSERT JSON
+
+With @INSERT@ statements, the new @JSON@ keyword can be used to enable inserting a @JSON@ encoded map as a single row.  The format of the @JSON@ map should generally match that returned by a @SELECT JSON@ statement on the same table.  In particular, case-sensitive column names should be surrounded with double quotes.  For example, to insert into a table with two columns named "myKey" and "value", you would do the following:
+
+bc(sample). 
+INSERT INTO mytable JSON '{"\"myKey\"": 0, "value": 0}'
+
+Any columns which are ommitted from the @JSON@ map will be defaulted to a @NULL@ value (which will result in a tombstone being created).
+
+h3(#jsonEncoding). JSON Encoding of Cassandra Data Types
+
+Where possible, Cassandra will represent and accept data types in their native @JSON@ representation.  Cassandra will also accept string representations matching the CQL literal format for all data types.  The following table describes the encodings that Cassandra will accept in @INSERT JSON@ values (and @fromJson()@ arguments) as well as the format Cassandra will use when returning data for @SELECT JSON@ statements (and @fromJson()@):
+
+|_. type    |_. formats accepted   |_. return format |_. notes|
+|@ascii@    |string                |string           |Uses JSON's @\u@ character escape|
+|@bigint@   |integer, string       |integer          |String must be valid 64 bit integer|
+|@blob@     |string                |string           |String should be 0x followed by an even number of hex digits|
+|@boolean@  |boolean, string       |boolean          |String must be "true" or "false"|
+|@date@     |string                |string           |Date in format @YYYY-MM-DD@, timezone UTC|
+|@decimal@  |integer, float, string|float            |May exceed 32 or 64-bit IEEE-754 floating point precision in client-side decoder|
+|@double@   |integer, float, string|float            |String must be valid integer or float|
+|@float@    |integer, float, string|float            |String must be valid integer or float|
+|@inet@     |string                |string           |IPv4 or IPv6 address|
+|@int@      |integer, string       |integer          |String must be valid 32 bit integer|
+|@text@     |string                |string           |Uses JSON's @\u@ character escape|
+|@time@     |string                |string           |Time of day in format @HH-MM-SS[.fffffffff]@|
+|@timestamp@|integer, string       |string           |A timestamp. Strings constant are allow to input timestamps as dates, see "Working with dates":#usingdates below for more information.  Datestamps with format @YYYY-MM-DD HH:MM:SS.SSS@ are returned.|
+|@timeuuid@ |string                |string           |Type 1 UUID. See "Constants":#constants for the UUID format|
+|@uuid@     |string                |string           |See "Constants":#constants for the UUID format|
+|@varchar@  |string                |string           |Uses JSON's @\u@ character escape|
+|@varint@   |integer, string       |integer          |Variable length; may overflow 32 or 64 bit integers in client-side decoder|
+
+h3(#fromJson). The fromJson() Function
+
+The @fromJson()@ function may be used similarly to @INSERT JSON@, but for a single column value.  It may only be used in the @VALUES@ clause of an @INSERT@ statement or as one of the column values in an @UPDATE@, @DELETE@, or @SELECT@ statement.  For example, it cannot be used in the selection clause of a @SELECT@ statement.
+
+h3(#toJson). The toJson() Function
+
+The @toJson()@ function may be used similarly to @SELECT JSON@, but for a single column value.  It may only be used in the selection clause of a @SELECT@ statement.
+
 h2(#appendixA). Appendix A: CQL Keywords
 
 CQL distinguishes between _reserved_ and _non-reserved_ keywords. Reserved keywords cannot be used as identifier, they are truly reserved for the language (but one can enclose a reserved keyword by double-quotes to use it as an identifier). Non-reserved keywords however only have a specific meaning in certain context but can used as identifer otherwise. The only _raison d'être_ of these non-reserved keywords is convenience: some keyword are non-reserved when it was always easy for the parser to decide whether they were used as keywords or not.
@@ -1654,7 +1714,7 @@ h3. 3.1.1
 h3. 3.1.0
 
 * "ALTER TABLE":#alterTableStmt @DROP@ option has been reenabled for CQL3 tables and has new semantics now: the space formerly used by dropped columns will now be eventually reclaimed (post-compaction). You should not readd previously dropped columns unless you use timestamps with microsecond precision (see "CASSANDRA-3919":https://issues.apache.org/jira/browse/CASSANDRA-3919 for more details).
-* @SELECT@ statement now supports aliases in select clause. Aliases in WHERE and ORDER BY clauses are not supported. See the "section on select"#selectStmt for details.
+* @SELECT@ statement now supports aliases in select clause. Aliases in WHERE and ORDER BY clauses are not supported. See the "section on select":#selectStmt for details.
 * @CREATE@ statements for @KEYSPACE@, @TABLE@ and @INDEX@ now supports an @IF NOT EXISTS@ condition. Similarly, @DROP@ statements support a @IF EXISTS@ condition.
 * @INSERT@ statements optionally supports a @IF NOT EXISTS@ condition and @UPDATE@ supports @IF@ conditions.
 
@@ -1682,7 +1742,6 @@ h3. 3.0.1
 * "Date strings":#usingtimestamps (and timestamps) are no longer accepted as valid @timeuuid@ values. Doing so was a bug in the sense that date string are not valid @timeuuid@, and it was thus resulting in "confusing behaviors":https://issues.apache.org/jira/browse/CASSANDRA-4936.  However, the following new methods have been added to help working with @timeuuid@: @now@, @minTimeuuid@, @maxTimeuuid@ , @dateOf@ and @unixTimestampOf@. See the "section dedicated to these methods":#usingtimeuuid for more detail.
 * "Float constants"#constants now support the exponent notation. In other words, @4.2E10@ is now a valid floating point value.
 
-
 h2. Versioning
 
 Versioning of the CQL language adheres to the "Semantic Versioning":http://semver.org guidelines. Versions take the form X.Y.Z where X, Y, and Z are integer values representing major, minor, and patch level respectively. There is no correlation between Cassandra release versions and the CQL language version.

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b02d1a/pylib/cqlshlib/cql3handling.py
----------------------------------------------------------------------
diff --git a/pylib/cqlshlib/cql3handling.py b/pylib/cqlshlib/cql3handling.py
index 592daad..ef18c2a 100644
--- a/pylib/cqlshlib/cql3handling.py
+++ b/pylib/cqlshlib/cql3handling.py
@@ -612,7 +612,7 @@ def working_on_keyspace(ctxt):
 syntax_rules += r'''
 <useStatement> ::= "USE" <keyspaceName>
                  ;
-<selectStatement> ::= "SELECT" <selectClause>
+<selectStatement> ::= "SELECT" ( "JSON" )? <selectClause>
                         "FROM" cf=<columnFamilyName>
                           ( "WHERE" <whereClause> )?
                           ( "ORDER" "BY" <orderByClause> ( "," <orderByClause> )* )?
@@ -692,10 +692,9 @@ explain_completion('selector', 'colname')
 
 syntax_rules += r'''
 <insertStatement> ::= "INSERT" "INTO" cf=<columnFamilyName>
-                               "(" [colname]=<cident> "," [colname]=<cident>
-                                   ( "," [colname]=<cident> )* ")"
-                      "VALUES" "(" [newval]=<term> valcomma="," [newval]=<term>
-                                   ( valcomma="," [newval]=<term> )* valcomma=")"
+                      ( ( "(" [colname]=<cident> ( "," [colname]=<cident> )* ")"
+                          "VALUES" "(" [newval]=<term> ( valcomma="," [newval]=<term> )* valcomma=")")
+                        | ("JSON" <stringLiteral>))
                       ( "IF" "NOT" "EXISTS")?
                       ( "USING" [insertopt]=<usingOption>
                                 ( "AND" [insertopt]=<usingOption> )* )?

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b02d1a/src/java/org/apache/cassandra/cql3/AbstractMarker.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/AbstractMarker.java b/src/java/org/apache/cassandra/cql3/AbstractMarker.java
index 990bf68..87344b6 100644
--- a/src/java/org/apache/cassandra/cql3/AbstractMarker.java
+++ b/src/java/org/apache/cassandra/cql3/AbstractMarker.java
@@ -57,7 +57,7 @@ public abstract class AbstractMarker extends Term.NonTerminal
             this.bindIndex = bindIndex;
         }
 
-        public AbstractMarker prepare(String keyspace, ColumnSpecification receiver) throws InvalidRequestException
+        public NonTerminal prepare(String keyspace, ColumnSpecification receiver) throws InvalidRequestException
         {
             if (!(receiver.type instanceof CollectionType))
                 return new Constants.Marker(bindIndex, receiver);

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b02d1a/src/java/org/apache/cassandra/cql3/ColumnCondition.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/ColumnCondition.java b/src/java/org/apache/cassandra/cql3/ColumnCondition.java
index a8c8cc2..21c1f79 100644
--- a/src/java/org/apache/cassandra/cql3/ColumnCondition.java
+++ b/src/java/org/apache/cassandra/cql3/ColumnCondition.java
@@ -32,6 +32,7 @@ import org.apache.cassandra.db.composites.Composite;
 import org.apache.cassandra.db.filter.ColumnSlice;
 import org.apache.cassandra.db.marshal.*;
 import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.serializers.CollectionSerializer;
 import org.apache.cassandra.transport.Server;
 import org.apache.cassandra.utils.ByteBufferUtil;
 
@@ -463,6 +464,7 @@ public class ColumnCondition
     static class CollectionBound extends Bound
     {
         private final Term.Terminal value;
+        private final QueryOptions options;
 
         private CollectionBound(ColumnCondition condition, QueryOptions options) throws InvalidRequestException
         {
@@ -470,6 +472,7 @@ public class ColumnCondition
             assert column.type.isCollection() && condition.collectionElement == null;
             assert condition.operator != Operator.IN;
             this.value = condition.value.bind(options);
+            this.options = options;
         }
 
         public boolean appliesTo(Composite rowPrefix, ColumnFamily current, final long now) throws InvalidRequestException
@@ -489,7 +492,7 @@ public class ColumnCondition
                         throw new InvalidRequestException(String.format("Invalid comparison with null for operator \"%s\"", operator));
                 }
 
-                return valueAppliesTo(type, iter, value, operator);
+                return valueAppliesTo(type, iter, value, operator, options);
             }
 
             // frozen collections
@@ -507,25 +510,31 @@ public class ColumnCondition
             // make sure we use v3 serialization format for comparison
             ByteBuffer conditionValue;
             if (type.kind == CollectionType.Kind.LIST)
-                conditionValue = ((Lists.Value) value).getWithProtocolVersion(Server.VERSION_3);
+                conditionValue = ((Lists.Value) value).get(Server.VERSION_3);
             else if (type.kind == CollectionType.Kind.SET)
-                conditionValue = ((Sets.Value) value).getWithProtocolVersion(Server.VERSION_3);
+                conditionValue = ((Sets.Value) value).get(Server.VERSION_3);
             else
-                conditionValue = ((Maps.Value) value).getWithProtocolVersion(Server.VERSION_3);
+                conditionValue = ((Maps.Value) value).get(Server.VERSION_3);
 
             return compareWithOperator(operator, type, conditionValue, cell.value());
         }
 
-        static boolean valueAppliesTo(CollectionType type, Iterator<Cell> iter, Term.Terminal value, Operator operator)
+        static boolean valueAppliesTo(CollectionType type, Iterator<Cell> iter, Term.Terminal value, Operator operator, QueryOptions options)
         {
             if (value == null)
                 return !iter.hasNext();
 
             switch (type.kind)
             {
-                case LIST: return listAppliesTo((ListType)type, iter, ((Lists.Value)value).elements, operator);
-                case SET: return setAppliesTo((SetType)type, iter, ((Sets.Value)value).elements, operator);
-                case MAP: return mapAppliesTo((MapType)type, iter, ((Maps.Value)value).map, operator);
+                case LIST:
+                    List<ByteBuffer> valueList = ((Lists.Value) value).elements;
+                    return listAppliesTo((ListType)type, iter, valueList, operator);
+                case SET:
+                    Set<ByteBuffer> valueSet = ((Sets.Value) value).elements;
+                    return setAppliesTo((SetType)type, iter, valueSet, operator);
+                case MAP:
+                    Map<ByteBuffer, ByteBuffer> valueMap = ((Maps.Value) value).map;
+                    return mapAppliesTo((MapType)type, iter, valueMap, operator);
             }
             throw new AssertionError();
         }
@@ -617,12 +626,14 @@ public class ColumnCondition
     public static class CollectionInBound extends Bound
     {
         private final List<Term.Terminal> inValues;
+        private final QueryOptions options;
 
         private CollectionInBound(ColumnCondition condition, QueryOptions options) throws InvalidRequestException
         {
             super(condition.column, condition.operator);
             assert column.type instanceof CollectionType && condition.collectionElement == null;
             assert condition.operator == Operator.IN;
+            this.options = options;
             inValues = new ArrayList<>();
             if (condition.inValues == null)
             {
@@ -680,7 +691,7 @@ public class ColumnCondition
                 List<Cell> cells = newArrayList(collectionColumns(name, current, now));
                 for (Term.Terminal value : inValues)
                 {
-                    if (CollectionBound.valueAppliesTo(type, cells.iterator(), value, Operator.EQ))
+                    if (CollectionBound.valueAppliesTo(type, cells.iterator(), value, Operator.EQ, options))
                         return true;
                 }
                 return false;
@@ -695,7 +706,7 @@ public class ColumnCondition
                         if (cell == null || !cell.isLive(now))
                             return true;
                     }
-                    else if (type.compare(((Term.CollectionTerminal)value).getWithProtocolVersion(Server.VERSION_3), cell.value()) == 0)
+                    else if (type.compare(value.get(Server.VERSION_3), cell.value()) == 0)
                     {
                         return true;
                     }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b02d1a/src/java/org/apache/cassandra/cql3/Constants.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/Constants.java b/src/java/org/apache/cassandra/cql3/Constants.java
index 69693bf..5f48160 100644
--- a/src/java/org/apache/cassandra/cql3/Constants.java
+++ b/src/java/org/apache/cassandra/cql3/Constants.java
@@ -49,22 +49,6 @@ public abstract class Constants
 
     public static final Term.Raw NULL_LITERAL = new Term.Raw()
     {
-        private final Term.Terminal NULL_VALUE = new Value(null)
-        {
-            @Override
-            public Terminal bind(QueryOptions options)
-            {
-                // We return null because that makes life easier for collections
-                return null;
-            }
-
-            @Override
-            public String toString()
-            {
-                return "null";
-            }
-        };
-
         public Term prepare(String keyspace, ColumnSpecification receiver) throws InvalidRequestException
         {
             if (!testAssignment(keyspace, receiver).isAssignable())
@@ -87,6 +71,22 @@ public abstract class Constants
         }
     };
 
+    public static final Term.Terminal NULL_VALUE = new Value(null)
+    {
+        @Override
+        public Terminal bind(QueryOptions options)
+        {
+            // We return null because that makes life easier for collections
+            return null;
+        }
+
+        @Override
+        public String toString()
+        {
+            return "null";
+        }
+    };
+
     public static class Literal implements Term.Raw
     {
         private final Type type;
@@ -256,7 +256,7 @@ public abstract class Constants
             this.bytes = bytes;
         }
 
-        public ByteBuffer get(QueryOptions options)
+        public ByteBuffer get(int protocolVersion)
         {
             return bytes;
         }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b02d1a/src/java/org/apache/cassandra/cql3/Cql.g
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/Cql.g b/src/java/org/apache/cassandra/cql3/Cql.g
index 405e265..e5b0efd 100644
--- a/src/java/org/apache/cassandra/cql3/Cql.g
+++ b/src/java/org/apache/cassandra/cql3/Cql.g
@@ -93,6 +93,13 @@ options {
         return marker;
     }
 
+    public Json.Marker newJsonBindVariables(ColumnIdentifier name)
+    {
+        Json.Marker marker = new Json.Marker(bindVariables.size());
+        bindVariables.add(name);
+        return marker;
+    }
+
     public void addErrorListener(ErrorListener listener)
     {
         this.listeners.add(listener);
@@ -283,9 +290,12 @@ selectStatement returns [SelectStatement.RawStatement expr]
         Term.Raw limit = null;
         Map<ColumnIdentifier.Raw, Boolean> orderings = new LinkedHashMap<ColumnIdentifier.Raw, Boolean>();
         boolean allowFiltering = false;
+        boolean isJson = false;
     }
-    : K_SELECT ( ( K_DISTINCT { isDistinct = true; } )? sclause=selectClause
-               | sclause=selectCountClause )
+    : K_SELECT 
+      ( K_JSON { isJson = true; } )?
+      ( ( K_DISTINCT { isDistinct = true; } )? sclause=selectClause
+        | sclause=selectCountClause )
       K_FROM cf=columnFamilyName
       ( K_WHERE wclause=whereClause )?
       ( K_ORDER K_BY orderByClause[orderings] ( ',' orderByClause[orderings] )* )?
@@ -294,7 +304,8 @@ selectStatement returns [SelectStatement.RawStatement expr]
       {
           SelectStatement.Parameters params = new SelectStatement.Parameters(orderings,
                                                                              isDistinct,
-                                                                             allowFiltering);
+                                                                             allowFiltering,
+                                                                             isJson);
           $expr = new SelectStatement.RawStatement(cf, params, sclause, wclause, limit);
       }
     ;
@@ -353,29 +364,49 @@ orderByClause[Map<ColumnIdentifier.Raw, Boolean> orderings]
  * USING TIMESTAMP <long>;
  *
  */
-insertStatement returns [UpdateStatement.ParsedInsert expr]
+insertStatement returns [ModificationStatement.Parsed expr]
+    : K_INSERT K_INTO cf=columnFamilyName
+        ( st1=normalInsertStatement[cf] { $expr = st1; }
+        | K_JSON st2=jsonInsertStatement[cf] { $expr = st2; })
+    ;
+
+normalInsertStatement [CFName cf] returns [UpdateStatement.ParsedInsert expr]
     @init {
         Attributes.Raw attrs = new Attributes.Raw();
         List<ColumnIdentifier.Raw> columnNames  = new ArrayList<ColumnIdentifier.Raw>();
         List<Term.Raw> values = new ArrayList<Term.Raw>();
         boolean ifNotExists = false;
     }
-    : K_INSERT K_INTO cf=columnFamilyName
-          '(' c1=cident { columnNames.add(c1); }  ( ',' cn=cident { columnNames.add(cn); } )* ')'
-        K_VALUES
-          '(' v1=term { values.add(v1); } ( ',' vn=term { values.add(vn); } )* ')'
+    : '(' c1=cident { columnNames.add(c1); }  ( ',' cn=cident { columnNames.add(cn); } )* ')'
+      K_VALUES
+      '(' v1=term { values.add(v1); } ( ',' vn=term { values.add(vn); } )* ')'
+      ( K_IF K_NOT K_EXISTS { ifNotExists = true; } )?
+      ( usingClause[attrs] )?
+      {
+          $expr = new UpdateStatement.ParsedInsert(cf, attrs, columnNames, values, ifNotExists);
+      }
+    ;
 
-        ( K_IF K_NOT K_EXISTS { ifNotExists = true; } )?
-        ( usingClause[attrs] )?
+jsonInsertStatement [CFName cf] returns [UpdateStatement.ParsedInsertJson expr]
+    @init {
+        Attributes.Raw attrs = new Attributes.Raw();
+        boolean ifNotExists = false;
+    }
+    : val=jsonValue
+      ( K_IF K_NOT K_EXISTS { ifNotExists = true; } )?
+      ( usingClause[attrs] )?
       {
-          $expr = new UpdateStatement.ParsedInsert(cf,
-                                                   attrs,
-                                                   columnNames,
-                                                   values,
-                                                   ifNotExists);
+          $expr = new UpdateStatement.ParsedInsertJson(cf, attrs, val, ifNotExists);
       }
     ;
 
+jsonValue returns [Json.Raw value]
+    :
+    | s=STRING_LITERAL { $value = new Json.Literal($s.text); }
+    | ':' id=ident     { $value = newJsonBindVariables(id); }
+    | QMARK            { $value = newJsonBindVariables(null); }
+    ;
+
 usingClause[Attributes.Raw attrs]
     : K_USING usingClauseObjective[attrs] ( K_AND usingClauseObjective[attrs] )*
     ;
@@ -1500,6 +1531,7 @@ basic_unreserved_keyword returns [String str]
         | K_LANGUAGE
         | K_NON
         | K_DETERMINISTIC
+        | K_JSON
         ) { $str = $k.text; }
     ;
 
@@ -1630,6 +1662,8 @@ K_OR:          O R;
 K_REPLACE:     R E P L A C E;
 K_DETERMINISTIC: D E T E R M I N I S T I C;
 
+K_JSON:        J S O N;
+
 // Case-insensitive alpha characters
 fragment A: ('a'|'A');
 fragment B: ('b'|'B');

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b02d1a/src/java/org/apache/cassandra/cql3/Json.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/Json.java b/src/java/org/apache/cassandra/cql3/Json.java
new file mode 100644
index 0000000..8929cc0
--- /dev/null
+++ b/src/java/org/apache/cassandra/cql3/Json.java
@@ -0,0 +1,329 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.cql3;
+
+import org.apache.cassandra.config.CFMetaData;
+import org.apache.cassandra.config.ColumnDefinition;
+import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.serializers.MarshalException;
+import org.codehaus.jackson.io.JsonStringEncoder;
+import org.codehaus.jackson.map.ObjectMapper;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.*;
+
+/** Term-related classes for INSERT JSON support. */
+public class Json
+{
+    public static final ObjectMapper JSON_OBJECT_MAPPER = new ObjectMapper();
+
+    public static final JsonStringEncoder JSON_STRING_ENCODER = new JsonStringEncoder();
+
+    public static final ColumnIdentifier JSON_COLUMN_ID = new ColumnIdentifier("[json]", true);
+
+    public interface Raw
+    {
+        public Prepared prepareAndCollectMarkers(CFMetaData metadata, Collection<ColumnDefinition> receivers, VariableSpecifications boundNames);
+    }
+
+    /**
+     * Represents a literal JSON string in an INSERT JSON statement.
+     * For example: INSERT INTO mytable (key, col) JSON '{"key": 0, "col": 0}';
+     */
+    public static class Literal implements Raw
+    {
+        private final String text;
+
+        public Literal(String text)
+        {
+            this.text = text;
+        }
+
+        public Prepared prepareAndCollectMarkers(CFMetaData metadata, Collection<ColumnDefinition> receivers, VariableSpecifications boundNames)
+        {
+            return new PreparedLiteral(metadata.ksName, parseJson(text, receivers));
+        }
+    }
+
+    /**
+     * Represents a marker for a JSON string in an INSERT JSON statement.
+     * For example: INSERT INTO mytable (key, col) JSON ?;
+     */
+    public static class Marker implements Raw
+    {
+        protected final int bindIndex;
+
+        public Marker(int bindIndex)
+        {
+            this.bindIndex = bindIndex;
+        }
+
+        public Prepared prepareAndCollectMarkers(CFMetaData metadata, Collection<ColumnDefinition> receivers, VariableSpecifications boundNames)
+        {
+            boundNames.add(bindIndex, makeReceiver(metadata));
+            return new PreparedMarker(metadata.ksName, bindIndex, receivers);
+        }
+
+        private ColumnSpecification makeReceiver(CFMetaData metadata)
+        {
+            return new ColumnSpecification(metadata.ksName, metadata.cfName, JSON_COLUMN_ID, UTF8Type.instance);
+        }
+    }
+
+    /**
+     * A prepared, full set of JSON values.
+     */
+    public static abstract class Prepared
+    {
+        private final String keyspace;
+
+        protected Prepared(String keyspace)
+        {
+            this.keyspace = keyspace;
+        }
+
+        protected abstract Term.Raw getRawTermForColumn(ColumnDefinition def);
+
+        public Term getPrimaryKeyValueForColumn(ColumnDefinition def)
+        {
+            // Note that we know we don't have to call collectMarkerSpecification since it has already been collected
+            return getRawTermForColumn(def).prepare(keyspace, def);
+        }
+
+        public Operation getSetOperationForColumn(ColumnDefinition def)
+        {
+            // Note that we know we don't have to call collectMarkerSpecification on the operation since we have
+            // already collected all we need.
+            return new Operation.SetValue(getRawTermForColumn(def)).prepare(keyspace, def);
+        }
+    }
+
+    /**
+     * A prepared literal set of JSON values
+     */
+    private static class PreparedLiteral extends Prepared
+    {
+        private final Map<ColumnIdentifier, Term> columnMap;
+
+        public PreparedLiteral(String keyspace, Map<ColumnIdentifier, Term> columnMap)
+        {
+            super(keyspace);
+            this.columnMap = columnMap;
+        }
+
+        protected Term.Raw getRawTermForColumn(ColumnDefinition def)
+        {
+            Term value = columnMap.get(def.name);
+            return value == null ? Constants.NULL_LITERAL : new ColumnValue(value);
+        }
+    }
+
+    /**
+     *  A prepared bind marker for a set of JSON values
+     */
+    private static class PreparedMarker extends Prepared
+    {
+        private final int bindIndex;
+        private final Collection<ColumnDefinition> columns;
+
+        private Map<ColumnIdentifier, Term> columnMap;
+
+        public PreparedMarker(String keyspace, int bindIndex, Collection<ColumnDefinition> columns)
+        {
+            super(keyspace);
+            this.bindIndex = bindIndex;
+            this.columns = columns;
+        }
+
+        protected DelayedColumnValue getRawTermForColumn(ColumnDefinition def)
+        {
+            return new DelayedColumnValue(this, def);
+        }
+
+        public void bind(QueryOptions options) throws InvalidRequestException
+        {
+            // this will be called once per column, so avoid duplicating work
+            if (columnMap != null)
+                return;
+
+            ByteBuffer value = options.getValues().get(bindIndex);
+            if (value == null)
+                throw new InvalidRequestException("Got null for INSERT JSON values");
+
+            columnMap = parseJson(UTF8Type.instance.getSerializer().deserialize(value), columns);
+        }
+
+        public Term getValue(ColumnDefinition def)
+        {
+            return columnMap.get(def.name);
+        }
+    }
+
+    /**
+     * A Terminal for a single column.
+     *
+     * Note that this is intrinsically an already prepared term, but this still implements Term.Raw so that we can
+     * easily use it to create raw operations.
+     */
+    private static class ColumnValue implements Term.Raw
+    {
+        private final Term term;
+
+        public ColumnValue(Term term)
+        {
+            this.term = term;
+        }
+
+        @Override
+        public Term prepare(String keyspace, ColumnSpecification receiver) throws InvalidRequestException
+        {
+            return term;
+        }
+
+        @Override
+        public TestResult testAssignment(String keyspace, ColumnSpecification receiver)
+        {
+            return TestResult.NOT_ASSIGNABLE;
+        }
+    }
+
+    /**
+     * A NonTerminal for a single column.
+     *
+     * As with {@code ColumnValue}, this is intrinsically a prepared term but implements Terms.Raw for convenience.
+     */
+    private static class DelayedColumnValue extends Term.NonTerminal implements Term.Raw
+    {
+        private final PreparedMarker marker;
+        private final ColumnDefinition column;
+
+        public DelayedColumnValue(PreparedMarker prepared, ColumnDefinition column)
+        {
+            this.marker = prepared;
+            this.column = column;
+        }
+
+        @Override
+        public Term prepare(String keyspace, ColumnSpecification receiver) throws InvalidRequestException
+        {
+            return this;
+        }
+
+        @Override
+        public TestResult testAssignment(String keyspace, ColumnSpecification receiver)
+        {
+            return TestResult.WEAKLY_ASSIGNABLE;
+        }
+
+        @Override
+        public void collectMarkerSpecification(VariableSpecifications boundNames)
+        {
+            // We've already collected what we should (and in practice this method is never called).
+        }
+
+        @Override
+        public boolean containsBindMarker()
+        {
+            return true;
+        }
+
+        @Override
+        public Terminal bind(QueryOptions options) throws InvalidRequestException
+        {
+            marker.bind(options);
+            Term term = marker.getValue(column);
+            return term == null ? null : term.bind(options);
+        }
+    }
+
+    /**
+     * Given a JSON string, return a map of columns to their values for the insert.
+     */
+    private static Map<ColumnIdentifier, Term> parseJson(String jsonString, Collection<ColumnDefinition> expectedReceivers)
+    {
+        try
+        {
+            Map<String, Object> valueMap = JSON_OBJECT_MAPPER.readValue(jsonString, Map.class);
+
+            if (valueMap == null)
+                throw new InvalidRequestException("Got null for INSERT JSON values");
+
+            handleCaseSensitivity(valueMap);
+
+            Map<ColumnIdentifier, Term> columnMap = new HashMap<>(expectedReceivers.size());
+            for (ColumnSpecification spec : expectedReceivers)
+            {
+                Object parsedJsonObject = valueMap.remove(spec.name.toString());
+                if (parsedJsonObject == null)
+                {
+                    columnMap.put(spec.name, null);
+                }
+                else
+                {
+                    try
+                    {
+                        columnMap.put(spec.name, spec.type.fromJSONObject(parsedJsonObject));
+                    }
+                    catch(MarshalException exc)
+                    {
+                        throw new InvalidRequestException(String.format("Error decoding JSON value for %s: %s", spec.name, exc.getMessage()));
+                    }
+                }
+            }
+
+            if (!valueMap.isEmpty())
+            {
+                throw new InvalidRequestException(String.format(
+                        "JSON values map contains unrecognized column: %s", valueMap.keySet().iterator().next()));
+            }
+
+            return columnMap;
+        }
+        catch (IOException exc)
+        {
+            throw new InvalidRequestException(String.format("Could not decode JSON string as a map: %s. (String was: %s)", exc.toString(), jsonString));
+        }
+        catch (MarshalException exc)
+        {
+            throw new InvalidRequestException(exc.getMessage());
+        }
+    }
+
+    /**
+     * Handles unquoting and case-insensitivity in map keys.
+     */
+    public static void handleCaseSensitivity(Map<String, Object> valueMap)
+    {
+        for (String mapKey : new ArrayList<>(valueMap.keySet()))
+        {
+            // if it's surrounded by quotes, remove them and preserve the case
+            if (mapKey.startsWith("\"") && mapKey.endsWith("\""))
+            {
+                valueMap.put(mapKey.substring(1, mapKey.length() - 1), valueMap.remove(mapKey));
+                continue;
+            }
+
+            // otherwise, lowercase it if needed
+            String lowered = mapKey.toLowerCase(Locale.US);
+            if (!mapKey.equals(lowered))
+                valueMap.put(lowered, valueMap.remove(mapKey));
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b02d1a/src/java/org/apache/cassandra/cql3/Lists.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/Lists.java b/src/java/org/apache/cassandra/cql3/Lists.java
index f478c15..ae2417e 100644
--- a/src/java/org/apache/cassandra/cql3/Lists.java
+++ b/src/java/org/apache/cassandra/cql3/Lists.java
@@ -27,10 +27,12 @@ import org.apache.cassandra.db.Cell;
 import org.apache.cassandra.db.ColumnFamily;
 import org.apache.cassandra.db.composites.CellName;
 import org.apache.cassandra.db.composites.Composite;
+import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.db.marshal.Int32Type;
 import org.apache.cassandra.db.marshal.ListType;
 import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.apache.cassandra.serializers.CollectionSerializer;
+import org.apache.cassandra.serializers.ListSerializer;
 import org.apache.cassandra.serializers.MarshalException;
 import org.apache.cassandra.transport.Server;
 import org.apache.cassandra.utils.ByteBufferUtil;
@@ -123,7 +125,7 @@ public abstract class Lists
         }
     }
 
-    public static class Value extends Term.MultiItemTerminal implements Term.CollectionTerminal
+    public static class Value extends Term.MultiItemTerminal
     {
         public final List<ByteBuffer> elements;
 
@@ -151,12 +153,7 @@ public abstract class Lists
             }
         }
 
-        public ByteBuffer get(QueryOptions options)
-        {
-            return getWithProtocolVersion(options.getProtocolVersion());
-        }
-
-        public ByteBuffer getWithProtocolVersion(int protocolVersion)
+        public ByteBuffer get(int protocolVersion)
         {
             return CollectionSerializer.pack(elements, elements.size(), protocolVersion);
         }
@@ -381,7 +378,6 @@ public abstract class Lists
         static void doAppend(Term t, ColumnFamily cf, Composite prefix, ColumnDefinition column, UpdateParameters params) throws InvalidRequestException
         {
             Term.Terminal value = t.bind(params.options);
-            Lists.Value listValue = (Lists.Value)value;
             if (column.type.isMultiCell())
             {
                 // If we append null, do nothing. Note that for Setter, we've
@@ -389,11 +385,10 @@ public abstract class Lists
                 if (value == null)
                     return;
 
-                List<ByteBuffer> toAdd = listValue.elements;
-                for (int i = 0; i < toAdd.size(); i++)
+                for (ByteBuffer buffer : ((Value) value).elements)
                 {
                     ByteBuffer uuid = ByteBuffer.wrap(UUIDGen.getTimeUUIDBytes());
-                    cf.addColumn(params.makeColumn(cf.getComparator().create(prefix, column, uuid), toAdd.get(i)));
+                    cf.addColumn(params.makeColumn(cf.getComparator().create(prefix, column, uuid), buffer));
                 }
             }
             else
@@ -403,7 +398,7 @@ public abstract class Lists
                 if (value == null)
                     cf.addAtom(params.makeTombstone(name));
                 else
-                    cf.addColumn(params.makeColumn(name, listValue.getWithProtocolVersion(Server.CURRENT_VERSION)));
+                    cf.addColumn(params.makeColumn(name, value.get(Server.CURRENT_VERSION)));
             }
         }
     }
@@ -422,10 +417,9 @@ public abstract class Lists
             if (value == null)
                 return;
 
-            assert value instanceof Lists.Value;
             long time = PrecisionTime.REFERENCE_TIME - (System.currentTimeMillis() - PrecisionTime.REFERENCE_TIME);
 
-            List<ByteBuffer> toAdd = ((Lists.Value)value).elements;
+            List<ByteBuffer> toAdd = ((Value) value).elements;
             for (int i = toAdd.size() - 1; i >= 0; i--)
             {
                 PrecisionTime pt = PrecisionTime.getNext(time);
@@ -463,13 +457,11 @@ public abstract class Lists
             if (value == null)
                 return;
 
-            assert value instanceof Lists.Value;
-
             // Note: below, we will call 'contains' on this toDiscard list for each element of existingList.
             // Meaning that if toDiscard is big, converting it to a HashSet might be more efficient. However,
             // the read-before-write this operation requires limits its usefulness on big lists, so in practice
             // toDiscard will be small and keeping a list will be more efficient.
-            List<ByteBuffer> toDiscard = ((Lists.Value)value).elements;
+            List<ByteBuffer> toDiscard = ((Value) value).elements;
             for (Cell cell : existingList)
             {
                 if (toDiscard.contains(cell.value()))
@@ -499,7 +491,7 @@ public abstract class Lists
                 throw new InvalidRequestException("Invalid null value for list index");
 
             List<Cell> existingList = params.getPrefetchedList(rowKey, column.name);
-            int idx = ByteBufferUtil.toInt(index.get(params.options));
+            int idx = ByteBufferUtil.toInt(index.get(params.options.getProtocolVersion()));
             if (existingList == null)
                 throw new InvalidRequestException("Attempted to delete an element from a list which is null");
             if (idx < 0 || idx >= existingList.size())

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b02d1a/src/java/org/apache/cassandra/cql3/Maps.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/Maps.java b/src/java/org/apache/cassandra/cql3/Maps.java
index 522a820..d91b9d8 100644
--- a/src/java/org/apache/cassandra/cql3/Maps.java
+++ b/src/java/org/apache/cassandra/cql3/Maps.java
@@ -18,22 +18,17 @@
 package org.apache.cassandra.cql3;
 
 import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.TreeMap;
+import java.util.*;
 
 import org.apache.cassandra.config.ColumnDefinition;
 import org.apache.cassandra.db.ColumnFamily;
 import org.apache.cassandra.db.composites.CellName;
 import org.apache.cassandra.db.composites.Composite;
+import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.db.marshal.MapType;
 import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.apache.cassandra.serializers.CollectionSerializer;
+import org.apache.cassandra.serializers.MapSerializer;
 import org.apache.cassandra.serializers.MarshalException;
 import org.apache.cassandra.transport.Server;
 import org.apache.cassandra.utils.FBUtilities;
@@ -146,7 +141,7 @@ public abstract class Maps
         }
     }
 
-    public static class Value extends Term.Terminal implements Term.CollectionTerminal
+    public static class Value extends Term.Terminal
     {
         public final Map<ByteBuffer, ByteBuffer> map;
 
@@ -173,12 +168,7 @@ public abstract class Maps
             }
         }
 
-        public ByteBuffer get(QueryOptions options)
-        {
-            return getWithProtocolVersion(options.getProtocolVersion());
-        }
-
-        public ByteBuffer getWithProtocolVersion(int protocolVersion)
+        public ByteBuffer get(int protocolVersion)
         {
             List<ByteBuffer> buffers = new ArrayList<>(2 * map.size());
             for (Map.Entry<ByteBuffer, ByteBuffer> entry : map.entrySet())
@@ -353,13 +343,13 @@ public abstract class Maps
         static void doPut(Term t, ColumnFamily cf, Composite prefix, ColumnDefinition column, UpdateParameters params) throws InvalidRequestException
         {
             Term.Terminal value = t.bind(params.options);
-            Maps.Value mapValue = (Maps.Value) value;
             if (column.type.isMultiCell())
             {
                 if (value == null)
                     return;
 
-                for (Map.Entry<ByteBuffer, ByteBuffer> entry : mapValue.map.entrySet())
+                Map<ByteBuffer, ByteBuffer> elements = ((Value) value).map;
+                for (Map.Entry<ByteBuffer, ByteBuffer> entry : elements.entrySet())
                 {
                     CellName cellName = cf.getComparator().create(prefix, column, entry.getKey());
                     cf.addColumn(params.makeColumn(cellName, entry.getValue()));
@@ -372,7 +362,7 @@ public abstract class Maps
                 if (value == null)
                     cf.addAtom(params.makeTombstone(cellName));
                 else
-                    cf.addColumn(params.makeColumn(cellName, mapValue.getWithProtocolVersion(Server.CURRENT_VERSION)));
+                    cf.addColumn(params.makeColumn(cellName, value.get(Server.CURRENT_VERSION)));
             }
         }
     }
@@ -391,7 +381,7 @@ public abstract class Maps
             if (key == null)
                 throw new InvalidRequestException("Invalid null map key");
 
-            CellName cellName = cf.getComparator().create(prefix, column, key.get(params.options));
+            CellName cellName = cf.getComparator().create(prefix, column, key.get(params.options.getProtocolVersion()));
             cf.addColumn(params.makeTombstone(cellName));
         }
     }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b02d1a/src/java/org/apache/cassandra/cql3/QueryOptions.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/QueryOptions.java b/src/java/org/apache/cassandra/cql3/QueryOptions.java
index 60fd339..98c2395 100644
--- a/src/java/org/apache/cassandra/cql3/QueryOptions.java
+++ b/src/java/org/apache/cassandra/cql3/QueryOptions.java
@@ -72,6 +72,11 @@ public abstract class QueryOptions
         return new DefaultQueryOptions(consistency, Collections.<ByteBuffer>emptyList(), false, SpecificOptions.DEFAULT, Server.VERSION_2);
     }
 
+    public static QueryOptions forProtocolVersion(int protocolVersion)
+    {
+        return new DefaultQueryOptions(null, null, true, null, protocolVersion);
+    }
+
     public static QueryOptions create(ConsistencyLevel consistency, List<ByteBuffer> values, boolean skipMetadata, int pageSize, PagingState pagingState, ConsistencyLevel serialConsistency)
     {
         return new DefaultQueryOptions(consistency, values, skipMetadata, new SpecificOptions(pageSize, pagingState, serialConsistency, -1L), 0);

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b02d1a/src/java/org/apache/cassandra/cql3/ResultSet.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/ResultSet.java b/src/java/org/apache/cassandra/cql3/ResultSet.java
index c0982c4..267c7c1 100644
--- a/src/java/org/apache/cassandra/cql3/ResultSet.java
+++ b/src/java/org/apache/cassandra/cql3/ResultSet.java
@@ -36,7 +36,6 @@ import org.apache.cassandra.service.pager.PagingState;
 public class ResultSet
 {
     public static final Codec codec = new Codec();
-    private static final ColumnIdentifier COUNT_COLUMN = new ColumnIdentifier("count", false);
 
     public final ResultMetadata metadata;
     public final List<List<ByteBuffer>> rows;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b02d1a/src/java/org/apache/cassandra/cql3/Sets.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/Sets.java b/src/java/org/apache/cassandra/cql3/Sets.java
index bd58237..e0a53ad 100644
--- a/src/java/org/apache/cassandra/cql3/Sets.java
+++ b/src/java/org/apache/cassandra/cql3/Sets.java
@@ -32,6 +32,7 @@ import org.apache.cassandra.db.marshal.SetType;
 import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.apache.cassandra.serializers.CollectionSerializer;
 import org.apache.cassandra.serializers.MarshalException;
+import org.apache.cassandra.serializers.SetSerializer;
 import org.apache.cassandra.transport.Server;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.FBUtilities;
@@ -131,7 +132,7 @@ public abstract class Sets
         }
     }
 
-    public static class Value extends Term.Terminal implements Term.CollectionTerminal
+    public static class Value extends Term.Terminal
     {
         public final SortedSet<ByteBuffer> elements;
 
@@ -158,14 +159,9 @@ public abstract class Sets
             }
         }
 
-        public ByteBuffer get(QueryOptions options)
+        public ByteBuffer get(int protocolVersion)
         {
-            return getWithProtocolVersion(options.getProtocolVersion());
-        }
-
-        public ByteBuffer getWithProtocolVersion(int protocolVersion)
-        {
-            return CollectionSerializer.pack(new ArrayList<>(elements), elements.size(), protocolVersion);
+            return CollectionSerializer.pack(elements, elements.size(), protocolVersion);
         }
 
         public boolean equals(SetType st, Value v)
@@ -279,14 +275,12 @@ public abstract class Sets
         static void doAdd(Term t, ColumnFamily cf, Composite prefix, ColumnDefinition column, UpdateParameters params) throws InvalidRequestException
         {
             Term.Terminal value = t.bind(params.options);
-            Sets.Value setValue = (Sets.Value)value;
             if (column.type.isMultiCell())
             {
                 if (value == null)
                     return;
 
-                Set<ByteBuffer> toAdd = setValue.elements;
-                for (ByteBuffer bb : toAdd)
+                for (ByteBuffer bb : ((Value) value).elements)
                 {
                     CellName cellName = cf.getComparator().create(prefix, column, bb);
                     cf.addColumn(params.makeColumn(cellName, ByteBufferUtil.EMPTY_BYTE_BUFFER));
@@ -299,7 +293,7 @@ public abstract class Sets
                 if (value == null)
                     cf.addAtom(params.makeTombstone(cellName));
                 else
-                    cf.addColumn(params.makeColumn(cellName, ((Value) value).getWithProtocolVersion(Server.CURRENT_VERSION)));
+                    cf.addColumn(params.makeColumn(cellName, value.get(Server.CURRENT_VERSION)));
             }
         }
     }
@@ -323,12 +317,10 @@ public abstract class Sets
             // This can be either a set or a single element
             Set<ByteBuffer> toDiscard = value instanceof Sets.Value
                                       ? ((Sets.Value)value).elements
-                                      : Collections.singleton(value.get(params.options));
+                                      : Collections.singleton(value.get(params.options.getProtocolVersion()));
 
             for (ByteBuffer bb : toDiscard)
-            {
                 cf.addColumn(params.makeTombstone(cf.getComparator().create(prefix, column, bb)));
-            }
         }
     }
 
@@ -346,7 +338,7 @@ public abstract class Sets
             if (elt == null)
                 throw new InvalidRequestException("Invalid null set element");
 
-            CellName cellName = cf.getComparator().create(prefix, column, elt.get(params.options));
+            CellName cellName = cf.getComparator().create(prefix, column, elt.get(params.options.getProtocolVersion()));
             cf.addColumn(params.makeTombstone(cellName));
         }
     }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b02d1a/src/java/org/apache/cassandra/cql3/Term.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/Term.java b/src/java/org/apache/cassandra/cql3/Term.java
index 7e20df8..fe8e51e 100644
--- a/src/java/org/apache/cassandra/cql3/Term.java
+++ b/src/java/org/apache/cassandra/cql3/Term.java
@@ -131,12 +131,13 @@ public interface Term
 
         /**
          * @return the serialized value of this terminal.
+         * @param protocolVersion
          */
-        public abstract ByteBuffer get(QueryOptions options);
+        public abstract ByteBuffer get(int protocolVersion) throws InvalidRequestException;
 
         public ByteBuffer bindAndGet(QueryOptions options) throws InvalidRequestException
         {
-            return get(options);
+            return get(options.getProtocolVersion());
         }
     }
 
@@ -145,12 +146,6 @@ public interface Term
         public abstract List<ByteBuffer> getElements();
     }
 
-    public interface CollectionTerminal
-    {
-        /** Gets the value of the collection when serialized with the given protocol version format */
-        public ByteBuffer getWithProtocolVersion(int protocolVersion);
-    }
-
     /**
      * A non terminal term, i.e. a term that can only be reduce to a byte buffer
      * at execution time.
@@ -171,7 +166,7 @@ public interface Term
         public ByteBuffer bindAndGet(QueryOptions options) throws InvalidRequestException
         {
             Terminal t = bind(options);
-            return t == null ? null : t.get(options);
+            return t == null ? null : t.get(options.getProtocolVersion());
         }
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b02d1a/src/java/org/apache/cassandra/cql3/Tuples.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/Tuples.java b/src/java/org/apache/cassandra/cql3/Tuples.java
index ca436a1..6d82218 100644
--- a/src/java/org/apache/cassandra/cql3/Tuples.java
+++ b/src/java/org/apache/cassandra/cql3/Tuples.java
@@ -153,7 +153,7 @@ public class Tuples
             return new Value(type.split(bytes));
         }
 
-        public ByteBuffer get(QueryOptions options)
+        public ByteBuffer get(int protocolVersion)
         {
             return TupleType.buildValue(elements);
         }
@@ -264,7 +264,7 @@ public class Tuples
             }
         }
 
-        public ByteBuffer get(QueryOptions options)
+        public ByteBuffer get(int protocolVersion)
         {
             throw new UnsupportedOperationException();
         }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b02d1a/src/java/org/apache/cassandra/cql3/UntypedResultSet.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/UntypedResultSet.java b/src/java/org/apache/cassandra/cql3/UntypedResultSet.java
index 657d536..072d3b7 100644
--- a/src/java/org/apache/cassandra/cql3/UntypedResultSet.java
+++ b/src/java/org/apache/cassandra/cql3/UntypedResultSet.java
@@ -73,8 +73,8 @@ public abstract class UntypedResultSet implements Iterable<UntypedResultSet.Row>
 
         public Row one()
         {
-            if (cqlRows.rows.size() != 1)
-                throw new IllegalStateException("One row required, " + cqlRows.rows.size() + " found");
+            if (cqlRows.size() != 1)
+                throw new IllegalStateException("One row required, " + cqlRows.size() + " found");
             return new Row(cqlRows.metadata.names, cqlRows.rows.get(0));
         }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b02d1a/src/java/org/apache/cassandra/cql3/functions/FromJsonFct.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/functions/FromJsonFct.java b/src/java/org/apache/cassandra/cql3/functions/FromJsonFct.java
new file mode 100644
index 0000000..2b9e8c6
--- /dev/null
+++ b/src/java/org/apache/cassandra/cql3/functions/FromJsonFct.java
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.cql3.functions;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.cassandra.cql3.Json;
+
+import org.apache.cassandra.cql3.QueryOptions;
+import org.apache.cassandra.db.marshal.*;
+import org.apache.cassandra.exceptions.FunctionExecutionException;
+import org.apache.cassandra.serializers.MarshalException;
+
+public class FromJsonFct extends NativeScalarFunction
+{
+    public static final FunctionName NAME = FunctionName.nativeFunction("fromjson");
+
+    private static final Map<AbstractType<?>, FromJsonFct> instances = new ConcurrentHashMap<>();
+
+    public static FromJsonFct getInstance(AbstractType<?> returnType)
+    {
+        FromJsonFct func = instances.get(returnType);
+        if (func == null)
+        {
+            func = new FromJsonFct(returnType);
+            instances.put(returnType, func);
+        }
+        return func;
+    }
+
+    private FromJsonFct(AbstractType<?> returnType)
+    {
+        super("fromjson", returnType, UTF8Type.instance);
+    }
+
+    public ByteBuffer execute(int protocolVersion, List<ByteBuffer> parameters)
+    {
+        assert parameters.size() == 1 : "Unexpectedly got " + parameters.size() + " arguments for fromJson()";
+        ByteBuffer argument = parameters.get(0);
+        if (argument == null)
+            return null;
+
+        String jsonArg = UTF8Type.instance.getSerializer().deserialize(argument);
+        try
+        {
+            Object object = Json.JSON_OBJECT_MAPPER.readValue(jsonArg, Object.class);
+            if (object == null)
+                return null;
+            return returnType.fromJSONObject(object).bindAndGet(QueryOptions.forProtocolVersion(protocolVersion));
+        }
+        catch (IOException exc)
+        {
+            throw new FunctionExecutionException(NAME, Collections.singletonList("text"), String.format("Could not decode JSON string '%s': %s", jsonArg, exc.toString()));
+        }
+        catch (MarshalException exc)
+        {
+            throw FunctionExecutionException.create(this, exc);
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b02d1a/src/java/org/apache/cassandra/cql3/functions/FunctionCall.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/functions/FunctionCall.java b/src/java/org/apache/cassandra/cql3/functions/FunctionCall.java
index 72ac63e..90ebaaf 100644
--- a/src/java/org/apache/cassandra/cql3/functions/FunctionCall.java
+++ b/src/java/org/apache/cassandra/cql3/functions/FunctionCall.java
@@ -62,14 +62,7 @@ public class FunctionCall extends Term.NonTerminal
     {
         List<ByteBuffer> buffers = new ArrayList<>(terms.size());
         for (Term t : terms)
-        {
-            // For now, we don't allow nulls as argument as no existing function needs it and it
-            // simplify things.
-            ByteBuffer val = t.bindAndGet(options);
-            if (val == null)
-                throw new InvalidRequestException(String.format("Invalid null value for argument to %s", fun));
-            buffers.add(val);
-        }
+            buffers.add(t.bindAndGet(options));
         return executeInternal(options.getProtocolVersion(), fun, buffers);
     }
 
@@ -85,8 +78,8 @@ public class FunctionCall extends Term.NonTerminal
         }
         catch (MarshalException e)
         {
-            throw new RuntimeException(String.format("Return of function %s (%s) is not a valid value for its declared return type %s", 
-                                                     fun, ByteBufferUtil.bytesToHex(result), fun.returnType().asCQL3Type()));
+            throw new RuntimeException(String.format("Return of function %s (%s) is not a valid value for its declared return type %s",
+                                                     fun, ByteBufferUtil.bytesToHex(result), fun.returnType().asCQL3Type()), e);
         }
     }
 
@@ -127,7 +120,7 @@ public class FunctionCall extends Term.NonTerminal
 
         public Term prepare(String keyspace, ColumnSpecification receiver) throws InvalidRequestException
         {
-            Function fun = Functions.get(keyspace, name, terms, receiver.ksName, receiver.cfName);
+            Function fun = Functions.get(keyspace, name, terms, receiver.ksName, receiver.cfName, receiver.type);
             if (fun == null)
                 throw new InvalidRequestException(String.format("Unknown function %s called", name));
             if (fun.isAggregate())
@@ -144,7 +137,7 @@ public class FunctionCall extends Term.NonTerminal
 
             if (fun.argTypes().size() != terms.size())
                 throw new InvalidRequestException(String.format("Incorrect number of arguments specified for function %s (expected %d, found %d)",
-                                                                fun.name(), fun.argTypes().size(), terms.size()));
+                                                                fun, fun.argTypes().size(), terms.size()));
 
             List<Term> parameters = new ArrayList<>(terms.size());
             boolean allTerminal = true;
@@ -170,7 +163,7 @@ public class FunctionCall extends Term.NonTerminal
             for (Term t : parameters)
             {
                 assert t instanceof Term.Terminal;
-                buffers.add(((Term.Terminal)t).get(QueryOptions.DEFAULT));
+                buffers.add(((Term.Terminal)t).get(QueryOptions.DEFAULT.getProtocolVersion()));
             }
 
             return executeInternal(Server.CURRENT_VERSION, fun, buffers);
@@ -184,7 +177,14 @@ public class FunctionCall extends Term.NonTerminal
             // later with a more helpful error message that if we were to return false here.
             try
             {
-                Function fun = Functions.get(keyspace, name, terms, receiver.ksName, receiver.cfName);
+                Function fun = Functions.get(keyspace, name, terms, receiver.ksName, receiver.cfName, receiver.type);
+
+                // Because fromJson() can return whatever type the receiver is, we'll always get EXACT_MATCH.  To
+                // handle potentially ambiguous function calls with fromJson() as an argument, always return
+                // WEAKLY_ASSIGNABLE to force the user to typecast if necessary
+                if (fun != null && fun.name().equals(FromJsonFct.NAME))
+                    return TestResult.WEAKLY_ASSIGNABLE;
+
                 if (fun != null && receiver.type.equals(fun.returnType()))
                     return AssignmentTestable.TestResult.EXACT_MATCH;
                 else if (fun == null || receiver.type.isValueCompatibleWith(fun.returnType()))

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b02d1a/src/java/org/apache/cassandra/cql3/functions/FunctionName.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/functions/FunctionName.java b/src/java/org/apache/cassandra/cql3/functions/FunctionName.java
index bb30040..d732efa 100644
--- a/src/java/org/apache/cassandra/cql3/functions/FunctionName.java
+++ b/src/java/org/apache/cassandra/cql3/functions/FunctionName.java
@@ -34,7 +34,7 @@ public final class FunctionName
     public FunctionName(String keyspace, String name)
     {
         assert name != null : "Name parameter must not be null";
-        this.keyspace = keyspace != null ? keyspace : null;
+        this.keyspace = keyspace;
         this.name = name;
     }
 
@@ -65,6 +65,15 @@ public final class FunctionName
             && Objects.equal(this.name, that.name);
     }
 
+    public final boolean equalsNativeFunction(FunctionName nativeFunction)
+    {
+        assert nativeFunction.keyspace.equals(SystemKeyspace.NAME);
+        if (this.hasKeyspace() && !this.keyspace.equals(SystemKeyspace.NAME))
+            return false;
+
+        return Objects.equal(this.name, nativeFunction.name);
+    }
+
     @Override
     public String toString()
     {

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b02d1a/src/java/org/apache/cassandra/cql3/functions/Functions.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/functions/Functions.java b/src/java/org/apache/cassandra/cql3/functions/Functions.java
index 09e360b..bc0b9d9 100644
--- a/src/java/org/apache/cassandra/cql3/functions/Functions.java
+++ b/src/java/org/apache/cassandra/cql3/functions/Functions.java
@@ -53,7 +53,7 @@ public abstract class Functions
 
         for (CQL3Type type : CQL3Type.Native.values())
         {
-            // Note: because text and varchar ends up being synonimous, our automatic makeToBlobFunction doesn't work
+            // Note: because text and varchar ends up being synonymous, our automatic makeToBlobFunction doesn't work
             // for varchar, so we special case it below. We also skip blob for obvious reasons.
             if (type == CQL3Type.Native.VARCHAR || type == CQL3Type.Native.BLOB)
                 continue;
@@ -101,18 +101,41 @@ public abstract class Functions
         return declared.get(name).size();
     }
 
+    /**
+     * @param keyspace the current keyspace
+     * @param name the name of the function
+     * @param providedArgs the arguments provided for the function call
+     * @param receiverKs the receiver's keyspace
+     * @param receiverCf the receiver's table
+     * @param receiverType if the receiver type is known (during inserts, for example), this should be the type of
+     *                     the receiver
+     * @throws InvalidRequestException
+     */
     public static Function get(String keyspace,
                                FunctionName name,
                                List<? extends AssignmentTestable> providedArgs,
                                String receiverKs,
-                               String receiverCf)
+                               String receiverCf,
+                               AbstractType<?> receiverType)
     throws InvalidRequestException
     {
-        if (name.hasKeyspace()
-            ? name.equals(TOKEN_FUNCTION_NAME)
-            : name.name.equals(TOKEN_FUNCTION_NAME.name))
+        if (name.equalsNativeFunction(TOKEN_FUNCTION_NAME))
             return new TokenFct(Schema.instance.getCFMetaData(receiverKs, receiverCf));
 
+        // The toJson() function can accept any type of argument, so instances of it are not pre-declared.  Instead,
+        // we create new instances as needed while handling selectors (which is the only place that toJson() is supported,
+        // due to needing to know the argument types in advance).
+        if (name.equalsNativeFunction(ToJsonFct.NAME))
+            throw new InvalidRequestException("toJson() may only be used within the selection clause of SELECT statements");
+
+        // Similarly, we can only use fromJson when we know the receiver type (such as inserts)
+        if (name.equalsNativeFunction(FromJsonFct.NAME))
+        {
+            if (receiverType == null)
+                throw new InvalidRequestException("fromJson() cannot be used in the selection clause of a SELECT statement");
+            return FromJsonFct.getInstance(receiverType);
+        }
+
         List<Function> candidates;
         if (!name.hasKeyspace())
         {

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b02d1a/src/java/org/apache/cassandra/cql3/functions/ToJsonFct.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/functions/ToJsonFct.java b/src/java/org/apache/cassandra/cql3/functions/ToJsonFct.java
new file mode 100644
index 0000000..bcb4559
--- /dev/null
+++ b/src/java/org/apache/cassandra/cql3/functions/ToJsonFct.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.cql3.functions;
+
+import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.utils.ByteBufferUtil;
+
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class ToJsonFct extends NativeScalarFunction
+{
+    public static final FunctionName NAME = FunctionName.nativeFunction("tojson");
+
+    private static final Map<AbstractType<?>, ToJsonFct> instances = new ConcurrentHashMap<>();
+
+    public static ToJsonFct getInstance(List<AbstractType<?>> argTypes) throws InvalidRequestException
+    {
+        if (argTypes.size() != 1)
+            throw new InvalidRequestException(String.format("toJson() only accepts one argument (got %d)", argTypes.size()));
+
+        AbstractType<?> fromType = argTypes.get(0);
+        ToJsonFct func = instances.get(fromType);
+        if (func == null)
+        {
+            func = new ToJsonFct(fromType);
+            instances.put(fromType, func);
+        }
+        return func;
+    }
+
+    private ToJsonFct(AbstractType<?> argType)
+    {
+        super("tojson", UTF8Type.instance, argType);
+    }
+
+    public ByteBuffer execute(int protocolVersion, List<ByteBuffer> parameters) throws InvalidRequestException
+    {
+        assert parameters.size() == 1 : "Expected 1 argument for toJson(), but got " + parameters.size();
+        ByteBuffer parameter = parameters.get(0);
+        if (parameter == null)
+            return ByteBufferUtil.bytes("null");
+
+        return ByteBufferUtil.bytes(argTypes.get(0).toJSONString(parameter, protocolVersion));
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c7b02d1a/src/java/org/apache/cassandra/cql3/selection/Selectable.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/selection/Selectable.java b/src/java/org/apache/cassandra/cql3/selection/Selectable.java
index c5ef857..4506111 100644
--- a/src/java/org/apache/cassandra/cql3/selection/Selectable.java
+++ b/src/java/org/apache/cassandra/cql3/selection/Selectable.java
@@ -27,6 +27,7 @@ import org.apache.cassandra.cql3.ColumnIdentifier;
 import org.apache.cassandra.cql3.functions.Function;
 import org.apache.cassandra.cql3.functions.FunctionName;
 import org.apache.cassandra.cql3.functions.Functions;
+import org.apache.cassandra.cql3.functions.ToJsonFct;
 import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.db.marshal.UserType;
 import org.apache.cassandra.exceptions.InvalidRequestException;
@@ -143,8 +144,14 @@ public abstract class Selectable
             SelectorFactories factories  =
                     SelectorFactories.createFactoriesAndCollectColumnDefinitions(args, cfm, defs);
 
-            // resolve built-in functions before user defined functions
-            Function fun = Functions.get(cfm.ksName, functionName, factories.newInstances(), cfm.ksName, cfm.cfName);
+            // We need to circumvent the normal function lookup process for toJson() because instances of the function
+            // are not pre-declared (because it can accept any type of argument).
+            Function fun;
+            if (functionName.equalsNativeFunction(ToJsonFct.NAME))
+                fun = ToJsonFct.getInstance(factories.getReturnTypes());
+            else
+                fun = Functions.get(cfm.ksName, functionName, factories.newInstances(), cfm.ksName, cfm.cfName, null);
+
             if (fun == null)
                 throw new InvalidRequestException(String.format("Unknown function '%s'", functionName));
             if (fun.returnType() == null)