4493d40a-92aa-4162-9dfc-bfc4ec0996e6
/
polardbxengine
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
polardbxengine/sql/sorting_iterator.cc

708 lines
24 KiB
C++
Raw Permalink Blame History

/* Copyright (c) 2018, 2019, Oracle and/or its affiliates. All rights reserved.
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License, version 2.0,
as published by the Free Software Foundation.
This program is also distributed with certain software (including
but not limited to OpenSSL) that is licensed under separate terms,
as designated in a particular file or component or in included license
documentation. The authors of MySQL hereby grant you an additional
permission to link the program and your derivative works with the
separately licensed software that they have included with MySQL.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License, version 2.0, for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA */
#include <stdio.h>
#include <sys/types.h>
#include <algorithm>
#include <cstring>
#include <memory>
#include <new>
#include "map_helpers.h"
#include "my_alloc.h"
#include "my_base.h"
#include "my_byteorder.h"
#include "my_dbug.h"
#include "my_inttypes.h"
#include "my_pointer_arithmetic.h"
#include "my_sys.h"
#include "my_thread_local.h"
#include "mysql/service_mysql_alloc.h"
#include "sql/basic_row_iterators.h"
#include "sql/field.h"
#include "sql/filesort.h" // Filesort
#include "sql/handler.h"
#include "sql/item.h"
#include "sql/mysqld.h" // stage_executing
#include "sql/psi_memory_key.h"
#include "sql/query_options.h"
#include "sql/sort_param.h"
#include "sql/sorting_iterator.h"
#include "sql/sql_class.h"
#include "sql/sql_const.h"
#include "sql/sql_executor.h"
#include "sql/sql_lex.h"
#include "sql/sql_opt_exec_shared.h"
#include "sql/sql_optimizer.h" // JOIN
#include "sql/sql_show.h" // get_schema_tables_result
#include "sql/sql_sort.h"
#include "sql/system_variables.h"
#include "sql/table.h"
#include "thr_lock.h"
#include "varlen_sort.h"
using std::string;
using std::vector;
SortFileIndirectIterator::SortFileIndirectIterator(THD *thd, TABLE *table,
IO_CACHE *tempfile,
bool request_cache,
bool ignore_not_found_rows,
ha_rows *examined_rows)
: TableRowIterator(thd, table),
m_io_cache(tempfile),
m_examined_rows(examined_rows),
m_record(table->record[0]),
m_ref_pos(table->file->ref),
m_ignore_not_found_rows(ignore_not_found_rows),
m_using_cache(request_cache),
m_ref_length(table->file->ref_length) {}
SortFileIndirectIterator::~SortFileIndirectIterator() {
(void)table()->file->ha_index_or_rnd_end();
close_cached_file(m_io_cache);
my_free(m_io_cache);
}
bool SortFileIndirectIterator::Init() {
// The sort's source iterator could have initialized an index
// read, and it won't call end until it's destroyed (which we
// can't do before destroying SortingIterator, since we may need
// to scan/sort multiple times). Thus, as a small hack, we need
// to reset it here.
table()->file->ha_index_or_rnd_end();
int error = table()->file->ha_rnd_init(0);
if (error) {
PrintError(error);
return true;
}
if (m_using_cache && thd()->variables.read_rnd_buff_size &&
!(table()->file->ha_table_flags() & HA_FAST_KEY_READ) &&
(table()->db_stat & HA_READ_ONLY ||
table()->reginfo.lock_type <= TL_READ_NO_INSERT) &&
(ulonglong)table()->s->reclength *
(table()->file->stats.records + table()->file->stats.deleted) >
(ulonglong)MIN_FILE_LENGTH_TO_USE_ROW_CACHE &&
m_io_cache->end_of_file / m_ref_length * table()->s->reclength >
(my_off_t)MIN_ROWS_TO_USE_TABLE_CACHE &&
!table()->s->blob_fields && m_ref_length <= MAX_REFLENGTH) {
m_using_cache = !InitCache();
} else {
m_using_cache = false;
}
DBUG_PRINT("info", ("using cache: %d", m_using_cache));
return false;
}
/**
Initialize caching of records from temporary file.
@retval
false OK, use caching.
true Buffer is too small, or cannot be allocated.
Skip caching, and read records directly from temporary file.
*/
bool SortFileIndirectIterator::InitCache() {
m_struct_length = 3 + MAX_REFLENGTH;
m_reclength = ALIGN_SIZE(table()->s->reclength + 1);
if (m_reclength < m_struct_length) m_reclength = ALIGN_SIZE(m_struct_length);
m_error_offset = table()->s->reclength;
m_cache_records =
thd()->variables.read_rnd_buff_size / (m_reclength + m_struct_length);
uint rec_cache_size = m_cache_records * m_reclength;
m_rec_cache_size = m_cache_records * m_ref_length;
if (m_cache_records <= 2) {
return true;
}
m_cache.reset((uchar *)my_malloc(
key_memory_READ_RECORD_cache,
rec_cache_size + m_cache_records * m_struct_length, MYF(0)));
if (m_cache == nullptr) {
return true;
}
DBUG_PRINT("info", ("Allocated buffer for %d records", m_cache_records));
m_read_positions = m_cache.get() + rec_cache_size;
m_cache_pos = m_cache_end = m_cache.get();
return false;
}
int SortFileIndirectIterator::Read() {
if (m_using_cache) {
return CachedRead();
} else {
return UncachedRead();
}
}
int SortFileIndirectIterator::UncachedRead() {
for (;;) {
if (my_b_read(m_io_cache, m_ref_pos, m_ref_length))
return -1; /* End of file */
int tmp = table()->file->ha_rnd_pos(m_record, m_ref_pos);
if (tmp == 0) {
if (m_examined_rows != nullptr) {
++*m_examined_rows;
}
return 0;
}
/* The following is extremely unlikely to happen */
if (tmp == HA_ERR_RECORD_DELETED ||
(tmp == HA_ERR_KEY_NOT_FOUND && m_ignore_not_found_rows))
continue;
return HandleError(tmp);
}
}
int SortFileIndirectIterator::CachedRead() {
uint i;
ulong length;
my_off_t rest_of_file;
int16 error;
uchar *position, *ref_position, *record_pos;
ulong record;
for (;;) {
if (m_cache_pos != m_cache_end) {
if (m_cache_pos[m_error_offset]) {
shortget(&error, m_cache_pos);
if (error == HA_ERR_KEY_NOT_FOUND && m_ignore_not_found_rows) {
m_cache_pos += m_reclength;
continue;
}
PrintError(error);
} else {
error = 0;
memcpy(m_record, m_cache_pos, (size_t)table()->s->reclength);
}
m_cache_pos += m_reclength;
if (error == 0 && m_examined_rows != nullptr) {
++*m_examined_rows;
}
return ((int)error);
}
length = m_rec_cache_size;
rest_of_file = m_io_cache->end_of_file - my_b_tell(m_io_cache);
if ((my_off_t)length > rest_of_file) length = (ulong)rest_of_file;
if (!length || my_b_read(m_io_cache, m_cache.get(), length)) {
DBUG_PRINT("info", ("Found end of file"));
return -1; /* End of file */
}
length /= m_ref_length;
position = m_cache.get();
ref_position = m_read_positions;
for (i = 0; i < length; i++, position += m_ref_length) {
memcpy(ref_position, position, (size_t)m_ref_length);
ref_position += MAX_REFLENGTH;
int3store(ref_position, (long)i);
ref_position += 3;
}
size_t ref_length = m_ref_length;
DBUG_ASSERT(ref_length <= MAX_REFLENGTH);
varlen_sort(m_read_positions, m_read_positions + length * m_struct_length,
m_struct_length, [ref_length](const uchar *a, const uchar *b) {
return memcmp(a, b, ref_length) < 0;
});
position = m_read_positions;
for (i = 0; i < length; i++) {
memcpy(m_ref_pos, position, (size_t)m_ref_length);
position += MAX_REFLENGTH;
record = uint3korr(position);
position += 3;
record_pos = m_cache.get() + record * m_reclength;
error = (int16)table()->file->ha_rnd_pos(record_pos, m_ref_pos);
if (error) {
record_pos[m_error_offset] = 1;
shortstore(record_pos, error);
DBUG_PRINT("error", ("Got error: %d:%d when reading row", my_errno(),
(int)error));
} else
record_pos[m_error_offset] = 0;
}
m_cache_pos = m_cache.get();
m_cache_end = m_cache_pos + length * m_reclength;
}
}
vector<string> SortFileIndirectIterator::DebugString() const {
// Not used, because sort result iterator is not decided at EXPLAIN time
// (we can't know whether the buffer would stay in RAM or not).
return {string("Read sorted data: Row IDs from file, records from ") +
table()->alias};
}
template <bool Packed_addon_fields>
SortFileIterator<Packed_addon_fields>::SortFileIterator(THD *thd, TABLE *table,
IO_CACHE *tempfile,
Filesort_info *sort,
ha_rows *examined_rows)
: TableRowIterator(thd, table),
m_rec_buf(sort->addon_fields->get_addon_buf()),
m_ref_length(sort->addon_fields->get_addon_buf_length()),
m_io_cache(tempfile),
m_sort(sort),
m_examined_rows(examined_rows) {}
template <bool Packed_addon_fields>
SortFileIterator<Packed_addon_fields>::~SortFileIterator() {
close_cached_file(m_io_cache);
my_free(m_io_cache);
}
/**
Read a result set record from a temporary file after sorting.
The function first reads the next sorted record from the temporary file.
into a buffer. If a success it calls a callback function that unpacks
the fields values use in the result set from this buffer into their
positions in the regular record buffer.
@tparam Packed_addon_fields Are the addon fields packed?
This is a compile-time constant, to avoid if (....) tests during execution.
@retval
0 Record successfully read.
@retval
-1 There is no record to be read anymore.
*/
template <bool Packed_addon_fields>
int SortFileIterator<Packed_addon_fields>::Read() {
uchar *destination = m_rec_buf;
if (Packed_addon_fields) {
const uint len_sz = Addon_fields::size_of_length_field;
// First read length of the record.
if (my_b_read(m_io_cache, destination, len_sz)) return -1;
uint res_length = Addon_fields::read_addon_length(destination);
DBUG_ASSERT(res_length > len_sz);
DBUG_ASSERT(m_sort->using_addon_fields());
// Then read the rest of the record.
if (my_b_read(m_io_cache, destination + len_sz, res_length - len_sz))
return -1; /* purecov: inspected */
} else {
if (my_b_read(m_io_cache, destination, m_ref_length)) return -1;
}
m_sort->unpack_addon_fields<Packed_addon_fields>(destination);
if (m_examined_rows != nullptr) {
++*m_examined_rows;
}
return 0;
}
template <bool Packed_addon_fields>
vector<string> SortFileIterator<Packed_addon_fields>::DebugString() const {
// Not used, because sort result iterator is not decided at EXPLAIN time
// (we can't know whether the buffer would stay in RAM or not).
return {string("Read sorted data from file (originally from ") +
table()->alias + ")"};
}
template <bool Packed_addon_fields>
SortBufferIterator<Packed_addon_fields>::SortBufferIterator(
THD *thd, TABLE *table, Filesort_info *sort, Sort_result *sort_result,
ha_rows *examined_rows)
: TableRowIterator(thd, table),
m_sort(sort),
m_sort_result(sort_result),
m_examined_rows(examined_rows) {}
template <bool Packed_addon_fields>
SortBufferIterator<Packed_addon_fields>::~SortBufferIterator() {
m_sort_result->sorted_result.reset();
m_sort_result->sorted_result_in_fsbuf = false;
}
template <bool Packed_addon_fields>
bool SortBufferIterator<Packed_addon_fields>::Init() {
m_unpack_counter = 0;
return false;
}
/**
Read a result set record from a buffer after sorting.
Get the next record from the filesort buffer,
then unpack the fields into their positions in the regular record buffer.
@tparam Packed_addon_fields Are the addon fields packed?
This is a compile-time constant, to avoid if (....) tests during execution.
TODO: consider templatizing on is_varlen as well.
Variable / Fixed size key is currently handled by
Filesort_info::get_start_of_payload
@retval
0 Record successfully read.
@retval
-1 There is no record to be read anymore.
*/
template <bool Packed_addon_fields>
int SortBufferIterator<Packed_addon_fields>::Read() {
if (m_unpack_counter ==
m_sort_result->found_records) // XXX send in as a parameter?
return -1; /* End of buffer */
uchar *record = m_sort->get_sorted_record(m_unpack_counter++);
uchar *payload = get_start_of_payload(m_sort, record);
m_sort->unpack_addon_fields<Packed_addon_fields>(payload);
if (m_examined_rows != nullptr) {
++*m_examined_rows;
}
return 0;
}
template <bool Packed_addon_fields>
vector<string> SortBufferIterator<Packed_addon_fields>::DebugString() const {
// Not used, because sort result iterator is not decided at EXPLAIN time
// (we can't know whether the buffer would stay in RAM or not).
return {string("Read sorted data from memory (originally from ") +
table()->alias + ")"};
}
SortBufferIndirectIterator::SortBufferIndirectIterator(
THD *thd, TABLE *table, Sort_result *sort_result,
bool ignore_not_found_rows, ha_rows *examined_rows)
: TableRowIterator(thd, table),
m_sort_result(sort_result),
m_ref_length(table->file->ref_length),
m_examined_rows(examined_rows),
m_record(table->record[0]),
m_ignore_not_found_rows(ignore_not_found_rows) {}
SortBufferIndirectIterator::~SortBufferIndirectIterator() {
m_sort_result->sorted_result.reset();
DBUG_ASSERT(!m_sort_result->sorted_result_in_fsbuf);
m_sort_result->sorted_result_in_fsbuf = false;
(void)table()->file->ha_index_or_rnd_end();
}
bool SortBufferIndirectIterator::Init() {
// The sort's source iterator could have initialized an index
// read, and it won't call end until it's destroyed (which we
// can't do before destroying SortingIterator, since we may need
// to scan/sort multiple times). Thus, as a small hack, we need
// to reset it here.
table()->file->ha_index_or_rnd_end();
int error = table()->file->ha_rnd_init(0);
if (error) {
PrintError(error);
return true;
}
m_cache_pos = m_sort_result->sorted_result.get();
m_cache_end =
m_cache_pos + m_sort_result->found_records * table()->file->ref_length;
return false;
}
int SortBufferIndirectIterator::Read() {
for (;;) {
if (m_cache_pos == m_cache_end) return -1; /* End of file */
uchar *cache_pos = m_cache_pos;
m_cache_pos += m_ref_length;
int tmp = table()->file->ha_rnd_pos(m_record, cache_pos);
if (tmp == 0) {
if (m_examined_rows != nullptr) {
++*m_examined_rows;
}
return 0;
}
/* The following is extremely unlikely to happen */
if (tmp == HA_ERR_RECORD_DELETED ||
(tmp == HA_ERR_KEY_NOT_FOUND && m_ignore_not_found_rows))
continue;
return HandleError(tmp);
}
}
vector<string> SortBufferIndirectIterator::DebugString() const {
// Not used, because sort result iterator is not decided at EXPLAIN time
// (we can't know whether the buffer would stay in RAM or not).
return {string("Read sorted data: Row IDs from memory, records from ") +
table()->alias};
}
SortingIterator::SortingIterator(THD *thd, Filesort *filesort,
unique_ptr_destroy_only<RowIterator> source,
ha_rows *examined_rows)
: RowIterator(thd),
m_filesort(filesort),
m_source_iterator(move(source)),
m_examined_rows(examined_rows) {}
SortingIterator::~SortingIterator() {
ReleaseBuffers();
CleanupAfterQuery();
}
void SortingIterator::CleanupAfterQuery() {
m_fs_info.free_sort_buffer();
my_free(m_fs_info.merge_chunks.array());
m_fs_info.merge_chunks = Merge_chunk_array(NULL, 0);
m_fs_info.addon_fields = NULL;
}
void SortingIterator::ReleaseBuffers() {
m_result_iterator.reset();
if (m_sort_result.io_cache) {
// NOTE: The io_cache is only owned by us if it were never used.
close_cached_file(m_sort_result.io_cache);
my_free(m_sort_result.io_cache);
m_sort_result.io_cache = nullptr;
}
m_sort_result.sorted_result.reset();
m_sort_result.sorted_result_in_fsbuf = false;
// Keep the sort buffer in m_fs_info.
}
bool SortingIterator::Init() {
QEP_TAB *qep_tab = m_filesort->qep_tab;
ReleaseBuffers();
// Both empty result and error count as errors. (TODO: Why? This is a legacy
// choice that doesn't always seem right to me, although it should nearly
// never happen in practice.)
if (DoSort(qep_tab) != 0) return true;
/*
Filesort has filtered rows already (see skip_record() in
find_all_keys()): so we can simply scan the cache, so have to set
quick=NULL.
But if we do this, we still need to delete the quick, now or later. We
cannot do it now: the dtor of quick_index_merge would do free_io_cache,
but the cache has to remain, because scan will read from it.
So we delay deletion: we just let the "quick" continue existing in
"quick_optim"; double benefit:
- EXPLAIN will show the "quick_optim"
- it will be deleted late enough.
There is an exception to the reasoning above. If the filtering condition
contains a condition triggered by Item_func_trig_cond::FOUND_MATCH
(i.e. QEP_TAB is inner to an outer join), the trigger variable is still
false at this stage, so the condition evaluated to true in skip_record()
and did not filter rows. In that case, we leave the condition in place for
the next stage (evaluate_join_record()). We can still delete the QUICK as
triggered conditions don't use that.
If you wonder how we can come here for such inner table: it can happen if
the outer table is constant (so the inner one is first-non-const) and a
window function requires sorting.
*/
qep_tab->set_quick(NULL);
if (!qep_tab->is_inner_table_of_outer_join()) qep_tab->set_condition(NULL);
// Prepare the result iterator for actually reading the data. Read()
// will proxy to it.
TABLE *table = qep_tab->table();
if (m_sort_result.io_cache && my_b_inited(m_sort_result.io_cache)) {
// Test if ref-records was used
if (m_fs_info.using_addon_fields()) {
DBUG_PRINT("info", ("using SortFileIterator"));
if (m_fs_info.addon_fields->using_packed_addons())
m_result_iterator.reset(
new (&m_result_iterator_holder.sort_file_packed_addons)
SortFileIterator<true>(thd(), table, m_sort_result.io_cache,
&m_fs_info, m_examined_rows));
else
m_result_iterator.reset(
new (&m_result_iterator_holder.sort_file)
SortFileIterator<false>(thd(), table, m_sort_result.io_cache,
&m_fs_info, m_examined_rows));
} else {
/*
m_fs_info->addon_field is checked because if we use addon fields,
it doesn't make sense to use cache - we don't read from the table
and m_fs_info->io_cache is read sequentially
*/
bool request_cache = !m_fs_info.using_addon_fields();
m_result_iterator.reset(
new (&m_result_iterator_holder.sort_file_indirect)
SortFileIndirectIterator(
thd(), table, m_sort_result.io_cache, request_cache,
/*ignore_not_found_rows=*/false, m_examined_rows));
}
m_sort_result.io_cache =
nullptr; // The result iterator has taken ownership.
} else {
DBUG_ASSERT(m_sort_result.has_result_in_memory());
if (m_fs_info.using_addon_fields()) {
DBUG_PRINT("info", ("using SortBufferIterator"));
DBUG_ASSERT(m_sort_result.sorted_result_in_fsbuf);
if (m_fs_info.addon_fields->using_packed_addons())
m_result_iterator.reset(
new (&m_result_iterator_holder.sort_buffer_packed_addons)
SortBufferIterator<true>(thd(), table, &m_fs_info,
&m_sort_result, m_examined_rows));
else
m_result_iterator.reset(
new (&m_result_iterator_holder.sort_buffer)
SortBufferIterator<false>(thd(), table, &m_fs_info,
&m_sort_result, m_examined_rows));
} else {
DBUG_PRINT("info", ("using SortBufferIndirectIterator (sort)"));
m_result_iterator.reset(
new (&m_result_iterator_holder.sort_buffer_indirect)
SortBufferIndirectIterator(thd(), table, &m_sort_result,
/*ignore_not_found_rows=*/false,
m_examined_rows));
}
}
return m_result_iterator->Init();
}
/*
Do the actual sort, by calling filesort. The result will be left in one of
several places depending on what sort strategy we chose; it is up to Init() to
figure out what happened and create the appropriate iterator to read from it.
RETURN VALUES
0 ok
-1 Some fatal error
1 No records
*/
int SortingIterator::DoSort(QEP_TAB *qep_tab) {
JOIN *join = qep_tab->join();
/*
One row, no need to sort. make_tmp_tables_info should already handle this.
ROLLUP generates one more row. So that is the only exception.
*/
if (join != nullptr) {
DBUG_ASSERT(
(!join->plan_is_const() || join->rollup.state != ROLLUP::STATE_NONE) &&
m_filesort);
}
TABLE *table = qep_tab->table();
DBUG_ASSERT(m_sort_result.io_cache == nullptr);
m_sort_result.io_cache =
(IO_CACHE *)my_malloc(key_memory_TABLE_sort_io_cache, sizeof(IO_CACHE),
MYF(MY_WME | MY_ZEROFILL));
// If table has a range, move it to select
if (qep_tab->quick() && qep_tab->ref().key >= 0) {
if (qep_tab->type() != JT_REF_OR_NULL && qep_tab->type() != JT_FT) {
DBUG_ASSERT(qep_tab->type() == JT_REF || qep_tab->type() == JT_EQ_REF);
// Update ref value
if (cp_buffer_from_ref(thd(), table, &qep_tab->ref()) &&
thd()->is_fatal_error())
return -1; // out of memory
}
}
if (join != nullptr && join->unit->root_iterator() == nullptr) {
/* Fill schema tables with data before filesort if it's necessary */
if ((join->select_lex->active_options() & OPTION_SCHEMA_TABLE) &&
get_schema_tables_result(join, PROCESSED_BY_CREATE_SORT_INDEX))
return -1;
}
ha_rows found_rows;
bool error = filesort(thd(), m_filesort, m_source_iterator.get(), &m_fs_info,
&m_sort_result, &found_rows);
qep_tab->set_records(found_rows); // For SQL_CALC_FOUND_ROWS
table->set_keyread(false); // Restore if we used indexes
if (qep_tab->type() == JT_FT)
table->file->ft_end();
else
table->file->ha_index_or_rnd_end();
return error;
}
template <bool Packed_addon_fields>
inline void Filesort_info::unpack_addon_fields(uchar *buff) {
Sort_addon_field *addonf = addon_fields->begin();
const uchar *start_of_record = buff + addonf->offset;
for (; addonf != addon_fields->end(); ++addonf) {
Field *field = addonf->field;
if (addonf->null_bit && (addonf->null_bit & buff[addonf->null_offset])) {
field->set_null();
continue;
}
field->set_notnull();
if (Packed_addon_fields)
start_of_record = field->unpack(field->ptr, start_of_record);
else
field->unpack(field->ptr, buff + addonf->offset);
}
}
vector<string> SortingIterator::DebugString() const {
string ret;
if (m_filesort->using_addon_fields()) {
ret = "Sort";
} else {
ret = "Sort row IDs";
}
if (m_filesort->m_remove_duplicates) {
ret += " with duplicate removal: ";
} else {
ret += ": ";
}
bool first = true;
for (unsigned i = 0; i < m_filesort->sort_order_length(); ++i) {
if (first) {
first = false;
} else {
ret += ", ";
}
const st_sort_field *order = &m_filesort->sortorder[i];
if (order->item) {
ret += ItemToString(order->item);
} else {
ret += order->field->table->alias;
ret += ".";
ret += order->field->field_name;
}
if (order->reverse) {
ret += " DESC";
}
}
if (m_filesort->limit != HA_POS_ERROR) {
char buf[256];
snprintf(buf, sizeof(buf), ", limit input to %llu row(s) per chunk",
m_filesort->limit);
ret += buf;
}
return {ret};
}
Reference in New Issue View Git Blame Copy Permalink