be/src/exec/vectorized/cross_join_node.cpp
/*
First, build a large chunk to contain the right table.
Then the right table is divided into two parts. 
The number of rows is a multiple of 4096 (big_Chunk) and small with less than 4096 rows(small_chunk).

right table is about _number_of_build_rows rows.
big_chunk's range is    [0 - _build_chunks_size)
small_chunk's range is  [_build_chunks_size - _number_of_build_rows)

For each chunk probe in the left table, probe_chunk, 
we need to make it iterate with the right table, as iteratoe big_chunk and small_chunk separately.

Our way is
step 1: For every row in probe_chunk, use it to iterate with big_chunk.
Step 2: For probe_chunk and small_chunk, use every row in smaller chunk to iterate with bigger chunk.

So far, probe_chunk is done, we will process next chunk.
*/
First, build a large chunk to contain the right table.
Then the right table is divided into two parts.
The number of rows is a multiple of 4096 (big_Chunk) and small with less than 4096 rows(small_chunk).

right table is about _number_of_build_rows rows.

For each chunk probe in the left table, probe_chunk,
we need to make it iterate with the right table, as iteratoe big_chunk and small_chunk separately.

Our way is
step 1: For every row in probe_chunk, use it to iterate with big_chunk.
Step 2: For probe_chunk and small_chunk, use every row in smaller chunk to iterate with bigger chunk.

So far, probe_chunk is done, we will process next chunk.
First build a large chunk to contain the right table . Then the right table is divided into two parts . The number of rows is a multiple of 4096 ( big_Chunk ) and small with less than 4096 rows ( small_chunk ) . right table is about _number_of_build_rows rows . For each chunk probe in the left table probe_chunk we need to make it iterate with the right table as iteratoe big_chunk and small_chunk separately . Our way is step 1 : For every row in probe_chunk use it to iterate with big_chunk . Step 2 : For probe_chunk and small_chunk use every row in smaller chunk to iterate with bigger chunk . So far probe_chunk is done we will process next chunk .
Status CrossJoinNode::get_next_internal(RuntimeState* state, ChunkPtr* chunk, bool* eos,
                                        ScopedTimer<MonotonicStopWatch>& probe_timer) {
    RETURN_IF_CANCELLED(state);
    *chunk = nullptr;
    if (_eos) {
        *eos = true;
        return Status::OK();
    }
    if (_build_chunk == nullptr || _build_chunk->num_rows() == 0) {
        _eos = true;
        *eos = true;
        return Status::OK();
    }
    for (;;) {
        if (_probe_chunk == nullptr || _probe_chunk->num_rows() == 0) {
            probe_timer.stop();
            RETURN_IF_ERROR(_get_next_probe_chunk(state));
            probe_timer.start();
            if (_eos) {
                if (*chunk == nullptr || (*chunk)->num_rows() < 1) {
                    *chunk = nullptr;
                    *eos = true;
                    return Status::OK();
                } else {
                    ExecNode::eval_conjuncts(_conjunct_ctxs, (*chunk).get());
                    break;
                }
            }
            continue;
        }
        if ((*chunk) == nullptr) {
            _init_chunk(chunk);
        }
        size_t row_count = config::vector_chunk_size - (*chunk)->num_rows();
        if (_probe_chunk_index == _probe_chunk->num_rows()) {
            if (_probe_chunk_index > _number_of_build_rows - _build_chunks_size) {
                if (row_count > _probe_chunk_index - _probe_rows_index) {
                    row_count = _probe_chunk_index - _probe_rows_index;
                }
                _copy_joined_rows_with_index_base_build(*chunk, row_count, _probe_rows_index, _build_rows_index);
                _probe_rows_index += row_count;
                if (_probe_rows_index == _probe_chunk_index) {
                    ++_build_rows_index;
                    _probe_rows_index = 0;
                }
                if (_build_rows_index >= _number_of_build_rows) {
                    _probe_chunk = nullptr;
                }
            } else {
                if (row_count > _number_of_build_rows - _build_rows_index) {
                    row_count = _number_of_build_rows - _build_rows_index;
                }
                _copy_joined_rows_with_index_base_probe(*chunk, row_count, _probe_rows_index, _build_rows_index);
                _build_rows_index += row_count;
                if (_build_rows_index == _number_of_build_rows) {
                    ++_probe_rows_index;
                    _build_rows_index = _build_chunks_size;
                }
                if (_probe_rows_index >= _probe_chunk_index) {
                    _probe_chunk = nullptr;
                }
            }
        } else if (_build_chunks_index < _build_chunks_size) {
            if (row_count > _build_chunks_size - _build_chunks_index) {
                row_count = _build_chunks_size - _build_chunks_index;
            }
            _copy_joined_rows_with_index_base_probe(*chunk, row_count, _probe_chunk_index, _build_chunks_index);
            _build_chunks_index += row_count;
        } else {
            DCHECK_EQ(_build_chunks_index, _build_chunks_size);
            if (_build_chunks_size != 0) {
                ++_probe_chunk_index;
                if (_probe_chunk_index < _probe_chunk->num_rows()) {
                    _build_chunks_index = 0;
                } else {
                    if (_build_chunks_size == _number_of_build_rows) {
                        _probe_chunk = nullptr;
                    } else {
                        _build_rows_index = _build_chunks_size;
                        _probe_rows_index = 0;
                    }
                }
            } else {
                _probe_chunk_index = _probe_chunk->num_rows();
                _build_rows_index = _build_chunks_size;
                _probe_rows_index = 0;
            }
            continue;
        }
        if ((*chunk)->num_rows() < config::vector_chunk_size) {
            continue;
        }
        ExecNode::eval_conjuncts(_conjunct_ctxs, (*chunk).get());
        break;
    }
    _num_rows_returned += (*chunk)->num_rows();
    if (reached_limit()) {
        (*chunk)->set_num_rows((*chunk)->num_rows() - (_num_rows_returned - _limit));
        _num_rows_returned = _limit;
        COUNTER_SET(_rows_returned_counter, _limit);
    } else {
        COUNTER_SET(_rows_returned_counter, _num_rows_returned);
    }
    DCHECK(!(*chunk)->has_const_column());
    DCHECK_CHUNK(*chunk);
    *eos = false;
    return Status::OK();
}
Status CrossJoinNode :: get_next_internal ( RuntimeState * state , ChunkPtr * chunk , bool * eos , ScopedTimer < MonotonicStopWatch > & probe_timer ) { RETURN_IF_CANCELLED ( state ) ; * chunk = nullptr ; if ( _eos ) { * eos = true ; return Status :: OK ( ) ; } if ( _build_chunk == nullptr || _build_chunk -> num_rows ( ) == 0 ) { _eos = true ; * eos = true ; return Status :: OK ( ) ; } for ( ; ; ) { if ( _probe_chunk == nullptr || _probe_chunk -> num_rows ( ) == 0 ) { probe_timer . stop ( ) ; RETURN_IF_ERROR ( _get_next_probe_chunk ( state ) ) ; probe_timer . start ( ) ; if ( _eos ) { if ( * chunk == nullptr || ( * chunk ) -> num_rows ( ) < 1 ) { * chunk = nullptr ; * eos = true ; return Status :: OK ( ) ; } else { ExecNode :: eval_conjuncts ( _conjunct_ctxs , ( * chunk ) . get ( ) ) ; break ; } } continue ; } if ( ( * chunk ) == nullptr ) { _init_chunk ( chunk ) ; } size_t row_count = config :: vector_chunk_size - ( * chunk ) -> num_rows ( ) ; if ( _probe_chunk_index == _probe_chunk -> num_rows ( ) ) { if ( _probe_chunk_index > _number_of_build_rows - _build_chunks_size ) { if ( row_count > _probe_chunk_index - _probe_rows_index ) { row_count = _probe_chunk_index - _probe_rows_index ; } _copy_joined_rows_with_index_base_build ( * chunk , row_count , _probe_rows_index , _build_rows_index ) ; _probe_rows_index += row_count ; if ( _probe_rows_index == _probe_chunk_index ) { ++ _build_rows_index ; _probe_rows_index = 0 ; } if ( _build_rows_index >= _number_of_build_rows ) { _probe_chunk = nullptr ; } } else { if ( row_count > _number_of_build_rows - _build_rows_index ) { row_count = _number_of_build_rows - _build_rows_index ; } _copy_joined_rows_with_index_base_probe ( * chunk , row_count , _probe_rows_index , _build_rows_index ) ; _build_rows_index += row_count ; if ( _build_rows_index == _number_of_build_rows ) { ++ _probe_rows_index ; _build_rows_index = _build_chunks_size ; } if ( _probe_rows_index >= _probe_chunk_index ) { _probe_chunk = nullptr ; } } } else if ( _build_chunks_index < _build_chunks_size ) { if ( row_count > _build_chunks_size - _build_chunks_index ) { row_count = _build_chunks_size - _build_chunks_index ; } _copy_joined_rows_with_index_base_probe ( * chunk , row_count , _probe_chunk_index , _build_chunks_index ) ; _build_chunks_index += row_count ; } else { DCHECK_EQ ( _build_chunks_index , _build_chunks_size ) ; if ( _build_chunks_size != 0 ) { ++ _probe_chunk_index ; if ( _probe_chunk_index < _probe_chunk -> num_rows ( ) ) { _build_chunks_index = 0 ; } else { if ( _build_chunks_size == _number_of_build_rows ) { _probe_chunk = nullptr ; } else { _build_rows_index = _build_chunks_size ; _probe_rows_index = 0 ; } } } else { _probe_chunk_index = _probe_chunk -> num_rows ( ) ; _build_rows_index = _build_chunks_size ; _probe_rows_index = 0 ; } continue ; } if ( ( * chunk ) -> num_rows ( ) < config :: vector_chunk_size ) { continue ; } ExecNode :: eval_conjuncts ( _conjunct_ctxs , ( * chunk ) . get ( ) ) ; break ; } _num_rows_returned += ( * chunk ) -> num_rows ( ) ; if ( reached_limit ( ) ) { ( * chunk ) -> set_num_rows ( ( * chunk ) -> num_rows ( ) - ( _num_rows_returned - _limit ) ) ; _num_rows_returned = _limit ; COUNTER_SET ( _rows_returned_counter , _limit ) ; } else { COUNTER_SET ( _rows_returned_counter , _num_rows_returned ) ; } DCHECK ( ! ( * chunk ) -> has_const_column ( ) ) ; DCHECK_CHUNK ( * chunk ) ; * eos = false ; return Status :: OK ( ) ; }
First, build a large chunk to contain the right table.
First build a large chunk to contain the right table .
// need to get probe_chunk // should output (*chunk) first before EOS // we need a valid probe chunk to initialize the new chunk. // need row_count to fill in chunk. // means we have scan all chunks of right tables. // we should scan all remain rows of right table. // once _probe_chunk_index == _probe_chunk->num_rows() is true, // this condition will always true for this _probe_chunk, // Until _probe_chunk be done. // step 2: // if left chunk is bigger than right, we shuld scan left based on right. // _probe_chunk is done with _build_chunk. // if remain rows of right is bigger than left, we should scan right based on left. // _probe_chunk is done with _build_chunk. // step 1: // we scan all chunks of right table. // step policy decision: // scan right chunk_size rows for next row of left chunk. // if right table is all about chunks, means _probe_chunk is done. // optimized for smaller right table < 4096 rows. // we get result chunk.