Backport #66898 to 24.3: [CI Fest] Better processing of broken parts and their projections (fixes rare cases of lost forever)

robot-clickhouse · robot-clickhouse · commit 2bc9a6ddfaa9 · 2024-07-30T19:06:35.000Z
diff --git a/src/Storages/MergeTree/DataPartsExchange.cpp b/src/Storages/MergeTree/DataPartsExchange.cpp
@@ -14,6 +14,9 @@
 #include <Storages/MergeTree/ReplicatedFetchList.h>
 #include <Storages/StorageReplicatedMergeTree.h>
 #include <Storages/MergeTree/DataPartStorageOnDiskFull.h>
+#include <Storages/MergeTree/MergeTreeData.h>
+#include <Storages/MergeTree/MergeTreeSettings.h>
+#include <Storages/MergeTree/checkDataPart.h>
 #include <Common/CurrentMetrics.h>
 #include <Common/NetException.h>
 #include <Disks/IO/createReadBufferFromFileBase.h>
@@ -210,14 +213,18 @@ void Service::processQuery(const HTMLForm & params, ReadBuffer & /*body*/, Write
     }
     catch (const Exception & e)
     {
-        if (e.code() != ErrorCodes::ABORTED && e.code() != ErrorCodes::CANNOT_WRITE_TO_OSTREAM)
+        if (e.code() != ErrorCodes::CANNOT_WRITE_TO_OSTREAM
+            && !isRetryableException(std::current_exception()))
+        {
             report_broken_part();
+        }
 
         throw;
     }
     catch (...)
     {
-        report_broken_part();
+        if (!isRetryableException(std::current_exception()))
+            report_broken_part();
         throw;
     }
 }
diff --git a/src/Storages/MergeTree/MergeTreeSequentialSource.cpp b/src/Storages/MergeTree/MergeTreeSequentialSource.cpp
@@ -14,16 +14,11 @@
 #include <Processors/QueryPlan/FilterStep.h>
 #include <Common/logger_useful.h>
 #include <Processors/Merges/Algorithms/MergeTreePartLevelInfo.h>
+#include <Storages/MergeTree/checkDataPart.h>
 
 namespace DB
 {
 
-namespace ErrorCodes
-{
-    extern const int MEMORY_LIMIT_EXCEEDED;
-}
-
-
 /// Lightweight (in terms of logic) stream for reading single part from
 /// MergeTree, used for merges and mutations.
 ///
@@ -275,7 +270,7 @@ try
 catch (...)
 {
     /// Suspicion of the broken part. A part is added to the queue for verification.
-    if (getCurrentExceptionCode() != ErrorCodes::MEMORY_LIMIT_EXCEEDED)
+    if (!isRetryableException(std::current_exception()))
         storage.reportBrokenPart(data_part);
     throw;
 }
diff --git a/src/Storages/MergeTree/checkDataPart.cpp b/src/Storages/MergeTree/checkDataPart.cpp
@@ -38,11 +38,13 @@ namespace ErrorCodes
     extern const int CANNOT_ALLOCATE_MEMORY;
     extern const int CANNOT_MUNMAP;
     extern const int CANNOT_MREMAP;
+    extern const int CANNOT_SCHEDULE_TASK;
     extern const int UNEXPECTED_FILE_IN_DATA_PART;
     extern const int NO_FILE_IN_DATA_PART;
     extern const int NETWORK_ERROR;
     extern const int SOCKET_TIMEOUT;
     extern const int BROKEN_PROJECTION;
+    extern const int ABORTED;
 }
 
 
@@ -88,11 +90,11 @@ bool isRetryableException(std::exception_ptr exception_ptr)
     }
     catch (const Exception & e)
     {
-        if (isNotEnoughMemoryErrorCode(e.code()))
-            return true;
-
-        if (e.code() == ErrorCodes::NETWORK_ERROR || e.code() == ErrorCodes::SOCKET_TIMEOUT)
-            return true;
+        return isNotEnoughMemoryErrorCode(e.code())
+            || e.code() == ErrorCodes::NETWORK_ERROR
+            || e.code() == ErrorCodes::SOCKET_TIMEOUT
+            || e.code() == ErrorCodes::CANNOT_SCHEDULE_TASK
+            || e.code() == ErrorCodes::ABORTED;
     }
     catch (const Poco::Net::NetException &)
     {
@@ -335,16 +337,21 @@ static IMergeTreeDataPart::Checksums checkDataPart(
         projections_on_disk.erase(projection_file);
     }
 
-    if (throw_on_broken_projection && !broken_projections_message.empty())
+    if (throw_on_broken_projection)
     {
-        throw Exception(ErrorCodes::BROKEN_PROJECTION, "{}", broken_projections_message);
-    }
+        if (!broken_projections_message.empty())
+        {
+            throw Exception(ErrorCodes::BROKEN_PROJECTION, "{}", broken_projections_message);
+        }
 
-    if (require_checksums && !projections_on_disk.empty())
-    {
-        throw Exception(ErrorCodes::UNEXPECTED_FILE_IN_DATA_PART,
-            "Found unexpected projection directories: {}",
-            fmt::join(projections_on_disk, ","));
+        /// This one is actually not broken, just redundant files on disk which
+        /// MergeTree will never use.
+        if (require_checksums && !projections_on_disk.empty())
+        {
+            throw Exception(ErrorCodes::UNEXPECTED_FILE_IN_DATA_PART,
+                "Found unexpected projection directories: {}",
+                fmt::join(projections_on_disk, ","));
+        }
     }
 
     if (is_cancelled())

Original file line number	Diff line number	Diff line change
`@@ -14,6 +14,9 @@`
`14`	`14`	`#include <Storages/MergeTree/ReplicatedFetchList.h>`
`15`	`15`	`#include <Storages/StorageReplicatedMergeTree.h>`
`16`	`16`	`#include <Storages/MergeTree/DataPartStorageOnDiskFull.h>`
	`17`	`+#include <Storages/MergeTree/MergeTreeData.h>`
	`18`	`+#include <Storages/MergeTree/MergeTreeSettings.h>`
	`19`	`+#include <Storages/MergeTree/checkDataPart.h>`
`17`	`20`	`#include <Common/CurrentMetrics.h>`
`18`	`21`	`#include <Common/NetException.h>`
`19`	`22`	`#include <Disks/IO/createReadBufferFromFileBase.h>`
`@@ -210,14 +213,18 @@ void Service::processQuery(const HTMLForm & params, ReadBuffer & /body/, Write`
`210`	`213`	`}`
`211`	`214`	`catch (const Exception & e)`
`212`	`215`	`{`
`213`		`- if (e.code() != ErrorCodes::ABORTED && e.code() != ErrorCodes::CANNOT_WRITE_TO_OSTREAM)`
	`216`	`+ if (e.code() != ErrorCodes::CANNOT_WRITE_TO_OSTREAM`
	`217`	`+ && !isRetryableException(std::current_exception()))`
	`218`	`+ {`
`214`	`219`	`report_broken_part();`
	`220`	`+ }`
`215`	`221`
`216`	`222`	`throw;`
`217`	`223`	`}`
`218`	`224`	`catch (...)`
`219`	`225`	`{`
`220`		`- report_broken_part();`
	`226`	`+ if (!isRetryableException(std::current_exception()))`
	`227`	`+ report_broken_part();`
`221`	`228`	`throw;`
`222`	`229`	`}`
`223`	`230`	`}`