Merge pull request ClickHouse#80038 from aalexfvk/reset_metadata_version_on_attach

Fix metadata version on attach part

Author: CheSema

src/Storages/MergeTree/IMergeTreeDataPart.cpp

@@ -828,7 +828,7 @@ ColumnsStatistics IMergeTreeDataPart::loadStatistics() const
     return result;
 }
 
-void IMergeTreeDataPart::loadColumnsChecksumsIndexes(bool require_columns_checksums, bool check_consistency)
+void IMergeTreeDataPart::loadColumnsChecksumsIndexes(bool require_columns_checksums, bool check_consistency, bool load_metadata_version)
 {
     /// Memory should not be limited during ATTACH TABLE query.
     /// This is already true at the server startup but must be also ensured for manual table ATTACH.
@@ -839,7 +839,7 @@ void IMergeTreeDataPart::loadColumnsChecksumsIndexes(bool require_columns_checks
     {
         if (!isStoredOnReadonlyDisk())
             loadUUID();
-        loadColumns(require_columns_checksums);
+        loadColumns(require_columns_checksums, load_metadata_version);
         loadColumnsSubstreams();
         loadChecksums(require_columns_checksums);
 
@@ -1228,6 +1228,19 @@ void IMergeTreeDataPart::writeVersionMetadata(const VersionMetadata & version_,
     }
 }
 
+void IMergeTreeDataPart::writeMetadataVersion(ContextPtr context, int32_t metadata_version_, bool sync)
+{
+    removeMetadataVersion();
+    {
+        auto out_metadata = getDataPartStorage().writeFile(METADATA_VERSION_FILE_NAME, 4096, context->getWriteSettings());
+        writeText(metadata_version_, *out_metadata);
+        out_metadata->finalize();
+        if (sync)
+            out_metadata->sync();
+    }
+    old_part_with_no_metadata_version_on_disk = false;
+}
+
 void IMergeTreeDataPart::removeDeleteOnDestroyMarker()
 {
     getDataPartStorage().removeFileIfExists(DELETE_ON_DESTROY_MARKER_FILE_NAME_DEPRECATED);
@@ -1612,7 +1625,7 @@ void IMergeTreeDataPart::loadUUID()
     }
 }
 
-void IMergeTreeDataPart::loadColumns(bool require)
+void IMergeTreeDataPart::loadColumns(bool require, bool load_metadata_version)
 {
     String path = fs::path(getDataPartStorage().getRelativePath()) / "columns.txt";
 
@@ -1656,20 +1669,24 @@ void IMergeTreeDataPart::loadColumns(bool require)
     if (auto in = readFileIfExists(SERIALIZATION_FILE_NAME))
         infos = SerializationInfoByName::readJSON(loaded_columns, settings, *in);
 
-    int32_t loaded_metadata_version;
-    if (auto in = readFileIfExists(METADATA_VERSION_FILE_NAME))
+    std::optional<int32_t> loaded_metadata_version;
+    if (load_metadata_version)
     {
-        readIntText(loaded_metadata_version, *in);
+        if (auto in = readFileIfExists(METADATA_VERSION_FILE_NAME))
+        {
+            readIntText(loaded_metadata_version.emplace(), *in);
+        }
     }
-    else
+
+    if (!loaded_metadata_version)
     {
         auto storage_metdata_snapshot = storage.getInMemoryMetadataPtr();
         loaded_metadata_version = storage_metdata_snapshot->getMetadataVersion();
         old_part_with_no_metadata_version_on_disk = true;
     }
 
-    LOG_DEBUG(storage.log, "Loaded metadata version {}", loaded_metadata_version);
-    setColumns(loaded_columns, infos, loaded_metadata_version);
+    LOG_DEBUG(storage.log, "Loaded metadata version {}", *loaded_metadata_version);
+    setColumns(loaded_columns, infos, *loaded_metadata_version);
 }
 
 void IMergeTreeDataPart::loadColumnsSubstreams()

src/Storages/MergeTree/IMergeTreeDataPart.h

@@ -137,6 +137,8 @@ class IMergeTreeDataPart : public std::enable_shared_from_this<IMergeTreeDataPar
 
     /// Version of metadata for part (columns, pk and so on)
     int32_t getMetadataVersion() const { return metadata_version; }
+    void setMetadataVersion(int32_t metadata_version_) noexcept { metadata_version = metadata_version_; }
+    void writeMetadataVersion(ContextPtr local_context, int32_t metadata_version, bool sync);
 
     const NamesAndTypesList & getColumns() const { return columns; }
     const ColumnsDescription & getColumnsDescription() const { return columns_description; }
@@ -164,7 +166,7 @@ class IMergeTreeDataPart : public std::enable_shared_from_this<IMergeTreeDataPar
 
     /// Initialize columns (from columns.txt if exists, or create from column files if not).
     /// Load various metadata into memory: checksums from checksums.txt, index if required, etc.
-    void loadColumnsChecksumsIndexes(bool require_columns_checksums, bool check_consistency);
+    void loadColumnsChecksumsIndexes(bool require_columns_checksums, bool check_consistency, bool load_metadata_version = true);
 
     void loadRowsCountFileForUnexpectedPart();
 
@@ -701,7 +703,7 @@ class IMergeTreeDataPart : public std::enable_shared_from_this<IMergeTreeDataPar
 
 
     /// Reads columns names and types from columns.txt
-    void loadColumns(bool require);
+    void loadColumns(bool require, bool load_metadata_version);
 
     /// Reads columns substreams from columns_substreams.txt (only in Compact parts).
     void loadColumnsSubstreams();

src/Storages/MergeTree/MergeTreeData.cpp

@@ -5473,14 +5473,7 @@ static void loadPartAndFixMetadataImpl(MergeTreeData::MutableDataPartPtr part, C
     /// Remove metadata version file and take it from table.
     /// Currently we cannot attach parts with different schema, so
     /// we can assume that it's equal to table's current schema.
-    part->removeMetadataVersion();
-    {
-        auto out_metadata = part->getDataPartStorage().writeFile(IMergeTreeDataPart::METADATA_VERSION_FILE_NAME, 4096, local_context->getWriteSettings());
-        writeText(metadata_version, *out_metadata);
-        out_metadata->finalize();
-        if (sync)
-            out_metadata->sync();
-    }
+    part->writeMetadataVersion(local_context, metadata_version, sync);
 
     part->loadColumnsChecksumsIndexes(false, true);
     part->modification_time = part->getDataPartStorage().getLastModified().epochTime();

src/Storages/StorageReplicatedMergeTree.cpp

@@ -193,6 +193,7 @@ namespace MergeTreeSetting
     extern const MergeTreeSettingsBool enable_the_endpoint_id_with_zookeeper_name_prefix;
     extern const MergeTreeSettingsFloat fault_probability_after_part_commit;
     extern const MergeTreeSettingsFloat fault_probability_before_part_commit;
+    extern const MergeTreeSettingsBool fsync_after_insert;
     extern const MergeTreeSettingsUInt64 index_granularity_bytes;
     extern const MergeTreeSettingsSeconds lock_acquire_timeout_for_background_operations;
     extern const MergeTreeSettingsUInt64 max_bytes_to_merge_at_max_space_in_pool;
@@ -2236,7 +2237,8 @@ MergeTreeData::MutableDataPartPtr StorageReplicatedMergeTree::attachPartHelperFo
 
             try
             {
-                part->loadColumnsChecksumsIndexes(true, true);
+                part->loadColumnsChecksumsIndexes(
+                    /* require_columns_checksums = */ true, /* check_consistency = */ true, /* load_metadata_version = */ false);
             }
             catch (const Exception&)
             {
@@ -2247,6 +2249,9 @@ MergeTreeData::MutableDataPartPtr StorageReplicatedMergeTree::attachPartHelperFo
 
             if (entry.part_checksum == part->checksums.getTotalChecksumHex())
             {
+                auto metadata_version = getInMemoryMetadataPtr()->getMetadataVersion();
+                part->writeMetadataVersion(getContext(), metadata_version, (*getSettings())[MergeTreeSetting::fsync_after_insert]);
+                part->setMetadataVersion(metadata_version);
                 part->modification_time = part->getDataPartStorage().getLastModified().epochTime();
                 return part;
             }

tests/integration/test_restore_replica/configs/fast_background_pool.xml

@@ -0,0 +1,9 @@
+<clickhouse>
+    <background_processing_pool_thread_sleep_seconds>1</background_processing_pool_thread_sleep_seconds>
+    <background_processing_pool_thread_sleep_seconds_random_part>0</background_processing_pool_thread_sleep_seconds_random_part>
+    <background_processing_pool_thread_sleep_seconds_if_nothing_to_do>0.0</background_processing_pool_thread_sleep_seconds_if_nothing_to_do>
+    <background_processing_pool_task_sleep_seconds_when_no_work_min>0</background_processing_pool_task_sleep_seconds_when_no_work_min>
+    <background_processing_pool_task_sleep_seconds_when_no_work_max>1</background_processing_pool_task_sleep_seconds_when_no_work_max>
+    <background_processing_pool_task_sleep_seconds_when_no_work_multiplier>1</background_processing_pool_task_sleep_seconds_when_no_work_multiplier>
+    <background_processing_pool_task_sleep_seconds_when_no_work_random_part>0</background_processing_pool_task_sleep_seconds_when_no_work_random_part>
+</clickhouse>

tests/integration/test_restore_replica/test.py

@@ -23,7 +23,7 @@ def fill_nodes(nodes):
 
 
 cluster = ClickHouseCluster(__file__)
-configs = ["configs/remote_servers.xml"]
+configs = ["configs/remote_servers.xml", "configs/fast_background_pool.xml"]
 
 node_1 = cluster.add_instance("replica1", with_zookeeper=True, main_configs=configs)
 node_2 = cluster.add_instance("replica2", with_zookeeper=True, main_configs=configs)
@@ -181,3 +181,49 @@ def test_restore_replica_alive_replicas(start_cluster):
     node_3.query("SYSTEM SYNC REPLICA test")
 
     check_after_restoration()
+
+
+def test_fix_metadata_version_on_attach_part_after_restore(start_cluster):
+    zk = cluster.get_kazoo_client("zoo1")
+
+    node_1.query("DROP TABLE IF EXISTS test_ttl ON CLUSTER 'test_cluster' SYNC")
+    for node in [node_1, node_2]:
+        node.query(
+            """
+            CREATE TABLE test_ttl(n UInt32, d DateTime)
+            ENGINE = ReplicatedMergeTree('/clickhouse/tables/test_ttl/', '{replica}')
+            TTL d + INTERVAL 5 SECOND DELETE
+            ORDER BY n PARTITION BY n % 10
+            SETTINGS merge_with_ttl_timeout = 0, number_of_free_entries_in_pool_to_execute_mutation = 0;
+        """.format(
+                replica=node.name
+            )
+        )
+
+    # Alter metadata to increment metadata version of the table
+    node_1.query("ALTER TABLE test_ttl ADD COLUMN Added1 UInt32")
+    node_1.query("ALTER TABLE test_ttl DROP COLUMN Added1")
+
+    node_1.query("SYSTEM STOP TTL MERGES test_ttl")
+
+    # Create a part
+    node_1.query("INSERT INTO test_ttl VALUES (1, now())")
+
+    # Delete root zk metadata path for the table
+    zk_rmr_with_retries(zk, "/clickhouse/tables/test_ttl")
+
+    # Restore replicas
+    node_1.query("SYSTEM RESTART REPLICA test_ttl")
+    node_1.query("SYSTEM RESTORE REPLICA test_ttl")
+    # Disable TTL merges on node_1. We expect it to be done on node_2
+    node_1.query("SYSTEM STOP TTL MERGES test_ttl")
+
+    node_2.query("SYSTEM RESTART REPLICA test_ttl")
+    node_2.query("SYSTEM RESTORE REPLICA test_ttl")
+
+    # TTL merge should work.
+    # Before fix it is failed due to metadata version discrepancy and we will have in the log:
+    # "Source part metadata version 2 is newer then the table metadata version 0. ALTER_METADATA is still in progress"
+    assert_eq_with_retry(
+        node_2, "SELECT count() FROM test_ttl", "0\n", retry_count=60, sleep_time=1
+    )