Intermediate changes

1 files changed, 11 insertions, 7 deletions

diff --git a/contrib/clickhouse/src/Interpreters/executeQuery.cpp b/contrib/clickhouse/src/Interpreters/executeQuery.cpp
index 32d1fd8507..498bb60b73 100644
--- a/contrib/clickhouse/src/Interpreters/executeQuery.cpp
+++ b/contrib/clickhouse/src/Interpreters/executeQuery.cpp
@@ -1261,7 +1261,8 @@ void executeQuery(
     bool allow_into_outfile,
     ContextMutablePtr context,
     SetResultDetailsFunc set_result_details,
-    const std::optional<FormatSettings> & output_format_settings)
+    const std::optional<FormatSettings> & output_format_settings,
+    HandleExceptionInOutputFormatFunc handle_exception_in_output_format)
 {
     PODArray<char> parse_buf;
     const char * begin;
@@ -1319,6 +1320,7 @@ void executeQuery(
 
     ASTPtr ast;
     BlockIO streams;
+    OutputFormatPtr output_format;
 
     std::tie(ast, streams) = executeQueryImpl(begin, end, context, false, QueryProcessingStage::Complete, &istr);
     auto & pipeline = streams.pipeline;
@@ -1361,30 +1363,30 @@ void executeQuery(
                                     ? getIdentifierName(ast_query_with_output->format)
                                     : context->getDefaultFormat();
 
-            auto out = FormatFactory::instance().getOutputFormatParallelIfPossible(
+            output_format = FormatFactory::instance().getOutputFormatParallelIfPossible(
                 format_name,
                 compressed_buffer ? *compressed_buffer : *out_buf,
                 materializeBlock(pipeline.getHeader()),
                 context,
                 output_format_settings);
 
-            out->setAutoFlush();
+            output_format->setAutoFlush();
 
             /// Save previous progress callback if any. TODO Do it more conveniently.
             auto previous_progress_callback = context->getProgressCallback();
 
             /// NOTE Progress callback takes shared ownership of 'out'.
-            pipeline.setProgressCallback([out, previous_progress_callback] (const Progress & progress)
+            pipeline.setProgressCallback([output_format, previous_progress_callback] (const Progress & progress)
             {
                 if (previous_progress_callback)
                     previous_progress_callback(progress);
-                out->onProgress(progress);
+                output_format->onProgress(progress);
             });
 
-            result_details.content_type = out->getContentType();
+            result_details.content_type = output_format->getContentType();
             result_details.format = format_name;
 
-            pipeline.complete(std::move(out));
+            pipeline.complete(output_format);
         }
         else
         {
@@ -1414,6 +1416,8 @@ void executeQuery(
     }
     catch (...)
     {
+        if (handle_exception_in_output_format && output_format)
+            handle_exception_in_output_format(*output_format);
         streams.onException();
         throw;
     }