Merge pull request ClickHouse#86988 from ClickHouse/backport/25.8/86932

Backport ClickHouse#86932 to 25.8: Process exception thrown from background pool

Author: clickhouse-gh[bot]

src/Storages/ObjectStorage/DataLakes/Iceberg/AvroForIcebergDeserializer.cpp

@@ -16,6 +16,7 @@
 namespace DB::ErrorCodes
 {
     extern const int ICEBERG_SPECIFICATION_VIOLATION;
+    extern const int INCORRECT_DATA;
 }
 
 namespace DB::Iceberg
@@ -24,9 +25,8 @@ namespace DB::Iceberg
 using namespace DB;
 
 AvroForIcebergDeserializer::AvroForIcebergDeserializer(
-    std::unique_ptr<ReadBufferFromFileBase> buffer_,
-    const std::string & manifest_file_path_,
-    const DB::FormatSettings & format_settings)
+    std::unique_ptr<ReadBufferFromFileBase> buffer_, const std::string & manifest_file_path_, const DB::FormatSettings & format_settings)
+try
     : buffer(std::move(buffer_))
     , manifest_file_path(manifest_file_path_)
 {
@@ -51,6 +51,10 @@ AvroForIcebergDeserializer::AvroForIcebergDeserializer(
     parsed_column = std::move(columns[0]);
     parsed_column_data_type = std::dynamic_pointer_cast<const DataTypeTuple>(data_type);
 }
+catch (const std::exception & e)
+{
+    throw Exception(ErrorCodes::INCORRECT_DATA, "Cannot read Iceberg avro manifest file '{}': {}", manifest_file_path_, e.what());
+}
 
 size_t AvroForIcebergDeserializer::rows() const
 {

src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergIterator.cpp

@@ -269,10 +269,24 @@ IcebergIterator::IcebergIterator(
           {
               while (!blocking_queue.isFinished())
               {
-                  auto info = data_files_iterator.next();
-                  if (!info.has_value())
+                  std::optional<ManifestFileEntry> entry;
+                  try
+                  {
+                      entry = data_files_iterator.next();
+                  }
+                  catch (...)
+                  {
+                      std::lock_guard lock(exception_mutex);
+                      if (!exception)
+                      {
+                          exception = std::current_exception();
+                      }
+                      blocking_queue.finish();
                       break;
-                  while (!blocking_queue.push(std::move(info.value())))
+                  }
+                  if (!entry.has_value())
+                      break;
+                  while (!blocking_queue.push(std::move(entry.value())))
                   {
                       if (blocking_queue.isFinished())
                       {
@@ -321,6 +335,15 @@ ObjectInfoPtr IcebergIterator::next(size_t)
         }
         return object_info;
     }
+    {
+        std::lock_guard lock(exception_mutex);
+        if (exception)
+        {
+            auto exception_message = getExceptionMessage(exception, true, true);
+            auto exception_code = getExceptionErrorCode(exception);
+            throw DB::Exception(exception_code, "Iceberg iterator is failed with exception: {}", exception_message);
+        }
+    }
     return nullptr;
 }
 

src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergIterator.h

@@ -109,6 +109,8 @@ class IcebergIterator : public IObjectIterator
     const String compression_method;
     std::vector<Iceberg::ManifestFileEntry> position_deletes_files;
     std::vector<Iceberg::ManifestFileEntry> equality_deletes_files;
+    std::exception_ptr exception;
+    std::mutex exception_mutex;
 };
 }
 

tests/queries/0_stateless/03611_verify_exception_in_iceberg_iterator.reference

@@ -0,0 +1,9 @@
+
+-- Tags: no-fasttest
+-- Tag no-fasttest: Depends on AWS
+
+-- Verify that exception is thrown when Iceberg table contains corrupted Avro manifest files.
+-- This test verifies not so much thrown error during avro files parsing but rather
+-- that we correctly process exceptions thrown from iceberg iterator in background execution.
+
+SELECT * FROM icebergS3('http://localhost:11111/test/corrupted_avro_files_test/', 'clickhouse', 'clickhouse') SETTINGS use_iceberg_metadata_files_cache = False; -- { serverError INCORRECT_DATA}

tests/queries/0_stateless/03611_verify_exception_in_iceberg_iterator.sql

@@ -0,0 +1 @@
+Intentionally corrupted file which pretends to have avro format. 

tests/queries/0_stateless/data_minio/corrupted_avro_files_test/metadata/83521e92-07a7-450d-ba44-3a179b730c85-m0.avro

@@ -0,0 +1,75 @@
+{
+  "format-version" : 2,
+  "table-uuid" : "a5920f9f-12d0-4773-87fc-569c00676258",
+  "location" : "s3a://test/corrupted_avro_files_test",
+  "last-sequence-number" : 1,
+  "last-updated-ms" : 1740666694866,
+  "last-column-id" : 2,
+  "current-schema-id" : 0,
+  "schemas" : [ {
+    "type" : "struct",
+    "schema-id" : 0,
+    "fields" : [ {
+      "id" : 1,
+      "name" : "order_number",
+      "required" : false,
+      "type" : "long"
+    }, {
+      "id" : 2,
+      "name" : "product_code",
+      "required" : false,
+      "type" : "string"
+    } ]
+  } ],
+  "default-spec-id" : 0,
+  "partition-specs" : [ {
+    "spec-id" : 0,
+    "fields" : [ ]
+  } ],
+  "last-partition-id" : 999,
+  "default-sort-order-id" : 0,
+  "sort-orders" : [ {
+    "order-id" : 0,
+    "fields" : [ ]
+  } ],
+  "properties" : {
+    "owner" : "divanik",
+    "option.format-version" : "2",
+    "write.parquet.compression-codec" : "zstd"
+  },
+  "current-snapshot-id" : 3858160944799047644,
+  "refs" : {
+    "main" : {
+      "snapshot-id" : 3858160944799047644,
+      "type" : "branch"
+    }
+  },
+  "snapshots" : [ {
+    "sequence-number" : 1,
+    "snapshot-id" : 3858160944799047644,
+    "timestamp-ms" : 1740666694866,
+    "summary" : {
+      "operation" : "append",
+      "spark.app.id" : "local-1740666666602",
+      "added-data-files" : "1",
+      "added-records" : "1",
+      "added-files-size" : "701",
+      "changed-partition-count" : "1",
+      "total-records" : "1",
+      "total-files-size" : "701",
+      "total-data-files" : "1",
+      "total-delete-files" : "0",
+      "total-position-deletes" : "0",
+      "total-equality-deletes" : "0"
+    },
+    "manifest-list" : "s3a://test/corrupted_avro_files_test/metadata/snap-3858160944799047644-1-83521e92-07a7-450d-ba44-3a179b730c85.avro",
+    "schema-id" : 0
+  } ],
+  "statistics" : [ ],
+  "partition-statistics" : [ ],
+  "snapshot-log" : [ {
+    "timestamp-ms" : 1740666694866,
+    "snapshot-id" : 3858160944799047644
+  } ],
+  "metadata-log" : []
+}