Merge pull request #19664 from ClickHouse/backport/20.12/19449

Backport #19449 to 20.12: Mark distributed batch as broken in case of empty data block in one of files

src/Client/Connection.cpp

@@ -53,6 +53,7 @@ namespace ErrorCodes
     extern const int UNKNOWN_PACKET_FROM_SERVER;
     extern const int SUPPORT_IS_DISABLED;
     extern const int BAD_ARGUMENTS;
+    extern const int EMPTY_DATA_PASSED;
 }
 
 
@@ -539,6 +540,9 @@ void Connection::sendPreparedData(ReadBuffer & input, size_t size, const String
 {
     /// NOTE 'Throttler' is not used in this method (could use, but it's not important right now).
 
+    if (input.eof())
+        throw Exception("Buffer is empty (some kind of corruption)", ErrorCodes::EMPTY_DATA_PASSED);
+
     writeVarUInt(Protocol::Client::Data, *out);
     writeStringBinary(name, *out);
 

src/Storages/Distributed/DirectoryMonitor.cpp

@@ -42,6 +42,7 @@ namespace ErrorCodes
     extern const int CHECKSUM_DOESNT_MATCH;
     extern const int TOO_LARGE_SIZE_COMPRESSED;
     extern const int ATTEMPT_TO_READ_AFTER_EOF;
+    extern const int EMPTY_DATA_PASSED;
 }
 
 
@@ -742,6 +743,7 @@ void StorageDistributedDirectoryMonitor::processFilesWithBatching(const std::map
 bool StorageDistributedDirectoryMonitor::isFileBrokenErrorCode(int code)
 {
     return code == ErrorCodes::CHECKSUM_DOESNT_MATCH
+        || code == ErrorCodes::EMPTY_DATA_PASSED
         || code == ErrorCodes::TOO_LARGE_SIZE_COMPRESSED
         || code == ErrorCodes::CANNOT_READ_ALL_DATA
         || code == ErrorCodes::UNKNOWN_CODEC