From f2ed9fcd7e95012ca1af7c49f0afbf33d092eef2 Mon Sep 17 00:00:00 2001 From: Tomasz Grabiec Date: Thu, 29 Jun 2023 01:12:35 +0200 Subject: [PATCH] schema_mutations, migration_manager: Ignore empty partitions in per-table digest Schema digest is calculated by querying for mutations of all schema tables, then compacting them so that all tombstones in them are dropped. However, even if the mutation becomes empty after compaction, we still feed its partition key. If the same mutations were compacted prior to the query, because the tombstones expire, we won't get any mutation at all and won't feed the partition key. So schema digest will change once an empty partition of some schema table is compacted away. Tombstones expire 7 days after schema change which introduces them. If one of the nodes is restarted after that, it will compute a different table schema digest on boot. This may cause performance problems. When sending a request from coordinator to replica, the replica needs schema_ptr of exact schema version request by the coordinator. If it doesn't know that version, it will request it from the coordinator and perform a full schema merge. This adds latency to every such request. Schema versions which are not referenced are currently kept in cache for only 1 second, so if request flow has low-enough rate, this situation results in perpetual schema pulls. After ae8d2a550d227de2221c310bd38704ad2e078b35, it is more liekly to run into this situation, because table creation generates tombstones for all schema tables relevant to the table, even the ones which will be otherwise empty for the new table (e.g. computed_columns). This change inroduces a cluster feature which when enabled will change digest calculation to be insensitive to expiry by ignoring empty partitions in digest calculation. When the feature is enabled, schema_ptrs are reloaded so that the window of discrepancy during transition is short and no rolling restart is required. A similar problem was fixed for per-node digest calculation in 18f484cc753d17d1e3658bcb5c73ed8f319d32e8. Per-table digest calculation was not fixed at that time because we didn't persist enabled features and they were not enabled early-enough on boot for us to depend on them in digest calculation. Now they are enabled before non-system tables are loaded so digest calculation can rely on cluster features. Fixes #4485. --- db/config.cc | 2 ++ db/config.hh | 1 + db/schema_features.hh | 7 ++++++- db/schema_tables.cc | 10 ++++++---- db/schema_tables.hh | 11 +++++++++-- gms/feature_service.cc | 4 ++++ gms/feature_service.hh | 1 + schema_mutations.cc | 23 +++++++++++++---------- schema_mutations.hh | 3 ++- service/migration_manager.cc | 8 ++++++++ test/boost/schema_change_test.cc | 13 ++++++------- test/boost/schema_registry_test.cc | 8 +++++--- tools/schema_loader.cc | 4 +++- 13 files changed, 66 insertions(+), 29 deletions(-) diff --git a/db/config.cc b/db/config.cc index dfce1310e5..19e5e1ccbb 100644 --- a/db/config.cc +++ b/db/config.cc @@ -881,6 +881,8 @@ db::config::config(std::shared_ptr exts) , uuid_sstable_identifiers_enabled(this, "uuid_sstable_identifiers_enabled", liveness::LiveUpdate, value_status::Used, true, "If set to true, each newly created sstable will have a UUID " "based generation identifier, and such files are not readable by previous Scylla versions.") + , table_digest_insensitive_to_expiry(this, "table_digest_insensitive_to_expiry", liveness::MustRestart, value_status::Used, true, + "When enabled, per-table schema digest calculation ignores empty partitions.") , enable_dangerous_direct_import_of_cassandra_counters(this, "enable_dangerous_direct_import_of_cassandra_counters", value_status::Used, false, "Only turn this option on if you want to import tables from Cassandra containing counters, and you are SURE that no counters in that table were created in a version earlier than Cassandra 2.1." " It is not enough to have ever since upgraded to newer versions of Cassandra. If you EVER used a version earlier than 2.1 in the cluster where these SSTables come from, DO NOT TURN ON THIS OPTION! You will corrupt your data. You have been warned.") , enable_shard_aware_drivers(this, "enable_shard_aware_drivers", value_status::Used, true, "Enable native transport drivers to use connection-per-shard for better performance") diff --git a/db/config.hh b/db/config.hh index cf4dbb81e4..322e539d50 100644 --- a/db/config.hh +++ b/db/config.hh @@ -357,6 +357,7 @@ public: named_value enable_sstables_md_format; named_value sstable_format; named_value uuid_sstable_identifiers_enabled; + named_value table_digest_insensitive_to_expiry; named_value enable_dangerous_direct_import_of_cassandra_counters; named_value enable_shard_aware_drivers; named_value enable_ipv6_dns_lookup; diff --git a/db/schema_features.hh b/db/schema_features.hh index c4d37f9c72..0c48ad260d 100644 --- a/db/schema_features.hh +++ b/db/schema_features.hh @@ -24,6 +24,10 @@ enum class schema_feature { PER_TABLE_PARTITIONERS, SCYLLA_KEYSPACES, SCYLLA_AGGREGATES, + + // When enabled, schema_mutations::digest() will skip empty mutations (with only tombstones), + // so that the digest remains the same after schema tables are compacted. + TABLE_DIGEST_INSENSITIVE_TO_EXPIRY, }; using schema_features = enum_set>; } diff --git a/db/schema_tables.cc b/db/schema_tables.cc index 02431f7498..aee556b384 100644 --- a/db/schema_tables.cc +++ b/db/schema_tables.cc @@ -102,8 +102,10 @@ namespace { }); } -schema_ctxt::schema_ctxt(const db::config& cfg, std::shared_ptr uts, replica::database* db) +schema_ctxt::schema_ctxt(const db::config& cfg, std::shared_ptr uts, + const gms::feature_service& features, replica::database* db) : _db(db) + , _features(features) , _extensions(cfg.extensions()) , _murmur3_partitioner_ignore_msb_bits(cfg.murmur3_partitioner_ignore_msb_bits()) , _schema_registry_grace_period(cfg.schema_registry_grace_period()) @@ -111,7 +113,7 @@ schema_ctxt::schema_ctxt(const db::config& cfg, std::shared_ptr& db) @@ -3127,7 +3129,7 @@ schema_ptr create_table_from_mutations(const schema_ctxt& ctxt, schema_mutations if (version) { builder.with_version(*version); } else { - builder.with_version(sm.digest()); + builder.with_version(sm.digest(ctxt.features().cluster_schema_features())); } if (auto partitioner = sm.partitioner()) { @@ -3347,7 +3349,7 @@ view_ptr create_view_from_mutations(const schema_ctxt& ctxt, schema_mutations sm if (version) { builder.with_version(*version); } else { - builder.with_version(sm.digest()); + builder.with_version(sm.digest(ctxt.features().cluster_schema_features())); } auto base_id = table_id(row.get_nonnull("base_table_id")); diff --git a/db/schema_tables.hh b/db/schema_tables.hh index f650f83bf7..de628558e9 100644 --- a/db/schema_tables.hh +++ b/db/schema_tables.hh @@ -14,6 +14,7 @@ #include "schema/schema_fwd.hh" #include "schema_features.hh" #include "utils/hashing.hh" +#include "gms/feature_service.hh" #include "schema_mutations.hh" #include "types/map.hh" #include "query-result-set.hh" @@ -66,7 +67,8 @@ class config; class schema_ctxt { public: - schema_ctxt(const config&, std::shared_ptr uts, replica::database* = nullptr); + schema_ctxt(const config&, std::shared_ptr uts, const gms::feature_service&, + replica::database* = nullptr); schema_ctxt(replica::database&); schema_ctxt(distributed&); schema_ctxt(distributed&); @@ -87,11 +89,16 @@ public: return *_user_types; } - replica::database* get_db() { + const gms::feature_service& features() const { + return _features; + } + + replica::database* get_db() const { return _db; } private: replica::database* _db; + const gms::feature_service& _features; const db::extensions& _extensions; const unsigned _murmur3_partitioner_ignore_msb_bits; const uint32_t _schema_registry_grace_period; diff --git a/gms/feature_service.cc b/gms/feature_service.cc index abd6711907..305b43733b 100644 --- a/gms/feature_service.cc +++ b/gms/feature_service.cc @@ -75,6 +75,9 @@ feature_config feature_config_from_db_config(const db::config& cfg, std::set(per_table_partitioners); f.set_if(keyspace_storage_options); f.set_if(aggregate_storage_options); + f.set_if(table_digest_insensitive_to_expiry); return f; } diff --git a/gms/feature_service.hh b/gms/feature_service.hh index d3ea900b65..ed62226d62 100644 --- a/gms/feature_service.hh +++ b/gms/feature_service.hh @@ -118,6 +118,7 @@ public: gms::feature secondary_indexes_on_static_columns { *this, "SECONDARY_INDEXES_ON_STATIC_COLUMNS"sv }; gms::feature tablets { *this, "TABLETS"sv }; gms::feature uuid_sstable_identifiers { *this, "UUID_SSTABLE_IDENTIFIERS"sv }; + gms::feature table_digest_insensitive_to_expiry { *this, "TABLE_DIGEST_INSENSITIVE_TO_EXPIRY"sv }; // A feature just for use in tests. It must not be advertised unless // the "features_enable_test_feature" injection is enabled. diff --git a/schema_mutations.cc b/schema_mutations.cc index 02ce9a643d..92525d2be0 100644 --- a/schema_mutations.cc +++ b/schema_mutations.cc @@ -49,7 +49,7 @@ void schema_mutations::copy_to(std::vector& dst) const { } } -table_schema_version schema_mutations::digest() const { +table_schema_version schema_mutations::digest(db::schema_features sf) const { if (_scylla_tables) { auto rs = query::result_set(*_scylla_tables); if (!rs.empty()) { @@ -62,16 +62,19 @@ table_schema_version schema_mutations::digest() const { } md5_hasher h; - db::schema_features sf = db::schema_features::full(); - // Disable this feature so that the digest remains compactible with Scylla - // versions prior to this feature. - // This digest affects the table schema version calculation and it's important - // that all nodes arrive at the same table schema version to avoid needless schema version - // pulls. Table schema versions are calculated on boot when we don't yet - // know all the cluster features, so we could get different table versions after reboot - // in an already upgraded cluster. - sf.remove(); + if (!sf.contains()) { + // Disable this feature so that the digest remains compactible with Scylla + // versions prior to this feature. + // This digest affects the table schema version calculation and it's important + // that all nodes arrive at the same table schema version to avoid needless schema version + // pulls. It used to be the case that when table schema versions were calculated on boot we + // didn't yet know all the cluster features, so we could get different table versions after reboot + // in an already upgraded cluster. However, they are now available, and if + // TABLE_DIGEST_INSENSITIVE_TO_EXPIRY is enabled, we can compute with DIGEST_INSENSITIVE_TO_EXPIRY + // enabled. + sf.remove(); + } db::schema_tables::feed_hash_for_schema_digest(h, _columnfamilies, sf); db::schema_tables::feed_hash_for_schema_digest(h, _columns, sf); diff --git a/schema_mutations.hh b/schema_mutations.hh index efc0c1a6dc..183082b6a8 100644 --- a/schema_mutations.hh +++ b/schema_mutations.hh @@ -12,6 +12,7 @@ #include "mutation/mutation.hh" #include "schema/schema_fwd.hh" #include "mutation/canonical_mutation.hh" +#include "db/schema_features.hh" // Commutative representation of table schema // Equality ignores tombstones. @@ -124,7 +125,7 @@ public: bool is_view() const; - table_schema_version digest() const; + table_schema_version digest(db::schema_features) const; std::optional partitioner() const; bool operator==(const schema_mutations&) const; diff --git a/service/migration_manager.cc b/service/migration_manager.cc index 0e8ba3fda8..bc95109d86 100644 --- a/service/migration_manager.cc +++ b/service/migration_manager.cc @@ -109,6 +109,14 @@ void migration_manager::init_messaging_service() _feature_listeners.push_back(_feat.cdc.when_enabled(update_schema)); _feature_listeners.push_back(_feat.per_table_partitioners.when_enabled(update_schema)); _feature_listeners.push_back(_feat.computed_columns.when_enabled(update_schema)); + + if (!_feat.table_digest_insensitive_to_expiry) { + _feature_listeners.push_back(_feat.table_digest_insensitive_to_expiry.when_enabled([this] { + (void) with_gate(_background_tasks, [this] { + return reload_schema(); + }); + })); + } } _messaging.register_definitions_update([this] (const rpc::client_info& cinfo, std::vector fm, rpc::optional> cm) { diff --git a/test/boost/schema_change_test.cc b/test/boost/schema_change_test.cc index cab8e7c85d..9a64f61c57 100644 --- a/test/boost/schema_change_test.cc +++ b/test/boost/schema_change_test.cc @@ -812,8 +812,6 @@ future<> test_schema_digest_does_not_change_with_disabled_features(sstring data_ expect_digest(sf, expected_digests[4]); - // FIXME: schema_mutations::digest() is still sensitive to expiry, so we can check versions only after forward_jump_clocks() - // otherwise the results would not be stable. expect_version("tests", "table1", expected_digests[5]); expect_version("ks", "tbl", expected_digests[6]); expect_version("ks", "tbl_view", expected_digests[7]); @@ -840,7 +838,8 @@ SEASTAR_TEST_CASE(test_schema_digest_does_not_change) { utils::UUID("de49e92f-a00d-3f24-8779-d07de26708cb"), }; return test_schema_digest_does_not_change_with_disabled_features("./test/resource/sstables/schema_digest_test", - std::set{"COMPUTED_COLUMNS", "CDC", "KEYSPACE_STORAGE_OPTIONS"}, std::move(expected_digests), [] (cql_test_env& e) {}); + std::set{"COMPUTED_COLUMNS", "CDC", "KEYSPACE_STORAGE_OPTIONS", "TABLE_DIGEST_INSENSITIVE_TO_EXPIRY"}, + std::move(expected_digests), [] (cql_test_env& e) {}); } SEASTAR_TEST_CASE(test_schema_digest_does_not_change_after_computed_columns) { @@ -857,7 +856,7 @@ SEASTAR_TEST_CASE(test_schema_digest_does_not_change_after_computed_columns) { utils::UUID("94606636-ae43-3e0a-b238-e7f0e33ef600"), }; return test_schema_digest_does_not_change_with_disabled_features("./test/resource/sstables/schema_digest_test_computed_columns", - std::set{"CDC", "KEYSPACE_STORAGE_OPTIONS"}, std::move(expected_digests), [] (cql_test_env& e) {}); + std::set{"CDC", "KEYSPACE_STORAGE_OPTIONS", "TABLE_DIGEST_INSENSITIVE_TO_EXPIRY"}, std::move(expected_digests), [] (cql_test_env& e) {}); } SEASTAR_TEST_CASE(test_schema_digest_does_not_change_with_functions) { @@ -875,7 +874,7 @@ SEASTAR_TEST_CASE(test_schema_digest_does_not_change_with_functions) { }; return test_schema_digest_does_not_change_with_disabled_features( "./test/resource/sstables/schema_digest_with_functions_test", - std::set{"CDC", "KEYSPACE_STORAGE_OPTIONS"}, + std::set{"CDC", "KEYSPACE_STORAGE_OPTIONS", "TABLE_DIGEST_INSENSITIVE_TO_EXPIRY"}, std::move(expected_digests), [] (cql_test_env& e) { e.execute_cql("create function twice(val int) called on null input returns int language lua as 'return 2 * val';").get(); @@ -900,7 +899,7 @@ SEASTAR_TEST_CASE(test_schema_digest_does_not_change_with_cdc_options) { }; return test_schema_digest_does_not_change_with_disabled_features( "./test/resource/sstables/schema_digest_test_cdc_options", - std::set{"KEYSPACE_STORAGE_OPTIONS"}, + std::set{"KEYSPACE_STORAGE_OPTIONS", "TABLE_DIGEST_INSENSITIVE_TO_EXPIRY"}, std::move(expected_digests), [] (cql_test_env& e) { e.execute_cql("create table tests.table_cdc (pk int primary key, c1 int, c2 int) with cdc = {'enabled':'true'};").get(); @@ -923,7 +922,7 @@ SEASTAR_TEST_CASE(test_schema_digest_does_not_change_with_keyspace_storage_optio }; return test_schema_digest_does_not_change_with_disabled_features( "./test/resource/sstables/schema_digest_test_keyspace_storage_options", - std::set{}, + std::set{"TABLE_DIGEST_INSENSITIVE_TO_EXPIRY"}, std::move(expected_digests), [] (cql_test_env& e) { e.execute_cql("create keyspace tests_s3 with replication = { 'class' : 'NetworkTopologyStrategy', 'replication_factor' : 1 }" diff --git a/test/boost/schema_registry_test.cc b/test/boost/schema_registry_test.cc index 09c61eed0d..862430258b 100644 --- a/test/boost/schema_registry_test.cc +++ b/test/boost/schema_registry_test.cc @@ -34,10 +34,12 @@ static schema_ptr random_schema() { struct dummy_init { std::unique_ptr config; + gms::feature_service fs; - dummy_init() { - config = std::make_unique(); - local_schema_registry().init(db::schema_ctxt(*config,std::make_shared())); + dummy_init() + : config(std::make_unique()) + , fs(gms::feature_config_from_db_config(*config)) { + local_schema_registry().init(db::schema_ctxt(*config, std::make_shared(), fs)); } }; diff --git a/tools/schema_loader.cc b/tools/schema_loader.cc index c4312a89aa..3ca25f53c6 100644 --- a/tools/schema_loader.cc +++ b/tools/schema_loader.cc @@ -534,7 +534,9 @@ schema_ptr do_load_schema_from_schema_tables(std::filesystem::path scylla_data_p db::config dbcfg; auto user_type_storage = std::make_shared(std::move(utm)); - db::schema_ctxt ctxt(dbcfg, user_type_storage); + gms::feature_service features(gms::feature_config_from_db_config(dbcfg)); + db::schema_ctxt ctxt(dbcfg, user_type_storage, features); + schema_mutations muts(std::move(*tables), std::move(*columns), std::move(view_virtual_columns), std::move(computed_columns), std::move(indexes), std::move(dropped_columns), std::move(scylla_tables)); return db::schema_tables::create_table_from_mutations(ctxt, muts);