diff --git a/build.xml b/build.xml
index 06638f6974..8df738ced2 100644
--- a/build.xml
+++ b/build.xml
@@ -134,13 +134,11 @@
     <!-- Check if all tests are being run or just one. If it's all tests don't spam the console with test output.
          If it's an individual test print the output from the test under the assumption someone is debugging the test
          and wants to know what is going on without having to context switch to the log file that is generated.
-         This may be overridden when running a single test by adding the -Dtest.brief.output property to the ant
-         command (its value is unimportant).
          Debug level output still needs to be retrieved from the log file.  -->
     <script language="javascript">
         if (project.getProperty("cassandra.keepBriefBrief") == null)
         {
-            if (project.getProperty("test.name").equals("*Test") || project.getProperty("test.brief.output"))
+            if (project.getProperty("test.name").equals("*Test"))
                 project.setProperty("cassandra.keepBriefBrief", "true");
             else
                 project.setProperty("cassandra.keepBriefBrief", "false");
diff --git a/lib/cassandra-driver-core-2.2.0-rc2-SNAPSHOT-shaded.jar b/lib/cassandra-driver-core-3.0.0-alpha2-188d996-SNAPSHOT-shaded.jar
similarity index 68%
rename from lib/cassandra-driver-core-2.2.0-rc2-SNAPSHOT-shaded.jar
rename to lib/cassandra-driver-core-3.0.0-alpha2-188d996-SNAPSHOT-shaded.jar
index edb926d191..14354bd9ec 100644
Binary files a/lib/cassandra-driver-core-2.2.0-rc2-SNAPSHOT-shaded.jar and b/lib/cassandra-driver-core-3.0.0-alpha2-188d996-SNAPSHOT-shaded.jar differ
diff --git a/lib/cassandra-driver-internal-only-2.6.0rc2.post0-1a480f1.zip b/lib/cassandra-driver-internal-only-3.0.0a1.post0-90a797e.zip
similarity index 76%
rename from lib/cassandra-driver-internal-only-2.6.0rc2.post0-1a480f1.zip
rename to lib/cassandra-driver-internal-only-3.0.0a1.post0-90a797e.zip
index a611501740..0331144fa9 100644
Binary files a/lib/cassandra-driver-internal-only-2.6.0rc2.post0-1a480f1.zip and b/lib/cassandra-driver-internal-only-3.0.0a1.post0-90a797e.zip differ
diff --git a/src/java/org/apache/cassandra/config/CFMetaData.java b/src/java/org/apache/cassandra/config/CFMetaData.java
index a830469f2a..e794901a27 100644
--- a/src/java/org/apache/cassandra/config/CFMetaData.java
+++ b/src/java/org/apache/cassandra/config/CFMetaData.java
@@ -30,7 +30,8 @@ import java.util.stream.Collectors;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.MoreObjects;
 import com.google.common.base.Objects;
-import com.google.common.collect.*;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Sets;
 import org.apache.commons.lang3.ArrayUtils;
 import org.apache.commons.lang3.builder.HashCodeBuilder;
 import org.apache.commons.lang3.builder.ToStringBuilder;
@@ -42,16 +43,16 @@ import org.apache.cassandra.cql3.QueryProcessor;
 import org.apache.cassandra.cql3.statements.CFStatement;
 import org.apache.cassandra.cql3.statements.CreateTableStatement;
 import org.apache.cassandra.db.*;
-import org.apache.cassandra.db.compaction.*;
+import org.apache.cassandra.db.compaction.AbstractCompactionStrategy;
 import org.apache.cassandra.db.index.SecondaryIndex;
 import org.apache.cassandra.db.marshal.*;
 import org.apache.cassandra.dht.IPartitioner;
-import org.apache.cassandra.exceptions.*;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.apache.cassandra.io.util.DataInputPlus;
 import org.apache.cassandra.io.util.DataOutputPlus;
 import org.apache.cassandra.schema.*;
 import org.apache.cassandra.utils.*;
-import org.apache.cassandra.utils.AbstractIterator;
 import org.github.jamm.Unmetered;
 
 /**
@@ -95,6 +96,7 @@ public final class CFMetaData
     private volatile Map<ByteBuffer, DroppedColumn> droppedColumns = new HashMap<>();
     private volatile Triggers triggers = Triggers.none();
     private volatile MaterializedViews materializedViews = MaterializedViews.none();
+    private volatile Indexes indexes = Indexes.none();
 
     /*
      * All CQL3 columns definition are stored in the columnMetadata map.
@@ -224,6 +226,12 @@ public final class CFMetaData
         return this;
     }
 
+    public CFMetaData indexes(Indexes indexes)
+    {
+        this.indexes = indexes;
+        return this;
+    }
+
     private CFMetaData(String keyspace,
                        String name,
                        UUID cfId,
@@ -302,6 +310,11 @@ public final class CFMetaData
         return materializedViews;
     }
 
+    public Indexes getIndexes()
+    {
+        return indexes;
+    }
+
     public static CFMetaData create(String ksName,
                                     String name,
                                     UUID cfId,
@@ -491,7 +504,8 @@ public final class CFMetaData
         return newCFMD.params(oldCFMD.params)
                       .droppedColumns(new HashMap<>(oldCFMD.droppedColumns))
                       .triggers(oldCFMD.triggers)
-                      .materializedViews(oldCFMD.materializedViews);
+                      .materializedViews(oldCFMD.materializedViews)
+                      .indexes(oldCFMD.indexes);
     }
 
     /**
@@ -502,18 +516,10 @@ public final class CFMetaData
      *
      * @return name of the index ColumnFamily
      */
-    public String indexColumnFamilyName(ColumnDefinition info)
+    public String indexColumnFamilyName(IndexMetadata info)
     {
         // TODO simplify this when info.index_name is guaranteed to be set
-        return cfName + Directories.SECONDARY_INDEX_NAME_SEPARATOR + (info.getIndexName() == null ? ByteBufferUtil.bytesToHex(info.name.bytes) : info.getIndexName());
-    }
-
-    /**
-     * The '.' char is the only way to identify if the CFMetadata is for a secondary index
-     */
-    public boolean isSecondaryIndex()
-    {
-        return cfName.contains(".");
+        return cfName + Directories.SECONDARY_INDEX_NAME_SEPARATOR + info.name;
     }
 
     /**
@@ -689,7 +695,8 @@ public final class CFMetaData
             && Objects.equal(columnMetadata, other.columnMetadata)
             && Objects.equal(droppedColumns, other.droppedColumns)
             && Objects.equal(triggers, other.triggers)
-            && Objects.equal(materializedViews, other.materializedViews);
+            && Objects.equal(materializedViews, other.materializedViews)
+            && Objects.equal(indexes, other.indexes);
     }
 
     @Override
@@ -707,6 +714,7 @@ public final class CFMetaData
             .append(droppedColumns)
             .append(triggers)
             .append(materializedViews)
+            .append(indexes)
             .toHashCode();
     }
 
@@ -752,6 +760,7 @@ public final class CFMetaData
 
         triggers = cfm.triggers;
         materializedViews = cfm.materializedViews;
+        indexes = cfm.indexes;
 
         logger.debug("application result is {}", this);
 
@@ -820,74 +829,11 @@ public final class CFMetaData
         return columnMetadata.get(name);
     }
 
-    public ColumnDefinition getColumnDefinitionForIndex(String indexName)
-    {
-        for (ColumnDefinition def : allColumns())
-        {
-            if (indexName.equals(def.getIndexName()))
-                return def;
-        }
-        return null;
-    }
-
-    /**
-     * Convert a null index_name to appropriate default name according to column status
-     */
-    public void addDefaultIndexNames() throws ConfigurationException
-    {
-        // if this is ColumnFamily update we need to add previously defined index names to the existing columns first
-        UUID cfId = Schema.instance.getId(ksName, cfName);
-        if (cfId != null)
-        {
-            CFMetaData cfm = Schema.instance.getCFMetaData(cfId);
-
-            for (ColumnDefinition newDef : allColumns())
-            {
-                if (!cfm.columnMetadata.containsKey(newDef.name.bytes) || newDef.getIndexType() == null)
-                    continue;
-
-                String oldIndexName = cfm.getColumnDefinition(newDef.name).getIndexName();
-
-                if (oldIndexName == null)
-                    continue;
-
-                if (newDef.getIndexName() != null && !oldIndexName.equals(newDef.getIndexName()))
-                    throw new ConfigurationException("Can't modify index name: was '" + oldIndexName + "' changed to '" + newDef.getIndexName() + "'.");
-
-                newDef.setIndexName(oldIndexName);
-            }
-        }
-
-        Set<String> existingNames = existingIndexNames(null);
-        for (ColumnDefinition column : allColumns())
-        {
-            if (column.getIndexType() != null && column.getIndexName() == null)
-            {
-                String baseName = getDefaultIndexName(cfName, column.name);
-                String indexName = baseName;
-                int i = 0;
-                while (existingNames.contains(indexName))
-                    indexName = baseName + '_' + (++i);
-                column.setIndexName(indexName);
-            }
-        }
-    }
-
-    public static String getDefaultIndexName(String cfName, ColumnIdentifier columnName)
-    {
-        return (cfName + "_" + columnName + "_idx").replaceAll("\\W", "");
-    }
-
     public static boolean isNameValid(String name)
     {
         return name != null && !name.isEmpty() && name.length() <= Schema.NAME_LENGTH && name.matches("\\w+");
     }
 
-    public static boolean isIndexNameValid(String name)
-    {
-        return name != null && !name.isEmpty() && name.matches("\\w+");
-    }
-
     public CFMetaData validate() throws ConfigurationException
     {
         rebuild();
@@ -921,49 +867,28 @@ public final class CFMetaData
                     throw new ConfigurationException("Cannot add a counter column (" + def.name + ") in a non counter column family");
         }
 
+        if (!indexes.isEmpty() && isSuper())
+            throw new ConfigurationException("Secondary indexes are not supported on super column families");
+
         // initialize a set of names NOT in the CF under consideration
-        Set<String> indexNames = existingIndexNames(cfName);
-        for (ColumnDefinition c : allColumns())
+        KeyspaceMetadata ksm = Schema.instance.getKSMetaData(ksName);
+        Set<String> indexNames = ksm == null ? new HashSet<>() : ksm.existingIndexNames(cfName);
+        for (IndexMetadata index : indexes)
         {
-            if (c.getIndexType() == null)
-            {
-                if (c.getIndexName() != null)
-                    throw new ConfigurationException("Index name cannot be set without index type");
-            }
-            else
-            {
-                if (isSuper())
-                    throw new ConfigurationException("Secondary indexes are not supported on super column families");
-                if (!isIndexNameValid(c.getIndexName()))
-                    throw new ConfigurationException("Illegal index name " + c.getIndexName());
-                // check index names against this CF _and_ globally
-                if (indexNames.contains(c.getIndexName()))
-                    throw new ConfigurationException("Duplicate index name " + c.getIndexName());
-                indexNames.add(c.getIndexName());
+            index.validate();
 
-                if (c.getIndexType() == IndexType.CUSTOM)
-                {
-                    if (c.getIndexOptions() == null || !c.hasIndexOption(SecondaryIndex.CUSTOM_INDEX_OPTION_NAME))
-                        throw new ConfigurationException("Required index option missing: " + SecondaryIndex.CUSTOM_INDEX_OPTION_NAME);
-                }
+            // check index names against this CF _and_ globally
+            if (indexNames.contains(index.name))
+                throw new ConfigurationException("Duplicate index name " + index.name);
+            indexNames.add(index.name);
 
-                // This method validates the column metadata but does not intialize the index
-                SecondaryIndex.createInstance(null, c);
-            }
+            // This method validates any custom options in the index metadata but does not intialize the index
+            SecondaryIndex.validate(this, index);
         }
 
         return this;
     }
 
-    private static Set<String> existingIndexNames(String cfToExclude)
-    {
-        Set<String> indexNames = new HashSet<>();
-        for (ColumnFamilyStore cfs : ColumnFamilyStore.all())
-            if (cfToExclude == null || !cfs.name.equals(cfToExclude))
-                for (ColumnDefinition cd : cfs.metadata.allColumns())
-                    indexNames.add(cd.getIndexName());
-        return indexNames;
-    }
 
     // The comparator to validate the definition name with thrift.
     public AbstractType<?> thriftColumnNameType()
@@ -1048,7 +973,7 @@ public final class CFMetaData
         {
             throw new InvalidRequestException(String.format("Cannot rename non PRIMARY KEY part %s", from));
         }
-        else if (def.isIndexed())
+        else if (getIndexes().hasIndexFor(def))
         {
             throw new InvalidRequestException(String.format("Cannot rename column %s because it is secondary indexed", from));
         }
@@ -1172,6 +1097,7 @@ public final class CFMetaData
                     .collect(Collectors.toSet());
     }
 
+
     @Override
     public String toString()
     {
@@ -1190,6 +1116,7 @@ public final class CFMetaData
             .append("droppedColumns", droppedColumns)
             .append("triggers", triggers)
             .append("materializedViews", materializedViews)
+            .append("indexes", indexes)
             .toString();
     }
 
diff --git a/src/java/org/apache/cassandra/config/ColumnDefinition.java b/src/java/org/apache/cassandra/config/ColumnDefinition.java
index 8d8a929818..6afd3e7999 100644
--- a/src/java/org/apache/cassandra/config/ColumnDefinition.java
+++ b/src/java/org/apache/cassandra/config/ColumnDefinition.java
@@ -29,7 +29,6 @@ import org.apache.cassandra.cql3.*;
 import org.apache.cassandra.db.rows.*;
 import org.apache.cassandra.db.marshal.*;
 import org.apache.cassandra.serializers.MarshalException;
-import org.apache.cassandra.exceptions.*;
 
 public class ColumnDefinition extends ColumnSpecification implements Comparable<ColumnDefinition>
 {
@@ -60,10 +59,6 @@ public class ColumnDefinition extends ColumnSpecification implements Comparable<
 
     public final Kind kind;
 
-    private String indexName;
-    private IndexType indexType;
-    private Map<String,String> indexOptions;
-
     /*
      * If the column comparator is a composite type, indicates to which
      * component this definition refers to. If null, the definition refers to
@@ -93,7 +88,7 @@ public class ColumnDefinition extends ColumnSpecification implements Comparable<
 
     public static ColumnDefinition partitionKeyDef(String ksName, String cfName, String name, AbstractType<?> validator, Integer componentIndex)
     {
-        return new ColumnDefinition(ksName, cfName, ColumnIdentifier.getInterned(name, true), validator, null, null, null, componentIndex, Kind.PARTITION_KEY);
+        return new ColumnDefinition(ksName, cfName, ColumnIdentifier.getInterned(name, true), validator, componentIndex, Kind.PARTITION_KEY);
     }
 
     public static ColumnDefinition clusteringKeyDef(CFMetaData cfm, ByteBuffer name, AbstractType<?> validator, Integer componentIndex)
@@ -103,7 +98,7 @@ public class ColumnDefinition extends ColumnSpecification implements Comparable<
 
     public static ColumnDefinition clusteringKeyDef(String ksName, String cfName, String name, AbstractType<?> validator, Integer componentIndex)
     {
-        return new ColumnDefinition(ksName, cfName, ColumnIdentifier.getInterned(name, true),  validator, null, null, null, componentIndex, Kind.CLUSTERING);
+        return new ColumnDefinition(ksName, cfName, ColumnIdentifier.getInterned(name, true),  validator, componentIndex, Kind.CLUSTERING);
     }
 
     public static ColumnDefinition regularDef(CFMetaData cfm, ByteBuffer name, AbstractType<?> validator)
@@ -113,7 +108,7 @@ public class ColumnDefinition extends ColumnSpecification implements Comparable<
 
     public static ColumnDefinition regularDef(String ksName, String cfName, String name, AbstractType<?> validator)
     {
-        return new ColumnDefinition(ksName, cfName, ColumnIdentifier.getInterned(name, true), validator, null, null, null, null, Kind.REGULAR);
+        return new ColumnDefinition(ksName, cfName, ColumnIdentifier.getInterned(name, true), validator, null, Kind.REGULAR);
     }
 
     public static ColumnDefinition staticDef(CFMetaData cfm, ByteBuffer name, AbstractType<?> validator)
@@ -127,26 +122,15 @@ public class ColumnDefinition extends ColumnSpecification implements Comparable<
              cfm.cfName,
              ColumnIdentifier.getInterned(name, cfm.getColumnDefinitionNameComparator(kind)),
              validator,
-             null,
-             null,
-             null,
              componentIndex,
              kind);
     }
 
-    public ColumnDefinition(String ksName, String cfName, ColumnIdentifier name, AbstractType<?> type, Integer componentIndex, Kind kind)
-    {
-        this(ksName, cfName, name, type, null, null, null, componentIndex, kind);
-    }
-
     @VisibleForTesting
     public ColumnDefinition(String ksName,
                             String cfName,
                             ColumnIdentifier name,
                             AbstractType<?> validator,
-                            IndexType indexType,
-                            Map<String, String> indexOptions,
-                            String indexName,
                             Integer componentIndex,
                             Kind kind)
     {
@@ -156,9 +140,7 @@ public class ColumnDefinition extends ColumnSpecification implements Comparable<
         assert componentIndex == null || kind.isPrimaryKeyKind(); // The componentIndex really only make sense for partition and clustering columns,
                                                                   // so make sure we don't sneak it for something else since it'd breaks equals()
         this.kind = kind;
-        this.indexName = indexName;
         this.componentIndex = componentIndex;
-        this.setIndexType(indexType, indexOptions);
         this.cellPathComparator = makeCellPathComparator(kind, validator);
         this.cellComparator = cellPathComparator == null ? ColumnData.comparator : (a, b) -> cellPathComparator.compare(a.path(), b.path());
         this.asymmetricCellPathComparator = cellPathComparator == null ? null : (a, b) -> cellPathComparator.compare(((Cell)a).path(), (CellPath) b);
@@ -193,17 +175,17 @@ public class ColumnDefinition extends ColumnSpecification implements Comparable<
 
     public ColumnDefinition copy()
     {
-        return new ColumnDefinition(ksName, cfName, name, type, indexType, indexOptions, indexName, componentIndex, kind);
+        return new ColumnDefinition(ksName, cfName, name, type, componentIndex, kind);
     }
 
     public ColumnDefinition withNewName(ColumnIdentifier newName)
     {
-        return new ColumnDefinition(ksName, cfName, newName, type, indexType, indexOptions, indexName, componentIndex, kind);
+        return new ColumnDefinition(ksName, cfName, newName, type, componentIndex, kind);
     }
 
     public ColumnDefinition withNewType(AbstractType<?> newType)
     {
-        return new ColumnDefinition(ksName, cfName, name, newType, indexType, indexOptions, indexName, componentIndex, kind);
+        return new ColumnDefinition(ksName, cfName, name, newType, componentIndex, kind);
     }
 
     public boolean isOnAllComponents()
@@ -255,16 +237,13 @@ public class ColumnDefinition extends ColumnSpecification implements Comparable<
             && Objects.equal(name, cd.name)
             && Objects.equal(type, cd.type)
             && Objects.equal(kind, cd.kind)
-            && Objects.equal(componentIndex, cd.componentIndex)
-            && Objects.equal(indexName, cd.indexName)
-            && Objects.equal(indexType, cd.indexType)
-            && Objects.equal(indexOptions, cd.indexOptions);
+            && Objects.equal(componentIndex, cd.componentIndex);
     }
 
     @Override
     public int hashCode()
     {
-        return Objects.hashCode(ksName, cfName, name, type, kind, componentIndex, indexName, indexType, indexOptions);
+        return Objects.hashCode(ksName, cfName, name, type, kind, componentIndex);
     }
 
     @Override
@@ -275,8 +254,6 @@ public class ColumnDefinition extends ColumnSpecification implements Comparable<
                       .add("type", type)
                       .add("kind", kind)
                       .add("componentIndex", componentIndex)
-                      .add("indexName", indexName)
-                      .add("indexType", indexType)
                       .toString();
     }
 
@@ -302,83 +279,6 @@ public class ColumnDefinition extends ColumnSpecification implements Comparable<
             return kind == Kind.STATIC;
     }
 
-    public ColumnDefinition apply(ColumnDefinition def)  throws ConfigurationException
-    {
-        assert kind == def.kind && Objects.equal(componentIndex, def.componentIndex);
-
-        if (getIndexType() != null && def.getIndexType() != null)
-        {
-            // If an index is set (and not drop by this update), the validator shouldn't be change to a non-compatible one
-            // (and we want true comparator compatibility, not just value one, since the validator is used by LocalPartitioner to order index rows)
-            if (!def.type.isCompatibleWith(type))
-                throw new ConfigurationException(String.format("Cannot modify validator to a non-order-compatible one for column %s since an index is set", name));
-
-            assert getIndexName() != null;
-            if (!getIndexName().equals(def.getIndexName()))
-                throw new ConfigurationException("Cannot modify index name: " + def.getIndexName());
-        }
-
-        return new ColumnDefinition(ksName,
-                                    cfName,
-                                    name,
-                                    def.type,
-                                    def.getIndexType(),
-                                    def.getIndexOptions(),
-                                    def.getIndexName(),
-                                    componentIndex,
-                                    kind);
-    }
-
-    public String getIndexName()
-    {
-        return indexName;
-    }
-
-    public ColumnDefinition setIndexName(String indexName)
-    {
-        this.indexName = indexName;
-        return this;
-    }
-
-    public ColumnDefinition setIndexType(IndexType indexType, Map<String,String> indexOptions)
-    {
-        this.indexType = indexType;
-        this.indexOptions = indexOptions;
-        return this;
-    }
-
-    public ColumnDefinition setIndex(String indexName, IndexType indexType, Map<String,String> indexOptions)
-    {
-        return setIndexName(indexName).setIndexType(indexType, indexOptions);
-    }
-
-    public boolean isIndexed()
-    {
-        return indexType != null;
-    }
-
-    public IndexType getIndexType()
-    {
-        return indexType;
-    }
-
-    public Map<String,String> getIndexOptions()
-    {
-        return indexOptions;
-    }
-
-    /**
-     * Checks if the index option with the specified name has been specified.
-     *
-     * @param name index option name
-     * @return <code>true</code> if the index option with the specified name has been specified, <code>false</code>
-     * otherwise.
-     */
-    public boolean hasIndexOption(String name)
-    {
-        return indexOptions != null && indexOptions.containsKey(name);
-    }
-
     /**
      * Converts the specified column definitions into column identifiers.
      *
diff --git a/src/java/org/apache/cassandra/config/IndexType.java b/src/java/org/apache/cassandra/config/IndexType.java
deleted file mode 100644
index d39dccba1b..0000000000
--- a/src/java/org/apache/cassandra/config/IndexType.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
- * 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.config;
-
-public enum IndexType
-{
-    KEYS,
-    CUSTOM,
-    COMPOSITES
-}
diff --git a/src/java/org/apache/cassandra/cql3/functions/JavaBasedUDFunction.java b/src/java/org/apache/cassandra/cql3/functions/JavaBasedUDFunction.java
index 126160d1a3..2aafeb9364 100644
--- a/src/java/org/apache/cassandra/cql3/functions/JavaBasedUDFunction.java
+++ b/src/java/org/apache/cassandra/cql3/functions/JavaBasedUDFunction.java
@@ -171,7 +171,7 @@ final class JavaBasedUDFunction extends UDFunction
         // javaParamTypes is just the Java representation for argTypes resp. argDataTypes
         Class<?>[] javaParamTypes = UDHelper.javaTypes(argDataTypes, calledOnNullInput);
         // javaReturnType is just the Java representation for returnType resp. returnDataType
-        Class<?> javaReturnType = returnDataType.asJavaClass();
+        Class<?> javaReturnType = UDHelper.asJavaClass(returnDataType);
 
         // put each UDF in a separate package to prevent cross-UDF code access
         String pkgName = BASE_PACKAGE + '.' + generateClassName(name, 'p');
diff --git a/src/java/org/apache/cassandra/cql3/functions/JavaUDF.java b/src/java/org/apache/cassandra/cql3/functions/JavaUDF.java
index 09b6d58dfb..47f5d47a47 100644
--- a/src/java/org/apache/cassandra/cql3/functions/JavaUDF.java
+++ b/src/java/org/apache/cassandra/cql3/functions/JavaUDF.java
@@ -22,8 +22,6 @@ import java.nio.ByteBuffer;
 import java.util.List;
 
 import com.datastax.driver.core.DataType;
-import com.datastax.driver.core.ProtocolVersion;
-import org.apache.cassandra.db.marshal.AbstractType;
 
 /**
  * Base class for all Java UDFs.
@@ -58,48 +56,48 @@ public abstract class JavaUDF
     protected float compose_float(int protocolVersion, int argIndex, ByteBuffer value)
     {
         assert value != null && value.remaining() > 0;
-        return (float) DataType.cfloat().deserialize(value, ProtocolVersion.fromInt(protocolVersion));
+        return (float) UDHelper.deserialize(DataType.cfloat(), protocolVersion, value);
     }
 
     // do not remove - used by generated Java UDFs
     protected double compose_double(int protocolVersion, int argIndex, ByteBuffer value)
     {
         assert value != null && value.remaining() > 0;
-        return (double) DataType.cdouble().deserialize(value, ProtocolVersion.fromInt(protocolVersion));
+        return (double) UDHelper.deserialize(DataType.cdouble(), protocolVersion, value);
     }
 
     // do not remove - used by generated Java UDFs
     protected byte compose_byte(int protocolVersion, int argIndex, ByteBuffer value)
     {
         assert value != null && value.remaining() > 0;
-        return (byte) DataType.tinyint().deserialize(value, ProtocolVersion.fromInt(protocolVersion));
+        return (byte) UDHelper.deserialize(DataType.tinyint(), protocolVersion, value);
     }
 
     // do not remove - used by generated Java UDFs
     protected short compose_short(int protocolVersion, int argIndex, ByteBuffer value)
     {
         assert value != null && value.remaining() > 0;
-        return (short) DataType.smallint().deserialize(value, ProtocolVersion.fromInt(protocolVersion));
+        return (short) UDHelper.deserialize(DataType.smallint(), protocolVersion, value);
     }
 
     // do not remove - used by generated Java UDFs
     protected int compose_int(int protocolVersion, int argIndex, ByteBuffer value)
     {
         assert value != null && value.remaining() > 0;
-        return (int) DataType.cint().deserialize(value, ProtocolVersion.fromInt(protocolVersion));
+        return (int) UDHelper.deserialize(DataType.cint(), protocolVersion, value);
     }
 
     // do not remove - used by generated Java UDFs
     protected long compose_long(int protocolVersion, int argIndex, ByteBuffer value)
     {
         assert value != null && value.remaining() > 0;
-        return (long) DataType.bigint().deserialize(value, ProtocolVersion.fromInt(protocolVersion));
+        return (long) UDHelper.deserialize(DataType.bigint(), protocolVersion, value);
     }
 
     // do not remove - used by generated Java UDFs
     protected boolean compose_boolean(int protocolVersion, int argIndex, ByteBuffer value)
     {
         assert value != null && value.remaining() > 0;
-        return (boolean) DataType.cboolean().deserialize(value, ProtocolVersion.fromInt(protocolVersion));
+        return (boolean) UDHelper.deserialize(DataType.cboolean(), protocolVersion, value);
     }
 }
diff --git a/src/java/org/apache/cassandra/cql3/functions/ScriptBasedUDFunction.java b/src/java/org/apache/cassandra/cql3/functions/ScriptBasedUDFunction.java
index c4faa72d8b..19ef769d21 100644
--- a/src/java/org/apache/cassandra/cql3/functions/ScriptBasedUDFunction.java
+++ b/src/java/org/apache/cassandra/cql3/functions/ScriptBasedUDFunction.java
@@ -185,7 +185,7 @@ final class ScriptBasedUDFunction extends UDFunction
         if (result == null)
             return null;
 
-        Class<?> javaReturnType = returnDataType.asJavaClass();
+        Class<?> javaReturnType = UDHelper.asJavaClass(returnDataType);
         Class<?> resultType = result.getClass();
         if (!javaReturnType.isAssignableFrom(resultType))
         {
diff --git a/src/java/org/apache/cassandra/cql3/functions/UDFunction.java b/src/java/org/apache/cassandra/cql3/functions/UDFunction.java
index 12b1f9380e..e21d8af8e9 100644
--- a/src/java/org/apache/cassandra/cql3/functions/UDFunction.java
+++ b/src/java/org/apache/cassandra/cql3/functions/UDFunction.java
@@ -36,14 +36,11 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
 import com.google.common.base.Objects;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.datastax.driver.core.DataType;
-import com.datastax.driver.core.ProtocolVersion;
 import com.datastax.driver.core.UserType;
-
 import org.apache.cassandra.config.Config;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.config.Schema;
@@ -274,6 +271,7 @@ public abstract class UDFunction extends AbstractFunction implements ScalarFunct
             ByteBuffer result = DatabaseDescriptor.enableUserDefinedFunctionsThreads()
                                 ? executeAsync(protocolVersion, parameters)
                                 : executeUserDefined(protocolVersion, parameters);
+
             Tracing.trace("Executed UDF {} in {}\u03bcs", name(), (System.nanoTime() - tStart) / 1000);
             return result;
         }
@@ -313,8 +311,8 @@ public abstract class UDFunction extends AbstractFunction implements ScalarFunct
             threadMXBean.getCurrentThreadCpuTime();
             //
             // Get the TypeCodec stuff in Java Driver initialized.
-            DataType.inet().format(InetAddress.getLoopbackAddress());
-            DataType.list(DataType.ascii()).format(Collections.emptyList());
+            UDHelper.codecRegistry.codecFor(DataType.inet()).format(InetAddress.getLoopbackAddress());
+            UDHelper.codecRegistry.codecFor(DataType.ascii()).format("");
         }
 
         void setup()
@@ -475,7 +473,7 @@ public abstract class UDFunction extends AbstractFunction implements ScalarFunct
 
     protected static Object compose(DataType[] argDataTypes, int protocolVersion, int argIndex, ByteBuffer value)
     {
-        return value == null ? null : argDataTypes[argIndex].deserialize(value, ProtocolVersion.fromInt(protocolVersion));
+        return value == null ? null : UDHelper.deserialize(argDataTypes[argIndex], protocolVersion, value);
     }
 
     /**
@@ -492,7 +490,7 @@ public abstract class UDFunction extends AbstractFunction implements ScalarFunct
 
     protected static ByteBuffer decompose(DataType dataType, int protocolVersion, Object value)
     {
-        return value == null ? null : dataType.serialize(value, ProtocolVersion.fromInt(protocolVersion));
+        return value == null ? null : UDHelper.serialize(dataType, protocolVersion, value);
     }
 
     @Override
diff --git a/src/java/org/apache/cassandra/cql3/functions/UDHelper.java b/src/java/org/apache/cassandra/cql3/functions/UDHelper.java
index d1d12e6795..df6ca1f5bb 100644
--- a/src/java/org/apache/cassandra/cql3/functions/UDHelper.java
+++ b/src/java/org/apache/cassandra/cql3/functions/UDHelper.java
@@ -23,9 +23,14 @@ import java.lang.reflect.Method;
 import java.nio.ByteBuffer;
 import java.util.List;
 
+import com.datastax.driver.core.CodecRegistry;
 import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.ProtocolVersion;
+import com.datastax.driver.core.TypeCodec;
+import com.datastax.driver.core.exceptions.InvalidTypeException;
 import org.apache.cassandra.cql3.CQL3Type;
 import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.transport.Server;
 
 /**
  * Helper class for User Defined Functions + Aggregates.
@@ -34,14 +39,16 @@ public final class UDHelper
 {
     // TODO make these c'tors and methods public in Java-Driver - see https://datastax-oss.atlassian.net/browse/JAVA-502
     static final MethodHandle methodParseOne;
+    static final CodecRegistry codecRegistry;
     static
     {
         try
         {
             Class<?> cls = Class.forName("com.datastax.driver.core.CassandraTypeParser");
-            Method m = cls.getDeclaredMethod("parseOne", String.class);
+            Method m = cls.getDeclaredMethod("parseOne", String.class, ProtocolVersion.class, CodecRegistry.class);
             m.setAccessible(true);
             methodParseOne = MethodHandles.lookup().unreflect(m);
+            codecRegistry = new CodecRegistry();
         }
         catch (Exception e)
         {
@@ -61,7 +68,7 @@ public final class UDHelper
         Class<?>[] paramTypes = new Class[dataTypes.length];
         for (int i = 0; i < paramTypes.length; i++)
         {
-            Class<?> clazz = dataTypes[i].asJavaClass();
+            Class<?> clazz = asJavaClass(dataTypes[i]);
             if (!calledOnNullInput)
             {
                 // only care about classes that can be used in a data type
@@ -108,7 +115,9 @@ public final class UDHelper
         CQL3Type cqlType = abstractType.asCQL3Type();
         try
         {
-            return (DataType) methodParseOne.invoke(cqlType.getType().toString());
+            return (DataType) methodParseOne.invoke(cqlType.getType().toString(),
+                                                    ProtocolVersion.fromInt(Server.CURRENT_VERSION),
+                                                    codecRegistry);
         }
         catch (RuntimeException | Error e)
         {
@@ -121,6 +130,25 @@ public final class UDHelper
         }
     }
 
+    public static Object deserialize(DataType dataType, int protocolVersion, ByteBuffer value)
+    {
+        return codecRegistry.codecFor(dataType).deserialize(value, ProtocolVersion.fromInt(protocolVersion));
+    }
+
+    public static ByteBuffer serialize(DataType dataType, int protocolVersion, Object value)
+    {
+        TypeCodec<Object> codec = codecRegistry.codecFor(dataType);
+        if (! codec.getJavaType().getRawType().isAssignableFrom(value.getClass()))
+            throw new InvalidTypeException("Invalid value for CQL type " + dataType.getName().toString());
+
+        return codec.serialize(value, ProtocolVersion.fromInt(protocolVersion));
+    }
+
+    public static Class<?> asJavaClass(DataType dataType)
+    {
+        return codecRegistry.codecFor(dataType).getJavaType().getRawType();
+    }
+
     public static boolean isNullOrEmpty(AbstractType<?> type, ByteBuffer bb)
     {
         return bb == null ||
diff --git a/src/java/org/apache/cassandra/cql3/statements/AlterTableStatement.java b/src/java/org/apache/cassandra/cql3/statements/AlterTableStatement.java
index b7e09d969f..a0136fab1a 100644
--- a/src/java/org/apache/cassandra/cql3/statements/AlterTableStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/AlterTableStatement.java
@@ -22,12 +22,13 @@ import java.util.List;
 import java.util.Map;
 
 import org.apache.cassandra.auth.Permission;
-import org.apache.cassandra.config.CFMetaData;
-import org.apache.cassandra.config.ColumnDefinition;
-import org.apache.cassandra.config.MaterializedViewDefinition;
-import org.apache.cassandra.config.Schema;
-import org.apache.cassandra.cql3.*;
-import org.apache.cassandra.db.marshal.*;
+import org.apache.cassandra.config.*;
+import org.apache.cassandra.cql3.CFName;
+import org.apache.cassandra.cql3.CQL3Type;
+import org.apache.cassandra.cql3.ColumnIdentifier;
+import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.db.marshal.CollectionType;
+import org.apache.cassandra.db.marshal.CounterColumnType;
 import org.apache.cassandra.exceptions.*;
 import org.apache.cassandra.schema.TableParams;
 import org.apache.cassandra.service.ClientState;
@@ -263,6 +264,12 @@ public class AlterTableStatement extends SchemaAlteringStatement
                         break;
                 }
 
+                // If a column is dropped which is the target of a secondary index
+                // we need to also drop the index.
+                cfm.getIndexes().get(def).ifPresent(indexMetadata -> {
+                    cfm.indexes(cfm.getIndexes().without(indexMetadata.name));
+                });
+
                 // If a column is dropped which is the target of a materialized view,
                 // then we need to drop the view.
                 // If a column is dropped which was selected into a materialized view,
diff --git a/src/java/org/apache/cassandra/cql3/statements/CreateIndexStatement.java b/src/java/org/apache/cassandra/cql3/statements/CreateIndexStatement.java
index cc808acc45..64bb1dad56 100644
--- a/src/java/org/apache/cassandra/cql3/statements/CreateIndexStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/CreateIndexStatement.java
@@ -20,6 +20,7 @@ package org.apache.cassandra.cql3.statements;
 import java.util.Collections;
 import java.util.Map;
 
+import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableMap;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -27,11 +28,15 @@ import org.slf4j.LoggerFactory;
 import org.apache.cassandra.auth.Permission;
 import org.apache.cassandra.config.CFMetaData;
 import org.apache.cassandra.config.ColumnDefinition;
-import org.apache.cassandra.config.IndexType;
 import org.apache.cassandra.config.Schema;
+import org.apache.cassandra.cql3.CFName;
+import org.apache.cassandra.cql3.IndexName;
 import org.apache.cassandra.db.marshal.MapType;
-import org.apache.cassandra.exceptions.*;
-import org.apache.cassandra.cql3.*;
+import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.exceptions.RequestValidationException;
+import org.apache.cassandra.exceptions.UnauthorizedException;
+import org.apache.cassandra.schema.IndexMetadata;
+import org.apache.cassandra.schema.Indexes;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.MigrationManager;
 import org.apache.cassandra.thrift.ThriftValidation;
@@ -91,21 +96,31 @@ public class CreateIndexStatement extends SchemaAlteringStatement
             validateTargetColumnIsMapIfIndexInvolvesKeys(isMap, target);
         }
 
-        if (cd.getIndexType() != null)
+        Indexes existingIndexes = cfm.getIndexes();
+        for (IndexMetadata index : existingIndexes)
         {
-            IndexTarget.TargetType prevType = IndexTarget.TargetType.fromColumnDefinition(cd);
-            if (isMap && target.type != prevType)
+            if (index.indexedColumn(cfm).equals(cd))
             {
-                String msg = "Cannot create index on %s(%s): an index on %s(%s) already exists and indexing " +
-                             "a map on more than one dimension at the same time is not currently supported";
-                throw new InvalidRequestException(String.format(msg,
-                                                                target.type, target.column,
-                                                                prevType, target.column));
-            }
+                IndexTarget.TargetType prevType = IndexTarget.TargetType.fromIndexMetadata(index, cfm);
+                if (isMap && target.type != prevType)
+                {
+                    String msg = "Cannot create index on %s(%s): an index on %s(%s) already exists and indexing " +
+                                 "a map on more than one dimension at the same time is not currently supported";
+                    throw new InvalidRequestException(String.format(msg,
+                                                                    target.type, target.column,
+                                                                    prevType, target.column));
+                }
 
-            if (ifNotExists)
-                return;
-            else
+                if (ifNotExists)
+                    return;
+                else
+                    throw new InvalidRequestException("Index already existss");
+            }
+        }
+
+        if (!Strings.isNullOrEmpty(indexName))
+        {
+            if (Schema.instance.getKSMetaData(keyspace()).existingIndexNames(null).contains(indexName))
                 throw new InvalidRequestException("Index already exists");
         }
 
@@ -166,15 +181,25 @@ public class CreateIndexStatement extends SchemaAlteringStatement
     {
         CFMetaData cfm = Schema.instance.getCFMetaData(keyspace(), columnFamily()).copy();
         IndexTarget target = rawTarget.prepare(cfm);
-        logger.debug("Updating column {} definition for index {}", target.column, indexName);
+
         ColumnDefinition cd = cfm.getColumnDefinition(target.column);
+        String acceptedName = indexName;
+        if (Strings.isNullOrEmpty(acceptedName))
+            acceptedName = Indexes.getAvailableIndexName(keyspace(), columnFamily(), cd.name);
 
-        if (cd.getIndexType() != null && ifNotExists)
-            return false;
+        for (IndexMetadata existing : cfm.getIndexes())
+            if (existing.indexedColumn(cfm).equals(cd) || existing.name.equals(acceptedName))
+                if (ifNotExists)
+                    return false;
+                else
+                    throw new InvalidRequestException("Index already exists");
 
+        IndexMetadata.IndexType indexType;
+        Map<String, String> indexOptions;
         if (properties.isCustom)
         {
-            cd.setIndexType(IndexType.CUSTOM, properties.getOptions());
+            indexType = IndexMetadata.IndexType.CUSTOM;
+            indexOptions = properties.getOptions();
         }
         else if (cfm.isCompound())
         {
@@ -184,15 +209,21 @@ public class CreateIndexStatement extends SchemaAlteringStatement
             // lives easier then.
             if (cd.type.isCollection() && cd.type.isMultiCell())
                 options = ImmutableMap.of(target.type.indexOption(), "");
-            cd.setIndexType(IndexType.COMPOSITES, options);
+            indexType = IndexMetadata.IndexType.COMPOSITES;
+            indexOptions = options;
         }
         else
         {
-            cd.setIndexType(IndexType.KEYS, Collections.<String, String>emptyMap());
+            indexType = IndexMetadata.IndexType.KEYS;
+            indexOptions = Collections.emptyMap();
         }
 
-        cd.setIndexName(indexName);
-        cfm.addDefaultIndexNames();
+
+
+        logger.debug("Updating index definition for {}", indexName);
+
+        IndexMetadata index = IndexMetadata.legacyIndex(cd, acceptedName, indexType, indexOptions);
+        cfm.indexes(cfm.getIndexes().with(index));
         MigrationManager.announceColumnFamilyUpdate(cfm, false, isLocalOnly);
         return true;
     }
diff --git a/src/java/org/apache/cassandra/cql3/statements/DropIndexStatement.java b/src/java/org/apache/cassandra/cql3/statements/DropIndexStatement.java
index 4aae9acaee..63a1200cd4 100644
--- a/src/java/org/apache/cassandra/cql3/statements/DropIndexStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/DropIndexStatement.java
@@ -19,9 +19,9 @@ package org.apache.cassandra.cql3.statements;
 
 import org.apache.cassandra.auth.Permission;
 import org.apache.cassandra.config.CFMetaData;
-import org.apache.cassandra.config.ColumnDefinition;
 import org.apache.cassandra.config.Schema;
-import org.apache.cassandra.cql3.*;
+import org.apache.cassandra.cql3.IndexName;
+import org.apache.cassandra.cql3.QueryOptions;
 import org.apache.cassandra.db.KeyspaceNotDefinedException;
 import org.apache.cassandra.exceptions.*;
 import org.apache.cassandra.schema.KeyspaceMetadata;
@@ -37,7 +37,7 @@ public class DropIndexStatement extends SchemaAlteringStatement
     public final boolean ifExists;
 
     // initialized in announceMigration()
-    private String indexedCF;
+    private CFMetaData indexedTable;
 
     public DropIndexStatement(IndexName indexName, boolean ifExists)
     {
@@ -48,23 +48,16 @@ public class DropIndexStatement extends SchemaAlteringStatement
 
     public String columnFamily()
     {
-        if (indexedCF != null)
-            return indexedCF;
+        if (indexedTable != null)
+            return indexedTable.cfName;
 
-        try
-        {
-            CFMetaData cfm = findIndexedCF();
-            return cfm == null ? null : cfm.cfName;
-        }
-        catch (InvalidRequestException ire)
-        {
-            throw new RuntimeException(ire);
-        }
+        indexedTable = lookupIndexedTable();
+        return indexedTable == null ? null : indexedTable.cfName;
     }
 
     public void checkAccess(ClientState state) throws UnauthorizedException, InvalidRequestException
     {
-        CFMetaData cfm = findIndexedCF();
+        CFMetaData cfm = lookupIndexedTable();
         if (cfm == null)
             return;
 
@@ -73,7 +66,7 @@ public class DropIndexStatement extends SchemaAlteringStatement
 
     public void validate(ClientState state)
     {
-        // validated in findIndexedCf()
+        // validated in lookupIndexedTable()
     }
 
     public Event.SchemaChange changeEvent()
@@ -85,57 +78,39 @@ public class DropIndexStatement extends SchemaAlteringStatement
     @Override
     public ResultMessage execute(QueryState state, QueryOptions options) throws RequestValidationException
     {
-        announceMigration(false);
-        return indexedCF == null ? null : new ResultMessage.SchemaChange(changeEvent());
+        return announceMigration(false) ? new ResultMessage.SchemaChange(changeEvent()) : null;
     }
 
     public boolean announceMigration(boolean isLocalOnly) throws InvalidRequestException, ConfigurationException
     {
-        CFMetaData cfm = findIndexedCF();
+        CFMetaData cfm = lookupIndexedTable();
         if (cfm == null)
             return false;
 
-        CFMetaData updatedCfm = updateCFMetadata(cfm);
-        indexedCF = updatedCfm.cfName;
+        indexedTable = cfm;
+        CFMetaData updatedCfm = cfm.copy();
+        updatedCfm.indexes(updatedCfm.getIndexes().without(indexName));
         MigrationManager.announceColumnFamilyUpdate(updatedCfm, false, isLocalOnly);
         return true;
     }
 
-    private CFMetaData updateCFMetadata(CFMetaData cfm)
+    private CFMetaData lookupIndexedTable()
     {
-        ColumnDefinition column = findIndexedColumn(cfm);
-        assert column != null;
-        CFMetaData cloned = cfm.copy();
-        ColumnDefinition toChange = cloned.getColumnDefinition(column.name);
-        assert toChange.getIndexName() != null && toChange.getIndexName().equals(indexName);
-        toChange.setIndexName(null);
-        toChange.setIndexType(null, null);
-        return cloned;
-    }
+        if (indexedTable != null)
+            return indexedTable;
 
-    private CFMetaData findIndexedCF() throws InvalidRequestException
-    {
         KeyspaceMetadata ksm = Schema.instance.getKSMetaData(keyspace());
         if (ksm == null)
             throw new KeyspaceNotDefinedException("Keyspace " + keyspace() + " does not exist");
 
-        for (CFMetaData cfm : ksm.tables)
-            if (findIndexedColumn(cfm) != null)
-                return cfm;
-
-        if (ifExists)
-            return null;
-        else
-            throw new InvalidRequestException("Index '" + indexName + "' could not be found in any of the tables of keyspace '" + keyspace() + '\'');
-    }
-
-    private ColumnDefinition findIndexedColumn(CFMetaData cfm)
-    {
-        for (ColumnDefinition column : cfm.allColumns())
-        {
-            if (column.getIndexType() != null && column.getIndexName() != null && column.getIndexName().equals(indexName))
-                return column;
-        }
-        return null;
+        return ksm.findIndexedTable(indexName)
+                  .orElseGet(() -> {
+                      if (ifExists)
+                          return null;
+                      else
+                          throw new InvalidRequestException(String.format("Index '%s' could not be found in any " +
+                                                                          "of the tables of keyspace '%s'",
+                                                                          indexName, keyspace()));
+                  });
     }
 }
diff --git a/src/java/org/apache/cassandra/cql3/statements/IndexTarget.java b/src/java/org/apache/cassandra/cql3/statements/IndexTarget.java
index d6023882bd..4abb0778cf 100644
--- a/src/java/org/apache/cassandra/cql3/statements/IndexTarget.java
+++ b/src/java/org/apache/cassandra/cql3/statements/IndexTarget.java
@@ -23,6 +23,7 @@ import org.apache.cassandra.config.CFMetaData;
 import org.apache.cassandra.config.ColumnDefinition;
 import org.apache.cassandra.cql3.ColumnIdentifier;
 import org.apache.cassandra.db.index.SecondaryIndex;
+import org.apache.cassandra.schema.IndexMetadata;
 
 public class IndexTarget
 {
@@ -98,17 +99,29 @@ public class IndexTarget
             }
         }
 
-        public static TargetType fromColumnDefinition(ColumnDefinition cd)
+        public static TargetType fromIndexMetadata(IndexMetadata index, CFMetaData cfm)
         {
-            Map<String, String> options = cd.getIndexOptions();
+            Map<String, String> options = index.options;
             if (options.containsKey(SecondaryIndex.INDEX_KEYS_OPTION_NAME))
+            {
                 return KEYS;
+            }
             else if (options.containsKey(SecondaryIndex.INDEX_ENTRIES_OPTION_NAME))
+            {
                 return KEYS_AND_VALUES;
-            else if (cd.type.isCollection() && !cd.type.isMultiCell())
-                return FULL;
+            }
             else
-                return VALUES;
+            {
+                ColumnDefinition cd = index.indexedColumn(cfm);
+                if (cd.type.isCollection() && !cd.type.isMultiCell())
+                {
+                    return FULL;
+                }
+                else
+                {
+                    return VALUES;
+                }
+            }
         }
     }
 }
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 202047b61d..a12de0a813 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -371,11 +371,8 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
         }
 
         // create the private ColumnFamilyStores for the secondary column indexes
-        for (ColumnDefinition info : metadata.allColumns())
-        {
-            if (info.getIndexType() != null)
-                indexManager.addIndexedColumn(info);
-        }
+        for (IndexMetadata info : metadata.getIndexes())
+            indexManager.addIndexedColumn(info);
 
         if (registerBookkeeping)
         {
@@ -571,15 +568,9 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
         }
 
         // also clean out any index leftovers.
-        for (ColumnDefinition def : metadata.allColumns())
-        {
-            if (def.isIndexed())
-            {
-                CFMetaData indexMetadata = SecondaryIndex.newIndexMetadata(metadata, def);
-                if (indexMetadata != null)
-                    scrubDataDirectories(indexMetadata);
-            }
-        }
+        for (IndexMetadata def : metadata.getIndexes())
+            if (!def.isCustom())
+                scrubDataDirectories(SecondaryIndex.newIndexMetadata(metadata, def));
     }
 
     // must be called after all sstables are loaded since row cache merges all row versions
diff --git a/src/java/org/apache/cassandra/db/index/AbstractSimplePerColumnSecondaryIndex.java b/src/java/org/apache/cassandra/db/index/AbstractSimplePerColumnSecondaryIndex.java
index eccd7e2987..a631b9a7d4 100644
--- a/src/java/org/apache/cassandra/db/index/AbstractSimplePerColumnSecondaryIndex.java
+++ b/src/java/org/apache/cassandra/db/index/AbstractSimplePerColumnSecondaryIndex.java
@@ -53,7 +53,7 @@ public abstract class AbstractSimplePerColumnSecondaryIndex extends PerColumnSec
 
         columnDef = columnDefs.iterator().next();
 
-        CFMetaData indexedCfMetadata = SecondaryIndex.newIndexMetadata(baseCfs.metadata, columnDef, getIndexKeyComparator());
+        CFMetaData indexedCfMetadata = SecondaryIndex.newIndexMetadata(baseCfs.metadata, indexMetadata, getIndexKeyComparator());
         indexCfs = ColumnFamilyStore.createColumnFamilyStore(baseCfs.keyspace,
                                                              indexedCfMetadata.cfName,
                                                              indexedCfMetadata,
diff --git a/src/java/org/apache/cassandra/db/index/SecondaryIndex.java b/src/java/org/apache/cassandra/db/index/SecondaryIndex.java
index 9221090bb1..430211215f 100644
--- a/src/java/org/apache/cassandra/db/index/SecondaryIndex.java
+++ b/src/java/org/apache/cassandra/db/index/SecondaryIndex.java
@@ -18,9 +18,7 @@
 package org.apache.cassandra.db.index;
 
 import java.nio.ByteBuffer;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.Set;
+import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Future;
 import java.util.concurrent.FutureTask;
@@ -32,7 +30,7 @@ import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.config.CFMetaData;
 import org.apache.cassandra.config.ColumnDefinition;
-import org.apache.cassandra.config.IndexType;
+import org.apache.cassandra.cql3.ColumnIdentifier;
 import org.apache.cassandra.cql3.Operator;
 import org.apache.cassandra.db.*;
 import org.apache.cassandra.db.rows.*;
@@ -47,6 +45,7 @@ import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.apache.cassandra.io.sstable.ReducingKeyIterator;
+import org.apache.cassandra.schema.IndexMetadata;
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.concurrent.Refs;
 
@@ -87,6 +86,8 @@ public abstract class SecondaryIndex
      */
     protected final Set<ColumnDefinition> columnDefs = Collections.newSetFromMap(new ConcurrentHashMap<ColumnDefinition,Boolean>());
 
+    protected IndexMetadata indexMetadata;
+
     /**
      * Perform any initialization work
      */
@@ -101,10 +102,10 @@ public abstract class SecondaryIndex
     public abstract void reload();
 
     /**
-     * Validates the index_options passed in the ColumnDef
+     * Validates the index_options passed in the IndexMetadata
      * @throws ConfigurationException
      */
-    public abstract void validateOptions() throws ConfigurationException;
+    public abstract void validateOptions(CFMetaData baseCfm, IndexMetadata def) throws ConfigurationException;
 
     /**
      * @return The name of the index
@@ -268,6 +269,13 @@ public abstract class SecondaryIndex
         return columnDefs;
     }
 
+    void setIndexMetadata(IndexMetadata indexDef)
+    {
+        this.indexMetadata = indexDef;
+        for (ColumnIdentifier col : indexDef.columns)
+            this.columnDefs.add(baseCfs.metadata.getColumnDefinition(col));
+    }
+
     void addColumnDef(ColumnDefinition columnDef)
     {
        columnDefs.add(columnDef);
@@ -312,45 +320,56 @@ public abstract class SecondaryIndex
      * It will validate the index_options before initializing.
      *
      * @param baseCfs the source of data for the Index
-     * @param cdef the meta information about this column (index_type, index_options, name, etc...)
+     * @param indexDef the meta information about this index (index_type, index_options, name, etc...)
      *
      * @return The secondary index instance for this column
      * @throws ConfigurationException
      */
-    public static SecondaryIndex createInstance(ColumnFamilyStore baseCfs, ColumnDefinition cdef) throws ConfigurationException
+    public static SecondaryIndex createInstance(ColumnFamilyStore baseCfs,
+                                                IndexMetadata indexDef) throws ConfigurationException
     {
-        SecondaryIndex index;
+        SecondaryIndex index = uninitializedInstance(baseCfs.metadata, indexDef);
+        index.validateOptions(baseCfs.metadata, indexDef);
+        index.setBaseCfs(baseCfs);
+        index.setIndexMetadata(indexDef);
 
-        switch (cdef.getIndexType())
+        return index;
+    }
+
+    public static void validate(CFMetaData baseMetadata,
+                                IndexMetadata indexDef) throws ConfigurationException
+    {
+        SecondaryIndex index = uninitializedInstance(baseMetadata, indexDef);
+        index.validateOptions(baseMetadata, indexDef);
+    }
+
+    private static SecondaryIndex uninitializedInstance(CFMetaData baseMetadata,
+                                                        IndexMetadata indexDef) throws ConfigurationException
+    {
+        if (indexDef.isKeys())
         {
-        case KEYS:
-            index = new KeysIndex();
-            break;
-        case COMPOSITES:
-            index = CompositesIndex.create(cdef);
-            break;
-        case CUSTOM:
-            assert cdef.getIndexOptions() != null;
-            String class_name = cdef.getIndexOptions().get(CUSTOM_INDEX_OPTION_NAME);
+            return new KeysIndex();
+        }
+        else if (indexDef.isComposites())
+        {
+            return CompositesIndex.create(indexDef, baseMetadata);
+        }
+        else if (indexDef.isCustom())
+        {
+            assert indexDef.options != null;
+            String class_name = indexDef.options.get(CUSTOM_INDEX_OPTION_NAME);
             assert class_name != null;
             try
             {
-                index = (SecondaryIndex) Class.forName(class_name).newInstance();
+                return (SecondaryIndex) Class.forName(class_name).newInstance();
             }
             catch (Exception e)
             {
                 throw new RuntimeException(e);
             }
-            break;
-            default:
-                throw new RuntimeException("Unknown index type: " + cdef.getIndexName());
         }
 
-        index.addColumnDef(cdef);
-        index.validateOptions();
-        index.setBaseCfs(baseCfs);
-
-        return index;
+        throw new AssertionError("Unknown index type: " + indexDef.name);
     }
 
     public abstract void validate(DecoratedKey partitionKey) throws InvalidRequestException;
@@ -372,32 +391,31 @@ public abstract class SecondaryIndex
     /**
      * Create the index metadata for the index on a given column of a given table.
      */
-    public static CFMetaData newIndexMetadata(CFMetaData baseMetadata, ColumnDefinition def)
+    public static CFMetaData newIndexMetadata(CFMetaData baseMetadata, IndexMetadata def)
     {
-        return newIndexMetadata(baseMetadata, def, def.type);
+        return newIndexMetadata(baseMetadata, def, def.indexedColumn(baseMetadata).type);
     }
 
     /**
      * Create the index metadata for the index on a given column of a given table.
      */
-    static CFMetaData newIndexMetadata(CFMetaData baseMetadata, ColumnDefinition def, AbstractType<?> comparator)
+    static CFMetaData newIndexMetadata(CFMetaData baseMetadata, IndexMetadata def, AbstractType<?> comparator)
     {
-        if (def.getIndexType() == IndexType.CUSTOM)
-            return null;
+        assert !def.isCustom();
 
         CFMetaData.Builder builder = CFMetaData.Builder.create(baseMetadata.ksName, baseMetadata.indexColumnFamilyName(def))
                                                        .withId(baseMetadata.cfId)
                                                        .withPartitioner(new LocalPartitioner(comparator))
-                                                       .addPartitionKey(def.name, def.type);
+                                                       .addPartitionKey(def.indexedColumn(baseMetadata).name, comparator);
 
-        if (def.getIndexType() == IndexType.COMPOSITES)
+        if (def.isComposites())
         {
             CompositesIndex.addIndexClusteringColumns(builder, baseMetadata, def);
         }
         else
         {
-            assert def.getIndexType() == IndexType.KEYS;
-            KeysIndex.addIndexClusteringColumns(builder, baseMetadata, def);
+            assert def.isKeys();
+            KeysIndex.addIndexClusteringColumns(builder, baseMetadata);
         }
 
         return builder.build().reloadIndexMetadataProperties(baseMetadata);
diff --git a/src/java/org/apache/cassandra/db/index/SecondaryIndexManager.java b/src/java/org/apache/cassandra/db/index/SecondaryIndexManager.java
index aaefc9c9f2..996c7301a0 100644
--- a/src/java/org/apache/cassandra/db/index/SecondaryIndexManager.java
+++ b/src/java/org/apache/cassandra/db/index/SecondaryIndexManager.java
@@ -30,7 +30,6 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.config.ColumnDefinition;
-import org.apache.cassandra.config.IndexType;
 import org.apache.cassandra.db.*;
 import org.apache.cassandra.db.rows.*;
 import org.apache.cassandra.db.filter.RowFilter;
@@ -39,6 +38,7 @@ import org.apache.cassandra.db.compaction.CompactionManager;
 import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.apache.cassandra.io.sstable.ReducingKeyIterator;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
+import org.apache.cassandra.schema.IndexMetadata;
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.concurrent.OpOrder;
 
@@ -107,14 +107,16 @@ public class SecondaryIndexManager
         for (ByteBuffer indexedColumn : indexedColumnNames)
         {
             ColumnDefinition def = baseCfs.metadata.getColumnDefinition(indexedColumn);
-            if (def == null || def.getIndexType() == null)
+            if (def == null || !baseCfs.metadata.getIndexes().get(def).isPresent())
                 removeIndexedColumn(indexedColumn);
         }
 
         // TODO: allow all ColumnDefinition type
-        for (ColumnDefinition cdef : baseCfs.metadata.allColumns())
-            if (cdef.getIndexType() != null && !indexedColumnNames.contains(cdef.name.bytes))
-                addIndexedColumn(cdef);
+        for (IndexMetadata indexDef : baseCfs.metadata.getIndexes())
+        {
+            if (!indexedColumnNames.contains(indexDef.indexedColumn(baseCfs.metadata).name.bytes))
+                addIndexedColumn(indexDef);
+        }
 
         for (SecondaryIndex index : allIndexes)
             index.reload();
@@ -230,17 +232,16 @@ public class SecondaryIndexManager
 
     /**
      * Adds and builds a index for a column
-     * @param cdef the column definition holding the index data
+     * @param indexDef the index metadata
      * @return a future which the caller can optionally block on signaling the index is built
      */
-    public synchronized Future<?> addIndexedColumn(ColumnDefinition cdef)
+    public synchronized Future<?> addIndexedColumn(IndexMetadata indexDef)
     {
+        ColumnDefinition cdef = indexDef.indexedColumn(baseCfs.metadata);
         if (indexesByColumn.containsKey(cdef.name.bytes))
             return null;
 
-        assert cdef.getIndexType() != null;
-
-        SecondaryIndex index = SecondaryIndex.createInstance(baseCfs, cdef);
+        SecondaryIndex index = SecondaryIndex.createInstance(baseCfs, indexDef);
 
         // Keep a single instance of the index per-cf for row level indexes
         // since we want all columns to be under the index
@@ -256,14 +257,14 @@ public class SecondaryIndexManager
             else
             {
                 index = currentIndex;
-                index.addColumnDef(cdef);
-                logger.info("Creating new index : {}",cdef);
+                index.setIndexMetadata(indexDef);
+                logger.info("Creating new index : {}",indexDef.name);
             }
         }
         else
         {
             // TODO: We sould do better than throw a RuntimeException
-            if (cdef.getIndexType() == IndexType.CUSTOM && index instanceof AbstractSimplePerColumnSecondaryIndex)
+            if (indexDef.isCustom() && index instanceof AbstractSimplePerColumnSecondaryIndex)
                 throw new RuntimeException("Cannot use a subclass of AbstractSimplePerColumnSecondaryIndex as a CUSTOM index, as they assume they are CFS backed");
             index.init();
         }
diff --git a/src/java/org/apache/cassandra/db/index/composites/CompositesIndex.java b/src/java/org/apache/cassandra/db/index/composites/CompositesIndex.java
index df45ea22ed..abc534444b 100644
--- a/src/java/org/apache/cassandra/db/index/composites/CompositesIndex.java
+++ b/src/java/org/apache/cassandra/db/index/composites/CompositesIndex.java
@@ -22,6 +22,7 @@ import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
 
+import org.apache.cassandra.schema.IndexMetadata;
 import org.apache.cassandra.utils.concurrent.OpOrder;
 import org.apache.cassandra.config.CFMetaData;
 import org.apache.cassandra.config.ColumnDefinition;
@@ -40,8 +41,9 @@ import org.apache.cassandra.exceptions.ConfigurationException;
  */
 public abstract class CompositesIndex extends AbstractSimplePerColumnSecondaryIndex
 {
-    public static CompositesIndex create(ColumnDefinition cfDef)
+    public static CompositesIndex create(IndexMetadata indexDef, CFMetaData baseMetadata)
     {
+        ColumnDefinition cfDef = indexDef.indexedColumn(baseMetadata);
         if (cfDef.type.isCollection() && cfDef.type.isMultiCell())
         {
             switch (((CollectionType)cfDef.type).kind)
@@ -51,9 +53,9 @@ public abstract class CompositesIndex extends AbstractSimplePerColumnSecondaryIn
                 case SET:
                     return new CompositesIndexOnCollectionKey();
                 case MAP:
-                    if (cfDef.hasIndexOption(SecondaryIndex.INDEX_KEYS_OPTION_NAME))
+                    if (indexDef.options.containsKey(SecondaryIndex.INDEX_KEYS_OPTION_NAME))
                         return new CompositesIndexOnCollectionKey();
-                    else if (cfDef.hasIndexOption(SecondaryIndex.INDEX_ENTRIES_OPTION_NAME))
+                    else if (indexDef.options.containsKey(SecondaryIndex.INDEX_ENTRIES_OPTION_NAME))
                         return new CompositesIndexOnCollectionKeyAndValue();
                     else
                         return new CompositesIndexOnCollectionValue();
@@ -74,13 +76,14 @@ public abstract class CompositesIndex extends AbstractSimplePerColumnSecondaryIn
         throw new AssertionError();
     }
 
-    public static void addIndexClusteringColumns(CFMetaData.Builder indexMetadata, CFMetaData baseMetadata, ColumnDefinition cfDef)
+    public static void addIndexClusteringColumns(CFMetaData.Builder indexMetadata, CFMetaData baseMetadata, IndexMetadata indexDef)
     {
+        ColumnDefinition cfDef = indexDef.indexedColumn(baseMetadata);
         if (cfDef.type.isCollection() && cfDef.type.isMultiCell())
         {
             CollectionType type = (CollectionType)cfDef.type;
             if (type.kind == CollectionType.Kind.LIST
-                || (type.kind == CollectionType.Kind.MAP && cfDef.hasIndexOption(SecondaryIndex.INDEX_VALUES_OPTION_NAME)))
+                || (type.kind == CollectionType.Kind.MAP && indexDef.options.containsKey(SecondaryIndex.INDEX_VALUES_OPTION_NAME)))
             {
                 CompositesIndexOnCollectionValue.addClusteringColumns(indexMetadata, baseMetadata, cfDef);
             }
@@ -125,10 +128,10 @@ public abstract class CompositesIndex extends AbstractSimplePerColumnSecondaryIn
         return new CompositesSearcher(baseCfs.indexManager, columns);
     }
 
-    public void validateOptions() throws ConfigurationException
+    public void validateOptions(CFMetaData baseCfm, IndexMetadata indexMetadata) throws ConfigurationException
     {
-        ColumnDefinition columnDef = columnDefs.iterator().next();
-        Map<String, String> options = new HashMap<String, String>(columnDef.getIndexOptions());
+        ColumnDefinition columnDef = indexMetadata.indexedColumn(baseCfm);
+        Map<String, String> options = new HashMap<String, String>(indexMetadata.options);
 
         // We used to have an option called "prefix_size" so skip it silently for backward compatibility sake.
         options.remove("prefix_size");
diff --git a/src/java/org/apache/cassandra/db/index/keys/KeysIndex.java b/src/java/org/apache/cassandra/db/index/keys/KeysIndex.java
index 478559a2df..384b29d8f1 100644
--- a/src/java/org/apache/cassandra/db/index/keys/KeysIndex.java
+++ b/src/java/org/apache/cassandra/db/index/keys/KeysIndex.java
@@ -27,6 +27,7 @@ import org.apache.cassandra.db.rows.*;
 import org.apache.cassandra.db.index.AbstractSimplePerColumnSecondaryIndex;
 import org.apache.cassandra.db.index.SecondaryIndexSearcher;
 import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.schema.IndexMetadata;
 import org.apache.cassandra.utils.concurrent.OpOrder;
 
 /**
@@ -39,7 +40,7 @@ import org.apache.cassandra.utils.concurrent.OpOrder;
  */
 public class KeysIndex extends AbstractSimplePerColumnSecondaryIndex
 {
-    public static void addIndexClusteringColumns(CFMetaData.Builder indexMetadata, CFMetaData baseMetadata, ColumnDefinition cfDef)
+    public static void addIndexClusteringColumns(CFMetaData.Builder indexMetadata, CFMetaData baseMetadata)
     {
         indexMetadata.addClusteringColumn("partition_key", baseMetadata.partitioner.partitionOrdering());
     }
@@ -82,7 +83,7 @@ public class KeysIndex extends AbstractSimplePerColumnSecondaryIndex
         return new KeysSearcher(baseCfs.indexManager, columns);
     }
 
-    public void validateOptions() throws ConfigurationException
+    public void validateOptions(CFMetaData baseCfm, IndexMetadata def) throws ConfigurationException
     {
         // no options used
     }
diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordReader.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordReader.java
index 8373a2ba2a..9ca657d08c 100644
--- a/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordReader.java
+++ b/src/java/org/apache/cassandra/hadoop/cql3/CqlRecordReader.java
@@ -27,6 +27,8 @@ import java.util.*;
 import com.google.common.base.Function;
 import com.google.common.base.Joiner;
 import com.google.common.base.Splitter;
+
+import com.datastax.driver.core.TypeCodec;
 import org.apache.cassandra.utils.AbstractIterator;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Maps;
@@ -339,12 +341,48 @@ public class CqlRecordReader extends RecordReader<Long, Row>
             return row.getObject(i);
         }
 
+        @Override
+        public <T> T get(int i, Class<T> aClass)
+        {
+            return row.get(i, aClass);
+        }
+
+        @Override
+        public <T> T get(int i, TypeToken<T> typeToken)
+        {
+            return row.get(i, typeToken);
+        }
+
+        @Override
+        public <T> T get(int i, TypeCodec<T> typeCodec)
+        {
+            return row.get(i, typeCodec);
+        }
+
         @Override
         public Object getObject(String s)
         {
             return row.getObject(s);
         }
 
+        @Override
+        public <T> T get(String s, Class<T> aClass)
+        {
+            return row.get(s, aClass);
+        }
+
+        @Override
+        public <T> T get(String s, TypeToken<T> typeToken)
+        {
+            return row.get(s, typeToken);
+        }
+
+        @Override
+        public <T> T get(String s, TypeCodec<T> typeCodec)
+        {
+            return row.get(s, typeCodec);
+        }
+
         @Override
         public boolean getBool(int i)
         {
diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
index 5502669c43..59085942fb 100644
--- a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
@@ -56,6 +56,7 @@ import org.apache.cassandra.io.util.*;
 import org.apache.cassandra.metrics.RestorableMeter;
 import org.apache.cassandra.metrics.StorageMetrics;
 import org.apache.cassandra.schema.CachingParams;
+import org.apache.cassandra.schema.IndexMetadata;
 import org.apache.cassandra.service.ActiveRepairService;
 import org.apache.cassandra.service.CacheService;
 import org.apache.cassandra.utils.*;
@@ -335,8 +336,11 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted<SS
         {
             int i = descriptor.cfname.indexOf(SECONDARY_INDEX_NAME_SEPARATOR);
             String parentName = descriptor.cfname.substring(0, i);
+            String indexName = descriptor.cfname.substring(i + 1);
             CFMetaData parent = Schema.instance.getCFMetaData(descriptor.ksname, parentName);
-            ColumnDefinition def = parent.getColumnDefinitionForIndex(descriptor.cfname.substring(i + 1));
+            IndexMetadata def = parent.getIndexes()
+                                      .get(indexName)
+                                      .orElseThrow(() -> new AssertionError("Could not find index metadata for index cf " + i));
             metadata = SecondaryIndex.newIndexMetadata(parent, def);
         }
         else
diff --git a/src/java/org/apache/cassandra/schema/IndexMetadata.java b/src/java/org/apache/cassandra/schema/IndexMetadata.java
new file mode 100644
index 0000000000..af07ceea49
--- /dev/null
+++ b/src/java/org/apache/cassandra/schema/IndexMetadata.java
@@ -0,0 +1,167 @@
+/*
+ * 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.schema;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.base.Objects;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+
+import org.apache.commons.lang3.builder.ToStringBuilder;
+
+import org.apache.cassandra.config.CFMetaData;
+import org.apache.cassandra.config.ColumnDefinition;
+import org.apache.cassandra.cql3.ColumnIdentifier;
+import org.apache.cassandra.db.index.SecondaryIndex;
+import org.apache.cassandra.exceptions.ConfigurationException;
+
+/**
+ * An immutable representation of secondary index metadata.
+ */
+public final class IndexMetadata
+{
+    public enum IndexType
+    {
+        KEYS, CUSTOM, COMPOSITES
+    }
+
+    public enum TargetType
+    {
+        COLUMN, ROW
+    }
+
+    public final String name;
+    public final IndexType indexType;
+    public final TargetType targetType;
+    public final Map<String, String> options;
+    public final Set<ColumnIdentifier> columns;
+
+    private IndexMetadata(String name,
+                          Map<String, String> options,
+                          IndexType indexType,
+                          TargetType targetType,
+                          Set<ColumnIdentifier> columns)
+    {
+        this.name = name;
+        this.options = options == null ? ImmutableMap.of() : ImmutableMap.copyOf(options);
+        this.indexType = indexType;
+        this.targetType = targetType;
+        this.columns = columns == null ? ImmutableSet.of() : ImmutableSet.copyOf(columns);
+    }
+
+    public static IndexMetadata legacyIndex(ColumnIdentifier column,
+                                              String name,
+                                              IndexType type,
+                                              Map<String, String> options)
+    {
+        return new IndexMetadata(name, options, type, TargetType.COLUMN, Collections.singleton(column));
+    }
+
+    public static IndexMetadata legacyIndex(ColumnDefinition column,
+                                              String name,
+                                              IndexType type,
+                                              Map<String, String> options)
+    {
+        return legacyIndex(column.name, name, type, options);
+    }
+
+    public static boolean isNameValid(String name)
+    {
+        return name != null && !name.isEmpty() && name.matches("\\w+");
+    }
+
+    // these will go away as part of #9459 as we enable real per-row indexes
+    public static String getDefaultIndexName(String cfName, ColumnIdentifier columnName)
+    {
+        return (cfName + "_" + columnName + "_idx").replaceAll("\\W", "");
+    }
+
+    public void validate()
+    {
+        if (!isNameValid(name))
+            throw new ConfigurationException("Illegal index name " + name);
+
+        if (indexType == null)
+            throw new ConfigurationException("Index type is null for index " + name);
+
+        if (targetType == null)
+            throw new ConfigurationException("Target type is null for index " + name);
+
+        if (indexType == IndexMetadata.IndexType.CUSTOM)
+            if (options == null || !options.containsKey(SecondaryIndex.CUSTOM_INDEX_OPTION_NAME))
+                throw new ConfigurationException(String.format("Required option missing for index %s : %s",
+                                                               name, SecondaryIndex.CUSTOM_INDEX_OPTION_NAME));
+    }
+
+    public ColumnDefinition indexedColumn(CFMetaData cfm)
+    {
+       return cfm.getColumnDefinition(columns.iterator().next());
+    }
+
+    public boolean isCustom()
+    {
+        return indexType == IndexType.CUSTOM;
+    }
+
+    public boolean isKeys()
+    {
+        return indexType == IndexType.KEYS;
+    }
+
+    public boolean isComposites()
+    {
+        return indexType == IndexType.COMPOSITES;
+    }
+
+    public int hashCode()
+    {
+        return Objects.hashCode(name, indexType, targetType, options, columns);
+    }
+
+    public boolean equals(Object obj)
+    {
+        if (obj == this)
+            return true;
+
+        if (!(obj instanceof IndexMetadata))
+            return false;
+
+        IndexMetadata other = (IndexMetadata)obj;
+
+        return Objects.equal(name, other.name)
+            && Objects.equal(indexType, other.indexType)
+            && Objects.equal(targetType, other.targetType)
+            && Objects.equal(options, other.options)
+            && Objects.equal(columns, other.columns);
+    }
+
+    public String toString()
+    {
+        return new ToStringBuilder(this)
+            .append("name", name)
+            .append("indexType", indexType)
+            .append("targetType", targetType)
+            .append("columns", columns)
+            .append("options", options)
+            .build();
+    }
+}
diff --git a/src/java/org/apache/cassandra/schema/Indexes.java b/src/java/org/apache/cassandra/schema/Indexes.java
new file mode 100644
index 0000000000..7c930b30e6
--- /dev/null
+++ b/src/java/org/apache/cassandra/schema/Indexes.java
@@ -0,0 +1,208 @@
+/*
+ * 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.schema;
+
+import java.util.*;
+
+import com.google.common.collect.ImmutableMap;
+
+import org.apache.cassandra.config.ColumnDefinition;
+import org.apache.cassandra.config.Schema;
+import org.apache.cassandra.cql3.ColumnIdentifier;
+
+import static com.google.common.collect.Iterables.filter;
+
+/**
+ * For backwards compatibility, in the first instance an IndexMetadata must have
+ * TargetType.COLUMN and its Set of target columns must contain only a single
+ * ColumnIdentifier. Hence, this is what is enforced by the public factory methods
+ * on IndexMetadata.
+ * These constraints, along with the internal datastructures here will be relaxed as
+ * support is added for multiple target columns per-index and for indexes with
+ * TargetType.ROW
+ */
+public class Indexes implements Iterable<IndexMetadata>
+{
+    // lookup for index by target column
+    private final ImmutableMap<ColumnIdentifier, IndexMetadata> indexes;
+
+    private Indexes(Builder builder)
+    {
+        ImmutableMap.Builder<ColumnIdentifier, IndexMetadata> internalBuilder = ImmutableMap.builder();
+        builder.indexes.build()
+                       .values()
+                       .stream()
+                       .forEach(def -> internalBuilder.put(def.columns.iterator().next(), def));
+        indexes = internalBuilder.build();
+    }
+
+    public static Builder builder()
+    {
+        return new Builder();
+    }
+
+    public static Indexes none()
+    {
+        return builder().build();
+    }
+
+    public Iterator<IndexMetadata> iterator()
+    {
+        return indexes.values().iterator();
+    }
+
+    public int size()
+    {
+        return indexes.size();
+    }
+
+    public boolean isEmpty()
+    {
+        return indexes.isEmpty();
+    }
+
+    /**
+     * Get the index with the specified name
+     *
+     * @param name a non-qualified index name
+     * @return an empty {@link Optional} if the named index is not found; a non-empty optional of {@link IndexMetadata} otherwise
+     */
+    public Optional<IndexMetadata> get(String name)
+    {
+        return indexes.values().stream().filter(def -> def.name.equals(name)).findFirst();
+    }
+
+    /**
+     * Answer true if contains an index with the specified name.
+     * @param name a non-qualified index name.
+     * @return true if the named index is found; false otherwise
+     */
+    public boolean has(String name)
+    {
+        return get(name).isPresent();
+    }
+
+    /**
+     * Get the index associated with the specified column. This may be removed or modified as support is added
+     * for indexes with multiple target columns and with TargetType.ROW
+     *
+     * @param column a column definition for which an {@link IndexMetadata} is being sought
+     * @return an empty {@link Optional} if the named index is not found; a non-empty optional of {@link IndexMetadata} otherwise
+     */
+    public Optional<IndexMetadata> get(ColumnDefinition column)
+    {
+        return Optional.ofNullable(indexes.get(column.name));
+    }
+
+    /**
+     * Answer true if an index is associated with the specified column.
+     * @param column
+     * @return
+     */
+    public boolean hasIndexFor(ColumnDefinition column)
+    {
+        return indexes.get(column.name) != null;
+    }
+
+    /**
+     * Create a SecondaryIndexes instance with the provided index added
+     */
+    public Indexes with(IndexMetadata index)
+    {
+        if (get(index.name).isPresent())
+            throw new IllegalStateException(String.format("Index %s already exists", index.name));
+
+        return builder().add(this).add(index).build();
+    }
+
+    /**
+     * Creates a SecondaryIndexes instance with the index with the provided name removed
+     */
+    public Indexes without(String name)
+    {
+        IndexMetadata index = get(name).orElseThrow(() -> new IllegalStateException(String.format("Index %s doesn't exist", name)));
+        return builder().add(filter(this, v -> v != index)).build();
+    }
+
+    /**
+     * Creates a SecondaryIndexes instance which contains an updated index definition
+     */
+    public Indexes replace(IndexMetadata index)
+    {
+        return without(index.name).with(index);
+    }
+
+    @Override
+    public boolean equals(Object o)
+    {
+        return this == o || (o instanceof Indexes && indexes.equals(((Indexes) o).indexes));
+    }
+
+    @Override
+    public int hashCode()
+    {
+        return indexes.hashCode();
+    }
+
+    @Override
+    public String toString()
+    {
+        return indexes.values().toString();
+    }
+
+    public static String getAvailableIndexName(String ksName, String cfName, ColumnIdentifier columnName)
+    {
+
+        KeyspaceMetadata ksm = Schema.instance.getKSMetaData(ksName);
+        Set<String> existingNames = ksm == null ? new HashSet<>() : ksm.existingIndexNames(null);
+        String baseName = IndexMetadata.getDefaultIndexName(cfName, columnName);
+        String acceptedName = baseName;
+        int i = 0;
+        while (existingNames.contains(acceptedName))
+            acceptedName = baseName + '_' + (++i);
+
+        return acceptedName;
+    }
+
+    public static final class Builder
+    {
+        final ImmutableMap.Builder<String, IndexMetadata> indexes = new ImmutableMap.Builder<>();
+
+        private Builder()
+        {
+        }
+
+        public Indexes build()
+        {
+            return new Indexes(this);
+        }
+
+        public Builder add(IndexMetadata index)
+        {
+            indexes.put(index.name, index);
+            return this;
+        }
+
+        public Builder add(Iterable<IndexMetadata> indexes)
+        {
+            indexes.forEach(this::add);
+            return this;
+        }
+    }
+}
diff --git a/src/java/org/apache/cassandra/schema/KeyspaceMetadata.java b/src/java/org/apache/cassandra/schema/KeyspaceMetadata.java
index 0ce22fbf9b..372ff6ede4 100644
--- a/src/java/org/apache/cassandra/schema/KeyspaceMetadata.java
+++ b/src/java/org/apache/cassandra/schema/KeyspaceMetadata.java
@@ -17,6 +17,10 @@
  */
 package org.apache.cassandra.schema;
 
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+
 import com.google.common.base.Objects;
 
 import org.apache.cassandra.config.CFMetaData;
@@ -78,6 +82,25 @@ public final class KeyspaceMetadata
         return new KeyspaceMetadata(name, params, tables, types, functions);
     }
 
+    public Set<String> existingIndexNames(String cfToExclude)
+    {
+        Set<String> indexNames = new HashSet<>();
+        for (CFMetaData table : tables)
+            if (cfToExclude == null || !table.cfName.equals(cfToExclude))
+                for (IndexMetadata index : table.getIndexes())
+                    indexNames.add(index.name);
+        return indexNames;
+    }
+
+    public Optional<CFMetaData> findIndexedTable(String indexName)
+    {
+        for (CFMetaData cfm : tables)
+            if (cfm.getIndexes().has(indexName))
+                return Optional.of(cfm);
+
+        return Optional.empty();
+    }
+
     @Override
     public int hashCode()
     {
diff --git a/src/java/org/apache/cassandra/schema/LegacySchemaMigrator.java b/src/java/org/apache/cassandra/schema/LegacySchemaMigrator.java
index 8dac03b1df..43f3dc39c7 100644
--- a/src/java/org/apache/cassandra/schema/LegacySchemaMigrator.java
+++ b/src/java/org/apache/cassandra/schema/LegacySchemaMigrator.java
@@ -121,7 +121,6 @@ public final class LegacySchemaMigrator
     private static void storeKeyspaceInNewSchemaTables(Keyspace keyspace)
     {
         Mutation mutation = SchemaKeyspace.makeCreateKeyspaceMutation(keyspace.name, keyspace.params, keyspace.timestamp);
-
         for (Table table : keyspace.tables)
             SchemaKeyspace.addTableToSchemaMutation(table.metadata, table.timestamp, true, mutation);
 
@@ -293,6 +292,16 @@ public final class LegacySchemaMigrator
                                                                         isStaticCompactTable,
                                                                         needsUpgrade);
 
+        Indexes indexes = createIndexesFromColumnRows(columnRows,
+                                                      ksName,
+                                                      cfName,
+                                                      rawComparator,
+                                                      subComparator,
+                                                      isSuper,
+                                                      isCQLTable,
+                                                      isStaticCompactTable,
+                                                      needsUpgrade);
+
         if (needsUpgrade)
         {
             addDefinitionForUpgrade(columnDefs,
@@ -315,6 +324,7 @@ public final class LegacySchemaMigrator
                                            false, // legacy schema did not contain views
                                            columnDefs,
                                            DatabaseDescriptor.getPartitioner());
+        cfm.indexes(indexes);
 
         if (tableRow.has("dropped_columns"))
             addDroppedColumns(cfm, rawComparator, tableRow.getMap("dropped_columns", UTF8Type.instance, LongType.instance));
@@ -530,29 +540,73 @@ public final class LegacySchemaMigrator
             if (isEmptyCompactValueColumn(row))
                 continue;
 
-            ColumnDefinition.Kind kind = deserializeKind(row.getString("type"));
-            if (needsUpgrade && isStaticCompactTable && kind == ColumnDefinition.Kind.REGULAR)
-                kind = ColumnDefinition.Kind.STATIC;
+            columns.add(createColumnFromColumnRow(row,
+                                                  keyspace,
+                                                  table,
+                                                  rawComparator,
+                                                  rawSubComparator,
+                                                  isSuper,
+                                                  isCQLTable,
+                                                  isStaticCompactTable,
+                                                  needsUpgrade));
+        }
 
-            Integer componentIndex = null;
-            // Note that the component_index is not useful for non-primary key parts (it never really in fact since there is
-            // no particular ordering of non-PK columns, we only used to use it as a simplification but that's not needed
-            // anymore)
-            if (kind.isPrimaryKeyKind() && row.has("component_index"))
-                componentIndex = row.getInt("component_index");
+        return columns;
+    }
 
-            // Note: we save the column name as string, but we should not assume that it is an UTF8 name, we
-            // we need to use the comparator fromString method
-            AbstractType<?> comparator = isCQLTable
-                                       ? UTF8Type.instance
-                                       : CompactTables.columnDefinitionComparator(kind, isSuper, rawComparator, rawSubComparator);
-            ColumnIdentifier name = ColumnIdentifier.getInterned(comparator.fromString(row.getString("column_name")), comparator);
+    private static ColumnDefinition createColumnFromColumnRow(UntypedResultSet.Row row,
+                                                              String keyspace,
+                                                              String table,
+                                                              AbstractType<?> rawComparator,
+                                                              AbstractType<?> rawSubComparator,
+                                                              boolean isSuper,
+                                                              boolean isCQLTable,
+                                                              boolean isStaticCompactTable,
+                                                              boolean needsUpgrade)
+    {
+        ColumnDefinition.Kind kind = deserializeKind(row.getString("type"));
+        if (needsUpgrade && isStaticCompactTable && kind == ColumnDefinition.Kind.REGULAR)
+            kind = ColumnDefinition.Kind.STATIC;
 
-            AbstractType<?> validator = parseType(row.getString("validator"));
+        Integer componentIndex = null;
+        // Note that the component_index is not useful for non-primary key parts (it never really in fact since there is
+        // no particular ordering of non-PK columns, we only used to use it as a simplification but that's not needed
+        // anymore)
+        if (kind.isPrimaryKeyKind() && row.has("component_index"))
+            componentIndex = row.getInt("component_index");
 
-            IndexType indexType = null;
+        // Note: we save the column name as string, but we should not assume that it is an UTF8 name, we
+        // we need to use the comparator fromString method
+        AbstractType<?> comparator = isCQLTable
+                                     ? UTF8Type.instance
+                                     : CompactTables.columnDefinitionComparator(kind, isSuper, rawComparator, rawSubComparator);
+        ColumnIdentifier name = ColumnIdentifier.getInterned(comparator.fromString(row.getString("column_name")), comparator);
+
+        AbstractType<?> validator = parseType(row.getString("validator"));
+
+        return new ColumnDefinition(keyspace, table, name, validator, componentIndex, kind);
+    }
+
+    private static Indexes createIndexesFromColumnRows(UntypedResultSet rows,
+                                                       String keyspace,
+                                                       String table,
+                                                       AbstractType<?> rawComparator,
+                                                       AbstractType<?> rawSubComparator,
+                                                       boolean isSuper,
+                                                       boolean isCQLTable,
+                                                       boolean isStaticCompactTable,
+                                                       boolean needsUpgrade)
+    {
+        Indexes.Builder indexes = Indexes.builder();
+
+        for (UntypedResultSet.Row row : rows)
+        {
+            IndexMetadata.IndexType indexType = null;
             if (row.has("index_type"))
-                indexType = IndexType.valueOf(row.getString("index_type"));
+                indexType = IndexMetadata.IndexType.valueOf(row.getString("index_type"));
+
+            if (indexType == null)
+                continue;
 
             Map<String, String> indexOptions = null;
             if (row.has("index_options"))
@@ -562,10 +616,20 @@ public final class LegacySchemaMigrator
             if (row.has("index_name"))
                 indexName = row.getString("index_name");
 
-            columns.add(new ColumnDefinition(keyspace, table, name, validator, indexType, indexOptions, indexName, componentIndex, kind));
+            ColumnDefinition column = createColumnFromColumnRow(row,
+                                                                keyspace,
+                                                                table,
+                                                                rawComparator,
+                                                                rawSubComparator,
+                                                                isSuper,
+                                                                isCQLTable,
+                                                                isStaticCompactTable,
+                                                                needsUpgrade);
+
+            indexes.add(IndexMetadata.legacyIndex(column, indexName, indexType, indexOptions));
         }
 
-        return columns;
+        return indexes.build();
     }
 
     private static ColumnDefinition.Kind deserializeKind(String kind)
diff --git a/src/java/org/apache/cassandra/schema/SchemaKeyspace.java b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java
index 4f3fe93e20..f77b2bcf3f 100644
--- a/src/java/org/apache/cassandra/schema/SchemaKeyspace.java
+++ b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java
@@ -24,6 +24,7 @@ import java.security.NoSuchAlgorithmException;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
+import java.util.stream.Collectors;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.MapDifference;
@@ -48,8 +49,6 @@ import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.concurrent.OpOrder;
 
 import static org.apache.cassandra.cql3.QueryProcessor.executeOnceInternal;
-import static org.apache.cassandra.utils.FBUtilities.fromJsonMap;
-import static org.apache.cassandra.utils.FBUtilities.json;
 
 /**
  * system_schema.* tables and methods for manipulating them.
@@ -73,9 +72,10 @@ public final class SchemaKeyspace
     public static final String TYPES = "types";
     public static final String FUNCTIONS = "functions";
     public static final String AGGREGATES = "aggregates";
+    public static final String INDEXES = "indexes";
 
     public static final List<String> ALL =
-        ImmutableList.of(KEYSPACES, TABLES, COLUMNS, TRIGGERS, MATERIALIZED_VIEWS, TYPES, FUNCTIONS, AGGREGATES);
+        ImmutableList.of(KEYSPACES, TABLES, COLUMNS, TRIGGERS, MATERIALIZED_VIEWS, TYPES, FUNCTIONS, AGGREGATES, INDEXES);
 
     private static final CFMetaData Keyspaces =
         compile(KEYSPACES,
@@ -119,9 +119,6 @@ public final class SchemaKeyspace
                 + "column_name text,"
                 + "column_name_bytes blob,"
                 + "component_index int,"
-                + "index_name text,"
-                + "index_options text,"
-                + "index_type text,"
                 + "type text,"
                 + "validator text,"
                 + "PRIMARY KEY ((keyspace_name), table_name, column_name))");
@@ -159,6 +156,19 @@ public final class SchemaKeyspace
                 + "included_columns list<text>,"
                 + "PRIMARY KEY ((keyspace_name), table_name, view_name))");
 
+    private static final CFMetaData Indexes =
+        compile(INDEXES,
+                "secondary index definitions",
+                "CREATE TABLE %s ("
+                + "keyspace_name text,"
+                + "table_name text,"
+                + "index_name text,"
+                + "index_type text,"
+                + "options map<text, text>,"
+                + "target_columns set<text>,"
+                + "target_type text,"
+                + "PRIMARY KEY ((keyspace_name), table_name, index_name))");
+
     private static final CFMetaData Types =
         compile(TYPES,
                 "user defined type definitions",
@@ -199,8 +209,8 @@ public final class SchemaKeyspace
                 + "state_type text,"
                 + "PRIMARY KEY ((keyspace_name), aggregate_name, signature))");
 
-    public static final List<CFMetaData> All =
-        ImmutableList.of(Keyspaces, Tables, Columns, Triggers, DroppedColumns, MaterializedViews, Types, Functions, Aggregates);
+    public static final List<CFMetaData> ALL_TABLE_METADATA =
+        ImmutableList.of(Keyspaces, Tables, Columns, Triggers, DroppedColumns, MaterializedViews, Types, Functions, Aggregates, Indexes);
 
     private static CFMetaData compile(String name, String description, String schema)
     {
@@ -211,7 +221,7 @@ public final class SchemaKeyspace
 
     public static KeyspaceMetadata metadata()
     {
-        return KeyspaceMetadata.create(NAME, KeyspaceParams.local(), org.apache.cassandra.schema.Tables.of(All));
+        return KeyspaceMetadata.create(NAME, KeyspaceParams.local(), org.apache.cassandra.schema.Tables.of(ALL_TABLE_METADATA));
     }
 
     /**
@@ -699,7 +709,7 @@ public final class SchemaKeyspace
         int nowInSec = FBUtilities.nowInSeconds();
         Mutation mutation = new Mutation(NAME, Keyspaces.decorateKey(getSchemaKSKey(keyspace.name)));
 
-        for (CFMetaData schemaTable : All)
+        for (CFMetaData schemaTable : ALL_TABLE_METADATA)
             mutation.add(PartitionUpdate.fullPartitionDelete(schemaTable, mutation.key(), timestamp, nowInSec));
 
         return mutation;
@@ -836,6 +846,9 @@ public final class SchemaKeyspace
 
             for (MaterializedViewDefinition materializedView: table.getMaterializedViews())
                 addMaterializedViewToSchemaMutation(table, materializedView, timestamp, mutation);
+
+            for (IndexMetadata index : table.getIndexes())
+                addIndexToSchemaMutation(table, index, timestamp, mutation);
         }
     }
 
@@ -921,17 +934,45 @@ public final class SchemaKeyspace
 
         // newly created materialized views
         for (MaterializedViewDefinition materializedView : materializedViewDiff.entriesOnlyOnRight().values())
-            addMaterializedViewToSchemaMutation(oldTable, materializedView, timestamp, mutation);
+            addMaterializedViewToSchemaMutation(newTable, materializedView, timestamp, mutation);
 
         // updated materialized views need to be updated
         for (MapDifference.ValueDifference<MaterializedViewDefinition> diff : materializedViewDiff.entriesDiffering().values())
         {
-            addUpdatedMaterializedViewDefinitionToSchemaMutation(oldTable, diff.rightValue(), timestamp, mutation);
+            addUpdatedMaterializedViewDefinitionToSchemaMutation(newTable, diff.rightValue(), timestamp, mutation);
+        }
+
+        MapDifference<String, IndexMetadata> indexesDiff = indexesDiff(oldTable.getIndexes(),
+                                                                       newTable.getIndexes());
+
+        // dropped indexes
+        for (IndexMetadata index : indexesDiff.entriesOnlyOnLeft().values())
+            dropIndexFromSchemaMutation(oldTable, index, timestamp, mutation);
+
+        // newly created indexes
+        for (IndexMetadata index : indexesDiff.entriesOnlyOnRight().values())
+            addIndexToSchemaMutation(newTable, index, timestamp, mutation);
+
+        // updated indexes need to be updated
+        for (MapDifference.ValueDifference<IndexMetadata> diff : indexesDiff.entriesDiffering().values())
+        {
+            addUpdatedIndexToSchemaMutation(newTable, diff.rightValue(), timestamp, mutation);
         }
 
         return mutation;
     }
 
+    private static MapDifference<String, IndexMetadata> indexesDiff(Indexes before, Indexes after)
+    {
+        Map<String, IndexMetadata> beforeMap = new HashMap<>();
+        before.forEach(i -> beforeMap.put(i.name, i));
+
+        Map<String, IndexMetadata> afterMap = new HashMap<>();
+        after.forEach(i -> afterMap.put(i.name, i));
+
+        return Maps.difference(beforeMap, afterMap);
+    }
+
     private static MapDifference<String, TriggerMetadata> triggersDiff(Triggers before, Triggers after)
     {
         Map<String, TriggerMetadata> beforeMap = new HashMap<>();
@@ -970,6 +1011,9 @@ public final class SchemaKeyspace
         for (MaterializedViewDefinition materializedView : table.getMaterializedViews())
             dropMaterializedViewFromSchemaMutation(table, materializedView, timestamp, mutation);
 
+        for (IndexMetadata index : table.getIndexes())
+            dropIndexFromSchemaMutation(table, index, timestamp, mutation);
+
         return mutation;
     }
 
@@ -1037,9 +1081,18 @@ public final class SchemaKeyspace
         MaterializedViews views =
             readSchemaPartitionForTableAndApply(MATERIALIZED_VIEWS, keyspace, table, SchemaKeyspace::createMaterializedViewsFromMaterializedViewsPartition);
 
-        return createTableFromTableRowAndColumns(row, columns).droppedColumns(droppedColumns)
-                                                              .triggers(triggers)
-                                                              .materializedViews(views);
+        CFMetaData cfm = createTableFromTableRowAndColumns(row, columns).droppedColumns(droppedColumns)
+                                                                        .triggers(triggers)
+                                                                        .materializedViews(views);
+
+        // the CFMetaData itself is required to build the collection of indexes as
+        // the column definitions are needed because we store only the name each
+        // index's target columns and this is not enough to reconstruct a ColumnIdentifier
+        org.apache.cassandra.schema.Indexes indexes =
+            readSchemaPartitionForTableAndApply(INDEXES, keyspace, table, rowIterator -> createIndexesFromIndexesPartition(cfm, rowIterator));
+        cfm.indexes(indexes);
+
+        return cfm;
     }
 
     public static CFMetaData createTableFromTableRowAndColumns(UntypedResultSet.Row row, List<ColumnDefinition> columns)
@@ -1107,9 +1160,6 @@ public final class SchemaKeyspace
              .add("validator", column.type.toString())
              .add("type", column.kind.toString().toLowerCase())
              .add("component_index", column.isOnAllComponents() ? null : column.position())
-             .add("index_name", column.getIndexName())
-             .add("index_type", column.getIndexType() == null ? null : column.getIndexType().toString())
-             .add("index_options", json(column.getIndexOptions()))
              .build();
     }
 
@@ -1141,19 +1191,7 @@ public final class SchemaKeyspace
 
         AbstractType<?> validator = parseType(row.getString("validator"));
 
-        IndexType indexType = null;
-        if (row.has("index_type"))
-            indexType = IndexType.valueOf(row.getString("index_type"));
-
-        Map<String, String> indexOptions = null;
-        if (row.has("index_options"))
-            indexOptions = fromJsonMap(row.getString("index_options"));
-
-        String indexName = null;
-        if (row.has("index_name"))
-            indexName = row.getString("index_name");
-
-        return new ColumnDefinition(keyspace, table, name, validator, indexType, indexOptions, indexName, componentIndex, kind);
+        return new ColumnDefinition(keyspace, table, name, validator, componentIndex, kind);
     }
 
     /*
@@ -1233,7 +1271,7 @@ public final class SchemaKeyspace
     }
 
     /*
-     * Global Index metadata serialization/deserialization.
+     * Materialized View metadata serialization/deserialization.
      */
 
     private static void addMaterializedViewToSchemaMutation(CFMetaData table, MaterializedViewDefinition materializedView, long timestamp, Mutation mutation)
@@ -1329,6 +1367,100 @@ public final class SchemaKeyspace
                                               includedColumns);
     }
 
+    /*
+     * Secondary Index metadata serialization/deserialization.
+     */
+
+    private static void addIndexToSchemaMutation(CFMetaData table,
+                                                 IndexMetadata index,
+                                                 long timestamp,
+                                                 Mutation mutation)
+    {
+        RowUpdateBuilder builder = new RowUpdateBuilder(Indexes, timestamp, mutation)
+                                   .clustering(table.cfName, index.name);
+
+        builder.add("index_type", index.indexType.toString());
+        builder.map("options", index.options);
+        builder.set("target_columns", index.columns.stream()
+                                                   .map(ColumnIdentifier::toString)
+                                                   .collect(Collectors.toSet()));
+        builder.add("target_type", index.targetType.toString());
+        builder.build();
+    }
+
+    private static void dropIndexFromSchemaMutation(CFMetaData table,
+                                                    IndexMetadata index,
+                                                    long timestamp,
+                                                    Mutation mutation)
+    {
+        RowUpdateBuilder.deleteRow(Indexes, timestamp, mutation, table.cfName, index.name);
+    }
+
+    private static void addUpdatedIndexToSchemaMutation(CFMetaData table,
+                                                        IndexMetadata index,
+                                                        long timestamp,
+                                                        Mutation mutation)
+    {
+        RowUpdateBuilder builder = new RowUpdateBuilder(Indexes, timestamp, mutation).clustering(table.cfName, index.name);
+
+        builder.add("index_type", index.indexType.toString());
+        builder.map("options", index.options);
+        builder.set("target_columns", index.columns.stream().map(ColumnIdentifier::toString).collect(Collectors.toSet()));
+        builder.add("target_type", index.targetType.toString());
+        builder.build();
+    }
+    /**
+     * Deserialize secondary indexes from storage-level representation.
+     *
+     * @param partition storage-level partition containing the index definitions
+     * @return the list of processed IndexMetadata
+     */
+    private static Indexes createIndexesFromIndexesPartition(CFMetaData cfm, RowIterator partition)
+    {
+        Indexes.Builder indexes = org.apache.cassandra.schema.Indexes.builder();
+        String query = String.format("SELECT * FROM %s.%s", NAME, INDEXES);
+        QueryProcessor.resultify(query, partition).forEach(row -> indexes.add(createIndexMetadataFromIndexesRow(cfm, row)));
+        return indexes.build();
+    }
+
+    private static IndexMetadata createIndexMetadataFromIndexesRow(CFMetaData cfm, UntypedResultSet.Row row)
+    {
+        String name = row.getString("index_name");
+        IndexMetadata.IndexType type = IndexMetadata.IndexType.valueOf(row.getString("index_type"));
+        IndexMetadata.TargetType targetType = IndexMetadata.TargetType.valueOf(row.getString("target_type"));
+        Map<String, String> options = row.getTextMap("options");
+        if (options == null)
+            options = Collections.emptyMap();
+
+        Set<String> targetColumnNames = row.getSet("target_columns", UTF8Type.instance);
+        assert targetType == IndexMetadata.TargetType.COLUMN : "Per row indexes with dynamic target columns are not supported yet";
+        assert targetColumnNames.size() == 1 : "Secondary indexes targetting multiple columns are not supported yet";
+
+        Set<ColumnIdentifier> targetColumns = new HashSet<>();
+        // if it's not a CQL table, we can't assume that the column name is utf8, so
+        // in that case we have to do a linear scan of the cfm's columns to get the matching one
+        if (targetColumnNames != null)
+        {
+            targetColumnNames.forEach(targetColumnName -> {
+                if (cfm.isCQLTable())
+                    targetColumns.add(ColumnIdentifier.getInterned(targetColumnName, true));
+                else
+                    findColumnIdentifierWithName(targetColumnName, cfm.allColumns()).ifPresent(targetColumns::add);
+            });
+        }
+        return IndexMetadata.legacyIndex(targetColumns.iterator().next(), name, type, options);
+    }
+
+    private static Optional<ColumnIdentifier> findColumnIdentifierWithName(String name,
+                                                                           Iterable<ColumnDefinition> columns)
+    {
+        for (ColumnDefinition column : columns)
+            if (column.name.toString().equals(name))
+                return Optional.of(column.name);
+
+        return Optional.empty();
+    }
+
     /*
      * UDF metadata serialization/deserialization.
      */
diff --git a/src/java/org/apache/cassandra/thrift/CassandraServer.java b/src/java/org/apache/cassandra/thrift/CassandraServer.java
index cb0667a5ab..7a2126c0f2 100644
--- a/src/java/org/apache/cassandra/thrift/CassandraServer.java
+++ b/src/java/org/apache/cassandra/thrift/CassandraServer.java
@@ -1857,7 +1857,6 @@ public class CassandraServer implements Cassandra.Iface
             cf_def.unsetId(); // explicitly ignore any id set by client (Hector likes to set zero)
             CFMetaData cfm = ThriftConversion.fromThrift(cf_def);
             cfm.params.compaction.validate();
-            cfm.addDefaultIndexNames();
 
             if (!cfm.getTriggers().isEmpty())
                 state().ensureIsSuper("Only superusers are allowed to add triggers.");
@@ -1921,7 +1920,6 @@ public class CassandraServer implements Cassandra.Iface
             {
                 cf_def.unsetId(); // explicitly ignore any id set by client (same as system_add_column_family)
                 CFMetaData cfm = ThriftConversion.fromThrift(cf_def);
-                cfm.addDefaultIndexNames();
 
                 if (!cfm.getTriggers().isEmpty())
                     state().ensureIsSuper("Only superusers are allowed to add triggers.");
@@ -2007,7 +2005,6 @@ public class CassandraServer implements Cassandra.Iface
 
             CFMetaData cfm = ThriftConversion.fromThriftForUpdate(cf_def, oldCfm);
             cfm.params.compaction.validate();
-            cfm.addDefaultIndexNames();
 
             if (!oldCfm.getTriggers().equals(cfm.getTriggers()))
                 state().ensureIsSuper("Only superusers are allowed to add or remove triggers.");
diff --git a/src/java/org/apache/cassandra/thrift/ThriftConversion.java b/src/java/org/apache/cassandra/thrift/ThriftConversion.java
index 1744177c51..a74bcea318 100644
--- a/src/java/org/apache/cassandra/thrift/ThriftConversion.java
+++ b/src/java/org/apache/cassandra/thrift/ThriftConversion.java
@@ -20,6 +20,7 @@ package org.apache.cassandra.thrift;
 import java.util.*;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Strings;
 import com.google.common.collect.Maps;
 
 import org.apache.cassandra.db.compaction.AbstractCompactionStrategy;
@@ -278,7 +279,23 @@ public class ThriftConversion
             // We do not allow Thrift materialized views, so we always set it to false
             boolean isMaterializedView = false;
 
-            CFMetaData newCFMD = CFMetaData.create(cf_def.keyspace, cf_def.name, cfId, isDense, isCompound, isSuper, isCounter, isMaterializedView, defs, DatabaseDescriptor.getPartitioner());
+            CFMetaData newCFMD = CFMetaData.create(cf_def.keyspace,
+                                                   cf_def.name,
+                                                   cfId,
+                                                   isDense,
+                                                   isCompound,
+                                                   isSuper,
+                                                   isCounter,
+                                                   isMaterializedView,
+                                                   defs,
+                                                   DatabaseDescriptor.getPartitioner());
+
+            // Convert any secondary indexes defined in the thrift column_metadata
+            newCFMD.indexes(indexDefsFromThrift(cf_def.keyspace,
+                                                cf_def.name,
+                                                rawComparator,
+                                                subComparator,
+                                                cf_def.column_metadata));
 
             if (cf_def.isSetGc_grace_seconds())
                 newCFMD.gcGraceSeconds(cf_def.gc_grace_seconds);
@@ -492,9 +509,6 @@ public class ThriftConversion
                                     cfName,
                                     ColumnIdentifier.getInterned(ByteBufferUtil.clone(thriftColumnDef.name), comparator),
                                     TypeParser.parse(thriftColumnDef.validation_class),
-                                    thriftColumnDef.index_type == null ? null : org.apache.cassandra.config.IndexType.valueOf(thriftColumnDef.index_type.name()),
-                                    thriftColumnDef.index_options,
-                                    thriftColumnDef.index_name,
                                     null,
                                     kind);
     }
@@ -516,16 +530,58 @@ public class ThriftConversion
         return defs;
     }
 
+    private static Indexes indexDefsFromThrift(String ksName,
+                                               String cfName,
+                                               AbstractType<?> thriftComparator,
+                                               AbstractType<?> thriftSubComparator,
+                                               List<ColumnDef> thriftDefs)
+    {
+        if (thriftDefs == null)
+            return Indexes.none();
+
+        Set<String> indexNames = new HashSet<>();
+        Indexes.Builder indexes = Indexes.builder();
+        for (ColumnDef def : thriftDefs)
+        {
+            if (def.isSetIndex_type())
+            {
+                ColumnDefinition column = fromThrift(ksName, cfName, thriftComparator, thriftSubComparator, def);
+
+                String indexName = def.getIndex_name();
+                // add a generated index name if none was supplied
+                if (Strings.isNullOrEmpty(indexName))
+                    indexName = Indexes.getAvailableIndexName(ksName, cfName, column.name);
+
+                if (indexNames.contains(indexName))
+                    throw new ConfigurationException("Duplicate index name " + indexName);
+
+                indexNames.add(indexName);
+
+                Map<String, String> indexOptions = def.getIndex_options();
+                IndexMetadata.IndexType indexType = IndexMetadata.IndexType.valueOf(def.index_type.name());
+
+                indexes.add(IndexMetadata.legacyIndex(column,
+                                                      indexName,
+                                                      indexType,
+                                                      indexOptions));
+            }
+        }
+        return indexes.build();
+    }
+
     @VisibleForTesting
-    public static ColumnDef toThrift(ColumnDefinition column)
+    public static ColumnDef toThrift(CFMetaData cfMetaData, ColumnDefinition column)
     {
         ColumnDef cd = new ColumnDef();
 
         cd.setName(ByteBufferUtil.clone(column.name.bytes));
         cd.setValidation_class(column.type.toString());
-        cd.setIndex_type(column.getIndexType() == null ? null : org.apache.cassandra.thrift.IndexType.valueOf(column.getIndexType().name()));
-        cd.setIndex_name(column.getIndexName());
-        cd.setIndex_options(column.getIndexOptions() == null ? null : Maps.newHashMap(column.getIndexOptions()));
+        Optional<IndexMetadata> index = cfMetaData.getIndexes().get(column);
+        index.ifPresent(def -> {
+            cd.setIndex_type(org.apache.cassandra.thrift.IndexType.valueOf(def.indexType.name()));
+            cd.setIndex_name(def.name);
+            cd.setIndex_options(def.options == null || def.options.isEmpty() ? null : Maps.newHashMap(def.options));
+        });
 
         return cd;
     }
@@ -535,7 +591,7 @@ public class ThriftConversion
         List<ColumnDef> thriftDefs = new ArrayList<>(columns.size());
         for (ColumnDefinition def : columns)
             if (def.isPartOfCellName(metadata.isCQLTable(), metadata.isSuper()))
-                thriftDefs.add(ThriftConversion.toThrift(def));
+                thriftDefs.add(ThriftConversion.toThrift(metadata, def));
         return thriftDefs;
     }
 
diff --git a/src/java/org/apache/cassandra/utils/NativeSSTableLoaderClient.java b/src/java/org/apache/cassandra/utils/NativeSSTableLoaderClient.java
index fb65e34caa..f86281688c 100644
--- a/src/java/org/apache/cassandra/utils/NativeSSTableLoaderClient.java
+++ b/src/java/org/apache/cassandra/utils/NativeSSTableLoaderClient.java
@@ -158,6 +158,6 @@ public class NativeSSTableLoaderClient extends SSTableLoader.Client
 
         AbstractType<?> validator = TypeParser.parse(row.getString("validator"));
 
-        return new ColumnDefinition(keyspace, table, name, validator, null, null, null, componentIndex, kind);
+        return new ColumnDefinition(keyspace, table, name, validator, componentIndex, kind);
     }
 }
diff --git a/test/unit/org/apache/cassandra/SchemaLoader.java b/test/unit/org/apache/cassandra/SchemaLoader.java
index e914f13de3..1c0dd76ffc 100644
--- a/test/unit/org/apache/cassandra/SchemaLoader.java
+++ b/test/unit/org/apache/cassandra/SchemaLoader.java
@@ -34,13 +34,8 @@ import org.apache.cassandra.db.index.SecondaryIndex;
 import org.apache.cassandra.db.marshal.*;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.gms.Gossiper;
-import org.apache.cassandra.schema.CompressionParams;
+import org.apache.cassandra.schema.*;
 import org.apache.cassandra.io.util.FileUtils;
-import org.apache.cassandra.schema.CachingParams;
-import org.apache.cassandra.schema.CompactionParams;
-import org.apache.cassandra.schema.KeyspaceMetadata;
-import org.apache.cassandra.schema.KeyspaceParams;
-import org.apache.cassandra.schema.Tables;
 import org.apache.cassandra.service.MigrationManager;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.FBUtilities;
@@ -280,9 +275,6 @@ public class SchemaLoader
                                     ColumnIdentifier.getInterned(IntegerType.instance.fromString("42"), IntegerType.instance),
                                     UTF8Type.instance,
                                     null,
-                                    null,
-                                    null,
-                                    null,
                                     ColumnDefinition.Kind.REGULAR);
     }
 
@@ -293,9 +285,6 @@ public class SchemaLoader
                                     ColumnIdentifier.getInterned("fortytwo", true),
                                     UTF8Type.instance,
                                     null,
-                                    null,
-                                    null,
-                                    null,
                                     ColumnDefinition.Kind.REGULAR);
     }
 
@@ -309,8 +298,16 @@ public class SchemaLoader
                 .addPartitionKey("key", AsciiType.instance)
                 .build();
 
-        return cfm.addOrReplaceColumnDefinition(ColumnDefinition.regularDef(ksName, cfName, "indexed", AsciiType.instance)
-                                                                .setIndex("indexe1", IndexType.CUSTOM, indexOptions));
+        ColumnDefinition indexedColumn = ColumnDefinition.regularDef(ksName, cfName, "indexed", AsciiType.instance);
+        cfm.addOrReplaceColumnDefinition(indexedColumn);
+
+        cfm.indexes(
+            cfm.getIndexes()
+               .with(IndexMetadata.legacyIndex(indexedColumn,
+                                               "indexe1",
+                                               IndexMetadata.IndexType.CUSTOM,
+                                               indexOptions)));
+        return cfm;
     }
 
     private static void useCompression(List<KeyspaceMetadata> schema)
@@ -415,8 +412,12 @@ public class SchemaLoader
                 .build();
 
         if (withIndex)
-            cfm.getColumnDefinition(new ColumnIdentifier("birthdate", true))
-               .setIndex("birthdate_key_index", IndexType.COMPOSITES, Collections.EMPTY_MAP);
+            cfm.indexes(
+                cfm.getIndexes()
+                    .with(IndexMetadata.legacyIndex(cfm.getColumnDefinition(new ColumnIdentifier("birthdate", true)),
+                                                    "birthdate_key_index",
+                                                    IndexMetadata.IndexType.COMPOSITES,
+                                                    Collections.EMPTY_MAP)));
 
         return cfm.compression(getCompressionParameters());
     }
@@ -431,8 +432,13 @@ public class SchemaLoader
                                            .build();
 
         if (withIndex)
-            cfm.getColumnDefinition(new ColumnIdentifier("birthdate", true))
-               .setIndex("birthdate_composite_index", IndexType.KEYS, Collections.EMPTY_MAP);
+            cfm.indexes(
+                cfm.getIndexes()
+                    .with(IndexMetadata.legacyIndex(cfm.getColumnDefinition(new ColumnIdentifier("birthdate", true)),
+                                                    "birthdate_composite_index",
+                                                    IndexMetadata.IndexType.KEYS,
+                                                    Collections.EMPTY_MAP)));
+
 
         return cfm.compression(getCompressionParameters());
     }
diff --git a/test/unit/org/apache/cassandra/config/ColumnDefinitionTest.java b/test/unit/org/apache/cassandra/config/ColumnDefinitionTest.java
index c875165ccd..933d2314f4 100644
--- a/test/unit/org/apache/cassandra/config/ColumnDefinitionTest.java
+++ b/test/unit/org/apache/cassandra/config/ColumnDefinitionTest.java
@@ -38,9 +38,7 @@ public class ColumnDefinitionTest
                          .addRegularColumn("val", AsciiType.instance)
                          .build();
 
-        ColumnDefinition cd0 = ColumnDefinition.staticDef(cfm, ByteBufferUtil.bytes("TestColumnDefinitionName0"), BytesType.instance)
-                                               .setIndex("random index name 0", IndexType.KEYS, null);
-
+        ColumnDefinition cd0 = ColumnDefinition.staticDef(cfm, ByteBufferUtil.bytes("TestColumnDefinitionName0"), BytesType.instance);
         ColumnDefinition cd1 = ColumnDefinition.staticDef(cfm, ByteBufferUtil.bytes("TestColumnDefinition1"), LongType.instance);
 
         testSerializeDeserialize(cfm, cd0);
@@ -49,7 +47,7 @@ public class ColumnDefinitionTest
 
     protected void testSerializeDeserialize(CFMetaData cfm, ColumnDefinition cd) throws Exception
     {
-        ColumnDefinition newCd = ThriftConversion.fromThrift(cfm.ksName, cfm.cfName, cfm.comparator.subtype(0), null, ThriftConversion.toThrift(cd));
+        ColumnDefinition newCd = ThriftConversion.fromThrift(cfm.ksName, cfm.cfName, cfm.comparator.subtype(0), null, ThriftConversion.toThrift(cfm, cd));
         Assert.assertNotSame(cd, newCd);
         Assert.assertEquals(cd.hashCode(), newCd.hashCode());
         Assert.assertEquals(cd, newCd);
diff --git a/test/unit/org/apache/cassandra/cql3/CQLTester.java b/test/unit/org/apache/cassandra/cql3/CQLTester.java
index 2bd390fef0..e540ec51ab 100644
--- a/test/unit/org/apache/cassandra/cql3/CQLTester.java
+++ b/test/unit/org/apache/cassandra/cql3/CQLTester.java
@@ -670,6 +670,11 @@ public abstract class CQLTester
 
     protected void assertRowsNet(int protocolVersion, ResultSet result, Object[]... rows)
     {
+        // necessary as we need cluster objects to supply CodecRegistry.
+        // It's reasonably certain that the network setup has already been done
+        // by the time we arrive at this point, but adding this check doesn't hurt
+        requireNetwork();
+
         if (result == null)
         {
             if (rows.length > 0)
@@ -692,19 +697,21 @@ public abstract class CQLTester
             for (int j = 0; j < meta.size(); j++)
             {
                 DataType type = meta.getType(j);
-                ByteBuffer expectedByteValue = type.serialize(expected[j], ProtocolVersion.fromInt(protocolVersion));
+                com.datastax.driver.core.TypeCodec<Object> codec = cluster[protocolVersion -1].getConfiguration()
+                                                                                              .getCodecRegistry()
+                                                                                              .codecFor(type);
+                ByteBuffer expectedByteValue = codec.serialize(expected[j], ProtocolVersion.fromInt(protocolVersion));
                 int expectedBytes = expectedByteValue.remaining();
                 ByteBuffer actualValue = actual.getBytesUnsafe(meta.getName(j));
                 int actualBytes = actualValue.remaining();
-
                 if (!Objects.equal(expectedByteValue, actualValue))
                     Assert.fail(String.format("Invalid value for row %d column %d (%s of type %s), " +
                                               "expected <%s> (%d bytes) but got <%s> (%d bytes) " +
                                               "(using protocol version %d)",
                                               i, j, meta.getName(j), type,
-                                              type.format(expected[j]),
+                                              codec.format(expected[j]),
                                               expectedBytes,
-                                              type.format(type.deserialize(actualValue, ProtocolVersion.fromInt(protocolVersion))),
+                                              codec.format(codec.deserialize(actualValue, ProtocolVersion.fromInt(protocolVersion))),
                                               actualBytes,
                                               protocolVersion));
             }
@@ -1228,6 +1235,12 @@ public abstract class CQLTester
         return m;
     }
 
+    protected com.datastax.driver.core.TupleType tupleTypeOf(int protocolVersion, DataType...types)
+    {
+        requireNetwork();
+        return cluster[protocolVersion -1].getMetadata().newTupleType(types);
+    }
+
     // Attempt to find an AbstracType from a value (for serialization/printing sake).
     // Will work as long as we use types we know of, which is good enough for testing
     private static AbstractType typeFor(Object value)
diff --git a/test/unit/org/apache/cassandra/cql3/IndexQueryPagingTest.java b/test/unit/org/apache/cassandra/cql3/IndexQueryPagingTest.java
index 35d57d9390..0d590b2837 100644
--- a/test/unit/org/apache/cassandra/cql3/IndexQueryPagingTest.java
+++ b/test/unit/org/apache/cassandra/cql3/IndexQueryPagingTest.java
@@ -78,7 +78,7 @@ public class IndexQueryPagingTest extends CQLTester
         // that all rows are returned, so we know that paging
         // of the results was involved.
         Session session = sessionNet(maxProtocolVersion);
-        Statement stmt = new SimpleStatement(String.format(cql, KEYSPACE + "." + currentTable()));
+        Statement stmt = session.newSimpleStatement(String.format(cql, KEYSPACE + "." + currentTable()));
         stmt.setFetchSize(rowCount - 1);
         assertEquals(rowCount, session.execute(stmt).all().size());
     }
diff --git a/test/unit/org/apache/cassandra/cql3/validation/entities/UFPureScriptTest.java b/test/unit/org/apache/cassandra/cql3/validation/entities/UFPureScriptTest.java
index 1563197d87..f43e335a95 100644
--- a/test/unit/org/apache/cassandra/cql3/validation/entities/UFPureScriptTest.java
+++ b/test/unit/org/apache/cassandra/cql3/validation/entities/UFPureScriptTest.java
@@ -183,10 +183,15 @@ public class UFPureScriptTest extends CQLTester
                    row(map));
 
         // same test - but via native protocol
-        TupleType tType = TupleType.of(DataType.cdouble(),
-                                       DataType.list(DataType.cdouble()),
-                                       DataType.set(DataType.text()),
-                                       DataType.map(DataType.cint(), DataType.cboolean()));
+        // we use protocol V3 here to encode the expected version because the server
+        // always serializes Collections using V3 - see CollectionSerializer's
+        // serialize and deserialize methods.
+        TupleType tType = tupleTypeOf(Server.VERSION_3,
+                                      DataType.cdouble(),
+                                      DataType.list(DataType.cdouble()),
+                                      DataType.set(DataType.text()),
+                                      DataType.map(DataType.cint(),
+                                                   DataType.cboolean()));
         TupleValue tup = tType.newValue(1d, list, set, map);
         for (int version = Server.VERSION_2; version <= maxProtocolVersion; version++)
         {
diff --git a/test/unit/org/apache/cassandra/cql3/validation/entities/UFTest.java b/test/unit/org/apache/cassandra/cql3/validation/entities/UFTest.java
index 2d9c540703..6bd03adaab 100644
--- a/test/unit/org/apache/cassandra/cql3/validation/entities/UFTest.java
+++ b/test/unit/org/apache/cassandra/cql3/validation/entities/UFTest.java
@@ -29,6 +29,7 @@ import java.util.TreeSet;
 import java.util.UUID;
 import java.security.AccessControlException;
 
+import com.google.common.collect.Lists;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -1157,10 +1158,15 @@ public class UFTest extends CQLTester
         assertRows(execute("SELECT " + fTup4 + "(tup) FROM %s WHERE key = 1"),
                    row(map));
 
-        TupleType tType = TupleType.of(DataType.cdouble(),
-                                       DataType.list(DataType.cdouble()),
-                                       DataType.set(DataType.text()),
-                                       DataType.map(DataType.cint(), DataType.cboolean()));
+        // same test - but via native protocol
+        // we use protocol V3 here to encode the expected version because the server
+        // always serializes Collections using V3 - see CollectionSerializer's
+        // serialize and deserialize methods.
+        TupleType tType = tupleTypeOf(Server.VERSION_3,
+                                      DataType.cdouble(),
+                                      DataType.list(DataType.cdouble()),
+                                      DataType.set(DataType.text()),
+                                      DataType.map(DataType.cint(), DataType.cboolean()));
         TupleValue tup = tType.newValue(1d, list, set, map);
         for (int version = Server.VERSION_2; version <= maxProtocolVersion; version++)
         {
diff --git a/test/unit/org/apache/cassandra/db/BatchlogManagerTest.java b/test/unit/org/apache/cassandra/db/BatchlogManagerTest.java
index 8c79689d26..568e23d611 100644
--- a/test/unit/org/apache/cassandra/db/BatchlogManagerTest.java
+++ b/test/unit/org/apache/cassandra/db/BatchlogManagerTest.java
@@ -95,8 +95,8 @@ public class BatchlogManagerTest
         InetAddress localhost = InetAddress.getByName("127.0.0.1");
         metadata.updateNormalToken(Util.token("A"), localhost);
         metadata.updateHostId(UUIDGen.getTimeUUID(), localhost);
-        Schema.instance.getColumnFamilyStoreInstance(SystemKeyspace.Batches.cfId).truncateBlocking();
-        Schema.instance.getColumnFamilyStoreInstance(SystemKeyspace.LegacyBatchlog.cfId).truncateBlocking();
+        Keyspace.open(SystemKeyspace.NAME).getColumnFamilyStore(SystemKeyspace.BATCHES).truncateBlocking();
+        Keyspace.open(SystemKeyspace.NAME).getColumnFamilyStore(SystemKeyspace.LEGACY_BATCHLOG).truncateBlocking();
     }
 
     @Test
diff --git a/test/unit/org/apache/cassandra/db/DirectoriesTest.java b/test/unit/org/apache/cassandra/db/DirectoriesTest.java
index e0786f9016..cb38e37d1f 100644
--- a/test/unit/org/apache/cassandra/db/DirectoriesTest.java
+++ b/test/unit/org/apache/cassandra/db/DirectoriesTest.java
@@ -30,16 +30,15 @@ import org.junit.BeforeClass;
 import org.junit.Test;
 
 import org.apache.cassandra.config.CFMetaData;
-import org.apache.cassandra.config.ColumnDefinition;
 import org.apache.cassandra.config.Config.DiskFailurePolicy;
 import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.config.IndexType;
 import org.apache.cassandra.db.Directories.DataDirectory;
 import org.apache.cassandra.db.index.SecondaryIndex;
 import org.apache.cassandra.db.marshal.UTF8Type;
 import org.apache.cassandra.io.sstable.Component;
 import org.apache.cassandra.io.sstable.Descriptor;
 import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.schema.IndexMetadata;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.io.FSWriteError;
 import org.apache.cassandra.utils.Pair;
@@ -162,9 +161,12 @@ public class DirectoriesTest
                                   .addPartitionKey("thekey", UTF8Type.instance)
                                   .addClusteringColumn("col", UTF8Type.instance)
                                   .build();
-        ColumnDefinition def = PARENT_CFM.getColumnDefinition(ByteBufferUtil.bytes("col"));
-        def.setIndex("idx", IndexType.KEYS, Collections.emptyMap());
-        CFMetaData INDEX_CFM = SecondaryIndex.newIndexMetadata(PARENT_CFM, def);
+        IndexMetadata indexDef = IndexMetadata.legacyIndex(PARENT_CFM.getColumnDefinition(ByteBufferUtil.bytes("col")),
+                                                           "idx",
+                                                           IndexMetadata.IndexType.KEYS,
+                                                           Collections.emptyMap());
+        PARENT_CFM.indexes(PARENT_CFM.getIndexes().with(indexDef));
+        CFMetaData INDEX_CFM = SecondaryIndex.newIndexMetadata(PARENT_CFM, indexDef);
         Directories parentDirectories = new Directories(PARENT_CFM);
         Directories indexDirectories = new Directories(INDEX_CFM);
         // secondary index has its own directory
diff --git a/test/unit/org/apache/cassandra/db/RangeTombstoneTest.java b/test/unit/org/apache/cassandra/db/RangeTombstoneTest.java
index 1538665d58..70fbfe3887 100644
--- a/test/unit/org/apache/cassandra/db/RangeTombstoneTest.java
+++ b/test/unit/org/apache/cassandra/db/RangeTombstoneTest.java
@@ -48,6 +48,7 @@ import org.apache.cassandra.db.marshal.UTF8Type;
 import org.apache.cassandra.db.marshal.Int32Type;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.io.sstable.metadata.StatsMetadata;
+import org.apache.cassandra.schema.IndexMetadata;
 import org.apache.cassandra.schema.KeyspaceParams;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.FBUtilities;
@@ -68,7 +69,12 @@ public class RangeTombstoneTest
         SchemaLoader.prepareServer();
         SchemaLoader.createKeyspace(KSNAME,
                                     KeyspaceParams.simple(1),
-                                    SchemaLoader.standardCFMD(KSNAME, CFNAME, 0, UTF8Type.instance, Int32Type.instance, Int32Type.instance));
+                                    SchemaLoader.standardCFMD(KSNAME,
+                                                              CFNAME,
+                                                              0,
+                                                              UTF8Type.instance,
+                                                              Int32Type.instance,
+                                                              Int32Type.instance));
     }
 
     @Test
@@ -453,7 +459,49 @@ public class RangeTombstoneTest
     @Test
     public void testRowWithRangeTombstonesUpdatesSecondaryIndex() throws Exception
     {
-        runCompactionWithRangeTombstoneAndCheckSecondaryIndex();
+        Keyspace table = Keyspace.open(KSNAME);
+        ColumnFamilyStore cfs = table.getColumnFamilyStore(CFNAME);
+        ByteBuffer key = ByteBufferUtil.bytes("k5");
+        ByteBuffer indexedColumnName = ByteBufferUtil.bytes("val");
+
+        cfs.truncateBlocking();
+        cfs.disableAutoCompaction();
+
+        ColumnDefinition cd = cfs.metadata.getColumnDefinition(indexedColumnName).copy();
+        IndexMetadata indexDef = IndexMetadata.legacyIndex(cd,
+                                                           "test_index",
+                                                           IndexMetadata.IndexType.CUSTOM,
+                                                           ImmutableMap.of(SecondaryIndex.CUSTOM_INDEX_OPTION_NAME,
+                                                                           TestIndex.class.getName()));
+
+        if (!cfs.metadata.getIndexes().get("test_index").isPresent())
+            cfs.metadata.indexes(cfs.metadata.getIndexes().with(indexDef));
+
+        Future<?> rebuild = cfs.indexManager.addIndexedColumn(indexDef);
+        // If rebuild there is, wait for the rebuild to finish so it doesn't race with the following insertions
+        if (rebuild != null)
+            rebuild.get();
+
+        TestIndex index = ((TestIndex)cfs.indexManager.getIndexForColumn(cd));
+        index.resetCounts();
+
+        UpdateBuilder builder = UpdateBuilder.create(cfs.metadata, key).withTimestamp(0);
+        for (int i = 0; i < 10; i++)
+            builder.newRow(i).add("val", i);
+        builder.applyUnsafe();
+        cfs.forceBlockingFlush();
+
+        new RowUpdateBuilder(cfs.metadata, 0, key).addRangeTombstone(0, 7).build().applyUnsafe();
+        cfs.forceBlockingFlush();
+
+        assertEquals(10, index.inserts.size());
+
+        CompactionManager.instance.performMaximal(cfs, false);
+
+        // compacted down to single sstable
+        assertEquals(1, cfs.getLiveSSTables().size());
+
+        assertEquals(8, index.deletes.size());
     }
 
     @Test
@@ -512,12 +560,19 @@ public class RangeTombstoneTest
         cfs.disableAutoCompaction();
 
         ColumnDefinition cd = cfs.metadata.getColumnDefinition(indexedColumnName).copy();
-        cd.setIndex("test_index", IndexType.CUSTOM, ImmutableMap.of(SecondaryIndex.CUSTOM_INDEX_OPTION_NAME, TestIndex.class.getName()));
-        Future<?> rebuild = cfs.indexManager.addIndexedColumn(cd);
+        IndexMetadata indexDef = IndexMetadata.legacyIndex(cd,
+                                                           "test_index",
+                                                           IndexMetadata.IndexType.CUSTOM,
+                                                           ImmutableMap.of(SecondaryIndex.CUSTOM_INDEX_OPTION_NAME,
+                                                                           TestIndex.class.getName()));
+
+        if (!cfs.metadata.getIndexes().get("test_index").isPresent())
+            cfs.metadata.indexes(cfs.metadata.getIndexes().with(indexDef));
+
+        Future<?> rebuild = cfs.indexManager.addIndexedColumn(indexDef);
         // If rebuild there is, wait for the rebuild to finish so it doesn't race with the following insertions
         if (rebuild != null)
             rebuild.get();
-
         TestIndex index = ((TestIndex)cfs.indexManager.getIndexForColumn(cd));
         index.resetCounts();
 
@@ -537,45 +592,6 @@ public class RangeTombstoneTest
         assertEquals(1, index.updates.size());
     }
 
-    private void runCompactionWithRangeTombstoneAndCheckSecondaryIndex() throws Exception
-    {
-        Keyspace table = Keyspace.open(KSNAME);
-        ColumnFamilyStore cfs = table.getColumnFamilyStore(CFNAME);
-        ByteBuffer key = ByteBufferUtil.bytes("k5");
-        ByteBuffer indexedColumnName = ByteBufferUtil.bytes("val");
-
-        cfs.truncateBlocking();
-        cfs.disableAutoCompaction();
-
-        ColumnDefinition cd = cfs.metadata.getColumnDefinition(indexedColumnName).copy();
-        cd.setIndex("test_index", IndexType.CUSTOM, ImmutableMap.of(SecondaryIndex.CUSTOM_INDEX_OPTION_NAME, TestIndex.class.getName()));
-        Future<?> rebuild = cfs.indexManager.addIndexedColumn(cd);
-        // If rebuild there is, wait for the rebuild to finish so it doesn't race with the following insertions
-        if (rebuild != null)
-            rebuild.get();
-
-        TestIndex index = ((TestIndex)cfs.indexManager.getIndexForColumn(cd));
-        index.resetCounts();
-
-        UpdateBuilder builder = UpdateBuilder.create(cfs.metadata, key).withTimestamp(0);
-        for (int i = 0; i < 10; i++)
-            builder.newRow(i).add("val", i);
-        builder.applyUnsafe();
-        cfs.forceBlockingFlush();
-
-        new RowUpdateBuilder(cfs.metadata, 0, key).addRangeTombstone(0, 7).build().applyUnsafe();
-        cfs.forceBlockingFlush();
-
-        assertEquals(10, index.inserts.size());
-
-        CompactionManager.instance.performMaximal(cfs, false);
-
-        // compacted down to single sstable
-        assertEquals(1, cfs.getLiveSSTables().size());
-
-        assertEquals(8, index.deletes.size());
-    }
-
     private static ByteBuffer bb(int i)
     {
         return ByteBufferUtil.bytes(i);
@@ -621,7 +637,7 @@ public class RangeTombstoneTest
 
         public void reload(){}
 
-        public void validateOptions() throws ConfigurationException{}
+        public void validateOptions(CFMetaData cfm, IndexMetadata def) throws ConfigurationException{}
 
         public String getIndexName(){ return "TestIndex";}
 
diff --git a/test/unit/org/apache/cassandra/db/SecondaryIndexTest.java b/test/unit/org/apache/cassandra/db/SecondaryIndexTest.java
index 0974fc4421..115b13cbac 100644
--- a/test/unit/org/apache/cassandra/db/SecondaryIndexTest.java
+++ b/test/unit/org/apache/cassandra/db/SecondaryIndexTest.java
@@ -32,7 +32,6 @@ import org.junit.Test;
 import org.apache.cassandra.SchemaLoader;
 import org.apache.cassandra.Util;
 import org.apache.cassandra.config.ColumnDefinition;
-import org.apache.cassandra.config.IndexType;
 import org.apache.cassandra.cql3.Operator;
 import org.apache.cassandra.db.index.SecondaryIndex;
 import org.apache.cassandra.db.index.SecondaryIndexSearcher;
@@ -40,6 +39,7 @@ import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.db.partitions.*;
 import org.apache.cassandra.db.rows.Row;
 import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.schema.IndexMetadata;
 import org.apache.cassandra.schema.KeyspaceParams;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.FBUtilities;
@@ -60,8 +60,8 @@ public class SecondaryIndexTest
         SchemaLoader.prepareServer();
         SchemaLoader.createKeyspace(KEYSPACE1,
                                     KeyspaceParams.simple(1),
-                                    SchemaLoader.compositeIndexCFMD(KEYSPACE1, WITH_COMPOSITE_INDEX, true).gcGraceSeconds(0),
-                                    SchemaLoader.compositeIndexCFMD(KEYSPACE1, COMPOSITE_INDEX_TO_BE_ADDED, false).gcGraceSeconds(0),
+                                    SchemaLoader.compositeIndexCFMD(KEYSPACE1, WITH_COMPOSITE_INDEX, true) .gcGraceSeconds(0),
+                                    SchemaLoader.compositeIndexCFMD(KEYSPACE1, COMPOSITE_INDEX_TO_BE_ADDED, false) .gcGraceSeconds(0),
                                     SchemaLoader.keysIndexCFMD(KEYSPACE1, WITH_KEYS_INDEX, true).gcGraceSeconds(0));
     }
 
@@ -426,8 +426,12 @@ public class SecondaryIndexTest
         new RowUpdateBuilder(cfs.metadata, 0, "k1").clustering("c").add("birthdate", 1L).build().applyUnsafe();
 
         ColumnDefinition old = cfs.metadata.getColumnDefinition(ByteBufferUtil.bytes("birthdate"));
-        old.setIndex("birthdate_index", IndexType.COMPOSITES, Collections.EMPTY_MAP);
-        Future<?> future = cfs.indexManager.addIndexedColumn(old);
+        IndexMetadata indexDef = IndexMetadata.legacyIndex(old,
+                                                           "birthdate_index",
+                                                           IndexMetadata.IndexType.COMPOSITES,
+                                                           Collections.EMPTY_MAP);
+        cfs.metadata.indexes(cfs.metadata.getIndexes().with(indexDef));
+        Future<?> future = cfs.indexManager.addIndexedColumn(indexDef);
         future.get();
         // we had a bug (CASSANDRA-2244) where index would get created but not flushed -- check for that
         ColumnDefinition cDef = cfs.metadata.getColumnDefinition(ByteBufferUtil.bytes("birthdate"));
@@ -441,7 +445,7 @@ public class SecondaryIndexTest
         assert !indexedCfs.isIndexBuilt(ByteBufferUtil.bytes("birthdate"));
 
         // rebuild & re-query
-        future = cfs.indexManager.addIndexedColumn(cDef);
+        future = cfs.indexManager.addIndexedColumn(indexDef);
         future.get();
         assertIndexedOne(cfs, ByteBufferUtil.bytes("birthdate"), 1L);
     }
@@ -477,8 +481,9 @@ public class SecondaryIndexTest
         if (count != 0)
             assertTrue(searchers.size() > 0);
 
-        try (ReadOrderGroup orderGroup = rc.startOrderGroup(); PartitionIterator iter = UnfilteredPartitionIterators.filter(searchers.get(0).search(rc, orderGroup), FBUtilities.nowInSeconds()))
-        {
+        try (ReadOrderGroup orderGroup = rc.startOrderGroup();
+             PartitionIterator iter = UnfilteredPartitionIterators.filter(searchers.get(0).search(rc, orderGroup),
+                                                                          FBUtilities.nowInSeconds())) {
             assertEquals(count, Util.size(iter));
         }
     }
diff --git a/test/unit/org/apache/cassandra/db/index/PerRowSecondaryIndexTest.java b/test/unit/org/apache/cassandra/db/index/PerRowSecondaryIndexTest.java
index 77de665cc3..8996dedf0a 100644
--- a/test/unit/org/apache/cassandra/db/index/PerRowSecondaryIndexTest.java
+++ b/test/unit/org/apache/cassandra/db/index/PerRowSecondaryIndexTest.java
@@ -41,6 +41,7 @@ import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
 import org.apache.cassandra.db.partitions.PartitionUpdate;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.schema.IndexMetadata;
 import org.apache.cassandra.schema.KeyspaceParams;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.FBUtilities;
@@ -205,7 +206,7 @@ public class PerRowSecondaryIndexTest
         }
 
         @Override
-        public void validateOptions() throws ConfigurationException
+        public void validateOptions(CFMetaData cfm, IndexMetadata def) throws ConfigurationException{}
         {
         }
 
diff --git a/test/unit/org/apache/cassandra/schema/DefsTest.java b/test/unit/org/apache/cassandra/schema/DefsTest.java
index 98a954c2be..c10865aa2b 100644
--- a/test/unit/org/apache/cassandra/schema/DefsTest.java
+++ b/test/unit/org/apache/cassandra/schema/DefsTest.java
@@ -34,13 +34,13 @@ import org.apache.cassandra.SchemaLoader;
 import org.apache.cassandra.Util;
 import org.apache.cassandra.config.CFMetaData;
 import org.apache.cassandra.config.ColumnDefinition;
-import org.apache.cassandra.config.IndexType;
 import org.apache.cassandra.config.Schema;
 import org.apache.cassandra.cql3.QueryProcessor;
 import org.apache.cassandra.cql3.UntypedResultSet;
 import org.apache.cassandra.db.ColumnFamilyStore;
 import org.apache.cassandra.db.Directories;
 import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.index.SecondaryIndex;
 import org.apache.cassandra.db.lifecycle.TransactionLog;
 import org.apache.cassandra.db.marshal.BytesType;
 import org.apache.cassandra.db.marshal.UTF8Type;
@@ -101,7 +101,7 @@ public class DefsTest
         for (int i = 0; i < 5; i++)
         {
             ByteBuffer name = ByteBuffer.wrap(new byte[] { (byte)i });
-            cfm.addColumnDefinition(ColumnDefinition.regularDef(cfm, name, BytesType.instance).setIndex(Integer.toString(i), IndexType.KEYS, null));
+            cfm.addColumnDefinition(ColumnDefinition.regularDef(cfm, name, BytesType.instance));
         }
 
         cfm.comment("No comment")
@@ -116,13 +116,11 @@ public class DefsTest
         CFMetaData cfNew = cfm.copy();
 
         // add one.
-        ColumnDefinition addIndexDef = ColumnDefinition.regularDef(cfm, ByteBuffer.wrap(new byte[] { 5 }), BytesType.instance)
-                                                       .setIndex("5", IndexType.KEYS, null);
+        ColumnDefinition addIndexDef = ColumnDefinition.regularDef(cfm, ByteBuffer.wrap(new byte[] { 5 }), BytesType.instance);
         cfNew.addColumnDefinition(addIndexDef);
 
         // remove one.
-        ColumnDefinition removeIndexDef = ColumnDefinition.regularDef(cfm, ByteBuffer.wrap(new byte[] { 0 }), BytesType.instance)
-                                                          .setIndex("0", IndexType.KEYS, null);
+        ColumnDefinition removeIndexDef = ColumnDefinition.regularDef(cfm, ByteBuffer.wrap(new byte[] { 0 }), BytesType.instance);
         assertTrue(cfNew.removeColumnDefinition(removeIndexDef));
 
         cfm.apply(cfNew);
@@ -505,19 +503,36 @@ public class DefsTest
         ColumnFamilyStore cfs = Keyspace.open(KEYSPACE6).getColumnFamilyStore(TABLE1i);
 
         // insert some data.  save the sstable descriptor so we can make sure it's marked for delete after the drop
-        QueryProcessor.executeInternal(String.format("INSERT INTO %s.%s (key, c1, birthdate, notbirthdate) VALUES (?, ?, ?, ?)",
-                                                     KEYSPACE6, TABLE1i),
+        QueryProcessor.executeInternal(String.format(
+                                                    "INSERT INTO %s.%s (key, c1, birthdate, notbirthdate) VALUES (?, ?, ?, ?)",
+                                                    KEYSPACE6,
+                                                    TABLE1i),
                                        "key0", "col0", 1L, 1L);
 
         cfs.forceBlockingFlush();
-        ColumnFamilyStore indexedCfs = cfs.indexManager.getIndexForColumn(cfs.metadata.getColumnDefinition(ByteBufferUtil.bytes("birthdate"))).getIndexCfs();
+        ColumnDefinition indexedColumn = cfs.metadata.getColumnDefinition(ByteBufferUtil.bytes("birthdate"));
+        SecondaryIndex index = cfs.indexManager.getIndexForColumn(indexedColumn);
+        ColumnFamilyStore indexedCfs = index.getIndexCfs();
         Descriptor desc = indexedCfs.getLiveSSTables().iterator().next().descriptor;
 
         // drop the index
         CFMetaData meta = cfs.metadata.copy();
-        ColumnDefinition cdOld = cfs.metadata.getColumnDefinition(ByteBufferUtil.bytes("birthdate"));
-        ColumnDefinition cdNew = ColumnDefinition.regularDef(meta, cdOld.name.bytes, cdOld.type);
-        meta.addOrReplaceColumnDefinition(cdNew);
+        // We currently have a mismatch between IndexMetadata.name (which is simply the name
+        // of the index) and what gets returned from SecondaryIndex#getIndexName() (usually, this
+        // defaults to <tablename>.<indexname>.
+        // IndexMetadata takes its lead from the prior implementation of ColumnDefinition.name
+        // which did not include the table name.
+        // This mismatch causes some other, long standing inconsistencies:
+        // nodetool rebuild_index <ks> <tbl> <idx>  - <idx> must be qualified, i.e. include the redundant table name
+        //                                            without it, the rebuild silently fails
+        // system.IndexInfo (which is also exposed over JMX as CF.BuildIndexes) uses the form <tbl>.<idx>
+        // cqlsh> describe index [<ks>.]<idx>  - here <idx> must not be qualified by the table name.
+        //
+        // This should get resolved as part of #9459 by better separating the index name from the
+        // name of it's underlying CFS (if it as one), as the comment in CFMetaData#indexColumnFamilyName promises
+        // Then we will be able to just use the value of SI#getIndexName() when removing an index from CFMetaData
+        IndexMetadata existing = meta.getIndexes().iterator().next();
+        meta.indexes(meta.getIndexes().without(existing.name));
         MigrationManager.announceColumnFamilyUpdate(meta, false);
 
         // check
diff --git a/test/unit/org/apache/cassandra/schema/LegacySchemaMigratorTest.java b/test/unit/org/apache/cassandra/schema/LegacySchemaMigratorTest.java
index 1b0fb124a9..a1f9570495 100644
--- a/test/unit/org/apache/cassandra/schema/LegacySchemaMigratorTest.java
+++ b/test/unit/org/apache/cassandra/schema/LegacySchemaMigratorTest.java
@@ -474,15 +474,39 @@ public class LegacySchemaMigratorTest
                     ? ""
                     : column.name.toString();
 
-        RowUpdateBuilder adder = new RowUpdateBuilder(SystemKeyspace.LegacyColumns, timestamp, mutation).clustering(table.cfName, name);
+        final RowUpdateBuilder adder = new RowUpdateBuilder(SystemKeyspace.LegacyColumns, timestamp, mutation).clustering(table.cfName, name);
 
         adder.add("validator", column.type.toString())
              .add("type", serializeKind(column.kind, table.isDense()))
-             .add("component_index", column.isOnAllComponents() ? null : column.position())
-             .add("index_name", column.getIndexName())
-             .add("index_type", column.getIndexType() == null ? null : column.getIndexType().toString())
-             .add("index_options", json(column.getIndexOptions()))
-             .build();
+             .add("component_index", column.isOnAllComponents() ? null : column.position());
+
+        Optional<IndexMetadata> index = findIndexForColumn(table.getIndexes(), table, column);
+        if (index.isPresent())
+        {
+            IndexMetadata i = index.get();
+            adder.add("index_name", i.name);
+            adder.add("index_type", i.indexType.toString());
+            adder.add("index_options", json(i.options));
+        }
+        else
+        {
+            adder.add("index_name", null);
+            adder.add("index_type", null);
+            adder.add("index_options", null);
+        }
+
+        adder.build();
+    }
+
+    private static Optional<IndexMetadata> findIndexForColumn(Indexes indexes,
+                                                                CFMetaData table,
+                                                                ColumnDefinition column)
+    {
+        for (IndexMetadata index : indexes)
+            if (index.indexedColumn(table).equals(column))
+                return Optional.of(index);
+
+        return Optional.empty();
     }
 
     private static String serializeKind(ColumnDefinition.Kind kind, boolean isDense)
diff --git a/tools/stress/src/org/apache/cassandra/stress/util/JavaDriverClient.java b/tools/stress/src/org/apache/cassandra/stress/util/JavaDriverClient.java
index d82ebfd499..8be1c967c2 100644
--- a/tools/stress/src/org/apache/cassandra/stress/util/JavaDriverClient.java
+++ b/tools/stress/src/org/apache/cassandra/stress/util/JavaDriverClient.java
@@ -142,7 +142,7 @@ public class JavaDriverClient
 
     public ResultSet execute(String query, org.apache.cassandra.db.ConsistencyLevel consistency)
     {
-        SimpleStatement stmt = new SimpleStatement(query);
+        SimpleStatement stmt = getSession().newSimpleStatement(query);
         stmt.setConsistencyLevel(from(consistency));
         return getSession().execute(stmt);
     }