Create your Gitee Account
Explore and code with more than 5 million developers,Free private repositories !:)
Sign up
Clone or download
sdb_idx.cc 19.50 KB
Copy Edit Web IDE Raw Blame History
Mars authored 2019-03-14 15:24 . Improve alteration of indexes. (#43)
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653
/* Copyright (c) 2018, SequoiaDB 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 as published by
the Free Software Foundation; version 2 of the License.
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 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 */
#ifndef MYSQL_SERVER
#define MYSQL_SERVER
#endif
#include "sdb_idx.h"
#include <myisampack.h>
#include <bson/bson.hpp>
#include "sdb_cl.h"
#include "sdb_errcode.h"
#include "sdb_def.h"
#include "sdb_log.h"
#include "sdb_util.h"
#include "sql_table.h"
static inline int get_variable_key_length(const uchar *A) {
return (int)(((uint16)(A[0])) + ((uint16)(A[1]) << 8));
}
int sdb_get_key_direction(ha_rkey_function find_flag) {
switch (find_flag) {
case HA_READ_KEY_EXACT:
case HA_READ_KEY_OR_NEXT:
case HA_READ_AFTER_KEY:
return 1;
case HA_READ_BEFORE_KEY:
case HA_READ_KEY_OR_PREV:
case HA_READ_PREFIX_LAST:
case HA_READ_PREFIX_LAST_OR_PREV:
return -1;
case HA_READ_PREFIX:
default:
return 1;
}
}
static my_bool is_field_indexable(const Field *field) {
switch (field->type()) {
case MYSQL_TYPE_TINY:
case MYSQL_TYPE_SHORT:
case MYSQL_TYPE_INT24:
case MYSQL_TYPE_LONG:
case MYSQL_TYPE_LONGLONG:
case MYSQL_TYPE_BIT:
case MYSQL_TYPE_FLOAT:
case MYSQL_TYPE_DOUBLE:
case MYSQL_TYPE_DECIMAL:
case MYSQL_TYPE_NEWDECIMAL:
case MYSQL_TYPE_YEAR:
case MYSQL_TYPE_DATE:
case MYSQL_TYPE_TIME:
case MYSQL_TYPE_DATETIME:
case MYSQL_TYPE_TIMESTAMP:
return true;
case MYSQL_TYPE_VARCHAR:
case MYSQL_TYPE_STRING:
case MYSQL_TYPE_VAR_STRING:
case MYSQL_TYPE_BLOB: {
if (!field->binary()) {
return true;
} else {
return false;
}
}
case MYSQL_TYPE_JSON:
default:
return false;
}
}
int sdb_create_index(const KEY *key_info, Sdb_cl &cl) {
const KEY_PART_INFO *key_part;
const KEY_PART_INFO *key_end;
int rc = 0;
bson::BSONObj key_obj;
my_bool is_unique = false, is_enforced = false;
bson::BSONObjBuilder key_obj_builder;
key_part = key_info->key_part;
key_end = key_part + key_info->user_defined_key_parts;
for (; key_part != key_end; ++key_part) {
if (!is_field_indexable(key_part->field)) {
rc = HA_ERR_UNSUPPORTED;
SDB_PRINT_ERROR(rc,
"column '%-.192s' cannot be used in key specification.",
key_part->field->field_name);
goto error;
}
// TODO: ASC or DESC
key_obj_builder.append(key_part->field->field_name, 1);
}
key_obj = key_obj_builder.obj();
is_unique = key_info->flags & HA_NOSAME;
is_enforced = (0 == strcmp(key_info->name, primary_key_name));
rc = cl.create_index(key_obj, key_info->name, is_unique, is_enforced);
if (rc) {
goto error;
}
done:
return rc;
error:
goto done;
}
int sdb_get_idx_order(KEY *key_info, bson::BSONObj &order,
int order_direction) {
int rc = SDB_ERR_OK;
const KEY_PART_INFO *key_part;
const KEY_PART_INFO *key_end;
bson::BSONObjBuilder obj_builder;
if (!key_info) {
rc = SDB_ERR_INVALID_ARG;
goto error;
}
key_part = key_info->key_part;
key_end = key_part + key_info->user_defined_key_parts;
for (; key_part != key_end; ++key_part) {
obj_builder.append(key_part->field->field_name, order_direction);
}
order = obj_builder.obj();
done:
return rc;
error:
goto done;
}
static void get_int_key_obj(const uchar *key_ptr, const KEY_PART_INFO *key_part,
const char *op_str, bson::BSONObj &obj) {
bson::BSONObjBuilder obj_builder;
Field *field = key_part->field;
const uchar *new_ptr = key_ptr + key_part->store_length - key_part->length;
longlong value = field->val_int(new_ptr);
if (value < 0 && ((Field_num *)field)->unsigned_flag) {
// overflow UINT64, so store as DECIMAL
bson::bsonDecimal decimal_val;
char buf[24] = {0};
sprintf(buf, "%llu", (uint64)value);
decimal_val.fromString(buf);
obj_builder.append(op_str, decimal_val);
} else if (value > INT_MAX32 || value < INT_MIN32) {
// overflow INT32, so store as INT64
obj_builder.append(op_str, (long long)value);
} else {
obj_builder.append(op_str, (int)value);
}
obj = obj_builder.obj();
}
static void get_float_key_obj(const uchar *key_ptr,
const KEY_PART_INFO *key_part, const char *op_str,
bson::BSONObj &obj) {
bson::BSONObjBuilder obj_builder;
Field *field = key_part->field;
const uchar *new_ptr = key_ptr + key_part->store_length - key_part->length;
const uchar *old_ptr = field->ptr;
field->ptr = (uchar *)new_ptr;
double value = field->val_real();
field->ptr = (uchar *)old_ptr;
obj_builder.append(op_str, value);
obj = obj_builder.obj();
}
static void get_decimal_key_obj(const uchar *key_ptr,
const KEY_PART_INFO *key_part,
const char *op_str, bson::BSONObj &obj) {
bson::BSONObjBuilder obj_builder;
String str_val;
const uchar *new_ptr = key_ptr + key_part->store_length - key_part->length;
key_part->field->val_str(&str_val, new_ptr);
obj_builder.appendDecimal(op_str, str_val.c_ptr());
obj = obj_builder.obj();
}
static int get_text_key_obj(const uchar *key_ptr, const KEY_PART_INFO *key_part,
const char *op_str, bson::BSONObj &obj) {
int rc = SDB_ERR_OK;
bson::BSONObjBuilder obj_builder;
String *str = NULL;
String org_str;
String conv_str;
int key_start_pos = key_part->store_length - key_part->length;
int key_length = 0;
const uint16 NULL_BITS = 1;
/*if key's length is variable, remove spaces filled by mysql from the end of
varibale key string. otherwise remove from the end of store_length.*/
if (key_part->store_length - key_part->length > NULL_BITS) {
key_length = key_part->null_bit ? get_variable_key_length(&key_ptr[1])
: get_variable_key_length(&key_ptr[0]);
} else {
key_length = key_part->length;
}
org_str.set((const char *)(key_ptr + key_start_pos), key_length,
key_part->field->charset());
str = &org_str;
if (!my_charset_same(org_str.charset(), &SDB_CHARSET) &&
!my_charset_same(org_str.charset(), &my_charset_bin)) {
rc = sdb_convert_charset(org_str, conv_str, &SDB_CHARSET);
if (rc) {
goto error;
}
str = &conv_str;
}
if ((key_part->key_part_flag & HA_PART_KEY_SEG) && str->length() > 0 &&
0 == strcmp("$et", op_str)) {
op_str = "$gte";
}
// strip trailing space for some special collates
if (0 == strcmp("$gte", op_str)) {
str->strip_sp();
}
obj_builder.appendStrWithNoTerminating(op_str, (const char *)(str->ptr()),
str->length());
obj = obj_builder.obj();
done:
return rc;
error:
goto done;
}
static int get_char_key_obj(const uchar *key_ptr, const KEY_PART_INFO *key_part,
const char *op_str, bson::BSONObj &obj) {
int rc = SDB_ERR_OK;
bson::BSONObjBuilder obj_builder;
String str_val, conv_str;
String *str;
const uchar *new_ptr = key_ptr + key_part->store_length - key_part->length;
str = key_part->field->val_str(&str_val, new_ptr);
if (NULL == str) {
rc = SDB_ERR_INVALID_ARG;
goto error;
}
if (!my_charset_same(str->charset(), &my_charset_bin)) {
if (!my_charset_same(str->charset(), &SDB_CHARSET)) {
rc = sdb_convert_charset(*str, conv_str, &SDB_CHARSET);
if (rc) {
goto error;
}
str = &conv_str;
}
if (MYSQL_TYPE_STRING == key_part->field->type() ||
MYSQL_TYPE_VAR_STRING == key_part->field->type()) {
// Trailing space of CHAR/ENUM/SET condition should be stripped.
str->strip_sp();
}
}
if ((key_part->key_part_flag & HA_PART_KEY_SEG) && str->length() > 0 &&
0 == strcmp("$et", op_str)) {
op_str = "$gte";
}
obj_builder.appendStrWithNoTerminating(op_str, (const char *)(str->ptr()),
str->length());
obj = obj_builder.obj();
done:
return rc;
error:
goto done;
}
static void get_date_key_obj(const uchar *key_ptr,
const KEY_PART_INFO *key_part, const char *op_str,
bson::BSONObj &obj) {
bson::BSONObjBuilder obj_builder;
struct tm tm_val;
Field *field = key_part->field;
const uchar *new_ptr = key_ptr + key_part->store_length - key_part->length;
const uchar *old_ptr = field->ptr;
field->ptr = (uchar *)new_ptr;
longlong date_val = ((Field_newdate *)field)->val_int();
field->ptr = (uchar *)old_ptr;
tm_val.tm_sec = 0;
tm_val.tm_min = 0;
tm_val.tm_hour = 0;
tm_val.tm_mday = date_val % 100;
date_val = date_val / 100;
tm_val.tm_mon = date_val % 100 - 1;
date_val = date_val / 100;
tm_val.tm_year = date_val - 1900;
tm_val.tm_wday = 0;
tm_val.tm_yday = 0;
tm_val.tm_isdst = 0;
time_t time_tmp = mktime(&tm_val);
bson::Date_t dt((longlong)(time_tmp * 1000));
obj_builder.appendDate(op_str, dt);
obj = obj_builder.obj();
}
static void get_datetime_key_obj(const uchar *key_ptr,
const KEY_PART_INFO *key_part,
const char *op_str, bson::BSONObj &obj) {
bson::BSONObjBuilder obj_builder;
const uchar *new_ptr = key_ptr + key_part->store_length - key_part->length;
String org_str, str_val;
key_part->field->val_str(&org_str, new_ptr);
sdb_convert_charset(org_str, str_val, &SDB_CHARSET);
obj_builder.appendStrWithNoTerminating(op_str, str_val.ptr(),
str_val.length());
obj = obj_builder.obj();
}
static void get_timestamp_key_obj(const uchar *key_ptr,
const KEY_PART_INFO *key_part,
const char *op_str, bson::BSONObj &obj) {
bson::BSONObjBuilder obj_builder;
struct timeval tm;
int warnings = 0;
Field *field = key_part->field;
const uchar *new_ptr = key_ptr + key_part->store_length - key_part->length;
const uchar *old_ptr = field->ptr;
bool is_null = field->is_null();
if (is_null) {
field->set_notnull();
}
field->ptr = (uchar *)new_ptr;
field->get_timestamp(&tm, &warnings);
field->ptr = (uchar *)old_ptr;
if (is_null) {
field->set_null();
}
obj_builder.appendTimestamp(op_str, tm.tv_sec * 1000, tm.tv_usec);
obj = obj_builder.obj();
}
static int get_key_part_value(const KEY_PART_INFO *key_part,
const uchar *key_ptr, const char *op_str,
bool ignore_text_key, bson::BSONObj &obj) {
int rc = SDB_ERR_OK;
switch (key_part->field->type()) {
case MYSQL_TYPE_TINY:
case MYSQL_TYPE_SHORT:
case MYSQL_TYPE_INT24:
case MYSQL_TYPE_LONG:
case MYSQL_TYPE_LONGLONG:
case MYSQL_TYPE_BIT:
case MYSQL_TYPE_YEAR: {
get_int_key_obj(key_ptr, key_part, op_str, obj);
break;
}
case MYSQL_TYPE_FLOAT:
case MYSQL_TYPE_DOUBLE:
case MYSQL_TYPE_TIME: {
get_float_key_obj(key_ptr, key_part, op_str, obj);
break;
}
case MYSQL_TYPE_DECIMAL:
case MYSQL_TYPE_NEWDECIMAL: {
get_decimal_key_obj(key_ptr, key_part, op_str, obj);
break;
}
case MYSQL_TYPE_DATE: {
get_date_key_obj(key_ptr, key_part, op_str, obj);
break;
}
case MYSQL_TYPE_DATETIME: {
get_datetime_key_obj(key_ptr, key_part, op_str, obj);
break;
}
case MYSQL_TYPE_TIMESTAMP: {
get_timestamp_key_obj(key_ptr, key_part, op_str, obj);
break;
}
case MYSQL_TYPE_STRING:
case MYSQL_TYPE_VAR_STRING: {
if (!key_part->field->binary()) {
if (!ignore_text_key) {
rc = get_char_key_obj(key_ptr, key_part, op_str, obj);
if (rc) {
goto error;
}
}
} else {
// TODO: process the binary
rc = HA_ERR_UNSUPPORTED;
goto error;
}
break;
}
case MYSQL_TYPE_VARCHAR:
case MYSQL_TYPE_BLOB: {
if (!key_part->field->binary()) {
if (!ignore_text_key) {
rc = get_text_key_obj(key_ptr, key_part, op_str, obj);
if (rc) {
goto error;
}
}
} else {
// TODO: process the binary
rc = HA_ERR_UNSUPPORTED;
goto error;
}
break;
}
case MYSQL_TYPE_JSON:
default:
rc = HA_ERR_UNSUPPORTED;
goto error;
}
done:
return rc;
error:
goto done;
}
static inline int create_condition(Field *field, const KEY_PART_INFO *key_part,
const uchar *key_ptr, const char *op_str,
bool ignore_text_key,
bson::BSONArrayBuilder &builder) {
int rc = SDB_ERR_OK;
bson::BSONObj op_obj;
rc = get_key_part_value(key_part, key_ptr, op_str, ignore_text_key, op_obj);
if (SDB_ERR_OK == rc) {
if (!op_obj.isEmpty()) {
bson::BSONObj cond = BSON(field->field_name << op_obj);
builder.append(cond);
}
}
return rc;
}
// This function is modified from ha_federated::create_where_from_key.
int sdb_create_condition_from_key(TABLE *table, KEY *key_info,
const key_range *start_key,
const key_range *end_key,
bool from_records_in_range, bool eq_range_arg,
bson::BSONObj &condition) {
int rc = SDB_ERR_OK;
const uchar *key_ptr;
uint remainder, length;
const key_range *ranges[2] = {start_key, end_key};
my_bitmap_map *old_map;
bson::BSONArrayBuilder builder;
bson::BSONArray array;
if (start_key == NULL && end_key == NULL) {
return rc;
}
old_map = dbug_tmp_use_all_columns(table, table->read_set);
for (uint i = 0; i <= 1; i++) {
const KEY_PART_INFO *key_part;
bool ignore_text_key = false;
if (ranges[i] == NULL) {
continue;
}
// ignore end key of prefix index and like
if (i > 0 && HA_READ_BEFORE_KEY != ranges[i]->flag) {
ignore_text_key = true;
}
for (key_part = key_info->key_part,
remainder = key_info->user_defined_key_parts,
length = ranges[i]->length, key_ptr = ranges[i]->key;
; remainder--, key_part++) {
Field *field = key_part->field;
uint store_length = key_part->store_length;
if (key_part->null_bit) {
if (*key_ptr) {
/*
We got "IS [NOT] NULL" condition against nullable column. We
distinguish between "IS NOT NULL" and "IS NULL" by flag. For
"IS NULL", flag is set to HA_READ_KEY_EXACT.
*/
int is_null;
switch (ranges[i]->flag) {
case HA_READ_KEY_EXACT:
case HA_READ_BEFORE_KEY:
case HA_READ_KEY_OR_PREV:
case HA_READ_PREFIX_LAST:
case HA_READ_PREFIX_LAST_OR_PREV:
is_null = 1;
break;
case HA_READ_AFTER_KEY:
is_null = i > 0 ? 1 : 0;
break;
case HA_READ_KEY_OR_NEXT:
// >= null means read all records
default:
goto prepare_for_next_key_part;
}
bson::BSONObj is_null_obj = BSON("$isnull" << is_null);
bson::BSONObj is_null_cond = BSON(field->field_name << is_null_obj);
builder.append(is_null_cond);
/*
We need to adjust pointer and length to be prepared for next
key part. As well as check if this was last key part.
*/
goto prepare_for_next_key_part;
}
}
switch (ranges[i]->flag) {
case HA_READ_KEY_EXACT: {
DBUG_PRINT("info", ("sequoiadb HA_READ_KEY_EXACT %d", i));
const char *op_str = from_records_in_range ? "$gte" : "$et";
rc = create_condition(field, key_part, key_ptr, op_str,
ignore_text_key, builder);
if (0 != rc) {
goto error;
}
break;
}
case HA_READ_AFTER_KEY: {
if (eq_range_arg) {
break;
}
DBUG_PRINT("info", ("sequoiadb HA_READ_AFTER_KEY %d", i));
if ((store_length >= length) || (i > 0)) /* for all parts of end key*/
{
// end_key : start_key
const char *op_str = i > 0 ? "$lte" : "$gt";
rc = create_condition(field, key_part, key_ptr, op_str,
ignore_text_key, builder);
if (0 != rc) {
goto error;
}
break;
}
}
case HA_READ_KEY_OR_NEXT: {
DBUG_PRINT("info", ("sequoiadb HA_READ_KEY_OR_NEXT %d", i));
const char *op_str = "$gte";
rc = create_condition(field, key_part, key_ptr, op_str,
ignore_text_key, builder);
if (0 != rc) {
goto error;
}
break;
}
case HA_READ_BEFORE_KEY: {
DBUG_PRINT("info", ("sequoiadb HA_READ_BEFORE_KEY %d", i));
if (store_length >= length) {
const char *op_str = "$lt";
rc = create_condition(field, key_part, key_ptr, op_str,
ignore_text_key, builder);
if (0 != rc) {
goto error;
}
break;
}
}
case HA_READ_KEY_OR_PREV:
case HA_READ_PREFIX_LAST:
case HA_READ_PREFIX_LAST_OR_PREV: {
DBUG_PRINT("info", ("sequoiadb HA_READ_KEY_OR_PREV %d", i));
const char *op_str = "$lte";
rc = create_condition(field, key_part, key_ptr, op_str,
ignore_text_key, builder);
if (0 != rc) {
goto error;
}
break;
}
default:
DBUG_PRINT("info", ("cannot handle flag %d", ranges[i]->flag));
rc = HA_ERR_UNSUPPORTED;
goto error;
}
prepare_for_next_key_part:
if (store_length >= length) {
break;
}
DBUG_PRINT("info", ("remainder %d", remainder));
DBUG_ASSERT(remainder > 1);
length -= store_length;
key_ptr += store_length;
}
}
dbug_tmp_restore_column_map(table->read_set, old_map);
array = builder.arr();
if (array.nFields() > 1) {
condition = BSON("$and" << array);
} else if (!array.isEmpty()) {
condition = array.firstElement().embeddedObject().getOwned();
}
return rc;
error:
dbug_tmp_restore_column_map(table->read_set, old_map);
return rc;
}
my_bool sdb_is_same_index(const KEY *a, const KEY *b) {
my_bool rs = false;
const KEY_PART_INFO *key_part_a = NULL;
const KEY_PART_INFO *key_part_b = NULL;
const char *field_name_a = NULL;
const char *field_name_b = NULL;
if (strcmp(a->name, b->name) != 0 ||
a->user_defined_key_parts != b->user_defined_key_parts ||
(a->flags & HA_NOSAME) != (b->flags & HA_NOSAME)) {
goto done;
}
key_part_a = a->key_part;
key_part_b = b->key_part;
for (uint i = 0; i < a->user_defined_key_parts; ++i) {
field_name_a = key_part_a->field->field_name;
field_name_b = key_part_b->field->field_name;
if (strcmp(field_name_a, field_name_b) != 0) {
goto done;
}
++key_part_a;
++key_part_b;
}
rs = true;
done:
return rs;
}

Comment ( 0 )

Sign in for post a comment