for (;;) {

int err = m_source->Read();

if (err != 0) return err;

bool matched = m_condition->val_int();

if (thd()->killed) {

thd()->send_kill_message();

return 1;

}

/* check for errors evaluating the condition */

if (thd()->is_error()) return 1;

if (!matched) {

m_source->UnlockRow();

continue;

}

// Successful row.

return 0;

}

if (m_source->Init()) {

return true;

}

if (m_offset > 0) {

m_seen_rows = m_limit;

m_needs_offset = true;

} else {

m_seen_rows = 0;

m_needs_offset = false;

}

return false;

if (m_seen_rows >= m_limit) {

// We either have hit our LIMIT, or we need to skip OFFSET rows.

// Check which one.

if (m_needs_offset) {

// We skip OFFSET rows here and not in Init(), since performance schema

// batch mode may not be set up by the executor before the first Read().

// This makes sure that

//

// a) we get the performance benefits of batch mode even when reading

// OFFSET rows, and

// b) we don't inadvertedly enable batch mode (e.g. through the

// NestedLoopIterator) during Init(), since the executor may not

// be ready to _disable_ it if it gets an error before first Read().

for (ha_rows row_idx = 0; row_idx < m_offset; ++row_idx) {

int err = m_source->Read();

if (err != 0) {

// Note that we'll go back into this loop if Init() is called again,

// and return the same error/EOF status.

return err;

}

if (m_skipped_rows != nullptr) {

++*m_skipped_rows;

}

m_source->UnlockRow();

}

m_seen_rows = m_offset;

m_needs_offset = false;

// Fall through to LIMIT testing.

}

if (m_seen_rows >= m_limit) {

// We really hit LIMIT (or hit LIMIT immediately after OFFSET finished),

// so EOF.

if (m_count_all_rows) {

// Count rows until the end or error (ignore the error if any).

while (m_source->Read() == 0) {

++*m_skipped_rows;

}

}

return -1;

}

}

const int result = m_source->Read();

if (m_reject_multiple_rows) {

if (result != 0) {

++m_seen_rows;

return result;

}

// We read a row. Check for scalar subquery cardinality violation

if (m_seen_rows - m_offset > 0) {

my_error(ER_SUBQUERY_NO_1_ROW, MYF(0));

return 1;

}

}

++m_seen_rows;

return result;

THD *thd, unique_ptr_destroy_only<RowIterator> source, JOIN *join,

TableCollection tables, bool rollup)

: RowIterator(thd),

m_source(std::move(source)),

m_join(join),

m_rollup(rollup),

m_tables(std::move(tables)) {

const size_t upper_data_length = ComputeRowSizeUpperBound(m_tables);

m_first_row_this_group.reserve(upper_data_length);

m_first_row_next_group.reserve(upper_data_length);

assert(!m_join->tmp_table_param.precomputed_group_by);

// Disable any leftover rollup items used in children.

m_current_rollup_position = -1;

SetRollupLevel(INT_MAX);

// If the iterator has been executed before, restore the state of

// the table buffers. This is needed for correctness if there is an

// EQRefIterator below this iterator, as the restoring of the

// previous group in Read() may have disturbed the cache in

// EQRefIterator.

if (!m_first_row_next_group.is_empty()) {

LoadIntoTableBuffers(

m_tables, pointer_cast<const uchar *>(m_first_row_next_group.ptr()));

m_first_row_next_group.length(0);

}

if (m_source->Init()) {

return true;

}

// If we have a HAVING after us, it needs to be evaluated within the context

// of the slice we're in (unless we're in the hypergraph optimizer, which

// doesn't use slices). However, we might have a sort before us, and

// SortingIterator doesn't set the slice except on Init(); it just keeps

// whatever was already set. When there is a temporary table after the HAVING,

// the slice coming from there might be wrongly set on Read(), and thus,

// we need to properly restore it before returning any rows.

//

// This is a hack. It would be good to get rid of the slice system altogether

// (the hypergraph join optimizer does not use it).

if (!(m_join->implicit_grouping || m_join->group_optimized_away) &&

!thd()->lex->using_hypergraph_optimizer()) {

m_output_slice = m_join->get_ref_item_slice();

}

m_seen_eof = false;

m_save_nullinfo = 0;

// Not really used, just to be sure.

m_last_unchanged_group_item_idx = 0;

m_state = READING_FIRST_ROW;

return false;

switch (m_state) {

case READING_FIRST_ROW: {

// Start the first group, if possible. (If we're not at the first row,

// we already saw the first row in the new group at the previous Read().)

int err = m_source->Read();

if (err == -1) {

m_seen_eof = true;

m_state = DONE_OUTPUTTING_ROWS;

if (m_join->grouped || m_join->group_optimized_away) {

SetRollupLevel(m_join->send_group_parts);

return -1;

} else {

// If there's no GROUP BY, we need to output a row even if there are

// no input rows.

// Calculate aggregate functions for no rows

for (Item *item : *m_join->get_current_fields()) {

if (!item->hidden ||

(item->type() == Item::SUM_FUNC_ITEM &&

down_cast<Item_sum *>(item)->aggr_query_block ==

m_join->query_block)) {

item->no_rows_in_result();

}

}

/*

if (m_join->clear_fields(&m_save_nullinfo)) {

return 1;

}

for (Item_sum **item = m_join->sum_funcs; *item != nullptr; ++item) {

(*item)->clear();

}

if (m_output_slice != -1) {

m_join->set_ref_item_slice(m_output_slice);

}

return 0;

}

}

if (err != 0) return err;

// Set the initial value of the group fields.

(void)update_item_cache_if_changed(m_join->group_fields);

StoreFromTableBuffers(m_tables, &m_first_row_next_group);

m_last_unchanged_group_item_idx = 0;

}

[[fallthrough]];

case LAST_ROW_STARTED_NEW_GROUP:

SetRollupLevel(m_join->send_group_parts);

// We don't need m_first_row_this_group for the old group anymore,

// but we'd like to reuse its buffer, so swap instead of std::move.

// (Testing for state == READING_FIRST_ROW and avoiding the swap

// doesn't seem to give any speed gains.)

swap(m_first_row_this_group, m_first_row_next_group);

LoadIntoTableBuffers(

m_tables, pointer_cast<const uchar *>(m_first_row_this_group.ptr()));

for (Item_sum **item = m_join->sum_funcs; *item != nullptr; ++item) {

if (m_rollup) {

if (down_cast<Item_rollup_sum_switcher *>(*item)

->reset_and_add_for_rollup(m_last_unchanged_group_item_idx))

return true;

} else {

if ((*item)->reset_and_add()) return true;

}

}

// Keep reading rows as long as they are part of the existing group.

for (;;) {

int err = m_source->Read();

if (err == 1) return 1; // Error.

if (err == -1) {

m_seen_eof = true;

// We need to be able to restore the table buffers in Init()

// if the iterator is reexecuted (can happen if it's inside

// a correlated subquery).

StoreFromTableBuffers(m_tables, &m_first_row_next_group);

// End of input rows; return the last group. (One would think this

// LoadIntoTableBuffers() call is unneeded, since the last row read

// would be from the last group, but there may be filters in-between

// us and whatever put data into the row buffers, and those filters

// may have caused other rows to be loaded before discarding them.)

LoadIntoTableBuffers(m_tables, pointer_cast<const uchar *>(

m_first_row_this_group.ptr()));

if (m_rollup && m_join->send_group_parts > 0) {

// Also output the final groups, including the total row

// (with NULLs in all fields).

SetRollupLevel(m_join->send_group_parts);

m_last_unchanged_group_item_idx = 0;

m_state = OUTPUTTING_ROLLUP_ROWS;

} else {

SetRollupLevel(m_join->send_group_parts);

m_state = DONE_OUTPUTTING_ROWS;

}

if (m_output_slice != -1) {

m_join->set_ref_item_slice(m_output_slice);

}

return 0;

}

int first_changed_idx =

update_item_cache_if_changed(m_join->group_fields);

if (first_changed_idx >= 0) {

// The group changed. Store the new row (we can't really use it yet;

// next Read() will deal with it), then load back the group values

// so that we can output a row for the current group.

// NOTE: This does not save and restore FTS information,

// so evaluating MATCH() on these rows may give the wrong result.

// (Storing the row ID and repositioning it with ha_rnd_pos()

// would, but we can't do the latter without disturbing

// ongoing scans. See bug #32565923.) For the old join optimizer,

// we generally solve this by inserting temporary tables or sorts

// (both of which restore the information correctly); for the

// hypergraph join optimizer, we add a special streaming step

// for MATCH columns.

StoreFromTableBuffers(m_tables, &m_first_row_next_group);

LoadIntoTableBuffers(m_tables, pointer_cast<const uchar *>(

m_first_row_this_group.ptr()));

// If we have rollup, we may need to output more than one row.

// Mark so that the next calls to Read() will return those rows.

//

// NOTE: first_changed_idx is the first group value that _changed_,

// while what we store is the last item that did _not_ change.

if (m_rollup) {

m_last_unchanged_group_item_idx = first_changed_idx + 1;

if (static_cast<unsigned>(first_changed_idx) <

m_join->send_group_parts - 1) {

SetRollupLevel(m_join->send_group_parts);

m_state = OUTPUTTING_ROLLUP_ROWS;

} else {

SetRollupLevel(m_join->send_group_parts);

m_state = LAST_ROW_STARTED_NEW_GROUP;

}

} else {

m_last_unchanged_group_item_idx = 0;

m_state = LAST_ROW_STARTED_NEW_GROUP;

}

if (m_output_slice != -1) {

m_join->set_ref_item_slice(m_output_slice);

}

return 0;

}

// Give the new values to all the new aggregate functions.

for (Item_sum **item = m_join->sum_funcs; *item != nullptr; ++item) {

if (m_rollup) {

if (down_cast<Item_rollup_sum_switcher *>(*item)

->aggregator_add_all()) {

return 1;

}

} else {

if ((*item)->aggregator_add()) {

return 1;

}

}

}

// We're still in the same group, so just loop back.

}

case OUTPUTTING_ROLLUP_ROWS:

SetRollupLevel(m_current_rollup_position - 1);

if (m_current_rollup_position <= m_last_unchanged_group_item_idx) {

// Done outputting rollup rows; on next Read() call, deal with the new

// group instead.

if (m_seen_eof) {

m_state = DONE_OUTPUTTING_ROWS;

} else {

m_state = LAST_ROW_STARTED_NEW_GROUP;

}

}

if (m_output_slice != -1) {

m_join->set_ref_item_slice(m_output_slice);

}

return 0;

case DONE_OUTPUTTING_ROWS:

if (m_save_nullinfo != 0) {

m_join->restore_fields(m_save_nullinfo);

m_save_nullinfo = 0;

}

SetRollupLevel(INT_MAX); // Higher-level iterators up above should not

// activate any rollup.

return -1;

}

assert(false);

return 1;

if (m_rollup && m_current_rollup_position != level) {

m_current_rollup_position = level;

for (Item_rollup_group_item *item : m_join->rollup_group_items) {

item->set_current_rollup_level(level);

}

for (Item_rollup_sum_switcher *item : m_join->rollup_sums) {

item->set_current_rollup_level(level);

}

}

if (m_source_outer->Init()) {

return true;

}

m_state = NEEDS_OUTER_ROW;

if (m_pfs_batch_mode) {

m_source_inner->EndPSIBatchModeIfStarted();

}

return false;

if (m_state == END_OF_ROWS) {

return -1;

}

for (;;) { // Termination condition within loop.

if (m_state == NEEDS_OUTER_ROW) {

int err = m_source_outer->Read();

if (err == 1) {

return 1; // Error.

}

if (err == -1) {

m_state = END_OF_ROWS;

return -1;

}

if (m_pfs_batch_mode) {

m_source_inner->StartPSIBatchMode();

}

// Init() could read the NULL row flags (e.g., when building a hash

// table), so unset them before instead of after.

m_source_inner->SetNullRowFlag(false);

if (m_source_inner->Init()) {

return 1;

}

m_state = READING_FIRST_INNER_ROW;

}

assert(m_state == READING_INNER_ROWS || m_state == READING_FIRST_INNER_ROW);

int err = m_source_inner->Read();

if (err != 0 && m_pfs_batch_mode) {

m_source_inner->EndPSIBatchModeIfStarted();

}

if (err == 1) {

return 1; // Error.

}

if (thd()->killed) { // Aborted by user.

thd()->send_kill_message();

return 1;

}

if (err == -1) {

// Out of inner rows for this outer row. If we are an outer join

// and never found any inner rows, return a null-complemented row.

// If not, skip that and go straight to reading a new outer row.

if ((m_join_type == JoinType::OUTER &&

m_state == READING_FIRST_INNER_ROW) ||

m_join_type == JoinType::ANTI) {

m_source_inner->SetNullRowFlag(true);

m_state = NEEDS_OUTER_ROW;

return 0;

} else {

m_state = NEEDS_OUTER_ROW;

continue;

}

}

// An inner row has been found.

if (m_join_type == JoinType::ANTI) {

// Anti-joins should stop scanning the inner side as soon as we see

// a row, without returning that row.

m_state = NEEDS_OUTER_ROW;

continue;

}

// We have a new row. Semijoins should stop after the first row;

// regular joins (inner and outer) should go on to scan the rest.

if (m_join_type == JoinType::SEMI) {

m_state = NEEDS_OUTER_ROW;

} else {

m_state = READING_INNER_ROWS;

}

return 0;

}

public:

struct TimeStamp {};

static TimeStamp Now() { return TimeStamp(); }

double GetFirstRowMs() const override {

assert(false);

return 0.0;

}

double GetLastRowMs() const override {

assert(false);

return 0.0;

}

uint64_t GetNumInitCalls() const override {

assert(false);

return 0;

}

uint64_t GetNumRows() const override {

assert(false);

return 0;

}

/*

void StopInit([[maybe_unused]] TimeStamp start_time) {}

void IncrementNumRows([[maybe_unused]] uint64_t materialized_rows) {}

void StopRead([[maybe_unused]] TimeStamp start_time,

[[maybe_unused]] bool read_ok) {}

public:

/**

MaterializeIterator(THD *thd,

Mem_root_array<materialize_iterator::QueryBlock>

query_blocks_to_materialize,

const MaterializePathParameters *path_params,

unique_ptr_destroy_only<RowIterator> table_iterator,

JOIN *join);

bool Init() override;

int Read() override;

void SetNullRowFlag(bool is_null_row) override {

m_table_iterator->SetNullRowFlag(is_null_row);

}

void StartPSIBatchMode() override { m_table_iterator->StartPSIBatchMode(); }

void EndPSIBatchModeIfStarted() override;

// The temporary table is private to us, so there's no need to worry about

// locks to other transactions.

void UnlockRow() override {}

const IteratorProfiler *GetProfiler() const override {

assert(thd()->lex->is_explain_analyze);

return &m_profiler;

}

const Profiler *GetTableIterProfiler() const {

return &m_table_iter_profiler;

}

private:

Mem_root_array<materialize_iterator::QueryBlock>

m_query_blocks_to_materialize;

unique_ptr_destroy_only<RowIterator> m_table_iterator;

/// If we are materializing a CTE, points to it (otherwise nullptr).

/// Used so that we see if some other iterator already materialized the table,

/// avoiding duplicate work.

Common_table_expr *m_cte;

/// The query expression we are materializing. For derived tables,

/// we materialize the entire query expression; for materialization within

/// a query expression (e.g. for sorting or for windowing functions),

/// we materialize only parts of it. Used to clear correlated CTEs within

/// the unit when we rematerialize, since they depend on values from

/// outside the query expression, and those values may have changed

/// since last materialization.

Query_expression *m_query_expression;

/// See constructor.

JOIN *const m_join;

/// The slice to set when accessing temporary table; used if anything upstream

/// (e.g. WHERE, HAVING) wants to evaluate values based on its contents.

/// See constructor.

const int m_ref_slice;

/// If true, we need to materialize anew for each Init() (because the contents

/// of the table will depend on some outer non-constant value).

const bool m_rematerialize;

/// See constructor.

const bool m_reject_multiple_rows;

/// See constructor.

const ha_rows m_limit_rows;

struct Invalidator {

const CacheInvalidatorIterator *iterator;

int64_t generation_at_last_materialize;

};

Mem_root_array<Invalidator> m_invalidators;

/**

Profiler m_profiler;

/**

Profiler m_table_iter_profiler;

/// Whether we are deduplicating using a hash field on the temporary

/// table. (This condition mirrors check_unique_constraint().)

/// If so, we compute a hash value for every row, look up all rows with

/// the same hash and manually compare them to the row we are trying to

/// insert.

///

/// Note that this is _not_ the common way of deduplicating as we go.

/// The common method is to have a regular index on the table

/// over the right columns, and in that case, ha_write_row() will fail

/// with an ignorable error, so that the row is ignored even though

/// check_unique_constraint() is not called. However, B-tree indexes

/// have limitations, in particular on length, that sometimes require us

/// to do this instead. See create_tmp_table() for details.

bool doing_hash_deduplication() const { return table()->hash_field; }

/// Whether we are deduplicating, whether through a hash field

/// or a regular unique index.

bool doing_deduplication() const;

bool MaterializeRecursive();

bool MaterializeQueryBlock(

const materialize_iterator::QueryBlock &query_block,

ha_rows *stored_rows);