From 2d9397de5807fa2cecb6df296b0e24dd7db22e68 Mon Sep 17 00:00:00 2001
From: Pekka Enberg <penberg@cloudius-systems.com>
Date: Tue, 16 Jun 2015 09:50:52 +0300
Subject: [PATCH] db/legacy_schema_tables: Store columns to system tables

Signed-off-by: Pekka Enberg <penberg@cloudius-systems.com>
---
 db/legacy_schema_tables.cc | 45 +++++++++++++++++++++++++-------------
 db/legacy_schema_tables.hh |  4 ++++
 2 files changed, 34 insertions(+), 15 deletions(-)

diff --git a/db/legacy_schema_tables.cc b/db/legacy_schema_tables.cc
index fa31209e3d..d3f2375cfb 100644
--- a/db/legacy_schema_tables.cc
+++ b/db/legacy_schema_tables.cc
@@ -1045,16 +1045,18 @@ std::vector<const char*> ALL { KEYSPACES, COLUMNFAMILIES, COLUMNS, TRIGGERS, USE
             adder.addMapEntry("dropped_columns", entry.getKey().toString(), entry.getValue());
 
         adder.add("is_dense", table.getIsDense());
+#endif
 
-        if (withColumnsAndTriggers)
-        {
-            for (ColumnDefinition column : table.allColumns())
-                addColumnToSchemaMutation(table, column, timestamp, mutation);
+        if (with_columns_and_triggers) {
+            for (auto&& column : table->all_columns_in_select_order()) {
+                add_column_to_schema_mutation(table, column, timestamp, pkey, mutations);
+            }
 
+#if 0
             for (TriggerDefinition trigger : table.getTriggers().values())
                 addTriggerToSchemaMutation(table, trigger, timestamp, mutation);
-        }
 #endif
+        }
         mutations.emplace_back(std::move(m));
     }
 
@@ -1292,31 +1294,44 @@ std::vector<const char*> ALL { KEYSPACES, COLUMNFAMILIES, COLUMNS, TRIGGERS, USE
             converted.put(new ColumnIdentifier(entry.getKey(), true), entry.getValue());
         return converted;
     }
+#endif
 
     /*
      * Column metadata serialization/deserialization.
      */
 
-    private static void addColumnToSchemaMutation(CFMetaData table, ColumnDefinition column, long timestamp, Mutation mutation)
+    void add_column_to_schema_mutation(schema_ptr table,
+                                       const column_definition& column,
+                                       api::timestamp_type timestamp,
+                                       const partition_key& pkey,
+                                       std::vector<mutation>& mutations)
     {
-        ColumnFamily cells = mutation.addOrGet(Columns);
-        Composite prefix = Columns.comparator.make(table.cfName, column.name.toString());
-        CFRowAdder adder = new CFRowAdder(cells, prefix, timestamp);
-
-        adder.add("validator", column.type.toString());
-        adder.add("type", serializeKind(column.kind));
+        schema_ptr s = columns();
+        mutation m{pkey, s};
+        auto ckey = clustering_key::from_exploded(*s, {to_bytes(table->cf_name()), column.name()});
+        m.set_clustered_cell(ckey, "validator", column.type->name(), timestamp);
+        m.set_clustered_cell(ckey, "type", serialize_kind(column.kind), timestamp);
+#if 0
         adder.add("component_index", column.isOnAllComponents() ? null : column.position());
         adder.add("index_name", column.getIndexName());
         adder.add("index_type", column.getIndexType() == null ? null : column.getIndexType().toString());
         adder.add("index_options", json(column.getIndexOptions()));
+#endif
+        mutations.emplace_back(std::move(m));
     }
 
-    private static String serializeKind(ColumnDefinition.Kind kind)
+    sstring serialize_kind(column_kind kind)
     {
-        // For backward compatibility we need to special case CLUSTERING_COLUMN
-        return kind == ColumnDefinition.Kind.CLUSTERING_COLUMN ? "clustering_key" : kind.toString().toLowerCase();
+        switch (kind) {
+        case column_kind::partition_key:  return "partition_key";
+        case column_kind::clustering_key: return "clustering_key";
+        case column_kind::static_column:  return "static";
+        case column_kind::regular_column: return "regular";
+        default:                          throw std::invalid_argument("unknown column kind");
+        }
     }
 
+#if 0
     private static ColumnDefinition.Kind deserializeKind(String kind)
     {
         if (kind.equalsIgnoreCase("clustering_key"))
diff --git a/db/legacy_schema_tables.hh b/db/legacy_schema_tables.hh
index 396dd6670c..116d505626 100644
--- a/db/legacy_schema_tables.hh
+++ b/db/legacy_schema_tables.hh
@@ -79,5 +79,9 @@ void add_table_to_schema_mutation(schema_ptr table, api::timestamp_type timestam
 
 schema_ptr create_table_from_table_row(const query::result_set_row& row);
 
+void add_column_to_schema_mutation(schema_ptr table, const column_definition& column, api::timestamp_type timestamp, const partition_key& pkey, std::vector<mutation>& mutations);
+
+sstring serialize_kind(column_kind kind);
+
 } // namespace legacy_schema_tables
 } // namespace db