Repository: ClickHouse/ClickHouse
File: src/Storages/ObjectStorageQueue/StorageObjectStorageQueue.cpp (C++)

Code before:
        /// FIXME:
        /// it is possible that MV is dropped just before we start the insert,
        /// but in this case we would not throw any exception, so
        /// data will not be inserted anywhere.
        InterpreterInsertQuery interpreter(
            insert,
            queue_context,
            /*allow_materialized_=*/ false,
            /*no_squash_=*/ false,
            /*no_destination=*/ true,
            /*async_insert*/ is_deduplication_v2);
        auto block_io = interpreter.execute();
        auto read_from_format_info = prepareReadingFromFormat(
            block_io.pipeline.getHeader().getNames(),
            storage_snapshot,
            queue_context,
            supportsSubsetOfColumns(queue_context),
            /*supports_tuple_elements*/ false,
            PrepareReadingFromFormatHiveParams {file_columns,
                hive_partition_columns_to_read_from_file_path.getNameToTypeMap()}
        );

        Pipes pipes;
        std::vector<std::shared_ptr<ObjectStorageQueueSource>> sources;

        pipes.reserve(threads);
        sources.reserve(threads);

        auto parser_shared_resources
            = std::make_shared<FormatParserSharedResources>(queue_context->getSettingsRef(), /*num_streams_=*/threads);

        auto processing_progress = std::make_shared<ProcessingProgress>();
        for (size_t i = 0; i < threads; ++i)
        {
            size_t processor_id = i * (streaming_tasks_index + 1);
            auto source = createSource(
                processor_id,
                read_from_format_info,
                parser_shared_resources,
                processing_progress,
                file_iterator,
                DBMS_DEFAULT_BUFFER_SIZE,
                queue_context,
                /*commit_once_processed=*/false);

            pipes.emplace_back(source);
            sources.emplace_back(source);
        }
        auto pipe = Pipe::unitePipes(std::move(pipes));

        block_io.pipeline.complete(std::move(pipe));

Review comment (suggestion):
This is also unclear line.
actually this wont have significant influence. But it is better to have it the same as first line.

Code after:
            streaming_file_iterator = createFileIterator(queue_context, nullptr);
        }
        file_iterator = streaming_file_iterator;
    }
    size_t total_rows = 0;

    const size_t processing_threads_num = getTableMetadata().processing_threads_num;
    const bool parallel_inserts = getTableMetadata().parallel_inserts;
    const size_t threads = parallel_inserts ? 1 : processing_threads_num;

    LOG_TEST(log, "Using {} processing threads (processing_threads_num: {}, parallel_inserts: {}, async deduplicate: {})",
        threads, processing_threads_num, parallel_inserts, is_deduplication_v2);

    while (!shutdown_called && !file_iterator->isFinished())
    {
        /// FIXME:
        /// it is possible that MV is dropped just before we start the insert,
        /// but in this case we would not throw any exception, so
        /// data will not be inserted anywhere.
        InterpreterInsertQuery interpreter(
            insert,
            queue_context,
            /*allow_materialized_=*/ false,
            /*no_squash_=*/ false,
            /*no_destination=*/ true,
            /*async_insert*/ is_deduplication_v2);
        auto block_io = interpreter.execute();
        auto read_from_format_info = prepareReadingFromFormat(
            block_io.pipeline.getHeader().getNames(),
            storage_snapshot,
            queue_context,
            supportsSubsetOfColumns(queue_context),
            /*supports_tuple_elements*/ false,
            PrepareReadingFromFormatHiveParams {file_columns,
                hive_partition_columns_to_read_from_file_path.getNameToTypeMap()}
        );

        Pipes pipes;
        std::vector<std::shared_ptr<ObjectStorageQueueSource>> sources;

        pipes.reserve(threads);
        sources.reserve(threads);

        auto parser_shared_resources
            = std::make_shared<FormatParserSharedResources>(queue_context->getSettingsRef(), /*num_streams_=*/threads);

        auto processing_progress = std::make_shared<ProcessingProgress>();
        for (size_t i = 0; i < threads; ++i)
        {
            size_t processor_id = i * (streaming_tasks_index + 1);
            auto source = createSource(