db/legacy_schema_tables: Store CF "is_dense" to system tables

Persist column family's "is_dense" value to system tables. Please note
that we throw an exception if "is_dense" is null upon read. That needs
to be fixed later by inferring the value from other information like
Origin does.

Signed-off-by: Pekka Enberg <penberg@cloudius-systems.com>

cql3/statements/create_index_statement.cc

@@ -130,7 +130,7 @@ cql3::statements::create_index_statement::validate(distributed<service::storage_
 
 
     // Origin TODO: we could lift that limitation
-    if ((schema->thrift().is_dense() || !schema->thrift().has_compound_comparator()) && cd->kind != column_kind::regular_column) {
+    if ((schema->is_dense() || !schema->thrift().has_compound_comparator()) && cd->kind != column_kind::regular_column) {
         throw exceptions::invalid_request_exception("Secondary indexes are not supported on PRIMARY KEY columns in COMPACT STORAGE tables");
     }
 

cql3/statements/create_table_statement.cc

@@ -115,8 +115,8 @@ void create_table_statement::apply_properties_to(schema_builder& builder) {
 #if 0
     cfmd.defaultValidator(defaultValidator)
         .addAllColumnDefinitions(getColumns(cfmd))
-        .isDense(isDense);
 #endif
+    builder.set_is_dense(_is_dense);
 
     add_column_metadata_from_aliases(builder, _key_aliases, _partition_key_types, column_kind::partition_key);
     add_column_metadata_from_aliases(builder, _column_aliases, _clustering_key_types, column_kind::clustering_key);
@@ -200,11 +200,9 @@ ::shared_ptr<parsed_statement::prepared> create_table_statement::raw_statement::
     }
     stmt->_partition_key_types = key_types;
 
-#if 0
     // Dense means that no part of the comparator stores a CQL column name. This means
     // COMPACT STORAGE with at least one columnAliases (otherwise it's a thrift "static" CF).
-    stmt.isDense = useCompactStorage && !columnAliases.isEmpty();
-#endif
+    stmt->_is_dense = _use_compact_storage && !_column_aliases.empty();
 
     // Handle column aliases
     if (_column_aliases.empty()) {

cql3/statements/create_table_statement.hh

@@ -54,9 +54,9 @@ class create_table_statement : public schema_altering_statement {
     std::vector<bytes> _column_aliases;
 #if 0
     private ByteBuffer valueAlias;
-
-    private boolean isDense;
 #endif
+    bool _is_dense;
+
     using column_map_type =
         std::unordered_map<::shared_ptr<column_identifier>,
                            data_type,

cql3/statements/modification_statement.cc

@@ -219,7 +219,7 @@ modification_statement::create_exploded_clustering_prefix_internal(const query_o
             //   (b) thrift static CF with non-composite comparator
             // Those tables don't have clustering columns so we wouldn't reach this code, thus
             // the check seems redundant.
-            if (require_full_clustering_key() && !s->thrift().is_dense()) {
+            if (require_full_clustering_key() && !s->is_dense()) {
                 throw exceptions::invalid_request_exception(sprint("Missing mandatory PRIMARY KEY part %s", def.name_as_text()));
             }
         } else if (first_empty_key) {

cql3/statements/update_statement.cc

@@ -32,7 +32,7 @@ namespace cql3 {
 namespace statements {
 
 void update_statement::add_update_for_key(mutation& m, const exploded_clustering_prefix& prefix, const update_parameters& params) {
-    if (s->thrift().is_dense()) {
+    if (s->is_dense()) {
         throw std::runtime_error("Dense tables not supported yet");
 #if 0
             if (prefix.isEmpty())

db/legacy_schema_tables.cc

@@ -1044,10 +1044,10 @@ std::vector<const char*> ALL { KEYSPACES, COLUMNFAMILIES, COLUMNS, TRIGGERS, USE
 
         for (Map.Entry<ColumnIdentifier, Long> entry : table.getDroppedColumns().entrySet())
             adder.addMapEntry("dropped_columns", entry.getKey().toString(), entry.getValue());
-
-        adder.add("is_dense", table.getIsDense());
 #endif
 
+        m.set_clustered_cell(ckey, "is_dense", table->is_dense(), timestamp);
+
         if (with_columns_and_triggers) {
             for (auto&& column : table->all_columns_in_select_order()) {
                 add_column_to_schema_mutation(table, column, timestamp, pkey, mutations);
@@ -1230,11 +1230,16 @@ std::vector<const char*> ALL { KEYSPACES, COLUMNFAMILIES, COLUMNS, TRIGGERS, USE
                                                                         fullRawComparator,
                                                                         cfType == ColumnFamilyType.Super*/);
 
-#if 0
-        boolean isDense = result.has("is_dense")
-                        ? result.getBoolean("is_dense")
-                        : CFMetaData.calculateIsDense(fullRawComparator, columnDefs);
+        bool is_dense;
+        if (table_row.has("is_dense")) {
+            is_dense = table_row.get_nonnull<bool>("is_dense");
+        } else {
+            // FIXME:
+            // is_dense = CFMetaData.calculateIsDense(fullRawComparator, columnDefs);
+            throw std::runtime_error("not implemented");
+        }
 
+#if 0
         CellNameType comparator = CellNames.fromAbstractType(fullRawComparator, isDense);
 
         // if we are upgrading, we use id generated from names initially
@@ -1243,8 +1248,10 @@ std::vector<const char*> ALL { KEYSPACES, COLUMNFAMILIES, COLUMNS, TRIGGERS, USE
                   : CFMetaData.generateLegacyCfId(ksName, cfName);
 
         CFMetaData cfm = new CFMetaData(ksName, cfName, cfType, comparator, cfId);
-        cfm.isDense(isDense);
+#endif
+        builder.set_is_dense(is_dense);
 
+#if 0
         cfm.readRepairChance(result.getDouble("read_repair_chance"));
         cfm.dcLocalReadRepairChance(result.getDouble("local_read_repair_chance"));
         cfm.gcGraceSeconds(result.getInt("gc_grace_seconds"));

schema.cc

@@ -239,11 +239,6 @@ generate_legacy_id(const sstring& ks_name, const sstring& cf_name) {
     return utils::UUID_gen::get_name_UUID(ks_name + cf_name);
 }
 
-bool thrift_schema::is_dense() const {
-    warn(unimplemented::cause::COMPACT_TABLES);
-    return false;
-}
-
 bool thrift_schema::has_compound_comparator() const {
     // until we "map" compact storage, at which point it might not be "true".
     warn(unimplemented::cause::COMPACT_TABLES);

schema.hh

@@ -116,7 +116,6 @@ public:
  */
 class thrift_schema {
 public:
-    bool is_dense() const;
     bool has_compound_comparator() const;
 };
 
@@ -145,6 +144,7 @@ private:
         data_type _regular_column_name_type;
         double _bloom_filter_fp_chance = 0.01;
         compression_parameters _compressor_params;
+        bool _is_dense = false;
     };
     raw_schema _raw;
     thrift_schema _thrift;
@@ -207,7 +207,9 @@ public:
     const compression_parameters& get_compressor_params() const {
         return _raw._compressor_params;
     }
-
+    bool is_dense() const {
+        return _raw._is_dense;
+    }
     thrift_schema& thrift() {
         return _thrift;
     }

schema_builder.hh

@@ -56,7 +56,9 @@ public:
     void set_compressor_params(const compression_parameters& cp) {
         _raw._compressor_params = cp;
     }
-
+    void set_is_dense(bool is_dense) {
+        _raw._is_dense = is_dense;
+    }
     column_definition& find_column(const cql3::column_identifier&);
     schema_builder& with_column(const column_definition& c);
     schema_builder& with_column(bytes name, data_type type, column_kind kind = column_kind::regular_column);

sstables/partition.cc

@@ -238,7 +238,7 @@ class mp_row_consumer : public row_consumer {
     { }
 
     void validate_row_marker() {
-        if (_schema->thrift().is_dense()) {
+        if (_schema->is_dense()) {
             throw malformed_sstable_exception("row marker found in dense table");
         }
     }