Repository: ClickHouse/ClickHouse
File: src/Processors/Transforms/JoiningTransform.cpp (C++)

Code before:
        has_input = input_chunk.hasRows() || join_result != nullptr;
    }
    else
    {
        if (!non_joined_blocks)
        {
            if (!finish_counter)
            {
                process_non_joined = false;
                return;
            }

            if (join->supportParallelNonJoinedBlocksProcessing() && num_streams > 1)
            {
                /// parallel mode
                /// all JoiningTransforms participate in non-joined row emission
                /// first, arrive at the barrier (increment counter exactly once)
                if (!finish_counter_incremented)
                {
                    finish_counter->isLast(); /// increment (return value ignored)
                    finish_counter_incremented = true;
                }

                /// Wait until all transforms have finished their input before reading non-joined rows
                /// because used_flags may still be updated by other transforms during the probe phase
                if (!finish_counter->isFinished())
                    return; /// yield — prepare() will return Status::Ready, so we'll be called again

                /// all transforms done — create a partitioned stream for this transform's bucket range
                non_joined_blocks = join->getNonJoinedBlocks(
                    inputs.front().getHeader(), outputs.front().getHeader(), max_block_size,
                    stream_index, num_streams);
            }
            else
            {
                /// sequential mode: only the last transform to finish proceeds
                if (!finish_counter->isLast())
                {
                    process_non_joined = false;
                    return;
                }

                non_joined_blocks = join->getNonJoinedBlocks(
                    inputs.front().getHeader(), outputs.front().getHeader(), max_block_size);
            }

            if (!non_joined_blocks)
            {
                process_non_joined = false;
                return;
            }

Review comment (suggestion):
I'm not 100% sure but since transform remains `Ready` nothing prevents executor from calling `prepare` -> `work` of same transform instance in a loop without calling any other transform. For example, when we have multistream pipeline but executing it with single threaded executor for any reason.

Code after:
        if (!non_joined_blocks)
        {
            if (!finish_counter || !finish_counter->isLast())
            {
                process_non_joined = false;
                return;
            }

            non_joined_blocks = join->getNonJoinedBlocks(
                inputs.front().getHeader(), outputs.front().getHeader(), max_block_size);

            if (!non_joined_blocks)
            {
                process_non_joined = false;
                return;
            }
        }

        Block block = non_joined_blocks->next();
        if (block.empty())
        {
            process_non_joined = false;
            return;
        }

        if (block.rows())
        {
            ProfileEvents::increment(ProfileEvents::JoinResultRowCount, block.rows());
            output_chunk = Chunk(block.getColumns(), block.rows());
        }
    }
}

/// transform should consume the input chunk and set the output chunk
/// if not all data is consumed it may be set to the chunk and transform will be called again
void JoiningTransform::transform(Chunk & chunk)
{
    if (!initialized)
    {
        initialized = true;

        if (join->alwaysReturnsEmptySet() && !on_totals)
        {
            stop_reading = true;
            chunk.clear();
            return;
        }
    }

    Block res;
    if (on_totals)