Preserve errors for INSERT into Distributed

Before this patch (and after #22208) the INSERT may fail with "Cannot
schedule a task" because the pool in DistributedBlockOutputStream
already throws exception and simply fail in writeSuffix().

src/Storages/Distributed/DistributedBlockOutputStream.cpp

@@ -477,7 +477,9 @@ void DistributedBlockOutputStream::writeSuffix()
         LOG_DEBUG(log, "It took {} sec. to insert {} blocks, {} rows per second. {}", elapsed, inserted_blocks, inserted_rows / elapsed, getCurrentStateDescription());
     };
 
-    if (insert_sync && pool)
+    /// Pool finished means that some exception had been thrown before,
+    /// and scheduling new jobs will return "Cannot schedule a task" error.
+    if (insert_sync && pool && !pool->finished())
     {
         finished_jobs_count = 0;
         try

tests/config/config.d/clusters.xml

@@ -28,5 +28,19 @@
                  </replica>
              </shard>
         </test_cluster_one_shard_two_replicas>
+        <test_cluster_two_replicas_different_databases>
+             <shard>
+                 <replica>
+                     <default_database>shard_0</default_database>
+                     <host>localhost</host>
+                     <port>9000</port>
+                 </replica>
+                 <replica>
+                     <default_database>shard_1</default_database>
+                     <host>localhost</host>
+                     <port>9000</port>
+                 </replica>
+             </shard>
+        </test_cluster_two_replicas_different_databases>
     </remote_servers>
 </yandex>

tests/queries/0_stateless/01850_dist_INSERT_preserve_error.sql

@@ -0,0 +1,15 @@
+create database if not exists shard_0;
+create database if not exists shard_1;
+
+drop table if exists dist_01850;
+drop table if exists shard_0.data_01850;
+
+create table shard_0.data_01850 (key Int) engine=Memory();
+create table dist_01850 (key Int) engine=Distributed('test_cluster_two_replicas_different_databases', /* default_database= */ '', data_01850, key);
+
+set insert_distributed_sync=1;
+set prefer_localhost_replica=0;
+insert into dist_01850 values (1); -- { serverError 60 }
+
+drop table if exists dist_01850;
+drop table shard_0.data_01850;

tests/queries/0_stateless/arcadia_skip_list.txt

@@ -232,3 +232,4 @@
 01804_dictionary_decimal256_type
 01801_s3_distributed
 01833_test_collation_alvarotuso
+01850_dist_INSERT_preserve_error

tests/queries/skip_list.json

@@ -710,6 +710,7 @@
         "01780_clickhouse_dictionary_source_loop",
         "01785_dictionary_element_count",
         "01802_test_postgresql_protocol_with_row_policy", /// Creates database and users
-        "01804_dictionary_decimal256_type"
+        "01804_dictionary_decimal256_type",
+        "01850_dist_INSERT_preserve_error" // uses cluster with different static databases shard_0/shard_1
     ]
 }