From 6bcd51f9236a9f5718be8c422af0cba63dfd611b Mon Sep 17 00:00:00 2001
From: Ilya Verbin <iverbin@tarantool.org>
Date: Wed, 17 May 2023 20:06:35 +0300
Subject: [PATCH] box: fix unique violation in functional index with nullable
parts
Currently is_nullable property of a functional index part disables the
unique property of the index. The bug is in func_index_compare(), which
compares functional keys first, and if they are equal it compares the
primary keys. This behaviour is correct only when some part of the key
is NULL (and for non-unique indexes), but for now the primary keys are
compared unconditionally. Fix this by checking for NULL key parts.
Closes #8587
NO_DOC=bugfix
---
...ation-in-functional-index-with-nullable.md | 4 +
src/box/tuple_compare.cc | 66 +++++++---
..._8587_func_index_unique_violation_test.lua | 106 ++++++++++++++++
test/unit/key_def.c | 113 +++++++++++++++++-
4 files changed, 270 insertions(+), 19 deletions(-)
create mode 100644 changelogs/unreleased/gh-8587-unique-violation-in-functional-index-with-nullable.md
create mode 100644 test/box-luatest/gh_8587_func_index_unique_violation_test.lua
diff --git a/changelogs/unreleased/gh-8587-unique-violation-in-functional-index-with-nullable.md b/changelogs/unreleased/gh-8587-unique-violation-in-functional-index-with-nullable.md
new file mode 100644
index 0000000000..7b7c8e20b5
--- /dev/null
+++ b/changelogs/unreleased/gh-8587-unique-violation-in-functional-index-with-nullable.md
@@ -0,0 +1,4 @@
+## bugfix/core
+
+* Fixed a bug when a tuple could be inserted even if it violates a `unique`
+ constraint of a functional index that has a nullable part (gh-8587).
diff --git a/src/box/tuple_compare.cc b/src/box/tuple_compare.cc
index 6bf9310fe0..fc34797a42 100644
--- a/src/box/tuple_compare.cc
+++ b/src/box/tuple_compare.cc
@@ -674,7 +674,7 @@ tuple_compare_slowpath(struct tuple *tuple_a, hint_t tuple_a_hint,
if (!is_nullable || !was_null_met)
return 0;
/*
- * Inxex parts are equal and contain NULLs. So use
+ * Index parts are equal and contain NULLs. So use
* extended parts only.
*/
end = key_def->parts + key_def->part_count;
@@ -819,17 +819,21 @@ tuple_compare_with_key_slowpath(struct tuple *tuple, hint_t tuple_hint,
* will point to the first field that differ or to the end of key or
* part_count + 1 field in order.
* Key arguments must not be NULL, allowed to be NULL if dereferenced.
+ * Set was_null_met to true if at least one part is NULL, to false otherwise.
*/
template<bool is_nullable>
static inline int
key_compare_and_skip_parts(const char **key_a, const char **key_b,
- uint32_t part_count, struct key_def *key_def)
+ uint32_t part_count, struct key_def *key_def,
+ bool *was_null_met)
{
assert(is_nullable == key_def->is_nullable);
assert(key_a != NULL && key_b != NULL);
assert((*key_a != NULL && *key_b != NULL) || part_count == 0);
struct key_part *part = key_def->parts;
int rc;
+ *was_null_met = false;
+
if (likely(part_count == 1)) {
enum mp_type a_type = mp_typeof(**key_a);
enum mp_type b_type = mp_typeof(**key_b);
@@ -838,8 +842,10 @@ key_compare_and_skip_parts(const char **key_a, const char **key_b,
part->coll);
} else if (a_type == MP_NIL) {
rc = b_type == MP_NIL ? 0 : -1;
+ *was_null_met = true;
} else if (b_type == MP_NIL) {
rc = 1;
+ *was_null_met = true;
} else {
rc = tuple_compare_field_with_type(*key_a, a_type,
*key_b, b_type,
@@ -867,9 +873,11 @@ key_compare_and_skip_parts(const char **key_a, const char **key_b,
enum mp_type a_type = mp_typeof(**key_a);
enum mp_type b_type = mp_typeof(**key_b);
if (a_type == MP_NIL) {
+ *was_null_met = true;
if (b_type != MP_NIL)
return -1;
} else if (b_type == MP_NIL) {
+ *was_null_met = true;
return 1;
} else {
rc = tuple_compare_field_with_type(*key_a, a_type,
@@ -886,10 +894,11 @@ key_compare_and_skip_parts(const char **key_a, const char **key_b,
template<bool is_nullable>
static inline int
key_compare_parts(const char *key_a, const char *key_b, uint32_t part_count,
- struct key_def *key_def)
+ struct key_def *key_def, bool *was_null_met)
{
return key_compare_and_skip_parts<is_nullable>(&key_a, &key_b,
- part_count, key_def);
+ part_count, key_def,
+ was_null_met);
}
template<bool is_nullable, bool has_optional_parts>
@@ -914,8 +923,9 @@ tuple_compare_with_key_sequential(struct tuple *tuple, hint_t tuple_hint,
assert(field_count >= part_count);
cmp_part_count = part_count;
}
+ bool unused;
rc = key_compare_parts<is_nullable>(tuple_key, key, cmp_part_count,
- key_def);
+ key_def, &unused);
if (!has_optional_parts || rc != 0)
return rc;
/*
@@ -949,12 +959,13 @@ key_compare(const char *key_a, uint32_t part_count_a, hint_t key_a_hint,
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);
+ bool unused;
if (! key_def->is_nullable) {
return key_compare_parts<false>(key_a, key_b, part_count,
- key_def);
+ key_def, &unused);
} else {
return key_compare_parts<true>(key_a, key_b, part_count,
- key_def);
+ key_def, &unused);
}
}
@@ -978,8 +989,10 @@ tuple_compare_sequential(struct tuple *tuple_a, hint_t tuple_a_hint,
if (!has_optional_parts && !is_nullable) {
assert(fc_a >= key_def->part_count);
assert(fc_b >= key_def->part_count);
+ bool unused;
return key_compare_parts<false>(key_a, key_b,
- key_def->part_count, key_def);
+ key_def->part_count, key_def,
+ &unused);
}
bool was_null_met = false;
struct key_part *part = key_def->parts;
@@ -1400,13 +1413,17 @@ static const comparator_with_key_signature cmp_wk_arr[] = {
/**
* A functional index tuple compare.
- * tuple_a_hint and tuple_b_hint are expected to be valid
- * pointers to functional key memory. These keys have been already
- * validated and are represented as a MsgPack array with exactly
- * func_index_part_count parts. The cmp_def has part_count > func_index_part_count,
- * since it was produced by key_def_merge() of the functional key part
- * and the primary key. So its tail parts are taken from primary
- * index key definition.
+ * tuple_a_hint and tuple_b_hint are expected to be valid pointers to functional
+ * key memory. These keys are represented as a MsgPack array with the number of
+ * elements equal to the functional index definition part_count (let's denote
+ * it as func_index_part_count). The keys have been already validated by
+ * key_list_iterator_next().
+ *
+ * In case of unique non-nullable index, the non-extended key_def is used as a
+ * cmp_def (see memtx_tree_index_new_tpl()). Otherwise, the extended cmp_def has
+ * part_count > func_index_part_count, since it was produced by key_def_merge()
+ * of the functional key part and the primary key. So its tail parts are taken
+ * from primary index key definition.
*/
template<bool is_nullable>
static inline int
@@ -1423,12 +1440,24 @@ func_index_compare(struct tuple *tuple_a, hint_t tuple_a_hint,
uint32_t part_count_a = mp_decode_array(&key_a);
assert(mp_typeof(*key_b) == MP_ARRAY);
uint32_t part_count_b = mp_decode_array(&key_b);
+ assert(part_count_a == part_count_b);
+ uint32_t key_part_count = part_count_a;
+ (void)part_count_b;
- uint32_t key_part_count = MIN(part_count_a, part_count_b);
+ bool was_null_met;
int rc = key_compare_parts<is_nullable>(key_a, key_b, key_part_count,
- cmp_def);
+ cmp_def, &was_null_met);
if (rc != 0)
return rc;
+ /*
+ * Tuples with nullified fields may violate the uniqueness constraint
+ * so if we encountered a nullified field while comparing the secondary
+ * key parts we must proceed with comparing the primary key parts even
+ * if the secondary index is unique.
+ */
+ if (key_part_count == cmp_def->unique_part_count &&
+ (!is_nullable || !was_null_met))
+ return 0;
/*
* Primary index definition key compare.
* It cannot contain nullable parts so the code is
@@ -1481,9 +1510,10 @@ func_index_compare_with_key(struct tuple *tuple, hint_t tuple_hint,
uint32_t tuple_key_count = mp_decode_array(&tuple_key);
uint32_t cmp_part_count = MIN(part_count, tuple_key_count);
cmp_part_count = MIN(cmp_part_count, key_def->part_count);
+ bool unused;
int rc = key_compare_and_skip_parts<is_nullable>(&tuple_key, &key,
cmp_part_count,
- key_def);
+ key_def, &unused);
if (rc != 0)
return rc;
/* Equals if nothing to compare. */
diff --git a/test/box-luatest/gh_8587_func_index_unique_violation_test.lua b/test/box-luatest/gh_8587_func_index_unique_violation_test.lua
new file mode 100644
index 0000000000..2ef8fcefed
--- /dev/null
+++ b/test/box-luatest/gh_8587_func_index_unique_violation_test.lua
@@ -0,0 +1,106 @@
+local t = require('luatest')
+
+local g = t.group('gh-8587', t.helpers.matrix{
+ is_unique = {true, false},
+ is_nullable = {true, false}})
+g.before_all(function(cg)
+ local server = require('luatest.server')
+ cg.server = server:new{alias = 'master'}
+ cg.server:start()
+end)
+
+g.after_all(function(cg)
+ cg.server:drop()
+end)
+
+g.after_each(function(cg)
+ cg.server:exec(function()
+ if box.space.test then box.space.test:drop() end
+ end)
+end)
+
+-- Test a simple functional index with all combinations of
+-- unique = {true, false} and is_nullable = {true, false}.
+g.test_simple = function(cg)
+ cg.server:exec(function(is_unique, is_nullable)
+ local opts = {format = {{name = 'id', type = 'unsigned'},
+ {name = 'f2', type = 'string',
+ is_nullable = is_nullable}}}
+ local s = box.schema.space.create('test', opts)
+ s:create_index('pk')
+
+ opts = {is_sandboxed = true,
+ is_deterministic = true,
+ body = "function(tuple) return {tuple[2]} end"}
+ box.schema.func.create('func_ret_f2', opts)
+
+ opts = {func = 'func_ret_f2',
+ unique = is_unique,
+ parts = {{field = 1, type = 'string',
+ is_nullable = is_nullable}}}
+ local idx = s:create_index('idx', opts)
+
+ if is_nullable then
+ s:insert{1, box.NULL}
+ s:insert{2, box.NULL}
+ t.assert_equals(idx:select(), {{1}, {2}})
+ end
+
+ s:insert{3, 'aa'}
+ if is_unique then
+ t.assert_error_msg_content_equals(
+ 'Duplicate key exists in unique index "idx" in space "test" ' ..
+ 'with old tuple - [3, "aa"] and new tuple - [4, "aa"]',
+ function() s:insert{4, 'aa'} end)
+ t.assert_equals(idx:select{'aa'}, {{3, 'aa'}})
+ else
+ s:insert{4, 'aa'}
+ t.assert_equals(idx:select{'aa'}, {{3, 'aa'}, {4, 'aa'}})
+ end
+ end, {cg.params.is_unique, cg.params.is_nullable})
+end
+
+-- Test a multikey multipart functional index.
+g.test_multikey_multipart = function(cg)
+ cg.server:exec(function(is_unique, is_nullable)
+ local s = box.schema.space.create('test')
+ s:format{{name = 'name', type = 'string'},
+ {name = 'address', type = 'string'},
+ {name = 'address2', type = 'string'}}
+ s:create_index('name', {parts = {{field = 1, type = 'string'}}})
+ local lua_code = [[function(tuple)
+ local address = string.split(tuple[2])
+ local ret = {}
+ for _, v in pairs(address) do
+ table.insert(ret, {utf8.upper(v), tuple[3]})
+ end
+ return ret
+ end]]
+ box.schema.func.create('address', {body = lua_code,
+ is_sandboxed = true,
+ is_deterministic = true,
+ opts = {is_multikey = true}})
+ s:create_index('addr', {unique = is_unique,
+ func = 'address',
+ parts = {{field = 1, type = 'string',
+ is_nullable = is_nullable},
+ {field = 2, type = 'string',
+ is_nullable = is_nullable}}})
+ s:insert{'James', 'SIS Building Lambeth London UK', '1'}
+ s:insert{'Sherlock', '221B Baker St Marylebone London NW1 6XE UK', '2'}
+ if is_unique then
+ t.assert_error_msg_content_equals(
+ 'Duplicate key exists in unique index "addr" in space "test"' ..
+ ' with old tuple - ["Sherlock", "221B Baker St Marylebone ' ..
+ 'London NW1 6XE UK", "2"] and new tuple - ["Sherlock2", "' ..
+ '221B Baker St Marylebone London NW1 6XE UK", "2"]',
+ function()
+ s:insert{'Sherlock2',
+ '221B Baker St Marylebone London NW1 6XE UK', '2'}
+ end)
+ else
+ s:insert{'Sherlock2',
+ '221B Baker St Marylebone London NW1 6XE UK', '2'}
+ end
+ end, {cg.params.is_unique, cg.params.is_nullable})
+end
diff --git a/test/unit/key_def.c b/test/unit/key_def.c
index b2cf0de1c2..292cf6e240 100644
--- a/test/unit/key_def.c
+++ b/test/unit/key_def.c
@@ -134,6 +134,114 @@ test_key_def_new_func(const char *format, ...)
return def;
}
+/**
+ * Checks that tuple_compare() -> func_index_compare() return value equals
+ * `expected`.
+ */
+static void
+test_check_tuple_compare_func(struct key_def *cmp_def,
+ struct tuple *tuple_a, struct tuple *func_key_a,
+ struct tuple *tuple_b, struct tuple *func_key_b,
+ int expected)
+{
+ int r = tuple_compare(tuple_a, (hint_t)func_key_a,
+ tuple_b, (hint_t)func_key_b, cmp_def);
+ r = r > 0 ? 1 : r < 0 ? -1 : 0;
+ is(r, expected, "func_index_compare(%s/%s, %s/%s) = %d, expected %d",
+ tuple_str(tuple_a), tuple_str(func_key_a),
+ tuple_str(tuple_b), tuple_str(func_key_b), r, expected);
+}
+
+static void
+test_func_compare(void)
+{
+ plan(6);
+ header();
+
+ struct key_def *func_def = test_key_def_new_func(
+ "[{%s%u%s%s%s%b}{%s%u%s%s%s%b}]",
+ "field", 0, "type", "string", "is_nullable", 1,
+ "field", 1, "type", "string", "is_nullable", 1);
+
+ struct key_def *pk_def = test_key_def_new(
+ "[{%s%u%s%s}]",
+ "field", 1, "type", "unsigned");
+
+ struct key_def *cmp_def = key_def_merge(func_def, pk_def);
+ /* Just like when `opts->is_unique == true`, see index_def_new(). */
+ cmp_def->unique_part_count = func_def->part_count;
+
+ struct testcase {
+ int expected_result;
+ struct tuple *tuple_a;
+ struct tuple *tuple_b;
+ struct tuple *func_key_a;
+ struct tuple *func_key_b;
+ };
+
+ struct testcase testcases[] = {
+ {
+ -1, /* func_key_a < func_key_b */
+ test_tuple_new("[%s%u%s]", "--", 0, "--"),
+ test_tuple_new("[%s%u%s]", "--", 0, "--"),
+ test_tuple_new("[%sNIL]", "aa"),
+ test_tuple_new("[%s%s]", "aa", "bb"),
+ }, {
+ 1, /* func_key_a > func_key_b */
+ test_tuple_new("[%s%u%s]", "--", 0, "--"),
+ test_tuple_new("[%s%u%s]", "--", 0, "--"),
+ test_tuple_new("[%s%s]", "aa", "bb"),
+ test_tuple_new("[%sNIL]", "aa"),
+ }, {
+ 0, /* func_key_a == func_key_b, pk not compared */
+ test_tuple_new("[%s%u%s]", "--", 10, "--"),
+ test_tuple_new("[%s%u%s]", "--", 20, "--"),
+ test_tuple_new("[%s%s]", "aa", "bb"),
+ test_tuple_new("[%s%s]", "aa", "bb"),
+ }, {
+ -1, /* func_key_a == func_key_b, pk_a < pk_b */
+ test_tuple_new("[%s%u%s]", "--", 30, "--"),
+ test_tuple_new("[%s%u%s]", "--", 40, "--"),
+ test_tuple_new("[%sNIL]", "aa"),
+ test_tuple_new("[%sNIL]", "aa"),
+ }, {
+ 1, /* func_key_a == func_key_b, pk_a > pk_b */
+ test_tuple_new("[%s%u%s]", "--", 60, "--"),
+ test_tuple_new("[%s%u%s]", "--", 50, "--"),
+ test_tuple_new("[%sNIL]", "aa"),
+ test_tuple_new("[%sNIL]", "aa"),
+ }, {
+ 0, /* func_key_a == func_key_b, pk_a == pk_b */
+ test_tuple_new("[%s%u%s]", "--", 70, "--"),
+ test_tuple_new("[%s%u%s]", "--", 70, "--"),
+ test_tuple_new("[%sNIL]", "aa"),
+ test_tuple_new("[%sNIL]", "aa"),
+ }
+ };
+
+ for (size_t i = 0; i < lengthof(testcases); i++) {
+ struct testcase *t = &testcases[i];
+ test_check_tuple_compare_func(cmp_def,
+ t->tuple_a, t->func_key_a,
+ t->tuple_b, t->func_key_b,
+ t->expected_result);
+ }
+
+ for (size_t i = 0; i < lengthof(testcases); i++) {
+ struct testcase *t = &testcases[i];
+ tuple_delete(t->tuple_a);
+ tuple_delete(t->tuple_b);
+ tuple_delete(t->func_key_a);
+ tuple_delete(t->func_key_b);
+ }
+ key_def_delete(func_def);
+ key_def_delete(pk_def);
+ key_def_delete(cmp_def);
+
+ footer();
+ check_plan();
+}
+
/**
* Checks that tuple_compare_with_key with cmp_def of functional index
* returns the same result as comparison of concatenated func and primary keys.
@@ -217,6 +325,8 @@ test_func_compare_with_key(void)
for (unsigned int i = 0; i < lengthof(keys); i++)
free(keys[i]);
tuple_delete(func_key);
+ tuple_delete(tuple);
+ tuple_delete(model);
key_def_delete(def);
key_def_delete(pk_def);
key_def_delete(cmp_def);
@@ -350,9 +460,10 @@ test_tuple_validate_key_parts_raw(void)
static int
test_main(void)
{
- plan(3);
+ plan(4);
header();
+ test_func_compare();
test_func_compare_with_key();
test_tuple_extract_key_raw_slowpath_nullable();
test_tuple_validate_key_parts_raw();
--
GitLab