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

Code before:
            {
                /// Continue processing.
                /// This failed file will be committed along with processed files.
                reader = {};
                progress->processed_files -= 1;
                continue;
            }
        }

        if (result)
        {
            const auto & object_metadata = reader.getObjectInfo()->getObjectMetadata();
            const auto row_offset = file_status->processed_rows.load();

            std::string dedup_token;
            if (add_deduplication_info)
            {
                /// Etag is quoted for some reason.
                std::string etag = object_metadata->etag;
                if (etag.size() > 2 && etag.front() == '\"' && etag.back() == '\"')
                    etag = etag.substr(1, etag.size() - 2);

                /// Create unique token per chunk: etag + row offset
                dedup_token = fmt::format("{}:{}", etag, row_offset);

                auto deduplication_info = DeduplicationInfo::create(/*async_insert*/true);
                deduplication_info->setUserToken(dedup_token, chunk.getNumRows());
                chunk.getChunkInfos().add(std::move(deduplication_info));
            }

            LOG_TEST(log, "Read {} rows from file {} (file offset: {}, deduplication token for chunk: {})",
                     chunk.getNumRows(), path, row_offset, dedup_token);

            file_status->processed_rows += chunk.getNumRows();
            progress->processed_rows += chunk.getNumRows();
            progress->processed_bytes += chunk.bytes();

            ProfileEvents::increment(ProfileEvents::ObjectStorageQueueReadRows, chunk.getNumRows());
            ProfileEvents::increment(ProfileEvents::ObjectStorageQueueReadBytes, chunk.bytes());

            if (!read_from_format_info.hive_partition_columns_to_read_from_file_path.empty())
            {
                HivePartitioningUtils::addPartitionColumnsToChunk(
                    chunk,
                    read_from_format_info.hive_partition_columns_to_read_from_file_path,
                    path);
            }

            VirtualColumnUtils::addRequestedFileLikeStorageVirtualsToChunk(
                chunk,
                read_from_format_info.requested_virtual_columns,

Review comment (suggestion):
It is the most unclear line.
Lets add a comment here. 
Something like that:
`when is_deduplication_v2 is true, all the source chunks are marked with DeduplicationInfo. Each chunk could have several pieces from different source files with unique deduplication token. In order to utilize deduplication with multiple tokens we set up insert pipeline as async_insert`

Code after:
             }

            if (mode == ObjectStorageQueueMode::ORDERED)
            {
                /// Stop processing and commit what is already processed.
                /// because we must preserve order.
                return {};
            }
            else
            {
                /// Continue processing.
                /// This failed file will be committed along with processed files.
                reader = {};
                progress->processed_files -= 1;
                continue;
            }
        }

        if (result)
        {
            const auto & object_metadata = reader.getObjectInfo()->getObjectMetadata();
            const auto row_offset = file_status->processed_rows.load();

            std::string dedup_token;
            if (add_deduplication_info)
            {
                /// Etag is quoted for some reason.
                std::string etag = object_metadata->etag;
                if (etag.size() > 2 && etag.front() == '\"' && etag.back() == '\"')
                    etag = etag.substr(1, etag.size() - 2);

                /// Create unique token per chunk: etag + row offset
                dedup_token = fmt::format("{}:{}", etag, row_offset);

                auto deduplication_info = DeduplicationInfo::create(/*async_insert*/true, insert_deduplication_version);
                deduplication_info->setUserToken(dedup_token, chunk.getNumRows());
                chunk.getChunkInfos().add(std::move(deduplication_info));
            }

            LOG_TEST(log, "Read {} rows from file {} (file offset: {}, deduplication token for chunk: {})",
                     chunk.getNumRows(), path, row_offset, dedup_token);

            file_status->processed_rows += chunk.getNumRows();
            progress->processed_rows += chunk.getNumRows();
            progress->processed_bytes += chunk.bytes();

            ProfileEvents::increment(ProfileEvents::ObjectStorageQueueReadRows, chunk.getNumRows());
            ProfileEvents::increment(ProfileEvents::ObjectStorageQueueReadBytes, chunk.bytes());

            if (!read_from_format_info.hive_partition_columns_to_read_from_file_path.empty())
            {