cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From alek...@apache.org
Subject [2/4] cassandra git commit: Use CQL type names in schema metadata tables
Date Wed, 04 Nov 2015 17:08:46 GMT
http://git-wip-us.apache.org/repos/asf/cassandra/blob/340df43f/src/java/org/apache/cassandra/schema/SchemaKeyspace.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/schema/SchemaKeyspace.java b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java
index 4ff8a23..e4e50a0 100644
--- a/src/java/org/apache/cassandra/schema/SchemaKeyspace.java
+++ b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java
@@ -17,14 +17,12 @@
  */
 package org.apache.cassandra.schema;
 
-import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.charset.CharacterCodingException;
 import java.security.MessageDigest;
 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;
@@ -47,9 +45,14 @@ import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.FBUtilities;
-import org.apache.cassandra.utils.concurrent.OpOrder;
+import org.apache.cassandra.utils.Pair;
 
+import static java.lang.String.format;
+
+import static java.util.stream.Collectors.toList;
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
 import static org.apache.cassandra.cql3.QueryProcessor.executeOnceInternal;
+import static org.apache.cassandra.schema.CQLTypeParser.parse;
 
 /**
  * system_schema.* tables and methods for manipulating them.
@@ -77,7 +80,6 @@ public final class SchemaKeyspace
     public static final String AGGREGATES = "aggregates";
     public static final String INDEXES = "indexes";
 
-
     public static final List<String> ALL =
         ImmutableList.of(KEYSPACES, TABLES, COLUMNS, TRIGGERS, VIEWS, TYPES, FUNCTIONS, AGGREGATES, INDEXES);
 
@@ -205,14 +207,13 @@ public final class SchemaKeyspace
                 "CREATE TABLE %s ("
                 + "keyspace_name text,"
                 + "function_name text,"
-                + "signature frozen<list<text>>,"
-                + "argument_names frozen<list<text>>,"
                 + "argument_types frozen<list<text>>,"
+                + "argument_names frozen<list<text>>,"
                 + "body text,"
                 + "language text,"
                 + "return_type text,"
                 + "called_on_null_input boolean,"
-                + "PRIMARY KEY ((keyspace_name), function_name, signature))");
+                + "PRIMARY KEY ((keyspace_name), function_name, argument_types))");
 
     private static final CFMetaData Aggregates =
         compile(AGGREGATES,
@@ -220,14 +221,13 @@ public final class SchemaKeyspace
                 "CREATE TABLE %s ("
                 + "keyspace_name text,"
                 + "aggregate_name text,"
-                + "signature frozen<list<text>>,"
                 + "argument_types frozen<list<text>>,"
                 + "final_func text,"
                 + "initcond blob,"
                 + "return_type text,"
                 + "state_func text,"
                 + "state_type text,"
-                + "PRIMARY KEY ((keyspace_name), aggregate_name, signature))");
+                + "PRIMARY KEY ((keyspace_name), aggregate_name, argument_types))");
 
     public static final List<CFMetaData> ALL_TABLE_METADATA =
         ImmutableList.of(Keyspaces, Tables, Columns, Triggers, DroppedColumns, Views, Types, Functions, Aggregates, Indexes);
@@ -267,35 +267,6 @@ public final class SchemaKeyspace
         makeCreateKeyspaceMutation(schema, timestamp + 1).apply();
     }
 
-    public static List<KeyspaceMetadata> readSchemaFromSystemTables()
-    {
-        ReadCommand cmd = getReadCommandForTableSchema(KEYSPACES);
-        try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); PartitionIterator schema = cmd.executeInternal(orderGroup))
-        {
-            List<KeyspaceMetadata> keyspaces = new ArrayList<>();
-
-            while (schema.hasNext())
-            {
-                try (RowIterator partition = schema.next())
-                {
-                    if (isSystemKeyspaceSchemaPartition(partition.partitionKey()))
-                        continue;
-
-                    DecoratedKey key = partition.partitionKey();
-
-                    readSchemaPartitionForKeyspaceAndApply(TYPES, key,
-                        types -> readSchemaPartitionForKeyspaceAndApply(TABLES, key,
-                        tables -> readSchemaPartitionForKeyspaceAndApply(VIEWS, key,
-                        views -> readSchemaPartitionForKeyspaceAndApply(FUNCTIONS, key,
-                        functions -> readSchemaPartitionForKeyspaceAndApply(AGGREGATES, key,
-                        aggregates -> keyspaces.add(createKeyspaceFromSchemaPartitions(partition, tables, views, types, functions, aggregates))))))
-                    );
-                }
-            }
-            return keyspaces;
-        }
-    }
-
     public static void truncate()
     {
         ALL.forEach(table -> getSchemaCFS(table).truncateBlocking());
@@ -397,336 +368,18 @@ public final class SchemaKeyspace
         }
     }
 
-    private static Map<DecoratedKey, FilteredPartition> readSchemaForKeyspaces(String schemaTableName, Set<String> keyspaceNames)
-    {
-        Map<DecoratedKey, FilteredPartition> schema = new HashMap<>();
-
-        for (String keyspaceName : keyspaceNames)
-        {
-            // We don't to return the RowIterator directly because we should guarantee that this iterator
-            // will be closed, and putting it in a Map make that harder/more awkward.
-            readSchemaPartitionForKeyspaceAndApply(schemaTableName, keyspaceName,
-                partition -> {
-                    if (!partition.isEmpty())
-                        schema.put(partition.partitionKey(), FilteredPartition.create(partition));
-                    return null;
-                }
-            );
-        }
-
-        return schema;
-    }
-
     private static ByteBuffer getSchemaKSKey(String ksName)
     {
         return AsciiType.instance.fromString(ksName);
     }
 
-    private static <T> T readSchemaPartitionForKeyspaceAndApply(String schemaTableName, String keyspaceName, Function<RowIterator, T> fct)
-    {
-        return readSchemaPartitionForKeyspaceAndApply(schemaTableName, getSchemaKSKey(keyspaceName), fct);
-    }
-
-    private static <T> T readSchemaPartitionForKeyspaceAndApply(String schemaTableName, ByteBuffer keyspaceKey, Function<RowIterator, T> fct)
-    {
-        ColumnFamilyStore store = getSchemaCFS(schemaTableName);
-        return readSchemaPartitionForKeyspaceAndApply(store, store.decorateKey(keyspaceKey), fct);
-    }
-
-    private static <T> T readSchemaPartitionForKeyspaceAndApply(String schemaTableName, DecoratedKey keyspaceKey, Function<RowIterator, T> fct)
-    {
-        return readSchemaPartitionForKeyspaceAndApply(getSchemaCFS(schemaTableName), keyspaceKey, fct);
-    }
-
-    private static <T> T readSchemaPartitionForKeyspaceAndApply(ColumnFamilyStore store, DecoratedKey keyspaceKey, Function<RowIterator, T> fct)
-    {
-        int nowInSec = FBUtilities.nowInSeconds();
-        try (OpOrder.Group op = store.readOrdering.start();
-             RowIterator partition = UnfilteredRowIterators.filter(SinglePartitionReadCommand.fullPartitionRead(store.metadata, nowInSec, keyspaceKey)
-                                                                                             .queryMemtableAndDisk(store, op), nowInSec))
-        {
-            return fct.apply(partition);
-        }
-    }
-
-    private static <T> T readSchemaPartitionForTableAndApply(String schemaTableName, String keyspaceName, String tableName, Function<RowIterator, T> fct)
-    {
-        ColumnFamilyStore store = getSchemaCFS(schemaTableName);
-
-        ClusteringComparator comparator = store.metadata.comparator;
-        Slices slices = Slices.with(comparator, Slice.make(comparator, tableName));
-        int nowInSec = FBUtilities.nowInSeconds();
-        try (OpOrder.Group op = store.readOrdering.start();
-             RowIterator partition =  UnfilteredRowIterators.filter(SinglePartitionReadCommand.create(store.metadata, nowInSec, getSchemaKSKey(keyspaceName), slices)
-                                                                                              .queryMemtableAndDisk(store, op), nowInSec))
-        {
-            return fct.apply(partition);
-        }
-    }
-
     private static boolean isSystemKeyspaceSchemaPartition(DecoratedKey partitionKey)
     {
         return Schema.isSystemKeyspace(UTF8Type.instance.compose(partitionKey.getKey()));
     }
 
-    /**
-     * Merge remote schema in form of mutations with local and mutate ks/cf metadata objects
-     * (which also involves fs operations on add/drop ks/cf)
-     *
-     * @param mutations the schema changes to apply
-     *
-     * @throws ConfigurationException If one of metadata attributes has invalid value
-     * @throws IOException If data was corrupted during transportation or failed to apply fs operations
-     */
-    public static synchronized void mergeSchemaAndAnnounceVersion(Collection<Mutation> mutations) throws ConfigurationException, IOException
-    {
-        mergeSchema(mutations);
-        Schema.instance.updateVersionAndAnnounce();
-    }
-
-    public static synchronized void mergeSchema(Collection<Mutation> mutations) throws IOException
-    {
-        // compare before/after schemas of the affected keyspaces only
-        Set<String> keyspaces = new HashSet<>(mutations.size());
-        for (Mutation mutation : mutations)
-            keyspaces.add(ByteBufferUtil.string(mutation.key().getKey()));
-
-        // current state of the schema
-        Map<DecoratedKey, FilteredPartition> oldKeyspaces = readSchemaForKeyspaces(KEYSPACES, keyspaces);
-        Map<DecoratedKey, FilteredPartition> oldColumnFamilies = readSchemaForKeyspaces(TABLES, keyspaces);
-        Map<DecoratedKey, FilteredPartition> oldViews = readSchemaForKeyspaces(VIEWS, keyspaces);
-        Map<DecoratedKey, FilteredPartition> oldTypes = readSchemaForKeyspaces(TYPES, keyspaces);
-        Map<DecoratedKey, FilteredPartition> oldFunctions = readSchemaForKeyspaces(FUNCTIONS, keyspaces);
-        Map<DecoratedKey, FilteredPartition> oldAggregates = readSchemaForKeyspaces(AGGREGATES, keyspaces);
-
-        mutations.forEach(Mutation::apply);
-
-        if (FLUSH_SCHEMA_TABLES)
-            flush();
-
-        // with new data applied
-        Map<DecoratedKey, FilteredPartition> newKeyspaces = readSchemaForKeyspaces(KEYSPACES, keyspaces);
-        Map<DecoratedKey, FilteredPartition> newColumnFamilies = readSchemaForKeyspaces(TABLES, keyspaces);
-        Map<DecoratedKey, FilteredPartition> newViews = readSchemaForKeyspaces(VIEWS, keyspaces);
-        Map<DecoratedKey, FilteredPartition> newTypes = readSchemaForKeyspaces(TYPES, keyspaces);
-        Map<DecoratedKey, FilteredPartition> newFunctions = readSchemaForKeyspaces(FUNCTIONS, keyspaces);
-        Map<DecoratedKey, FilteredPartition> newAggregates = readSchemaForKeyspaces(AGGREGATES, keyspaces);
-
-        Set<String> keyspacesToDrop = mergeKeyspaces(oldKeyspaces, newKeyspaces);
-        mergeTables(oldColumnFamilies, newColumnFamilies);
-        mergeViews(oldViews, newViews);
-        mergeTypes(oldTypes, newTypes);
-        mergeFunctions(oldFunctions, newFunctions);
-        mergeAggregates(oldAggregates, newAggregates);
-
-        // it is safe to drop a keyspace only when all nested ColumnFamilies where deleted
-        keyspacesToDrop.forEach(Schema.instance::dropKeyspace);
-    }
-
-    private static Set<String> mergeKeyspaces(Map<DecoratedKey, FilteredPartition> before, Map<DecoratedKey, FilteredPartition> after)
-    {
-        for (FilteredPartition newPartition : after.values())
-        {
-            String name = AsciiType.instance.compose(newPartition.partitionKey().getKey());
-            KeyspaceParams params = createKeyspaceParamsFromSchemaPartition(newPartition.rowIterator());
-
-            FilteredPartition oldPartition = before.remove(newPartition.partitionKey());
-            if (oldPartition == null || oldPartition.isEmpty())
-                Schema.instance.addKeyspace(KeyspaceMetadata.create(name, params));
-            else
-                Schema.instance.updateKeyspace(name, params);
-        }
-
-        // What's remain in old is those keyspace that are not in updated, i.e. the dropped ones.
-        return asKeyspaceNamesSet(before.keySet());
-    }
-
-    private static Set<String> asKeyspaceNamesSet(Set<DecoratedKey> keys)
-    {
-        Set<String> names = new HashSet<>(keys.size());
-        for (DecoratedKey key : keys)
-            names.add(AsciiType.instance.compose(key.getKey()));
-        return names;
-    }
-
-    private static void mergeTables(Map<DecoratedKey, FilteredPartition> before, Map<DecoratedKey, FilteredPartition> after)
-    {
-        diffSchema(before, after, new Differ()
-        {
-            public void onDropped(UntypedResultSet.Row oldRow)
-            {
-                Schema.instance.dropTable(oldRow.getString("keyspace_name"), oldRow.getString("table_name"));
-            }
-
-            public void onAdded(UntypedResultSet.Row newRow)
-            {
-                Schema.instance.addTable(createTableFromTableRow(newRow));
-            }
-
-            public void onUpdated(UntypedResultSet.Row oldRow, UntypedResultSet.Row newRow)
-            {
-                Schema.instance.updateTable(newRow.getString("keyspace_name"), newRow.getString("table_name"));
-            }
-        });
-    }
-
-    private static void mergeViews(Map<DecoratedKey, FilteredPartition> before, Map<DecoratedKey, FilteredPartition> after)
-    {
-        diffSchema(before, after, new Differ()
-        {
-            public void onDropped(UntypedResultSet.Row oldRow)
-            {
-                Schema.instance.dropView(oldRow.getString("keyspace_name"), oldRow.getString("view_name"));
-            }
-
-            public void onAdded(UntypedResultSet.Row newRow)
-            {
-                Schema.instance.addView(createViewFromViewRow(newRow));
-            }
-
-            public void onUpdated(UntypedResultSet.Row oldRow, UntypedResultSet.Row newRow)
-            {
-                Schema.instance.updateView(newRow.getString("keyspace_name"), newRow.getString("view_name"));
-            }
-        });
-    }
-
-    private static void mergeTypes(Map<DecoratedKey, FilteredPartition> before, Map<DecoratedKey, FilteredPartition> after)
-    {
-        diffSchema(before, after, new Differ()
-        {
-            public void onDropped(UntypedResultSet.Row oldRow)
-            {
-                Schema.instance.dropType(createTypeFromRow(oldRow));
-            }
-
-            public void onAdded(UntypedResultSet.Row newRow)
-            {
-                Schema.instance.addType(createTypeFromRow(newRow));
-            }
-
-            public void onUpdated(UntypedResultSet.Row oldRow, UntypedResultSet.Row newRow)
-            {
-                Schema.instance.updateType(createTypeFromRow(newRow));
-            }
-        });
-    }
-
-    private static void mergeFunctions(Map<DecoratedKey, FilteredPartition> before, Map<DecoratedKey, FilteredPartition> after)
-    {
-        diffSchema(before, after, new Differ()
-        {
-            public void onDropped(UntypedResultSet.Row oldRow)
-            {
-                Schema.instance.dropFunction(createFunctionFromFunctionRow(oldRow));
-            }
-
-            public void onAdded(UntypedResultSet.Row newRow)
-            {
-                Schema.instance.addFunction(createFunctionFromFunctionRow(newRow));
-            }
-
-            public void onUpdated(UntypedResultSet.Row oldRow, UntypedResultSet.Row newRow)
-            {
-                Schema.instance.updateFunction(createFunctionFromFunctionRow(newRow));
-            }
-        });
-    }
-
-    private static void mergeAggregates(Map<DecoratedKey, FilteredPartition> before, Map<DecoratedKey, FilteredPartition> after)
-    {
-        diffSchema(before, after, new Differ()
-        {
-            public void onDropped(UntypedResultSet.Row oldRow)
-            {
-                Schema.instance.dropAggregate(createAggregateFromAggregateRow(oldRow));
-            }
-
-            public void onAdded(UntypedResultSet.Row newRow)
-            {
-                Schema.instance.addAggregate(createAggregateFromAggregateRow(newRow));
-            }
-
-            public void onUpdated(UntypedResultSet.Row oldRow, UntypedResultSet.Row newRow)
-            {
-                Schema.instance.updateAggregate(createAggregateFromAggregateRow(newRow));
-            }
-        });
-    }
-
-    public interface Differ
-    {
-        void onDropped(UntypedResultSet.Row oldRow);
-        void onAdded(UntypedResultSet.Row newRow);
-        void onUpdated(UntypedResultSet.Row oldRow, UntypedResultSet.Row newRow);
-    }
-
-    private static void diffSchema(Map<DecoratedKey, FilteredPartition> before, Map<DecoratedKey, FilteredPartition> after, Differ differ)
-    {
-        for (FilteredPartition newPartition : after.values())
-        {
-            CFMetaData metadata = newPartition.metadata();
-            DecoratedKey key = newPartition.partitionKey();
-
-            FilteredPartition oldPartition = before.remove(key);
-
-            if (oldPartition == null || oldPartition.isEmpty())
-            {
-                // Means everything is to be added
-                for (Row row : newPartition)
-                    differ.onAdded(UntypedResultSet.Row.fromInternalRow(metadata, key, row));
-                continue;
-            }
-
-            Iterator<Row> oldIter = oldPartition.iterator();
-            Iterator<Row> newIter = newPartition.iterator();
-
-            Row oldRow = oldIter.hasNext() ? oldIter.next() : null;
-            Row newRow = newIter.hasNext() ? newIter.next() : null;
-            while (oldRow != null && newRow != null)
-            {
-                int cmp = metadata.comparator.compare(oldRow.clustering(), newRow.clustering());
-                if (cmp < 0)
-                {
-                    differ.onDropped(UntypedResultSet.Row.fromInternalRow(metadata, key, oldRow));
-                    oldRow = oldIter.hasNext() ? oldIter.next() : null;
-                }
-                else if (cmp > 0)
-                {
-
-                    differ.onAdded(UntypedResultSet.Row.fromInternalRow(metadata, key, newRow));
-                    newRow = newIter.hasNext() ? newIter.next() : null;
-                }
-                else
-                {
-                    if (!oldRow.equals(newRow))
-                        differ.onUpdated(UntypedResultSet.Row.fromInternalRow(metadata, key, oldRow), UntypedResultSet.Row.fromInternalRow(metadata, key, newRow));
-
-                    oldRow = oldIter.hasNext() ? oldIter.next() : null;
-                    newRow = newIter.hasNext() ? newIter.next() : null;
-                }
-            }
-
-            while (oldRow != null)
-            {
-                differ.onDropped(UntypedResultSet.Row.fromInternalRow(metadata, key, oldRow));
-                oldRow = oldIter.hasNext() ? oldIter.next() : null;
-            }
-            while (newRow != null)
-            {
-                differ.onAdded(UntypedResultSet.Row.fromInternalRow(metadata, key, newRow));
-                newRow = newIter.hasNext() ? newIter.next() : null;
-            }
-        }
-
-        // What remains is those keys that were only in before.
-        for (FilteredPartition partition : before.values())
-            for (Row row : partition)
-                differ.onDropped(UntypedResultSet.Row.fromInternalRow(partition.metadata(), partition.partitionKey(), row));
-    }
-
     /*
-     * Keyspace metadata serialization/deserialization.
+     * Schema entities to mutations
      */
 
     public static Mutation makeCreateKeyspaceMutation(String name, KeyspaceParams params, long timestamp)
@@ -761,46 +414,6 @@ public final class SchemaKeyspace
         return mutation;
     }
 
-    private static KeyspaceMetadata createKeyspaceFromSchemaPartitions(RowIterator serializedParams,
-                                                                       RowIterator serializedTables,
-                                                                       RowIterator serializedViews,
-                                                                       RowIterator serializedTypes,
-                                                                       RowIterator serializedFunctions,
-                                                                       RowIterator serializedAggregates)
-    {
-        String name = AsciiType.instance.compose(serializedParams.partitionKey().getKey());
-
-        KeyspaceParams params = createKeyspaceParamsFromSchemaPartition(serializedParams);
-        Tables tables = createTablesFromTablesPartition(serializedTables);
-        Views views = createViewsFromViewsPartition(serializedViews);
-        Types types = createTypesFromPartition(serializedTypes);
-
-        Collection<UDFunction> udfs = createFunctionsFromFunctionsPartition(serializedFunctions);
-        Functions functions = org.apache.cassandra.schema.Functions.builder().add(udfs).build();
-        functions = createAggregatesFromAggregatesPartition(functions, serializedAggregates);
-
-        return KeyspaceMetadata.create(name, params, tables, views, types, functions);
-    }
-
-    /**
-     * Deserialize only Keyspace attributes without nested tables or types
-     *
-     * @param partition Keyspace attributes in serialized form
-     */
-
-    private static KeyspaceParams createKeyspaceParamsFromSchemaPartition(RowIterator partition)
-    {
-        String query = String.format("SELECT * FROM %s.%s", NAME, KEYSPACES);
-        UntypedResultSet.Row row = QueryProcessor.resultify(query, partition).one();
-
-        return KeyspaceParams.create(row.getBoolean(KeyspaceParams.Option.DURABLE_WRITES.toString()),
-                                     row.getFrozenTextMap(KeyspaceParams.Option.REPLICATION.toString()));
-    }
-
-    /*
-     * User type metadata serialization/deserialization.
-     */
-
     public static Mutation makeCreateTypeMutation(KeyspaceMetadata keyspace, UserType type, long timestamp)
     {
         // Include the serialized keyspace in case the target node missed a CREATE KEYSPACE migration (see CASSANDRA-5631).
@@ -813,8 +426,8 @@ public final class SchemaKeyspace
     {
         RowUpdateBuilder adder = new RowUpdateBuilder(Types, timestamp, mutation)
                                  .clustering(type.getNameAsString())
-                                 .frozenList("field_names", type.fieldNames().stream().map(SchemaKeyspace::bbToString).collect(Collectors.toList()))
-                                 .frozenList("field_types", type.fieldTypes().stream().map(AbstractType::toString).collect(Collectors.toList()));
+                                 .frozenList("field_names", type.fieldNames().stream().map(SchemaKeyspace::bbToString).collect(toList()))
+                                 .frozenList("field_types", type.fieldTypes().stream().map(AbstractType::asCQL3Type).map(CQL3Type::toString).collect(toList()));
 
         adder.build();
     }
@@ -838,36 +451,6 @@ public final class SchemaKeyspace
         return RowUpdateBuilder.deleteRow(Types, timestamp, mutation, type.name);
     }
 
-    private static Types createTypesFromPartition(RowIterator partition)
-    {
-        String query = String.format("SELECT * FROM %s.%s", NAME, TYPES);
-        Types.Builder types = org.apache.cassandra.schema.Types.builder();
-        QueryProcessor.resultify(query, partition).forEach(row -> types.add(createTypeFromRow(row)));
-        return types.build();
-    }
-
-    private static UserType createTypeFromRow(UntypedResultSet.Row row)
-    {
-        String keyspace = row.getString("keyspace_name");
-        ByteBuffer name = ByteBufferUtil.bytes(row.getString("type_name"));
-        List<String> rawColumns = row.getFrozenList("field_names", UTF8Type.instance);
-        List<String> rawTypes = row.getFrozenList("field_types", UTF8Type.instance);
-
-        List<ByteBuffer> columns = new ArrayList<>(rawColumns.size());
-        for (String rawColumn : rawColumns)
-            columns.add(ByteBufferUtil.bytes(rawColumn));
-
-        List<AbstractType<?>> types = new ArrayList<>(rawTypes.size());
-        for (String rawType : rawTypes)
-            types.add(parseType(rawType));
-
-        return new UserType(keyspace, name, columns, types);
-    }
-
-    /*
-     * Table metadata serialization/deserialization.
-     */
-
     public static Mutation makeCreateTableMutation(KeyspaceMetadata keyspace, CFMetaData table, long timestamp)
     {
         // Include the serialized keyspace in case the target node missed a CREATE KEYSPACE migration (see CASSANDRA-5631).
@@ -990,9 +573,7 @@ public final class SchemaKeyspace
 
         // updated indexes need to be updated
         for (MapDifference.ValueDifference<IndexMetadata> diff : indexesDiff.entriesDiffering().values())
-        {
             addUpdatedIndexToSchemaMutation(newTable, diff.rightValue(), timestamp, mutation);
-        }
 
         return mutation;
     }
@@ -1038,274 +619,59 @@ public final class SchemaKeyspace
         return mutation;
     }
 
-    public static CFMetaData createTableFromName(String keyspace, String table)
+    private static void addColumnToSchemaMutation(CFMetaData table, ColumnDefinition column, long timestamp, Mutation mutation)
     {
-        return readSchemaPartitionForTableAndApply(TABLES, keyspace, table, partition ->
-        {
-            if (partition.isEmpty())
-                throw new RuntimeException(String.format("%s:%s not found in the schema definitions keyspace.", keyspace, table));
+        RowUpdateBuilder adder = new RowUpdateBuilder(Columns, timestamp, mutation).clustering(table.cfName, column.name.toString());
+
+        AbstractType<?> type = column.type;
+        if (type instanceof ReversedType)
+            type = ((ReversedType) type).baseType;
 
-            return createTableFromTablePartition(partition);
-        });
+        adder.add("column_name_bytes", column.name.bytes)
+             .add("kind", column.kind.toString().toLowerCase())
+             .add("position", column.isOnAllComponents() ? ColumnDefinition.NO_POSITION : column.position())
+             .add("clustering_order", column.clusteringOrder().toString().toLowerCase())
+             .add("type", type.asCQL3Type().toString())
+             .build();
     }
 
-    /**
-     * Deserialize tables from low-level schema representation, all of them belong to the same keyspace
-     */
-    private static Tables createTablesFromTablesPartition(RowIterator partition)
+    private static void dropColumnFromSchemaMutation(CFMetaData table, ColumnDefinition column, long timestamp, Mutation mutation)
     {
-        String query = String.format("SELECT * FROM %s.%s", NAME, TABLES);
-        Tables.Builder tables = org.apache.cassandra.schema.Tables.builder();
-        QueryProcessor.resultify(query, partition).forEach(row -> tables.add(createTableFromTableRow(row)));
-        return tables.build();
+        // Note: we do want to use name.toString(), not name.bytes directly for backward compatibility (For CQL3, this won't make a difference).
+        RowUpdateBuilder.deleteRow(Columns, timestamp, mutation, table.cfName, column.name.toString());
     }
 
-    private static List<ColumnDefinition> createColumnsFromColumnsPartition(RowIterator serializedColumns)
+    private static void addDroppedColumnToSchemaMutation(CFMetaData table, CFMetaData.DroppedColumn column, long timestamp, Mutation mutation)
     {
-        String query = String.format("SELECT * FROM %s.%s", NAME, COLUMNS);
-        return createColumnsFromColumnRows(QueryProcessor.resultify(query, serializedColumns));
+        RowUpdateBuilder adder = new RowUpdateBuilder(DroppedColumns, timestamp, mutation).clustering(table.cfName, column.name);
+
+        adder.add("dropped_time", new Date(TimeUnit.MICROSECONDS.toMillis(column.droppedTime)))
+             .add("type", expandUserTypes(column.type).asCQL3Type().toString())
+             .build();
     }
 
-    private static CFMetaData createTableFromTablePartition(RowIterator partition)
+    private static void addTriggerToSchemaMutation(CFMetaData table, TriggerMetadata trigger, long timestamp, Mutation mutation)
     {
-        String query = String.format("SELECT * FROM %s.%s", NAME, TABLES);
-        return createTableFromTableRow(QueryProcessor.resultify(query, partition).one());
+        new RowUpdateBuilder(Triggers, timestamp, mutation)
+            .clustering(table.cfName, trigger.name)
+            .frozenMap("options", Collections.singletonMap("class", trigger.classOption))
+            .build();
     }
 
-    public static CFMetaData createTableFromTablePartitionAndColumnsPartition(RowIterator tablePartition,
-                                                                              RowIterator columnsPartition)
+    private static void dropTriggerFromSchemaMutation(CFMetaData table, TriggerMetadata trigger, long timestamp, Mutation mutation)
     {
-        List<ColumnDefinition> columns = createColumnsFromColumnsPartition(columnsPartition);
-        String query = String.format("SELECT * FROM %s.%s", NAME, TABLES);
-        return createTableFromTableRowAndColumns(QueryProcessor.resultify(query, tablePartition).one(), columns);
+        RowUpdateBuilder.deleteRow(Triggers, timestamp, mutation, table.cfName, trigger.name);
     }
 
-    /**
-     * Deserialize table metadata from low-level representation
-     *
-     * @return Metadata deserialized from schema
-     */
-    private static CFMetaData createTableFromTableRow(UntypedResultSet.Row row)
+    public static Mutation makeCreateViewMutation(KeyspaceMetadata keyspace, ViewDefinition view, long timestamp)
     {
-        String keyspace = row.getString("keyspace_name");
-        String table = row.getString("table_name");
-
-        List<ColumnDefinition> columns =
-            readSchemaPartitionForTableAndApply(COLUMNS, keyspace, table, SchemaKeyspace::createColumnsFromColumnsPartition);
-
-        Map<ByteBuffer, CFMetaData.DroppedColumn> droppedColumns =
-            readSchemaPartitionForTableAndApply(DROPPED_COLUMNS, keyspace, table, SchemaKeyspace::createDroppedColumnsFromDroppedColumnsPartition);
-
-        Triggers triggers =
-            readSchemaPartitionForTableAndApply(TRIGGERS, keyspace, table, SchemaKeyspace::createTriggersFromTriggersPartition);
-
-        CFMetaData cfm = createTableFromTableRowAndColumns(row, columns).droppedColumns(droppedColumns)
-                                                                        .triggers(triggers);
-
-        // 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;
+        // Include the serialized keyspace in case the target node missed a CREATE KEYSPACE migration (see CASSANDRA-5631).
+        Mutation mutation = makeCreateKeyspaceMutation(keyspace.name, keyspace.params, timestamp);
+        addViewToSchemaMutation(view, timestamp, true, mutation);
+        return mutation;
     }
 
-    public static CFMetaData createTableFromTableRowAndColumns(UntypedResultSet.Row row, List<ColumnDefinition> columns)
-    {
-        String keyspace = row.getString("keyspace_name");
-        String table = row.getString("table_name");
-        UUID id = row.getUUID("id");
-
-        Set<CFMetaData.Flag> flags = row.has("flags")
-                                   ? CFMetaData.flagsFromStrings(row.getFrozenSet("flags", UTF8Type.instance))
-                                   : Collections.emptySet();
-
-        boolean isSuper = flags.contains(CFMetaData.Flag.SUPER);
-        boolean isCounter = flags.contains(CFMetaData.Flag.COUNTER);
-        boolean isDense = flags.contains(CFMetaData.Flag.DENSE);
-        boolean isCompound = flags.contains(CFMetaData.Flag.COMPOUND);
-
-        return CFMetaData.create(keyspace,
-                                 table,
-                                 id,
-                                 isDense,
-                                 isCompound,
-                                 isSuper,
-                                 isCounter,
-                                 false,
-                                 columns,
-                                 DatabaseDescriptor.getPartitioner())
-                         .params(createTableParamsFromRow(row));
-    }
-
-    private static TableParams createTableParamsFromRow(UntypedResultSet.Row row)
-    {
-        TableParams.Builder builder = TableParams.builder();
-
-        builder.bloomFilterFpChance(row.getDouble("bloom_filter_fp_chance"))
-               .caching(CachingParams.fromMap(row.getFrozenTextMap("caching")))
-               .comment(row.getString("comment"))
-               .compaction(CompactionParams.fromMap(row.getFrozenTextMap("compaction")))
-               .compression(CompressionParams.fromMap(row.getFrozenTextMap("compression")))
-               .dcLocalReadRepairChance(row.getDouble("dclocal_read_repair_chance"))
-               .defaultTimeToLive(row.getInt("default_time_to_live"))
-               .gcGraceSeconds(row.getInt("gc_grace_seconds"))
-               .maxIndexInterval(row.getInt("max_index_interval"))
-               .memtableFlushPeriodInMs(row.getInt("memtable_flush_period_in_ms"))
-               .minIndexInterval(row.getInt("min_index_interval"))
-               .readRepairChance(row.getDouble("read_repair_chance"))
-               .crcCheckChance(row.getDouble("crc_check_chance"))
-               .speculativeRetry(SpeculativeRetryParam.fromString(row.getString("speculative_retry")));
-
-        if (row.has("extensions"))
-            builder.extensions(row.getFrozenMap("extensions", UTF8Type.instance, BytesType.instance));
-
-        return builder.build();
-    }
-
-    /*
-     * Column metadata serialization/deserialization.
-     */
-
-    private static void addColumnToSchemaMutation(CFMetaData table, ColumnDefinition column, long timestamp, Mutation mutation)
-    {
-        RowUpdateBuilder adder = new RowUpdateBuilder(Columns, timestamp, mutation).clustering(table.cfName, column.name.toString());
-
-        AbstractType<?> type = column.type;
-        if (type instanceof ReversedType)
-            type = ((ReversedType) type).baseType;
-
-        adder.add("column_name_bytes", column.name.bytes)
-             .add("kind", column.kind.toString().toLowerCase())
-             .add("position", column.isOnAllComponents() ? ColumnDefinition.NO_POSITION : column.position())
-             .add("clustering_order", column.clusteringOrder().toString().toLowerCase())
-             .add("type", type.toString())
-             .build();
-    }
-
-    private static void dropColumnFromSchemaMutation(CFMetaData table, ColumnDefinition column, long timestamp, Mutation mutation)
-    {
-        // Note: we do want to use name.toString(), not name.bytes directly for backward compatibility (For CQL3, this won't make a difference).
-        RowUpdateBuilder.deleteRow(Columns, timestamp, mutation, table.cfName, column.name.toString());
-    }
-
-    private static List<ColumnDefinition> createColumnsFromColumnRows(UntypedResultSet rows)
-{
-        List<ColumnDefinition> columns = new ArrayList<>(rows.size());
-        rows.forEach(row -> columns.add(createColumnFromColumnRow(row)));
-        return columns;
-    }
-
-    private static ColumnDefinition createColumnFromColumnRow(UntypedResultSet.Row row)
-    {
-        String keyspace = row.getString("keyspace_name");
-        String table = row.getString("table_name");
-
-        ColumnIdentifier name = ColumnIdentifier.getInterned(row.getBytes("column_name_bytes"), row.getString("column_name"));
-
-        ColumnDefinition.Kind kind = ColumnDefinition.Kind.valueOf(row.getString("kind").toUpperCase());
-
-        int position = row.getInt("position");
-        ClusteringOrder order = ClusteringOrder.valueOf(row.getString("clustering_order").toUpperCase());
-
-        AbstractType<?> type = parseType(row.getString("type"));
-        if (order == ClusteringOrder.DESC)
-            type = ReversedType.getInstance(type);
-
-        return new ColumnDefinition(keyspace, table, name, type, position, kind);
-    }
-
-    /*
-     * Dropped column metadata serialization/deserialization.
-     */
-
-    private static void addDroppedColumnToSchemaMutation(CFMetaData table, CFMetaData.DroppedColumn column, long timestamp, Mutation mutation)
-    {
-        RowUpdateBuilder adder = new RowUpdateBuilder(DroppedColumns, timestamp, mutation).clustering(table.cfName, column.name);
-
-        adder.add("dropped_time", new Date(TimeUnit.MICROSECONDS.toMillis(column.droppedTime)))
-             .add("type", column.type.toString())
-             .build();
-    }
-
-    private static Map<ByteBuffer, CFMetaData.DroppedColumn> createDroppedColumnsFromDroppedColumnsPartition(RowIterator serializedColumns)
-    {
-        String query = String.format("SELECT * FROM %s.%s", NAME, DROPPED_COLUMNS);
-        Map<ByteBuffer, CFMetaData.DroppedColumn> columns = new HashMap<>();
-        for (CFMetaData.DroppedColumn column : createDroppedColumnsFromDroppedColumnRows(QueryProcessor.resultify(query, serializedColumns)))
-            columns.put(UTF8Type.instance.decompose(column.name), column);
-        return columns;
-    }
-
-    private static List<CFMetaData.DroppedColumn> createDroppedColumnsFromDroppedColumnRows(UntypedResultSet rows)
-    {
-        List<CFMetaData.DroppedColumn> columns = new ArrayList<>(rows.size());
-        rows.forEach(row -> columns.add(createDroppedColumnFromDroppedColumnRow(row)));
-        return columns;
-    }
-
-    private static CFMetaData.DroppedColumn createDroppedColumnFromDroppedColumnRow(UntypedResultSet.Row row)
-    {
-        String name = row.getString("column_name");
-        AbstractType<?> type = TypeParser.parse(row.getString("type"));
-        long droppedTime = TimeUnit.MILLISECONDS.toMicros(row.getLong("dropped_time"));
-
-        return new CFMetaData.DroppedColumn(name, type, droppedTime);
-    }
-
-    /*
-     * Trigger metadata serialization/deserialization.
-     */
-
-    private static void addTriggerToSchemaMutation(CFMetaData table, TriggerMetadata trigger, long timestamp, Mutation mutation)
-    {
-        new RowUpdateBuilder(Triggers, timestamp, mutation)
-            .clustering(table.cfName, trigger.name)
-            .frozenMap("options", Collections.singletonMap("class", trigger.classOption))
-            .build();
-    }
-
-    private static void dropTriggerFromSchemaMutation(CFMetaData table, TriggerMetadata trigger, long timestamp, Mutation mutation)
-    {
-        RowUpdateBuilder.deleteRow(Triggers, timestamp, mutation, table.cfName, trigger.name);
-    }
-
-    /**
-     * Deserialize triggers from storage-level representation.
-     *
-     * @param partition storage-level partition containing the trigger definitions
-     * @return the list of processed TriggerDefinitions
-     */
-    private static Triggers createTriggersFromTriggersPartition(RowIterator partition)
-    {
-        Triggers.Builder triggers = org.apache.cassandra.schema.Triggers.builder();
-        String query = String.format("SELECT * FROM %s.%s", NAME, TRIGGERS);
-        QueryProcessor.resultify(query, partition).forEach(row -> triggers.add(createTriggerFromTriggerRow(row)));
-        return triggers.build();
-    }
-
-    private static TriggerMetadata createTriggerFromTriggerRow(UntypedResultSet.Row row)
-    {
-        String name = row.getString("trigger_name");
-        String classOption = row.getFrozenTextMap("options").get("class");
-        return new TriggerMetadata(name, classOption);
-    }
-
-    /*
-     * View metadata serialization/deserialization.
-     */
-
-    public static Mutation makeCreateViewMutation(KeyspaceMetadata keyspace, ViewDefinition view, long timestamp)
-    {
-        // Include the serialized keyspace in case the target node missed a CREATE KEYSPACE migration (see CASSANDRA-5631).
-        Mutation mutation = makeCreateKeyspaceMutation(keyspace.name, keyspace.params, timestamp);
-        addViewToSchemaMutation(view, timestamp, true, mutation);
-        return mutation;
-    }
-
-    private static void addViewToSchemaMutation(ViewDefinition view, long timestamp, boolean includeColumns, Mutation mutation)
+    private static void addViewToSchemaMutation(ViewDefinition view, long timestamp, boolean includeColumns, Mutation mutation)
     {
         RowUpdateBuilder builder = new RowUpdateBuilder(Views, timestamp, mutation)
             .clustering(view.viewName);
@@ -1363,9 +729,7 @@ public final class SchemaKeyspace
 
         // columns that are no longer needed
         for (ColumnDefinition column : columnDiff.entriesOnlyOnLeft().values())
-        {
             dropColumnFromSchemaMutation(oldView.metadata, column, timestamp, mutation);
-        }
 
         // newly added columns
         for (ColumnDefinition column : columnDiff.entriesOnlyOnRight().values())
@@ -1390,122 +754,303 @@ public final class SchemaKeyspace
         return mutation;
     }
 
-    public static ViewDefinition createViewFromName(String keyspace, String view)
+    private static void addIndexToSchemaMutation(CFMetaData table,
+                                                 IndexMetadata index,
+                                                 long timestamp,
+                                                 Mutation mutation)
     {
-        return readSchemaPartitionForTableAndApply(VIEWS, keyspace, view, partition ->
-        {
-            if (partition.isEmpty())
-                throw new RuntimeException(String.format("%s:%s not found in the schema definitions keyspace.", keyspace, view));
+        RowUpdateBuilder builder = new RowUpdateBuilder(Indexes, timestamp, mutation).clustering(table.cfName, index.name);
 
-            return createViewFromViewPartition(partition);
-        });
+        builder.add("kind", index.kind.toString());
+        builder.frozenMap("options", index.options);
+        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 ViewDefinition createViewFromViewPartition(RowIterator partition)
+    private static void addUpdatedIndexToSchemaMutation(CFMetaData table,
+                                                        IndexMetadata index,
+                                                        long timestamp,
+                                                        Mutation mutation)
     {
-        String query = String.format("SELECT * FROM %s.%s", NAME, VIEWS);
-        return createViewFromViewRow(QueryProcessor.resultify(query, partition).one());
+        addIndexToSchemaMutation(table, index, timestamp, mutation);
     }
 
-    /**
-     * Deserialize views from storage-level representation.
-     *
-     * @param partition storage-level partition containing the view definitions
-     * @return the list of processed ViewDefinitions
+    public static Mutation makeCreateFunctionMutation(KeyspaceMetadata keyspace, UDFunction function, long timestamp)
+    {
+        // Include the serialized keyspace in case the target node missed a CREATE KEYSPACE migration (see CASSANDRA-5631).
+        Mutation mutation = makeCreateKeyspaceMutation(keyspace.name, keyspace.params, timestamp);
+        addFunctionToSchemaMutation(function, timestamp, mutation);
+        return mutation;
+    }
+
+    static void addFunctionToSchemaMutation(UDFunction function, long timestamp, Mutation mutation)
+    {
+        RowUpdateBuilder adder =
+            new RowUpdateBuilder(Functions, timestamp, mutation).clustering(function.name().name, functionArgumentsList(function));
+
+        adder.add("body", function.body())
+             .add("language", function.language())
+             .add("return_type", function.returnType().asCQL3Type().toString())
+             .add("called_on_null_input", function.isCalledOnNullInput())
+             .frozenList("argument_names", function.argNames().stream().map((c) -> bbToString(c.bytes)).collect(toList()));
+
+        adder.build();
+    }
+
+    private static List<String> functionArgumentsList(AbstractFunction fun)
+    {
+        return fun.argTypes()
+                  .stream()
+                  .map(AbstractType::asCQL3Type)
+                  .map(CQL3Type::toString)
+                  .collect(toList());
+    }
+
+    public static Mutation makeDropFunctionMutation(KeyspaceMetadata keyspace, UDFunction function, long timestamp)
+    {
+        // Include the serialized keyspace in case the target node missed a CREATE KEYSPACE migration (see CASSANDRA-5631).
+        Mutation mutation = makeCreateKeyspaceMutation(keyspace.name, keyspace.params, timestamp);
+        return RowUpdateBuilder.deleteRow(Functions, timestamp, mutation, function.name().name, functionArgumentsList(function));
+    }
+
+    public static Mutation makeCreateAggregateMutation(KeyspaceMetadata keyspace, UDAggregate aggregate, long timestamp)
+    {
+        // Include the serialized keyspace in case the target node missed a CREATE KEYSPACE migration (see CASSANDRA-5631).
+        Mutation mutation = makeCreateKeyspaceMutation(keyspace.name, keyspace.params, timestamp);
+        addAggregateToSchemaMutation(aggregate, timestamp, mutation);
+        return mutation;
+    }
+
+    static void addAggregateToSchemaMutation(UDAggregate aggregate, long timestamp, Mutation mutation)
+    {
+        RowUpdateBuilder adder =
+            new RowUpdateBuilder(Aggregates, timestamp, mutation) .clustering(aggregate.name().name, functionArgumentsList(aggregate));
+
+        adder.add("return_type", aggregate.returnType().asCQL3Type().toString())
+             .add("state_func", aggregate.stateFunction().name().name)
+             .add("state_type", aggregate.stateType() != null ? aggregate.stateType().asCQL3Type().toString() : null)
+             .add("final_func", aggregate.finalFunction() != null ? aggregate.finalFunction().name().name : null)
+             .add("initcond", aggregate.initialCondition())
+             .build();
+    }
+
+    public static Mutation makeDropAggregateMutation(KeyspaceMetadata keyspace, UDAggregate aggregate, long timestamp)
+    {
+        // Include the serialized keyspace in case the target node missed a CREATE KEYSPACE migration (see CASSANDRA-5631).
+        Mutation mutation = makeCreateKeyspaceMutation(keyspace.name, keyspace.params, timestamp);
+        return RowUpdateBuilder.deleteRow(Aggregates, timestamp, mutation, aggregate.name().name, functionArgumentsList(aggregate));
+    }
+
+    /*
+     * Fetching schema
      */
-    private static Views createViewsFromViewsPartition(RowIterator partition)
+
+    public static Keyspaces fetchNonSystemKeyspaces()
     {
-        Views.Builder views = org.apache.cassandra.schema.Views.builder();
-        String query = String.format("SELECT * FROM %s.%s", NAME, VIEWS);
-        for (UntypedResultSet.Row row : QueryProcessor.resultify(query, partition))
+        return fetchKeyspacesWithout(Schema.SYSTEM_KEYSPACE_NAMES);
+    }
+
+    private static Keyspaces fetchKeyspacesWithout(Set<String> excludedKeyspaceNames)
+    {
+        String query = format("SELECT keyspace_name FROM %s.%s", NAME, KEYSPACES);
+
+        Keyspaces.Builder keyspaces = org.apache.cassandra.schema.Keyspaces.builder();
+        for (UntypedResultSet.Row row : query(query))
         {
-            ViewDefinition view = createViewFromViewRow(row);
-            views.add(view);
+            String keyspaceName = row.getString("keyspace_name");
+            if (!excludedKeyspaceNames.contains(keyspaceName))
+                keyspaces.add(fetchKeyspace(keyspaceName));
         }
-        return views.build();
+        return keyspaces.build();
     }
 
-    private static ViewDefinition createViewFromViewRow(UntypedResultSet.Row row)
+    private static Keyspaces fetchKeyspacesOnly(Set<String> includedKeyspaceNames)
     {
-        String keyspace = row.getString("keyspace_name");
-        String view = row.getString("view_name");
-        UUID id = row.getUUID("id");
-        UUID baseTableId = row.getUUID("base_table_id");
-        String baseTableName = row.getString("base_table_name");
-        boolean includeAll = row.getBoolean("include_all_columns");
-        String whereClause = row.getString("where_clause");
+        /*
+         * We know the keyspace names we are going to query, but we still want to run the SELECT IN
+         * query, to filter out the keyspaces that had been dropped by the applied mutation set.
+         */
+        String query = format("SELECT keyspace_name FROM %s.%s WHERE keyspace_name IN ?", NAME, KEYSPACES);
 
-        List<ColumnDefinition> columns =
-            readSchemaPartitionForTableAndApply(COLUMNS, keyspace, view, SchemaKeyspace::createColumnsFromColumnsPartition);
+        Keyspaces.Builder keyspaces = org.apache.cassandra.schema.Keyspaces.builder();
+        for (UntypedResultSet.Row row : query(query, new ArrayList<>(includedKeyspaceNames)))
+            keyspaces.add(fetchKeyspace(row.getString("keyspace_name")));
+        return keyspaces.build();
+    }
 
-        Map<ByteBuffer, CFMetaData.DroppedColumn> droppedColumns =
-            readSchemaPartitionForTableAndApply(DROPPED_COLUMNS, keyspace, view, SchemaKeyspace::createDroppedColumnsFromDroppedColumnsPartition);
+    private static KeyspaceMetadata fetchKeyspace(String keyspaceName)
+    {
+        KeyspaceParams params = fetchKeyspaceParams(keyspaceName);
+        Types types = fetchTypes(keyspaceName);
+        Tables tables = fetchTables(keyspaceName, types);
+        Views views = fetchViews(keyspaceName, types);
+        Functions functions = fetchFunctions(keyspaceName, types);
+        return KeyspaceMetadata.create(keyspaceName, params, tables, views, types, functions);
+    }
 
-        CFMetaData cfm = CFMetaData.create(keyspace,
-                                           view,
-                                           id,
-                                           false,
-                                           true,
-                                           false,
-                                           false,
-                                           true,
-                                           columns,
-                                           DatabaseDescriptor.getPartitioner())
-                                   .params(createTableParamsFromRow(row))
-                                   .droppedColumns(droppedColumns);
+    private static KeyspaceParams fetchKeyspaceParams(String keyspaceName)
+    {
+        String query = format("SELECT * FROM %s.%s WHERE keyspace_name = ?", NAME, KEYSPACES);
+
+        UntypedResultSet.Row row = query(query, keyspaceName).one();
+        boolean durableWrites = row.getBoolean(KeyspaceParams.Option.DURABLE_WRITES.toString());
+        Map<String, String> replication = row.getFrozenTextMap(KeyspaceParams.Option.REPLICATION.toString());
+        return KeyspaceParams.create(durableWrites, replication);
+    }
 
-        String rawSelect = View.buildSelectStatement(baseTableName, columns, whereClause);
-        SelectStatement.RawStatement rawStatement = (SelectStatement.RawStatement) QueryProcessor.parseStatement(rawSelect);
+    private static Types fetchTypes(String keyspaceName)
+    {
+        String query = format("SELECT * FROM %s.%s WHERE keyspace_name = ?", NAME, TYPES);
 
-        return new ViewDefinition(keyspace, view, baseTableId, baseTableName, includeAll, rawStatement, whereClause, cfm);
+        Types.RawBuilder types = org.apache.cassandra.schema.Types.rawBuilder(keyspaceName);
+        for (UntypedResultSet.Row row : query(query, keyspaceName))
+        {
+            String name = row.getString("type_name");
+            List<String> fieldNames = row.getFrozenList("field_names", UTF8Type.instance);
+            List<String> fieldTypes = row.getFrozenList("field_types", UTF8Type.instance);
+            types.add(name, fieldNames, fieldTypes);
+        }
+        return types.build();
     }
 
-    /*
-     * Secondary Index metadata serialization/deserialization.
-     */
+    private static Tables fetchTables(String keyspaceName, Types types)
+    {
+        String query = format("SELECT table_name FROM %s.%s WHERE keyspace_name = ?", NAME, TABLES);
 
-    private static void addIndexToSchemaMutation(CFMetaData table,
-                                                 IndexMetadata index,
-                                                 long timestamp,
-                                                 Mutation mutation)
+        Tables.Builder tables = org.apache.cassandra.schema.Tables.builder();
+        for (UntypedResultSet.Row row : query(query, keyspaceName))
+            tables.add(fetchTable(keyspaceName, row.getString("table_name"), types));
+        return tables.build();
+    }
+
+    private static CFMetaData fetchTable(String keyspaceName, String tableName, Types types)
     {
-        RowUpdateBuilder builder = new RowUpdateBuilder(Indexes, timestamp, mutation).clustering(table.cfName, index.name);
+        String query = String.format("SELECT * FROM %s.%s WHERE keyspace_name = ? AND table_name = ?", NAME, TABLES);
+        UntypedResultSet rows = query(query, keyspaceName, tableName);
+        if (rows.isEmpty())
+            throw new RuntimeException(String.format("%s:%s not found in the schema definitions keyspace.", keyspaceName, tableName));
+        UntypedResultSet.Row row = rows.one();
 
-        builder.add("kind", index.kind.toString());
-        builder.frozenMap("options", index.options);
-        builder.build();
+        UUID id = row.getUUID("id");
+
+        Set<CFMetaData.Flag> flags = CFMetaData.flagsFromStrings(row.getFrozenSet("flags", UTF8Type.instance));
+
+        boolean isSuper = flags.contains(CFMetaData.Flag.SUPER);
+        boolean isCounter = flags.contains(CFMetaData.Flag.COUNTER);
+        boolean isDense = flags.contains(CFMetaData.Flag.DENSE);
+        boolean isCompound = flags.contains(CFMetaData.Flag.COMPOUND);
+
+        List<ColumnDefinition> columns = fetchColumns(keyspaceName, tableName, types);
+        Map<ByteBuffer, CFMetaData.DroppedColumn> droppedColumns = fetchDroppedColumns(keyspaceName, tableName);
+        Indexes indexes = fetchIndexes(keyspaceName, tableName);
+        Triggers triggers = fetchTriggers(keyspaceName, tableName);
+
+        return CFMetaData.create(keyspaceName,
+                                 tableName,
+                                 id,
+                                 isDense,
+                                 isCompound,
+                                 isSuper,
+                                 isCounter,
+                                 false,
+                                 columns,
+                                 DatabaseDescriptor.getPartitioner())
+                         .params(createTableParamsFromRow(row))
+                         .droppedColumns(droppedColumns)
+                         .indexes(indexes)
+                         .triggers(triggers);
+    }
+
+    public static TableParams createTableParamsFromRow(UntypedResultSet.Row row)
+    {
+        return TableParams.builder()
+                          .bloomFilterFpChance(row.getDouble("bloom_filter_fp_chance"))
+                          .caching(CachingParams.fromMap(row.getFrozenTextMap("caching")))
+                          .comment(row.getString("comment"))
+                          .compaction(CompactionParams.fromMap(row.getFrozenTextMap("compaction")))
+                          .compression(CompressionParams.fromMap(row.getFrozenTextMap("compression")))
+                          .dcLocalReadRepairChance(row.getDouble("dclocal_read_repair_chance"))
+                          .defaultTimeToLive(row.getInt("default_time_to_live"))
+                          .extensions(row.getFrozenMap("extensions", UTF8Type.instance, BytesType.instance))
+                          .gcGraceSeconds(row.getInt("gc_grace_seconds"))
+                          .maxIndexInterval(row.getInt("max_index_interval"))
+                          .memtableFlushPeriodInMs(row.getInt("memtable_flush_period_in_ms"))
+                          .minIndexInterval(row.getInt("min_index_interval"))
+                          .readRepairChance(row.getDouble("read_repair_chance"))
+                          .crcCheckChance(row.getDouble("crc_check_chance"))
+                          .speculativeRetry(SpeculativeRetryParam.fromString(row.getString("speculative_retry")))
+                          .build();
+    }
+
+    private static List<ColumnDefinition> fetchColumns(String keyspace, String table, Types types)
+    {
+        String query = format("SELECT * FROM %s.%s WHERE keyspace_name = ? AND table_name = ?", NAME, COLUMNS);
+        List<ColumnDefinition> columns = new ArrayList<>();
+        query(query, keyspace, table).forEach(row -> columns.add(createColumnFromRow(row, types)));
+        return columns;
     }
 
-    private static void dropIndexFromSchemaMutation(CFMetaData table,
-                                                    IndexMetadata index,
-                                                    long timestamp,
-                                                    Mutation mutation)
+    public static ColumnDefinition createColumnFromRow(UntypedResultSet.Row row, Types types)
     {
-        RowUpdateBuilder.deleteRow(Indexes, timestamp, mutation, table.cfName, index.name);
+        String keyspace = row.getString("keyspace_name");
+        String table = row.getString("table_name");
+
+        ColumnIdentifier name = ColumnIdentifier.getInterned(row.getBytes("column_name_bytes"), row.getString("column_name"));
+
+        ColumnDefinition.Kind kind = ColumnDefinition.Kind.valueOf(row.getString("kind").toUpperCase());
+
+        int position = row.getInt("position");
+        ClusteringOrder order = ClusteringOrder.valueOf(row.getString("clustering_order").toUpperCase());
+
+        AbstractType<?> type = parse(keyspace, row.getString("type"), types);
+        if (order == ClusteringOrder.DESC)
+            type = ReversedType.getInstance(type);
+
+        return new ColumnDefinition(keyspace, table, name, type, position, kind);
     }
 
-    private static void addUpdatedIndexToSchemaMutation(CFMetaData table,
-                                                        IndexMetadata index,
-                                                        long timestamp,
-                                                        Mutation mutation)
+    private static Map<ByteBuffer, CFMetaData.DroppedColumn> fetchDroppedColumns(String keyspace, String table)
     {
-        addIndexToSchemaMutation(table, index, timestamp, mutation);
+        String query = format("SELECT * FROM %s.%s WHERE keyspace_name = ? AND table_name = ?", NAME, DROPPED_COLUMNS);
+        Map<ByteBuffer, CFMetaData.DroppedColumn> columns = new HashMap<>();
+        for (UntypedResultSet.Row row : query(query, keyspace, table))
+        {
+            CFMetaData.DroppedColumn column = createDroppedColumnFromRow(row);
+            columns.put(UTF8Type.instance.decompose(column.name), column);
+        }
+        return columns;
     }
-    /**
-     * 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)
+
+    private static CFMetaData.DroppedColumn createDroppedColumnFromRow(UntypedResultSet.Row row)
+    {
+        String keyspace = row.getString("keyspace_name");
+        String name = row.getString("column_name");
+        /*
+         * we never store actual UDT names in dropped column types (so that we can safely drop types if nothing refers to
+         * them anymore), so before storing dropped columns in schema we expand UDTs to tuples. See expandUserTypes method.
+         * Because of that, we can safely pass Types.none() to parse()
+         */
+        AbstractType<?> type = parse(keyspace, row.getString("type"), org.apache.cassandra.schema.Types.none());
+        long droppedTime = TimeUnit.MILLISECONDS.toMicros(row.getLong("dropped_time"));
+        return new CFMetaData.DroppedColumn(name, type, droppedTime);
+    }
+
+    private static Indexes fetchIndexes(String keyspace, String table)
     {
+        String query = String.format("SELECT * FROM %s.%s WHERE keyspace_name = ? AND table_name = ?", NAME, INDEXES);
         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(row)));
+        query(query, keyspace, table).forEach(row -> indexes.add(createIndexMetadataFromRow(row)));
         return indexes.build();
     }
 
-    private static IndexMetadata createIndexMetadataFromIndexesRow(UntypedResultSet.Row row)
+    private static IndexMetadata createIndexMetadataFromRow(UntypedResultSet.Row row)
     {
         String name = row.getString("index_name");
         IndexMetadata.Kind type = IndexMetadata.Kind.valueOf(row.getString("kind"));
@@ -1513,66 +1058,104 @@ public final class SchemaKeyspace
         return IndexMetadata.fromSchemaMetadata(name, type, options);
     }
 
-    /*
-     * UDF metadata serialization/deserialization.
-     */
+    private static Triggers fetchTriggers(String keyspace, String table)
+    {
+        String query = String.format("SELECT * FROM %s.%s WHERE keyspace_name = ? AND table_name = ?", NAME, TRIGGERS);
+        Triggers.Builder triggers = org.apache.cassandra.schema.Triggers.builder();
+        query(query, keyspace, table).forEach(row -> triggers.add(createTriggerFromRow(row)));
+        return triggers.build();
+    }
 
-    public static Mutation makeCreateFunctionMutation(KeyspaceMetadata keyspace, UDFunction function, long timestamp)
+    private static TriggerMetadata createTriggerFromRow(UntypedResultSet.Row row)
     {
-        // Include the serialized keyspace in case the target node missed a CREATE KEYSPACE migration (see CASSANDRA-5631).
-        Mutation mutation = makeCreateKeyspaceMutation(keyspace.name, keyspace.params, timestamp);
-        addFunctionToSchemaMutation(function, timestamp, mutation);
-        return mutation;
+        String name = row.getString("trigger_name");
+        String classOption = row.getFrozenTextMap("options").get("class");
+        return new TriggerMetadata(name, classOption);
     }
 
-    static void addFunctionToSchemaMutation(UDFunction function, long timestamp, Mutation mutation)
+    private static Views fetchViews(String keyspaceName, Types types)
     {
-        RowUpdateBuilder adder = new RowUpdateBuilder(Functions, timestamp, mutation)
-                                 .clustering(function.name().name, functionSignatureWithTypes(function));
+        String query = format("SELECT view_name FROM %s.%s WHERE keyspace_name = ?", NAME, VIEWS);
 
-        adder.add("body", function.body())
-             .add("language", function.language())
-             .add("return_type", function.returnType().toString())
-             .add("called_on_null_input", function.isCalledOnNullInput())
-             .frozenList("argument_names", function.argNames().stream().map((c) -> bbToString(c.bytes)).collect(Collectors.toList()))
-             .frozenList("argument_types", function.argTypes().stream().map(AbstractType::toString).collect(Collectors.toList()));
+        Views.Builder views = org.apache.cassandra.schema.Views.builder();
+        for (UntypedResultSet.Row row : query(query, keyspaceName))
+            views.add(fetchView(keyspaceName, row.getString("view_name"), types));
+        return views.build();
+    }
 
-        adder.build();
+    private static ViewDefinition fetchView(String keyspaceName, String viewName, Types types)
+    {
+        String query = String.format("SELECT * FROM %s.%s WHERE keyspace_name = ? AND view_name = ?", NAME, VIEWS);
+        UntypedResultSet rows = query(query, keyspaceName, viewName);
+        if (rows.isEmpty())
+            throw new RuntimeException(String.format("%s:%s not found in the schema definitions keyspace.", keyspaceName, viewName));
+        UntypedResultSet.Row row = rows.one();
+
+        UUID id = row.getUUID("id");
+        UUID baseTableId = row.getUUID("base_table_id");
+        String baseTableName = row.getString("base_table_name");
+        boolean includeAll = row.getBoolean("include_all_columns");
+        String whereClause = row.getString("where_clause");
+
+        List<ColumnDefinition> columns = fetchColumns(keyspaceName, viewName, types);
+
+        Map<ByteBuffer, CFMetaData.DroppedColumn> droppedColumns = fetchDroppedColumns(keyspaceName, viewName);
+
+        CFMetaData cfm = CFMetaData.create(keyspaceName,
+                                           viewName,
+                                           id,
+                                           false,
+                                           true,
+                                           false,
+                                           false,
+                                           true,
+                                           columns,
+                                           DatabaseDescriptor.getPartitioner())
+                                   .params(createTableParamsFromRow(row))
+                                   .droppedColumns(droppedColumns);
+
+            String rawSelect = View.buildSelectStatement(baseTableName, columns, whereClause);
+            SelectStatement.RawStatement rawStatement = (SelectStatement.RawStatement) QueryProcessor.parseStatement(rawSelect);
+
+            return new ViewDefinition(keyspaceName, viewName, baseTableId, baseTableName, includeAll, rawStatement, whereClause, cfm);
     }
 
-    public static Mutation makeDropFunctionMutation(KeyspaceMetadata keyspace, UDFunction function, long timestamp)
+    private static Functions fetchFunctions(String keyspaceName, Types types)
     {
-        // Include the serialized keyspace in case the target node missed a CREATE KEYSPACE migration (see CASSANDRA-5631).
-        Mutation mutation = makeCreateKeyspaceMutation(keyspace.name, keyspace.params, timestamp);
-        return RowUpdateBuilder.deleteRow(Functions, timestamp, mutation, function.name().name, functionSignatureWithTypes(function));
+        Functions udfs = fetchUDFs(keyspaceName, types);
+        Functions udas = fetchUDAs(keyspaceName, udfs, types);
+
+        return org.apache.cassandra.schema.Functions.builder()
+                                                    .add(udfs)
+                                                    .add(udas)
+                                                    .build();
     }
 
-    private static Collection<UDFunction> createFunctionsFromFunctionsPartition(RowIterator partition)
+    private static Functions fetchUDFs(String keyspaceName, Types types)
     {
-        List<UDFunction> functions = new ArrayList<>();
-        String query = String.format("SELECT * FROM %s.%s", NAME, FUNCTIONS);
-        for (UntypedResultSet.Row row : QueryProcessor.resultify(query, partition))
-            functions.add(createFunctionFromFunctionRow(row));
-        return functions;
+        String query = format("SELECT * FROM %s.%s WHERE keyspace_name = ?", NAME, FUNCTIONS);
+
+        Functions.Builder functions = org.apache.cassandra.schema.Functions.builder();
+        for (UntypedResultSet.Row row : query(query, keyspaceName))
+            functions.add(createUDFFromRow(row, types));
+        return functions.build();
     }
 
-    private static UDFunction createFunctionFromFunctionRow(UntypedResultSet.Row row)
+    private static UDFunction createUDFFromRow(UntypedResultSet.Row row, Types types)
     {
         String ksName = row.getString("keyspace_name");
         String functionName = row.getString("function_name");
         FunctionName name = new FunctionName(ksName, functionName);
 
         List<ColumnIdentifier> argNames = new ArrayList<>();
-        if (row.has("argument_names"))
-            for (String arg : row.getFrozenList("argument_names", UTF8Type.instance))
-                argNames.add(new ColumnIdentifier(arg, true));
+        for (String arg : row.getFrozenList("argument_names", UTF8Type.instance))
+            argNames.add(new ColumnIdentifier(arg, true));
 
         List<AbstractType<?>> argTypes = new ArrayList<>();
-        if (row.has("argument_types"))
-            for (String type : row.getFrozenList("argument_types", UTF8Type.instance))
-                argTypes.add(parseType(type));
+        for (String type : row.getFrozenList("argument_types", UTF8Type.instance))
+            argTypes.add(parse(ksName, type, types));
 
-        AbstractType<?> returnType = parseType(row.getString("return_type"));
+        AbstractType<?> returnType = parse(ksName, row.getString("return_type"), types);
 
         String language = row.getString("language");
         String body = row.getString("body");
@@ -1609,70 +1192,33 @@ public final class SchemaKeyspace
         }
     }
 
-    /*
-     * Aggregate UDF metadata serialization/deserialization.
-     */
-
-    public static Mutation makeCreateAggregateMutation(KeyspaceMetadata keyspace, UDAggregate aggregate, long timestamp)
-    {
-        // Include the serialized keyspace in case the target node missed a CREATE KEYSPACE migration (see CASSANDRA-5631).
-        Mutation mutation = makeCreateKeyspaceMutation(keyspace.name, keyspace.params, timestamp);
-        addAggregateToSchemaMutation(aggregate, timestamp, mutation);
-        return mutation;
-    }
-
-    static void addAggregateToSchemaMutation(UDAggregate aggregate, long timestamp, Mutation mutation)
-    {
-        RowUpdateBuilder adder = new RowUpdateBuilder(Aggregates, timestamp, mutation)
-                                 .clustering(aggregate.name().name, functionSignatureWithTypes(aggregate));
-
-        adder.add("return_type", aggregate.returnType().toString())
-             .add("state_func", aggregate.stateFunction().name().name)
-             .add("state_type", aggregate.stateType() != null ? aggregate.stateType().toString() : null)
-             .add("final_func", aggregate.finalFunction() != null ? aggregate.finalFunction().name().name : null)
-             .add("initcond", aggregate.initialCondition())
-             .frozenList("argument_types", aggregate.argTypes().stream().map(AbstractType::toString).collect(Collectors.toList()))
-             .build();
-    }
-
-    private static Functions createAggregatesFromAggregatesPartition(Functions functions, RowIterator partition)
+    private static Functions fetchUDAs(String keyspaceName, Functions udfs, Types types)
     {
-        String query = String.format("SELECT * FROM %s.%s", NAME, AGGREGATES);
-        for (UntypedResultSet.Row row : QueryProcessor.resultify(query, partition))
-            functions = functions.with(createAggregateFromAggregateRow(functions, row));
-        return functions;
-    }
+        String query = format("SELECT * FROM %s.%s WHERE keyspace_name = ?", NAME, AGGREGATES);
 
-    private static UDAggregate createAggregateFromAggregateRow(UntypedResultSet.Row row)
-    {
-        return createAggregateFromAggregateRow(Schema.instance.getKSMetaData(row.getString("keyspace_name")).functions, row);
+        Functions.Builder aggregates = org.apache.cassandra.schema.Functions.builder();
+        for (UntypedResultSet.Row row : query(query, keyspaceName))
+            aggregates.add(createUDAFromRow(row, udfs, types));
+        return aggregates.build();
     }
 
-    private static UDAggregate createAggregateFromAggregateRow(Functions functions, UntypedResultSet.Row row)
+    private static UDAggregate createUDAFromRow(UntypedResultSet.Row row, Functions functions, Types types)
     {
         String ksName = row.getString("keyspace_name");
         String functionName = row.getString("aggregate_name");
         FunctionName name = new FunctionName(ksName, functionName);
 
-        List<String> types = row.getFrozenList("argument_types", UTF8Type.instance);
+        List<AbstractType<?>> argTypes =
+            row.getFrozenList("argument_types", UTF8Type.instance)
+               .stream()
+               .map(t -> parse(ksName, t, types))
+               .collect(toList());
 
-        List<AbstractType<?>> argTypes;
-        if (types == null)
-        {
-            argTypes = Collections.emptyList();
-        }
-        else
-        {
-            argTypes = new ArrayList<>(types.size());
-            for (String type : types)
-                argTypes.add(parseType(type));
-        }
-
-        AbstractType<?> returnType = parseType(row.getString("return_type"));
+        AbstractType<?> returnType = parse(ksName, row.getString("return_type"), types);
 
         FunctionName stateFunc = new FunctionName(ksName, (row.getString("state_func")));
         FunctionName finalFunc = row.has("final_func") ? new FunctionName(ksName, row.getString("final_func")) : null;
-        AbstractType<?> stateType = row.has("state_type") ? parseType(row.getString("state_type")) : null;
+        AbstractType<?> stateType = row.has("state_type") ? parse(ksName, row.getString("state_type"), types) : null;
         ByteBuffer initcond = row.has("initcond") ? row.getBytes("initcond") : null;
 
         try
@@ -1685,30 +1231,171 @@ public final class SchemaKeyspace
         }
     }
 
-    public static Mutation makeDropAggregateMutation(KeyspaceMetadata keyspace, UDAggregate aggregate, long timestamp)
+    private static UntypedResultSet query(String query, Object... variables)
     {
-        // Include the serialized keyspace in case the target node missed a CREATE KEYSPACE migration (see CASSANDRA-5631).
-        Mutation mutation = makeCreateKeyspaceMutation(keyspace.name, keyspace.params, timestamp);
-        return RowUpdateBuilder.deleteRow(Aggregates, timestamp, mutation, aggregate.name().name, functionSignatureWithTypes(aggregate));
+        return executeInternal(query, variables);
+    }
+
+    /*
+     * Merging schema
+     */
+
+    /**
+     * Merge remote schema in form of mutations with local and mutate ks/cf metadata objects
+     * (which also involves fs operations on add/drop ks/cf)
+     *
+     * @param mutations the schema changes to apply
+     *
+     * @throws ConfigurationException If one of metadata attributes has invalid value
+     */
+    public static synchronized void mergeSchemaAndAnnounceVersion(Collection<Mutation> mutations) throws ConfigurationException
+    {
+        mergeSchema(mutations);
+        Schema.instance.updateVersionAndAnnounce();
+    }
+
+    public static synchronized void mergeSchema(Collection<Mutation> mutations)
+    {
+        // only compare the keyspaces affected by this set of schema mutations
+        Set<String> affectedKeyspaces =
+        mutations.stream()
+                 .map(m -> UTF8Type.instance.compose(m.key().getKey()))
+                 .collect(Collectors.toSet());
+
+        // fetch the current state of schema for the affected keyspaces only
+        Keyspaces before = Schema.instance.getKeyspaces(affectedKeyspaces);
+
+        // apply the schema mutations and flush
+        mutations.forEach(Mutation::apply);
+        if (FLUSH_SCHEMA_TABLES)
+            flush();
+
+        // fetch the new state of schema from schema tables (not applied to Schema.instance yet)
+        Keyspaces after = fetchKeyspacesOnly(affectedKeyspaces);
+
+        // deal with the diff
+        MapDifference<String, KeyspaceMetadata> keyspacesDiff = before.diff(after);
+
+        // dropped keyspaces
+        for (KeyspaceMetadata keyspace : keyspacesDiff.entriesOnlyOnLeft().values())
+        {
+            keyspace.functions.udas().forEach(Schema.instance::dropAggregate);
+            keyspace.functions.udfs().forEach(Schema.instance::dropFunction);
+            keyspace.views.forEach(v -> Schema.instance.dropView(v.ksName, v.viewName));
+            keyspace.tables.forEach(t -> Schema.instance.dropTable(t.ksName, t.cfName));
+            keyspace.types.forEach(Schema.instance::dropType);
+            Schema.instance.dropKeyspace(keyspace.name);
+        }
+
+        // new keyspaces
+        for (KeyspaceMetadata keyspace : keyspacesDiff.entriesOnlyOnRight().values())
+        {
+            Schema.instance.addKeyspace(KeyspaceMetadata.create(keyspace.name, keyspace.params));
+            keyspace.types.forEach(Schema.instance::addType);
+            keyspace.tables.forEach(Schema.instance::addTable);
+            keyspace.views.forEach(Schema.instance::addView);
+            keyspace.functions.udfs().forEach(Schema.instance::addFunction);
+            keyspace.functions.udas().forEach(Schema.instance::addAggregate);
+        }
+
+        // updated keyspaces
+        for (Map.Entry<String, MapDifference.ValueDifference<KeyspaceMetadata>> diff : keyspacesDiff.entriesDiffering().entrySet())
+            updateKeyspace(diff.getKey(), diff.getValue().leftValue(), diff.getValue().rightValue());
+    }
+
+    private static void updateKeyspace(String keyspaceName, KeyspaceMetadata keyspaceBefore, KeyspaceMetadata keyspaceAfter)
+    {
+        // calculate the deltas
+        MapDifference<String, CFMetaData> tablesDiff = keyspaceBefore.tables.diff(keyspaceAfter.tables);
+        MapDifference<String, ViewDefinition> viewsDiff = keyspaceBefore.views.diff(keyspaceAfter.views);
+        MapDifference<ByteBuffer, UserType> typesDiff = keyspaceBefore.types.diff(keyspaceAfter.types);
+
+        Map<Pair<FunctionName, List<String>>, UDFunction> udfsBefore = new HashMap<>();
+        keyspaceBefore.functions.udfs().forEach(f -> udfsBefore.put(Pair.create(f.name(), functionArgumentsList(f)), f));
+        Map<Pair<FunctionName, List<String>>, UDFunction> udfsAfter = new HashMap<>();
+        keyspaceAfter.functions.udfs().forEach(f -> udfsAfter.put(Pair.create(f.name(), functionArgumentsList(f)), f));
+        MapDifference<Pair<FunctionName, List<String>>, UDFunction> udfsDiff = Maps.difference(udfsBefore, udfsAfter);
+
+        Map<Pair<FunctionName, List<String>>, UDAggregate> udasBefore = new HashMap<>();
+        keyspaceBefore.functions.udas().forEach(f -> udasBefore.put(Pair.create(f.name(), functionArgumentsList(f)), f));
+        Map<Pair<FunctionName, List<String>>, UDAggregate> udasAfter = new HashMap<>();
+        keyspaceAfter.functions.udas().forEach(f -> udasAfter.put(Pair.create(f.name(), functionArgumentsList(f)), f));
+        MapDifference<Pair<FunctionName, List<String>>, UDAggregate> udasDiff = Maps.difference(udasBefore, udasAfter);
+
+        // update keyspace params, if changed
+        if (!keyspaceBefore.params.equals(keyspaceAfter.params))
+            Schema.instance.updateKeyspace(keyspaceName, keyspaceAfter.params);
+
+        // drop everything removed
+        udasDiff.entriesOnlyOnLeft().values().forEach(Schema.instance::dropAggregate);
+        udfsDiff.entriesOnlyOnLeft().values().forEach(Schema.instance::dropFunction);
+        viewsDiff.entriesOnlyOnLeft().values().forEach(v -> Schema.instance.dropView(v.ksName, v.viewName));
+        tablesDiff.entriesOnlyOnLeft().values().forEach(t -> Schema.instance.dropTable(t.ksName, t.cfName));
+        typesDiff.entriesOnlyOnLeft().values().forEach(Schema.instance::dropType);
+
+        // add everything created
+        typesDiff.entriesOnlyOnRight().values().forEach(Schema.instance::addType);
+        tablesDiff.entriesOnlyOnRight().values().forEach(Schema.instance::addTable);
+        viewsDiff.entriesOnlyOnRight().values().forEach(Schema.instance::addView);
+        udfsDiff.entriesOnlyOnRight().values().forEach(Schema.instance::addFunction);
+        udasDiff.entriesOnlyOnRight().values().forEach(Schema.instance::addAggregate);
+
+        // update everything altered
+        for (MapDifference.ValueDifference<UserType> diff : typesDiff.entriesDiffering().values())
+            Schema.instance.updateType(diff.rightValue());
+        for (MapDifference.ValueDifference<CFMetaData> diff : tablesDiff.entriesDiffering().values())
+            Schema.instance.updateTable(diff.rightValue());
+        for (MapDifference.ValueDifference<ViewDefinition> diff : viewsDiff.entriesDiffering().values())
+            Schema.instance.updateView(diff.rightValue());
+        for (MapDifference.ValueDifference<UDFunction> diff : udfsDiff.entriesDiffering().values())
+            Schema.instance.updateFunction(diff.rightValue());
+        for (MapDifference.ValueDifference<UDAggregate> diff : udasDiff.entriesDiffering().values())
+            Schema.instance.updateAggregate(diff.rightValue());
     }
 
-    private static AbstractType<?> parseType(String str)
+    /*
+     * Type parsing and transformation
+     */
+
+    /*
+     * Recursively replaces any instances of UserType with an equivalent TupleType.
+     * We do it for dropped_columns, to allow safely dropping unused user types without retaining any references
+     * in dropped_columns.
+     */
+    private static AbstractType<?> expandUserTypes(AbstractType<?> original)
     {
-        return TypeParser.parse(str);
+        if (original instanceof UserType)
+            return new TupleType(expandUserTypes(((UserType) original).fieldTypes()));
+
+        if (original instanceof TupleType)
+            return new TupleType(expandUserTypes(((TupleType) original).allTypes()));
+
+        if (original instanceof ListType<?>)
+            return ListType.getInstance(expandUserTypes(((ListType<?>) original).getElementsType()), original.isMultiCell());
+
+        if (original instanceof MapType<?,?>)
+        {
+            MapType<?, ?> mt = (MapType<?, ?>) original;
+            return MapType.getInstance(expandUserTypes(mt.getKeysType()), expandUserTypes(mt.getValuesType()), mt.isMultiCell());
+        }
+
+        if (original instanceof SetType<?>)
+            return SetType.getInstance(expandUserTypes(((SetType<?>) original).getElementsType()), original.isMultiCell());
+
+        // this is very unlikely to ever happen, but it's better to be safe than sorry
+        if (original instanceof ReversedType<?>)
+            return ReversedType.getInstance(expandUserTypes(((ReversedType) original).baseType));
+
+        if (original instanceof CompositeType)
+            return CompositeType.getInstance(expandUserTypes(original.getComponents()));
+
+        return original;
     }
 
-    // We allow method overloads, so a function is not uniquely identified by its name only, but
-    // also by its argument types. To distinguish overloads of given function name in the schema
-    // we use a "signature" which is just a list of it's CQL argument types (we could replace that by
-    // using a "signature" UDT that would be comprised of the function name and argument types,
-    // which we could then use as clustering column. But as we haven't yet used UDT in system tables,
-    // We'll leave that decision to #6717).
-    public static ByteBuffer functionSignatureWithTypes(AbstractFunction fun)
+    private static List<AbstractType<?>> expandUserTypes(List<AbstractType<?>> types)
     {
-        ListType<String> list = ListType.getInstance(UTF8Type.instance, false);
-        List<String> strList = new ArrayList<>(fun.argTypes().size());
-        for (AbstractType<?> argType : fun.argTypes())
-            strList.add(argType.asCQL3Type().toString());
-        return list.decompose(strList);
+        return types.stream()
+                    .map(SchemaKeyspace::expandUserTypes)
+                    .collect(toList());
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/340df43f/src/java/org/apache/cassandra/schema/Tables.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/schema/Tables.java b/src/java/org/apache/cassandra/schema/Tables.java
index 151697d..4f728d4 100644
--- a/src/java/org/apache/cassandra/schema/Tables.java
+++ b/src/java/org/apache/cassandra/schema/Tables.java
@@ -23,6 +23,8 @@ import java.util.Optional;
 import javax.annotation.Nullable;
 
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.MapDifference;
+import com.google.common.collect.Maps;
 
 import org.apache.cassandra.config.CFMetaData;
 
@@ -115,6 +117,11 @@ public final class Tables implements Iterable<CFMetaData>
         return builder().add(filter(this, t -> t != table)).build();
     }
 
+    MapDifference<String, CFMetaData> diff(Tables other)
+    {
+        return Maps.difference(tables, other.tables);
+    }
+
     @Override
     public boolean equals(Object o)
     {


Mime
View raw message