mariadb/storage/tokudb/ha_tokudb_update.cc
Monty 5a759d31f7 Changing field::field_name and Item::name to LEX_CSTRING
Benefits of this patch:
- Removed a lot of calls to strlen(), especially for field_string
- Strings generated by parser are now const strings, less chance of
  accidently changing a string
- Removed a lot of calls with LEX_STRING as parameter (changed to pointer)
- More uniform code
- Item::name_length was not kept up to date. Now fixed
- Several bugs found and fixed (Access to null pointers,
  access of freed memory, wrong arguments to printf like functions)
- Removed a lot of casts from (const char*) to (char*)

Changes:
- This caused some ABI changes
  - lex_string_set now uses LEX_CSTRING
  - Some fucntions are now taking const char* instead of char*
- Create_field::change and after changed to LEX_CSTRING
- handler::connect_string, comment and engine_name() changed to LEX_CSTRING
- Checked printf() related calls to find bugs. Found and fixed several
  errors in old code.
- A lot of changes from LEX_STRING to LEX_CSTRING, especially related to
  parsing and events.
- Some changes from LEX_STRING and LEX_STRING & to LEX_CSTRING*
- Some changes for char* to const char*
- Added printf argument checking for my_snprintf()
- Introduced null_clex_str, star_clex_string, temp_lex_str to simplify
  code
- Added item_empty_name and item_used_name to be able to distingush between
  items that was given an empty name and items that was not given a name
  This is used in sql_yacc.yy to know when to give an item a name.
- select table_name."*' is not anymore same as table_name.*
- removed not used function Item::rename()
- Added comparision of item->name_length before some calls to
  my_strcasecmp() to speed up comparison
- Moved Item_sp_variable::make_field() from item.h to item.cc
- Some minimal code changes to avoid copying to const char *
- Fixed wrong error message in wsrep_mysql_parse()
- Fixed wrong code in find_field_in_natural_join() where real_item() was
  set when it shouldn't
- ER_ERROR_ON_RENAME was used with extra arguments.
- Removed some (wrong) ER_OUTOFMEMORY, as alloc_root will already
  give the error.

TODO:
- Check possible unsafe casts in plugin/auth_examples/qa_auth_interface.c
- Change code to not modify LEX_CSTRING for database name
  (as part of lower_case_table_names)
2017-04-23 22:35:46 +03:00

1135 lines
34 KiB
C++

/* -*- mode: C++; c-basic-offset: 4; indent-tabs-mode: nil -*- */
// vim: ft=cpp:expandtab:ts=8:sw=4:softtabstop=4:
#ident "$Id$"
/*======
This file is part of TokuDB
Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
TokuDBis is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License, version 2,
as published by the Free Software Foundation.
TokuDB 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 for more details.
You should have received a copy of the GNU General Public License
along with TokuDB. If not, see <http://www.gnu.org/licenses/>.
======= */
#ident "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."
#if TOKU_INCLUDE_UPSERT
// Point updates and upserts
// Restrictions:
// No triggers
// Statement or mixed replication
// Primary key must be defined
// Simple and compound primary key
// Int, char and varchar primary key types
// No updates on fields that are part of any key
// No clustering keys
// Integer and char field updates
// Update expressions:
// x = constant
// x = x + constant
// x = x - constant
// x = if (x=0,0,x-1)
// x = x + values(x)
// Session variable disables slow updates and slow upserts
// Future features:
// Support more primary key types
// Force statement logging for fast updates
// Support clustering keys using broadcast updates
// Support primary key ranges using multicast messages
// Support more complicated update expressions
// Replace field_offset
// Debug function to dump an Item
static void dump_item(Item* item) {
fprintf(stderr, "%u", item->type());
switch (item->type()) {
case Item::FUNC_ITEM: {
Item_func* func = static_cast<Item_func*>(item);
uint n = func->argument_count();
Item** arguments = func->arguments();
fprintf(
stderr,
":func=%u,%s,%u(",
func->functype(),
func->func_name(),
n);
for (uint i = 0; i < n ; i++) {
dump_item(arguments[i]);
if (i < n-1)
fprintf(stderr,",");
}
fprintf(stderr, ")");
break;
}
case Item::INT_ITEM: {
Item_int* int_item = static_cast<Item_int*>(item);
fprintf(stderr, ":int=%lld", int_item->val_int());
break;
}
case Item::STRING_ITEM: {
Item_string* str_item = static_cast<Item_string*>(item);
fprintf(stderr, ":str=%s", str_item->val_str(NULL)->c_ptr());
break;
}
case Item::FIELD_ITEM: {
Item_field* field_item = static_cast<Item_field*>(item);
fprintf(
stderr,
":field=%s.%s.%s",
field_item->db_name,
field_item->table_name,
field_item->field_name.str);
break;
}
case Item::COND_ITEM: {
Item_cond* cond_item = static_cast<Item_cond*>(item);
fprintf(stderr, ":cond=%s(\n", cond_item->func_name());
List_iterator<Item> li(*cond_item->argument_list());
Item* list_item;
while ((list_item = li++)) {
dump_item(list_item);
fprintf(stderr, "\n");
}
fprintf(stderr, ")\n");
break;
}
case Item::INSERT_VALUE_ITEM: {
Item_insert_value* value_item = static_cast<Item_insert_value*>(item);
fprintf(stderr, ":insert_value");
dump_item(value_item->arg);
break;
}
default:
fprintf(stderr, ":unsupported\n");
break;
}
}
// Debug function to dump an Item list
static void dump_item_list(const char* h, List<Item> &l) {
fprintf(stderr, "%s elements=%u\n", h, l.elements);
List_iterator<Item> li(l);
Item* item;
while ((item = li++) != NULL) {
dump_item(item);
fprintf(stderr, "\n");
}
}
// Find a Field by its Item name
static Field* find_field_by_name(TABLE* table, Item* item) {
if (item->type() != Item::FIELD_ITEM)
return NULL;
Item_field* field_item = static_cast<Item_field*>(item);
#if 0
if (strcmp(table->s->db.str, field_item->db_name) != 0 ||
strcmp(table->s->table_name.str, field_item->table_name) != 0)
return NULL;
Field *found_field = NULL;
for (uint i = 0; i < table->s->fields; i++) {
Field *test_field = table->s->field[i];
if (strcmp(field_item->field_name.str, test_field->field_name.str) == 0) {
found_field = test_field;
break;
}
}
return found_field;
#else
// item->field may be a shortcut instead of the above table lookup
return field_item->field;
#endif
}
// Return the starting offset in the value for a particular index (selected by idx) of a
// particular field (selected by expand_field_num).
// This only works for fixed length fields
static uint32_t fixed_field_offset(
uint32_t null_bytes,
KEY_AND_COL_INFO* kc_info,
uint idx,
uint expand_field_num) {
uint32_t offset = null_bytes;
for (uint i = 0; i < expand_field_num; i++) {
if (bitmap_is_set(&kc_info->key_filters[idx], i))
continue;
offset += kc_info->field_lengths[i];
}
return offset;
}
static uint32_t var_field_index(
TABLE* table,
KEY_AND_COL_INFO* kc_info,
uint idx,
uint field_num) {
assert_always(field_num < table->s->fields);
uint v_index = 0;
for (uint i = 0; i < table->s->fields; i++) {
if (bitmap_is_set(&kc_info->key_filters[idx], i))
continue;
if (kc_info->length_bytes[i]) {
if (i == field_num)
break;
v_index++;
}
}
return v_index;
}
static uint32_t blob_field_index(
TABLE* table,
KEY_AND_COL_INFO* kc_info,
uint idx,
uint field_num) {
assert_always(field_num < table->s->fields);
uint b_index;
for (b_index = 0; b_index < kc_info->num_blobs; b_index++) {
if (kc_info->blob_fields[b_index] == field_num)
break;
}
assert_always(b_index < kc_info->num_blobs);
return b_index;
}
// Determine if an update operation can be offloaded to the storage engine.
// The update operation consists of a list of update expressions
// (fields[i] = values[i]), and a list of where conditions (conds).
// The function returns 0 if the update is handled in the storage engine.
// Otherwise, an error is returned.
int ha_tokudb::fast_update(
THD* thd,
List<Item>& update_fields,
List<Item>& update_values,
Item* conds) {
TOKUDB_HANDLER_DBUG_ENTER("");
int error = 0;
if (TOKUDB_UNLIKELY(TOKUDB_DEBUG_FLAGS(TOKUDB_DEBUG_UPSERT))) {
dump_item_list("fields", update_fields);
dump_item_list("values", update_values);
if (conds) {
fprintf(stderr, "conds\n"); dump_item(conds); fprintf(stderr, "\n");
}
}
if (update_fields.elements < 1 ||
update_fields.elements != update_values.elements) {
error = ENOTSUP; // something is fishy with the parameters
goto return_error;
}
if (!check_fast_update(thd, update_fields, update_values, conds)) {
error = ENOTSUP;
goto check_error;
}
error = send_update_message(
update_fields,
update_values,
conds,
transaction);
if (error != 0) {
goto check_error;
}
check_error:
if (error != 0) {
if (tokudb::sysvars::disable_slow_update(thd) != 0)
error = HA_ERR_UNSUPPORTED;
if (error != ENOTSUP)
print_error(error, MYF(0));
}
return_error:
TOKUDB_HANDLER_DBUG_RETURN(error);
}
// Return true if an expression is a simple int expression or a simple function
// of +- int expression.
static bool check_int_result(Item* item) {
Item::Type t = item->type();
if (t == Item::INT_ITEM) {
return true;
} else if (t == Item::FUNC_ITEM) {
Item_func* item_func = static_cast<Item_func*>(item);
if (strcmp(item_func->func_name(), "+") != 0 &&
strcmp(item_func->func_name(), "-") != 0)
return false;
if (item_func->argument_count() != 1)
return false;
Item** arguments = item_func->arguments();
if (arguments[0]->type() != Item::INT_ITEM)
return false;
return true;
} else
return false;
}
// check that an item is an insert value item with the same field name
static bool check_insert_value(Item* item, const char* field_name) {
if (item->type() != Item::INSERT_VALUE_ITEM)
return false;
Item_insert_value* value_item = static_cast<Item_insert_value*>(item);
if (value_item->arg->type() != Item::FIELD_ITEM)
return false;
Item_field* arg = static_cast<Item_field*>(value_item->arg);
if (strcmp(field_name, arg->field_name.str) != 0)
return false;
return true;
}
// Return true if an expression looks like field_name op constant.
static bool check_x_op_constant(
const char* field_name,
Item* item,
const char* op,
Item** item_constant,
bool allow_insert_value) {
if (item->type() != Item::FUNC_ITEM)
return false;
Item_func* item_func = static_cast<Item_func*>(item);
if (strcmp(item_func->func_name(), op) != 0)
return false;
Item** arguments = item_func->arguments();
uint n = item_func->argument_count();
if (n != 2)
return false;
if (arguments[0]->type() != Item::FIELD_ITEM)
return false;
Item_field* arg0 = static_cast<Item_field*>(arguments[0]);
if (strcmp(field_name, arg0->field_name.str) != 0)
return false;
if (!check_int_result(arguments[1]))
if (!(allow_insert_value &&
check_insert_value(arguments[1], field_name)))
return false;
*item_constant = arguments[1];
return true;
}
// Return true if an expression looks like field_name = constant
static bool check_x_equal_0(const char *field_name, Item *item) {
Item* item_constant;
if (!check_x_op_constant(field_name, item, "=", &item_constant, false))
return false;
if (item_constant->type() != Item::INT_ITEM ||
item_constant->val_int() != 0)
return false;
return true;
}
// Return true if an expression looks like fieldname - 1
static bool check_x_minus_1(const char* field_name, Item* item) {
Item* item_constant;
if (!check_x_op_constant(field_name, item, "-", &item_constant, false))
return false;
if (item_constant->type() != Item::INT_ITEM ||
item_constant->val_int() != 1)
return false;
return true;
}
// Return true if an expression looks like if(fieldname=0, 0, fieldname-1) and
// the field named by fieldname is an unsigned int.
static bool check_decr_floor_expression(Field* lhs_field, Item* item) {
if (item->type() != Item::FUNC_ITEM)
return false;
Item_func* item_func = static_cast<Item_func*>(item);
if (strcmp(item_func->func_name(), "if") != 0)
return false;
Item** arguments = item_func->arguments();
uint n = item_func->argument_count();
if (n != 3)
return false;
if (!check_x_equal_0(lhs_field->field_name.str, arguments[0]))
return false;
if (arguments[1]->type() != Item::INT_ITEM || arguments[1]->val_int() != 0)
return false;
if (!check_x_minus_1(lhs_field->field_name.str, arguments[2]))
return false;
if (!(lhs_field->flags & UNSIGNED_FLAG))
return false;
return true;
}
// Check if lhs = rhs expression is simple. Return true if it is.
static bool check_update_expression(
Item* lhs_item,
Item* rhs_item,
TABLE* table,
bool allow_insert_value) {
Field* lhs_field = find_field_by_name(table, lhs_item);
if (lhs_field == NULL)
return false;
if (!lhs_field->part_of_key.is_clear_all())
return false;
enum_field_types lhs_type = lhs_field->type();
Item::Type rhs_type = rhs_item->type();
switch (lhs_type) {
case MYSQL_TYPE_TINY:
case MYSQL_TYPE_SHORT:
case MYSQL_TYPE_INT24:
case MYSQL_TYPE_LONG:
case MYSQL_TYPE_LONGLONG:
if (check_int_result(rhs_item))
return true;
Item* item_constant;
if (check_x_op_constant(
lhs_field->field_name.str,
rhs_item,
"+",
&item_constant,
allow_insert_value))
return true;
if (check_x_op_constant(
lhs_field->field_name.str,
rhs_item,
"-",
&item_constant,
allow_insert_value))
return true;
if (check_decr_floor_expression(lhs_field, rhs_item))
return true;
break;
case MYSQL_TYPE_STRING:
if (rhs_type == Item::INT_ITEM || rhs_type == Item::STRING_ITEM)
return true;
break;
case MYSQL_TYPE_VARCHAR:
case MYSQL_TYPE_BLOB:
if (rhs_type == Item::STRING_ITEM)
return true;
break;
default:
break;
}
return false;
}
// Check that all update expressions are simple. Return true if they are.
static bool check_all_update_expressions(
List<Item>& fields,
List<Item>& values,
TABLE* table,
bool allow_insert_value) {
List_iterator<Item> lhs_i(fields);
List_iterator<Item> rhs_i(values);
while (1) {
Item* lhs_item = lhs_i++;
if (lhs_item == NULL)
break;
Item* rhs_item = rhs_i++;
assert_always(rhs_item != NULL);
if (!check_update_expression(
lhs_item,
rhs_item,
table,
allow_insert_value))
return false;
}
return true;
}
static bool full_field_in_key(TABLE* table, Field* field) {
assert_always(table->s->primary_key < table->s->keys);
KEY* key = &table->s->key_info[table->s->primary_key];
for (uint i = 0; i < key->user_defined_key_parts; i++) {
KEY_PART_INFO* key_part = &key->key_part[i];
if (strcmp(field->field_name.str, key_part->field->field_name.str) == 0) {
return key_part->length == field->field_length;
}
}
return false;
}
// Check that an expression looks like fieldname = constant, fieldname is part
// of the primary key, and the named field is an int, char or varchar type.
// Return true if it does.
static bool check_pk_field_equal_constant(
Item* item,
TABLE* table,
MY_BITMAP* pk_fields) {
if (item->type() != Item::FUNC_ITEM)
return false;
Item_func* func = static_cast<Item_func*>(item);
if (strcmp(func->func_name(), "=") != 0)
return false;
uint n = func->argument_count();
if (n != 2)
return false;
Item** arguments = func->arguments();
Field* field = find_field_by_name(table, arguments[0]);
if (field == NULL)
return false;
if (!bitmap_test_and_clear(pk_fields, field->field_index))
return false;
switch (field->type()) {
case MYSQL_TYPE_TINY:
case MYSQL_TYPE_SHORT:
case MYSQL_TYPE_INT24:
case MYSQL_TYPE_LONG:
case MYSQL_TYPE_LONGLONG:
return arguments[1]->type() == Item::INT_ITEM ||
arguments[1]->type() == Item::STRING_ITEM;
case MYSQL_TYPE_STRING:
case MYSQL_TYPE_VARCHAR:
return full_field_in_key(table, field) &&
(arguments[1]->type() == Item::INT_ITEM ||
arguments[1]->type() == Item::STRING_ITEM);
default:
return false;
}
}
// Check that the where condition covers all of the primary key components
// with fieldname = constant expressions. Return true if it does.
static bool check_point_update(Item* conds, TABLE* table) {
bool result = false;
if (conds == NULL)
return false; // no where condition on the update
if (table->s->primary_key >= table->s->keys)
return false; // no primary key defined
// use a bitmap of the primary key fields to keep track of those fields
// that are covered by the where conditions
MY_BITMAP pk_fields;
if (bitmap_init(&pk_fields, NULL, table->s->fields, FALSE)) // 1 -> failure
return false;
KEY *key = &table->s->key_info[table->s->primary_key];
for (uint i = 0; i < key->user_defined_key_parts; i++)
bitmap_set_bit(&pk_fields, key->key_part[i].field->field_index);
switch (conds->type()) {
case Item::FUNC_ITEM:
result = check_pk_field_equal_constant(conds, table, &pk_fields);
break;
case Item::COND_ITEM: {
Item_cond* cond_item = static_cast<Item_cond*>(conds);
if (strcmp(cond_item->func_name(), "and") != 0)
break;
List_iterator<Item> li(*cond_item->argument_list());
Item* list_item;
result = true;
while (result == true && (list_item = li++)) {
result = check_pk_field_equal_constant(
list_item,
table,
&pk_fields);
}
break;
}
default:
break;
}
if (!bitmap_is_clear_all(&pk_fields))
result = false;
bitmap_free(&pk_fields);
return result;
}
// Return true if there are any clustering keys (except the primary).
// Precompute this when the table is opened.
static bool clustering_keys_exist(TABLE *table) {
for (uint keynr = 0; keynr < table->s->keys; keynr++) {
if (keynr != table->s->primary_key &&
key_is_clustering(&table->s->key_info[keynr]))
return true;
}
return false;
}
static bool is_strict_mode(THD* thd) {
#if 50600 <= MYSQL_VERSION_ID && MYSQL_VERSION_ID <= 50699
return thd->is_strict_mode();
#else
return tokudb_test(thd->variables.sql_mode & (MODE_STRICT_TRANS_TABLES | MODE_STRICT_ALL_TABLES));
#endif
}
// Check if an update operation can be handled by this storage engine.
// Return true if it can.
bool ha_tokudb::check_fast_update(
THD* thd,
List<Item>& fields,
List<Item>& values,
Item* conds) {
if (!transaction)
return false;
// avoid strict mode arithmetic overflow issues
if (is_strict_mode(thd))
return false;
// no triggers
if (table->triggers)
return false;
// no binlog
if (mysql_bin_log.is_open() &&
!(thd->variables.binlog_format == BINLOG_FORMAT_STMT ||
thd->variables.binlog_format == BINLOG_FORMAT_MIXED))
return false;
// no clustering keys (need to broadcast an increment into the clustering
// keys since we are selecting with the primary key)
if (clustering_keys_exist(table))
return false;
if (!check_all_update_expressions(fields, values, table, false))
return false;
if (!check_point_update(conds, table))
return false;
return true;
}
static void marshall_varchar_descriptor(
tokudb::buffer& b,
TABLE* table,
KEY_AND_COL_INFO* kc_info,
uint key_num) {
b.append_ui<uint32_t>('v');
b.append_ui<uint32_t>(
table->s->null_bytes + kc_info->mcp_info[key_num].fixed_field_size);
uint32_t var_offset_bytes = kc_info->mcp_info[key_num].len_of_offsets;
b.append_ui<uint32_t>(var_offset_bytes);
b.append_ui<uint32_t>(
var_offset_bytes == 0 ? 0 : kc_info->num_offset_bytes);
}
static void marshall_blobs_descriptor(
tokudb::buffer& b,
TABLE* table,
KEY_AND_COL_INFO* kc_info) {
b.append_ui<uint32_t>('b');
uint32_t n = kc_info->num_blobs;
b.append_ui<uint32_t>(n);
for (uint i = 0; i < n; i++) {
uint blob_field_index = kc_info->blob_fields[i];
assert_always(blob_field_index < table->s->fields);
uint8_t blob_field_length =
table->s->field[blob_field_index]->row_pack_length();
b.append(&blob_field_length, sizeof blob_field_length);
}
}
static inline uint32_t get_null_bit_position(uint32_t null_bit);
// evaluate the int value of an item
static longlong item_val_int(Item* item) {
Item::Type t = item->type();
if (t == Item::INSERT_VALUE_ITEM) {
Item_insert_value* value_item = static_cast<Item_insert_value*>(item);
return value_item->arg->val_int();
} else
return item->val_int();
}
// Marshall update operations to a buffer.
static void marshall_update(
tokudb::buffer& b,
Item* lhs_item,
Item* rhs_item,
TABLE* table,
TOKUDB_SHARE* share) {
// figure out the update operation type (again)
Field* lhs_field = find_field_by_name(table, lhs_item);
assert_always(lhs_field); // we found it before, so this should work
// compute the update info
uint32_t field_type;
uint32_t field_null_num = 0;
if (lhs_field->real_maybe_null()) {
uint32_t field_num = lhs_field->field_index;
field_null_num =
((field_num/8)*8 + get_null_bit_position(lhs_field->null_bit)) + 1;
}
uint32_t offset;
void* v_ptr = NULL;
uint32_t v_length;
uint32_t update_operation;
longlong v_ll;
String v_str;
switch (lhs_field->type()) {
case MYSQL_TYPE_TINY:
case MYSQL_TYPE_SHORT:
case MYSQL_TYPE_INT24:
case MYSQL_TYPE_LONG:
case MYSQL_TYPE_LONGLONG: {
Field_num* lhs_num = static_cast<Field_num*>(lhs_field);
field_type = lhs_num->unsigned_flag ? UPDATE_TYPE_UINT : UPDATE_TYPE_INT;
offset =
fixed_field_offset(
table->s->null_bytes,
&share->kc_info,
table->s->primary_key,
lhs_field->field_index);
switch (rhs_item->type()) {
case Item::INT_ITEM: {
update_operation = '=';
v_ll = rhs_item->val_int();
v_length = lhs_field->pack_length();
v_ptr = &v_ll;
break;
}
case Item::FUNC_ITEM: {
Item_func* rhs_func = static_cast<Item_func*>(rhs_item);
Item** arguments = rhs_func->arguments();
// we only support one if function for now, and it is a
// decrement with floor.
if (strcmp(rhs_func->func_name(), "if") == 0) {
update_operation = '-';
v_ll = 1;
} else if (rhs_func->argument_count() == 1) {
update_operation = '=';
v_ll = rhs_func->val_int();
} else {
update_operation = rhs_func->func_name()[0];
v_ll = item_val_int(arguments[1]);
}
v_length = lhs_field->pack_length();
v_ptr = &v_ll;
break;
}
default:
assert_unreachable();
}
break;
}
case MYSQL_TYPE_STRING: {
update_operation = '=';
field_type =
lhs_field->binary() ? UPDATE_TYPE_BINARY : UPDATE_TYPE_CHAR;
offset =
fixed_field_offset(
table->s->null_bytes,
&share->kc_info,
table->s->primary_key,
lhs_field->field_index);
v_str = *rhs_item->val_str(&v_str);
v_length = v_str.length();
if (v_length >= lhs_field->pack_length()) {
v_length = lhs_field->pack_length();
v_str.length(v_length); // truncate
} else {
v_length = lhs_field->pack_length();
uchar pad_char =
lhs_field->binary() ? 0 : lhs_field->charset()->pad_char;
v_str.fill(lhs_field->pack_length(), pad_char); // pad
}
v_ptr = v_str.c_ptr();
break;
}
case MYSQL_TYPE_VARCHAR: {
update_operation = '=';
field_type =
lhs_field->binary() ? UPDATE_TYPE_VARBINARY : UPDATE_TYPE_VARCHAR;
offset =
var_field_index(
table,
&share->kc_info,
table->s->primary_key,
lhs_field->field_index);
v_str = *rhs_item->val_str(&v_str);
v_length = v_str.length();
if (v_length >= lhs_field->row_pack_length()) {
v_length = lhs_field->row_pack_length();
v_str.length(v_length); // truncate
}
v_ptr = v_str.c_ptr();
break;
}
case MYSQL_TYPE_BLOB: {
update_operation = '=';
field_type = lhs_field->binary() ? UPDATE_TYPE_BLOB : UPDATE_TYPE_TEXT;
offset =
blob_field_index(
table,
&share->kc_info,
table->s->primary_key,
lhs_field->field_index);
v_str = *rhs_item->val_str(&v_str);
v_length = v_str.length();
if (v_length >= lhs_field->max_data_length()) {
v_length = lhs_field->max_data_length();
v_str.length(v_length); // truncate
}
v_ptr = v_str.c_ptr();
break;
}
default:
assert_unreachable();
}
// marshall the update fields into the buffer
b.append_ui<uint32_t>(update_operation);
b.append_ui<uint32_t>(field_type);
b.append_ui<uint32_t>(field_null_num);
b.append_ui<uint32_t>(offset);
b.append_ui<uint32_t>(v_length);
b.append(v_ptr, v_length);
}
// Save an item's value into the appropriate field. Return 0 if successful.
static int save_in_field(Item* item, TABLE* table) {
assert_always(item->type() == Item::FUNC_ITEM);
Item_func *func = static_cast<Item_func*>(item);
assert_always(strcmp(func->func_name(), "=") == 0);
uint n = func->argument_count();
assert_always(n == 2);
Item **arguments = func->arguments();
assert_always(arguments[0]->type() == Item::FIELD_ITEM);
Item_field *field_item = static_cast<Item_field*>(arguments[0]);
my_bitmap_map *old_map = dbug_tmp_use_all_columns(table, table->write_set);
int error = arguments[1]->save_in_field(field_item->field, 0);
dbug_tmp_restore_column_map(table->write_set, old_map);
return error;
}
static void count_update_types(
Field* lhs_field,
uint* num_varchars,
uint* num_blobs) {
switch (lhs_field->type()) {
case MYSQL_TYPE_VARCHAR:
*num_varchars += 1;
break;
case MYSQL_TYPE_BLOB:
*num_blobs += 1;
break;
default:
break;
}
}
// Generate an update message for an update operation and send it into the
// primary tree. Return 0 if successful.
int ha_tokudb::send_update_message(
List<Item>& update_fields,
List<Item>& update_values,
Item* conds,
DB_TXN* txn) {
int error;
// Save the primary key from the where conditions
Item::Type t = conds->type();
if (t == Item::FUNC_ITEM) {
error = save_in_field(conds, table);
} else if (t == Item::COND_ITEM) {
Item_cond* cond_item = static_cast<Item_cond*>(conds);
List_iterator<Item> li(*cond_item->argument_list());
Item* list_item;
error = 0;
while (error == 0 && (list_item = li++)) {
error = save_in_field(list_item, table);
}
} else {
assert_unreachable();
}
if (error)
return error;
// put the primary key into key_buff and wrap it with key_dbt
DBT key_dbt;
bool has_null;
create_dbt_key_from_table(
&key_dbt,
primary_key,
key_buff,
table->record[0],
&has_null);
// construct the update message
tokudb::buffer update_message;
uint8_t op = UPDATE_OP_UPDATE_2;
update_message.append(&op, sizeof op);
uint32_t num_updates = update_fields.elements;
uint num_varchars = 0, num_blobs = 0;
if (1) {
List_iterator<Item> lhs_i(update_fields);
Item* lhs_item;
while ((lhs_item = lhs_i++)) {
if (lhs_item == NULL)
break;
Field* lhs_field = find_field_by_name(table, lhs_item);
assert_always(lhs_field); // we found it before, so this should work
count_update_types(lhs_field, &num_varchars, &num_blobs);
}
if (num_varchars > 0 || num_blobs > 0)
num_updates++;
if (num_blobs > 0)
num_updates++;
}
// append the updates
update_message.append_ui<uint32_t>(num_updates);
if (num_varchars > 0 || num_blobs > 0)
marshall_varchar_descriptor(
update_message,
table,
&share->kc_info,
table->s->primary_key);
if (num_blobs > 0)
marshall_blobs_descriptor(update_message, table, &share->kc_info);
List_iterator<Item> lhs_i(update_fields);
List_iterator<Item> rhs_i(update_values);
while (error == 0) {
Item* lhs_item = lhs_i++;
if (lhs_item == NULL)
break;
Item* rhs_item = rhs_i++;
assert_always(rhs_item != NULL);
marshall_update(update_message, lhs_item, rhs_item, table, share);
}
share->_num_DBs_lock.lock_read();
// hot index in progress
if (share->num_DBs > table->s->keys + tokudb_test(hidden_primary_key)) {
error = ENOTSUP; // run on the slow path
} else {
// send the update message
DBT update_dbt; memset(&update_dbt, 0, sizeof update_dbt);
update_dbt.data = update_message.data();
update_dbt.size = update_message.size();
error =
share->key_file[primary_key]->update(
share->key_file[primary_key],
txn,
&key_dbt,
&update_dbt,
0);
}
share->_num_DBs_lock.unlock();
return error;
}
// Determine if an upsert operation can be offloaded to the storage engine.
// An upsert consists of a row and a list of update expressions
// (update_fields[i] = update_values[i]).
// The function returns 0 if the upsert is handled in the storage engine.
// Otherwise, an error code is returned.
int ha_tokudb::upsert(
THD* thd,
List<Item>& update_fields,
List<Item>& update_values) {
TOKUDB_HANDLER_DBUG_ENTER("");
int error = 0;
if (TOKUDB_UNLIKELY(TOKUDB_DEBUG_FLAGS(TOKUDB_DEBUG_UPSERT))) {
fprintf(stderr, "upsert\n");
dump_item_list("update_fields", update_fields);
dump_item_list("update_values", update_values);
}
// not an upsert or something is fishy with the parameters
if (update_fields.elements < 1 ||
update_fields.elements != update_values.elements) {
error = ENOTSUP;
goto return_error;
}
if (!check_upsert(thd, update_fields, update_values)) {
error = ENOTSUP;
goto check_error;
}
error = send_upsert_message(thd, update_fields, update_values, transaction);
if (error != 0) {
goto check_error;
}
check_error:
if (error != 0) {
if (tokudb::sysvars::disable_slow_upsert(thd) != 0)
error = HA_ERR_UNSUPPORTED;
if (error != ENOTSUP)
print_error(error, MYF(0));
}
return_error:
TOKUDB_HANDLER_DBUG_RETURN(error);
}
// Check if an upsert can be handled by this storage engine.
// Return true if it can.
bool ha_tokudb::check_upsert(
THD* thd,
List<Item>& update_fields,
List<Item>& update_values) {
if (!transaction)
return false;
// avoid strict mode arithmetic overflow issues
if (is_strict_mode(thd))
return false;
// no triggers
if (table->triggers)
return false;
// primary key must exist
if (table->s->primary_key >= table->s->keys)
return false;
// no secondary keys
if (table->s->keys > 1)
return false;
// no binlog
if (mysql_bin_log.is_open() &&
!(thd->variables.binlog_format == BINLOG_FORMAT_STMT ||
thd->variables.binlog_format == BINLOG_FORMAT_MIXED))
return false;
if (!check_all_update_expressions(
update_fields,
update_values,
table,
true))
return false;
return true;
}
// Generate an upsert message and send it into the primary tree.
// Return 0 if successful.
int ha_tokudb::send_upsert_message(
THD* thd,
List<Item>& update_fields,
List<Item>& update_values,
DB_TXN* txn) {
int error = 0;
// generate primary key
DBT key_dbt;
bool has_null;
create_dbt_key_from_table(
&key_dbt,
primary_key,
primary_key_buff,
table->record[0],
&has_null);
// generate packed row
DBT row;
error = pack_row(&row, (const uchar *) table->record[0], primary_key);
if (error)
return error;
tokudb::buffer update_message;
// append the operation
uint8_t op = UPDATE_OP_UPSERT_2;
update_message.append(&op, sizeof op);
// append the row
update_message.append_ui<uint32_t>(row.size);
update_message.append(row.data, row.size);
uint32_t num_updates = update_fields.elements;
uint num_varchars = 0, num_blobs = 0;
if (1) {
List_iterator<Item> lhs_i(update_fields);
Item* lhs_item;
while ((lhs_item = lhs_i++)) {
if (lhs_item == NULL)
break;
Field* lhs_field = find_field_by_name(table, lhs_item);
assert_always(lhs_field); // we found it before, so this should work
count_update_types(lhs_field, &num_varchars, &num_blobs);
}
if (num_varchars > 0 || num_blobs > 0)
num_updates++;
if (num_blobs > 0)
num_updates++;
}
// append the updates
update_message.append_ui<uint32_t>(num_updates);
if (num_varchars > 0 || num_blobs > 0)
marshall_varchar_descriptor(
update_message,
table, &share->kc_info,
table->s->primary_key);
if (num_blobs > 0)
marshall_blobs_descriptor(update_message, table, &share->kc_info);
List_iterator<Item> lhs_i(update_fields);
List_iterator<Item> rhs_i(update_values);
while (1) {
Item* lhs_item = lhs_i++;
if (lhs_item == NULL)
break;
Item* rhs_item = rhs_i++;
assert_always(rhs_item != NULL);
marshall_update(update_message, lhs_item, rhs_item, table, share);
}
share->_num_DBs_lock.lock_read();
// hot index in progress
if (share->num_DBs > table->s->keys + tokudb_test(hidden_primary_key)) {
error = ENOTSUP; // run on the slow path
} else {
// send the upsert message
DBT update_dbt; memset(&update_dbt, 0, sizeof update_dbt);
update_dbt.data = update_message.data();
update_dbt.size = update_message.size();
error =
share->key_file[primary_key]->update(
share->key_file[primary_key],
txn,
&key_dbt,
&update_dbt,
0);
}
share->_num_DBs_lock.unlock();
return error;
}
#endif