diff --git a/src/box/errcode.h b/src/box/errcode.h
index d68148a66fbc1cd0d9aac8f4cc7dcf5255916216..c9bca47f8b15801e0e6d29e07c97139f1ba91fc2 100644
--- a/src/box/errcode.h
+++ b/src/box/errcode.h
@@ -76,7 +76,7 @@ struct errcode_record {
/* 21 */_(ER_PROC_RET, "msgpack.encode: can not encode Lua type '%s'") \
/* 22 */_(ER_TUPLE_NOT_ARRAY, "Tuple/Key must be MsgPack array") \
/* 23 */_(ER_FIELD_TYPE, "Tuple field %u type does not match one required by operation: expected %s") \
- /* 24 */_(ER_FIELD_TYPE_MISMATCH, "Ambiguous field type in index '%s', key part %u. Requested type is %s but the field has previously been defined as %s") \
+ /* 24 */_(ER_FIELD_TYPE_MISMATCH, "Ambiguous field type in index '%s', field %u. Requested type is %s but the field has previously been defined as %s") \
/* 25 */_(ER_SPLICE, "SPLICE error on field %u: %s") \
/* 26 */_(ER_UPDATE_ARG_TYPE, "Argument type in operation '%c' on field %u does not match field type: expected %s") \
/* 27 */_(ER_TUPLE_IS_TOO_LONG, "Tuple is too long %u") \
diff --git a/src/box/key_def.h b/src/box/key_def.h
index 7eb82ff8911bdcbbbc3fb03dd93020d7c13df81d..bebe0b5a2bcbbec29ae163c5e08468cff1228e95 100644
--- a/src/box/key_def.h
+++ b/src/box/key_def.h
@@ -246,8 +246,9 @@ struct key_def {
/** Index type. */
enum index_type type;
struct key_opts opts;
- /** comparators */
+ /** tuple <-> tuple comparison function */
tuple_compare_t tuple_compare;
+ /** tuple <-> key comparison function */
tuple_compare_with_key_t tuple_compare_with_key;
/** The size of the 'parts' array. */
uint32_t part_count;
@@ -454,6 +455,24 @@ int
key_validate_parts(struct key_def *key_def, const char *key,
uint32_t part_count);
+/**
+ * Return true if @a key_def defines a sequential key without
+ * holes starting from the first field. In other words, for all
+ * key parts key_def->parts[part_id].fieldno == part_id.
+ * @param key_def key_def
+ * @retval true key_def is sequential
+ * @retval false otherwise
+ */
+static inline bool
+key_def_is_sequential(const struct key_def *key_def)
+{
+ for (uint32_t part_id = 0; part_id < key_def->part_count; part_id++) {
+ if (key_def->parts[part_id].fieldno != part_id)
+ return false;
+ }
+ return true;
+}
+
/** A helper table for key_mp_type_validate */
extern const uint32_t key_mp_type[];
diff --git a/src/box/tuple_compare.cc b/src/box/tuple_compare.cc
index 9906381628687018bd21767cdcf6123422484525..905170e81bda48d151c256039ffd342ca06acaf2 100644
--- a/src/box/tuple_compare.cc
+++ b/src/box/tuple_compare.cc
@@ -253,22 +253,15 @@ tuple_compare_slowpath(const struct tuple *tuple_a, const struct tuple *tuple_b,
tuple_field_map(tuple_b), key_def);
}
-int
-key_compare(const char *key_a, uint32_t part_count_a,
- const char *key_b, uint32_t part_count_b,
- const struct key_def *key_def)
+static inline int
+key_compare_parts(const char *key_a, const char *key_b, uint32_t part_count,
+ const struct key_part *parts)
{
- assert(key_a != NULL || part_count_a == 0);
- assert(key_b != NULL || part_count_b == 0);
- assert(part_count_a <= key_def->part_count);
- assert(part_count_b <= key_def->part_count);
- const struct key_part *part = key_def->parts;
- uint32_t part_count;
- part_count = MIN(MIN(part_count_a, key_def->part_count), part_count_b);
+ assert((key_a != NULL && key_b != NULL) || part_count == 0);
+ const struct key_part *part = parts;
if (likely(part_count == 1))
return tuple_compare_field(key_a, key_b, part->type);
- const struct key_part *end;
- end = part + part_count;
+ const struct key_part *end = part + part_count;
int r = 0; /* Part count can be 0 in wildcard searches. */
for (; part < end; part++) {
r = tuple_compare_field(key_a, key_b, part->type);
@@ -280,6 +273,50 @@ key_compare(const char *key_a, uint32_t part_count_a,
return r;
}
+int
+key_compare(const char *key_a, uint32_t part_count_a,
+ const char *key_b, uint32_t part_count_b,
+ const struct key_def *key_def)
+{
+ assert(part_count_a <= key_def->part_count);
+ assert(part_count_b <= key_def->part_count);
+ uint32_t part_count = MIN(part_count_a, part_count_b);
+ assert(part_count <= key_def->part_count);
+ return key_compare_parts(key_a, key_b, part_count, key_def->parts);
+}
+
+static int
+tuple_compare_with_key_sequential(const struct tuple *tuple, const char *key,
+ uint32_t part_count,
+ const struct key_def *key_def)
+{
+ assert(key_def_is_sequential(key_def));
+ const char *tuple_key = tuple_data(tuple);
+ uint32_t tuple_field_count = mp_decode_array(&tuple_key);
+ assert(tuple_field_count >= key_def->part_count);
+ assert(part_count <= key_def->part_count);
+ (void) tuple_field_count;
+ return key_compare_parts(tuple_key, key, part_count, key_def->parts);
+}
+
+static int
+tuple_compare_sequential(const struct tuple *tuple_a,
+ const struct tuple *tuple_b,
+ const struct key_def *key_def)
+{
+ assert(key_def_is_sequential(key_def));
+ const char *key_a = tuple_data(tuple_a);
+ uint32_t field_count_a = mp_decode_array(&key_a);
+ assert(field_count_a >= key_def->part_count);
+ (void) field_count_a;
+ const char *key_b = tuple_data(tuple_b);
+ uint32_t field_count_b = mp_decode_array(&key_b);
+ assert(field_count_b >= key_def->part_count);
+ (void) field_count_b;
+ return key_compare_parts(key_a, key_b, key_def->part_count,
+ key_def->parts);
+}
+
static inline int
tuple_compare_with_key_slowpath_raw(const struct tuple_format *format,
const char *tuple, const uint32_t *field_map,
@@ -511,6 +548,8 @@ tuple_compare_create(const struct key_def *def) {
if (i == def->part_count && cmp_arr[k].p[i * 2] == UINT32_MAX)
return cmp_arr[k].f;
}
+ if (key_def_is_sequential(def))
+ return tuple_compare_sequential;
return tuple_compare_slowpath;
}
@@ -702,16 +741,16 @@ tuple_compare_with_key_create(const struct key_def *def)
uint32_t i = 0;
for (; i < def->part_count; i++) {
-
if (def->parts[i].fieldno != cmp_wk_arr[k].p[i * 2] ||
def->parts[i].type != cmp_wk_arr[k].p[i * 2 + 1]) {
-
break;
}
}
if (i == def->part_count)
return cmp_wk_arr[k].f;
}
+ if (key_def_is_sequential(def))
+ return tuple_compare_with_key_sequential;
return tuple_compare_with_key_slowpath;
}
diff --git a/src/box/tuple_compare.h b/src/box/tuple_compare.h
index 7c382ce67718ad72c24900d99d67197a93a4aab3..d146347693391bf1a152352d641e059404aed306 100644
--- a/src/box/tuple_compare.h
+++ b/src/box/tuple_compare.h
@@ -32,6 +32,7 @@
*/
#include <stddef.h>
#include <stdint.h>
+#include <stdbool.h>
#include "key_def.h"
diff --git a/src/box/tuple_format.c b/src/box/tuple_format.c
index 2bcdb5d617c3b3f611d19d68c538c8cf744ae138..dfec11c035b9b813f293175e24b41f8eea88f791 100644
--- a/src/box/tuple_format.c
+++ b/src/box/tuple_format.c
@@ -42,29 +42,54 @@ static uint32_t formats_size = 0, formats_capacity = 0;
static int
field_type_create(struct tuple_format *format, struct rlist *key_list)
{
+ if (format->field_count == 0)
+ return 0;
+
/* There may be fields between indexed fields (gaps). */
- for (uint32_t i = 0; i < format->field_count; i++)
+ for (uint32_t i = 0; i < format->field_count; i++) {
format->fields[i].type = FIELD_TYPE_ANY;
+ format->fields[i].offset_slot = TUPLE_OFFSET_SLOT_MISSING;
+ }
+
+ int current_slot = 0;
struct key_def *key_def;
/* extract field type info */
rlist_foreach_entry(key_def, key_list, link) {
- for (uint32_t i = 0; i < key_def->part_count; i++) {
- assert(key_def->parts[i].fieldno < format->field_count);
- enum field_type set_type = key_def->parts[i].type;
- enum field_type *fmt_type =
- &format->fields[key_def->parts[i].fieldno].type;
- if (*fmt_type != FIELD_TYPE_ANY &&
- *fmt_type != set_type) {
+ bool is_sequential = key_def_is_sequential(key_def);
+ const struct key_part *part = key_def->parts;
+ const struct key_part *parts_end = part + key_def->part_count;
+ for (; part < parts_end; part++) {
+ assert(part->fieldno < format->field_count);
+ struct tuple_field_format *field =
+ &format->fields[part->fieldno];
+ if (field->type != FIELD_TYPE_ANY &&
+ field->type != part->type) {
diag_set(ClientError, ER_FIELD_TYPE_MISMATCH,
- key_def->name, i + TUPLE_INDEX_BASE,
- field_type_strs[set_type],
- field_type_strs[*fmt_type]);
+ key_def->name,
+ part->fieldno + TUPLE_INDEX_BASE,
+ field_type_strs[part->type],
+ field_type_strs[field->type]);
return -1;
}
- *fmt_type = set_type;
+ field->type = part->type;
+
+ /*
+ * In the tuple, store only offsets necessary
+ * to access fields of non-sequential keys.
+ * First field is always simply accessible,
+ * so we don't store offset for it.
+ */
+ if (field->offset_slot == TUPLE_OFFSET_SLOT_MISSING &&
+ !is_sequential && part->fieldno > 0) {
+ field->offset_slot = --current_slot;
+ }
}
}
+
+ assert(format->fields[0].offset_slot == TUPLE_OFFSET_SLOT_MISSING);
+ assert((uint32_t) (-current_slot * sizeof(uint32_t)) <= UINT16_MAX);
+ format->field_map_size = -current_slot * sizeof(uint32_t);
return 0;
}
@@ -170,31 +195,6 @@ tuple_format_new(struct rlist *key_list, struct tuple_format_vtab *vtab)
tuple_format_delete(format);
return NULL;
}
- /* Set up offset slots */
- if (format->field_count == 0) {
- /* Nothing to store */
- format->field_map_size = 0;
- return format;
- }
- /**
- * First field is always simply accessible,
- * so we don't store offset for it
- */
- format->fields[0].offset_slot = INT32_MAX;
-
- int current_slot = 0;
- for (uint32_t i = 1; i < format->field_count; i++) {
- /*
- * In the tuple, store only offsets necessary to
- * quickly access indexed fields.
- */
- if (format->fields[i].type == FIELD_TYPE_ANY)
- format->fields[i].offset_slot = INT32_MAX;
- else
- format->fields[i].offset_slot = --current_slot;
- }
- assert((uint32_t) (-current_slot * sizeof(uint32_t)) <= UINT16_MAX);
- format->field_map_size = -current_slot * sizeof(uint32_t);
return format;
}
@@ -236,7 +236,7 @@ tuple_init_field_map(const struct tuple_format *format, uint32_t *field_map,
if (key_mp_type_validate(format->fields[i].type, mp_type,
ER_FIELD_TYPE, i + TUPLE_INDEX_BASE))
return -1;
- if (format->fields[i].offset_slot < 0)
+ if (format->fields[i].offset_slot != TUPLE_OFFSET_SLOT_MISSING)
field_map[format->fields[i].offset_slot] =
(uint32_t) (pos - tuple);
mp_next(&pos);
diff --git a/src/box/tuple_format.h b/src/box/tuple_format.h
index ee2a17842a6d91fce6f06b445f86aa3ce4eb77ed..58e73b44a25f29d8e552fad126c9a1e4cbad8b24 100644
--- a/src/box/tuple_format.h
+++ b/src/box/tuple_format.h
@@ -60,6 +60,12 @@ enum { FORMAT_REF_MAX = INT32_MAX};
*/
enum { TUPLE_INDEX_BASE = 1 };
+/*
+ * A special value to indicate that tuple format doesn't store
+ * an offset for a field_id.
+ */
+enum { TUPLE_OFFSET_SLOT_MISSING = INT32_MAX };
+
/**
* @brief Tuple field format
* Support structure for struct tuple_format.
@@ -226,10 +232,9 @@ tuple_field_raw(const struct tuple_format *format, const char *tuple,
return tuple;
}
- if (format->fields[field_no].offset_slot != INT32_MAX) {
- int32_t slot = format->fields[field_no].offset_slot;
- return tuple + field_map[slot];
- }
+ int32_t offset_slot = format->fields[field_no].offset_slot;
+ if (offset_slot != TUPLE_OFFSET_SLOT_MISSING)
+ return tuple + field_map[offset_slot];
}
ERROR_INJECT(ERRINJ_TUPLE_FIELD, return NULL);
uint32_t field_count = mp_decode_array(&tuple);
diff --git a/test/box/alter_limits.result b/test/box/alter_limits.result
index 68c3b8b2035d81cdc9a091829305abc4e83eebac..e928fee2497d960413d32c1ad27dec0ab28db89e 100644
--- a/test/box/alter_limits.result
+++ b/test/box/alter_limits.result
@@ -526,8 +526,8 @@ index = s:create_index('t1', { type = 'hash' })
-- field type contradicts field type of another index
index = s:create_index('t2', { type = 'hash', parts = { 1, 'string' }})
---
-- error: Ambiguous field type in index 't2', key part 1. Requested type is string
- but the field has previously been defined as unsigned
+- error: Ambiguous field type in index 't2', field 1. Requested type is string but
+ the field has previously been defined as unsigned
...
-- ok
index = s:create_index('t2', { type = 'hash', parts = { 2, 'string' }})
@@ -800,7 +800,7 @@ s.index.primary:select{}
-- ambiguous field type
index = s:create_index('string', { type = 'tree', unique = false, parts = { 2, 'string'}})
---
-- error: Ambiguous field type in index 'string', key part 1. Requested type is string
+- error: Ambiguous field type in index 'string', field 2. Requested type is string
but the field has previously been defined as unsigned
...
-- create index on a non-existing field
diff --git a/test/box/errinj.result b/test/box/errinj.result
index 655baec98bd80604cedba65c58ddcc217d54a019..01f762841e6ae250d50fe7fba94364becf51cba4 100644
--- a/test/box/errinj.result
+++ b/test/box/errinj.result
@@ -500,6 +500,90 @@ tostring(t[1]) .. tostring(t[2]) ..tostring(t[3]) .. tostring(t[4]) .. tostring(
s:drop()
---
...
+--
+-- gh-2046: don't store offsets for sequential multi-parts keys
+--
+s = box.schema.space.create('test')
+---
+...
+_ = s:create_index('seq2', { parts = { 1, 'unsigned', 2, 'unsigned' }})
+---
+...
+_ = s:create_index('seq3', { parts = { 1, 'unsigned', 2, 'unsigned', 3, 'unsigned' }})
+---
+...
+_ = s:create_index('seq5', { parts = { 1, 'unsigned', 2, 'unsigned', 3, 'unsigned', 4, 'scalar', 5, 'number' }})
+---
+...
+_ = s:create_index('rnd1', { parts = { 3, 'unsigned' }})
+---
+...
+errinj.set("ERRINJ_TUPLE_FIELD", true)
+---
+- ok
+...
+tuple = s:insert({1, 2, 3, 4, 5, 6, 7, 8, 9, 10})
+---
+...
+tuple
+---
+- [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
+...
+tuple[1] -- not-null, always accessible
+---
+- 1
+...
+tuple[2] -- null, doesn't have offset
+---
+- null
+...
+tuple[3] -- not null, has offset
+---
+- 3
+...
+tuple[4] -- null, doesn't have offset
+---
+- null
+...
+tuple[5] -- null, doesn't have offset
+---
+- null
+...
+s.index.seq2:select({1})
+---
+- - [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
+...
+s.index.seq2:select({1, 2})
+---
+- - [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
+...
+s.index.seq3:select({1})
+---
+- - [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
+...
+s.index.seq3:select({1, 2, 3})
+---
+- - [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
+...
+s.index.seq5:select({1})
+---
+- - [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
+...
+s.index.seq5:select({1, 2, 3, 4, 5})
+---
+- - [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
+...
+s.index.rnd1:select({3})
+---
+- - [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
+...
+errinj.set("ERRINJ_TUPLE_FIELD", false)
+---
+- ok
+...
+s:drop()
+---
+...
space = box.schema.space.create('test')
---
...
diff --git a/test/box/errinj.test.lua b/test/box/errinj.test.lua
index 3e71875d45f971ed1137ae2018059ed9abee9fb4..aafe5f822994b4a325fc16afb2affde3fc623f11 100644
--- a/test/box/errinj.test.lua
+++ b/test/box/errinj.test.lua
@@ -154,6 +154,33 @@ tostring(t[1]) .. tostring(t[2]) ..tostring(t[3]) .. tostring(t[4]) .. tostring(
-- Cleanup
s:drop()
+--
+-- gh-2046: don't store offsets for sequential multi-parts keys
+--
+s = box.schema.space.create('test')
+_ = s:create_index('seq2', { parts = { 1, 'unsigned', 2, 'unsigned' }})
+_ = s:create_index('seq3', { parts = { 1, 'unsigned', 2, 'unsigned', 3, 'unsigned' }})
+_ = s:create_index('seq5', { parts = { 1, 'unsigned', 2, 'unsigned', 3, 'unsigned', 4, 'scalar', 5, 'number' }})
+_ = s:create_index('rnd1', { parts = { 3, 'unsigned' }})
+
+errinj.set("ERRINJ_TUPLE_FIELD", true)
+tuple = s:insert({1, 2, 3, 4, 5, 6, 7, 8, 9, 10})
+tuple
+tuple[1] -- not-null, always accessible
+tuple[2] -- null, doesn't have offset
+tuple[3] -- not null, has offset
+tuple[4] -- null, doesn't have offset
+tuple[5] -- null, doesn't have offset
+s.index.seq2:select({1})
+s.index.seq2:select({1, 2})
+s.index.seq3:select({1})
+s.index.seq3:select({1, 2, 3})
+s.index.seq5:select({1})
+s.index.seq5:select({1, 2, 3, 4, 5})
+s.index.rnd1:select({3})
+errinj.set("ERRINJ_TUPLE_FIELD", false)
+s:drop()
+
space = box.schema.space.create('test')
_ = space:create_index('pk')
errinj.set("ERRINJ_WAL_WRITE", true)