[Tarantool-patches] [PATCH v3] key_def: support composite types extraction

[Tarantool-patches] [PATCH v3] key_def: support composite types extraction

[Ilya Kosarev]

key_def didn't support key definitions with array, map, varbinary & any
fields. Thus they couldn't be extracted with
key_def_object:extract_key(). Since the restriction existed due to
impossibility of such types comparison, this patch removes the
restriction for the fields extraction and only leaves it for
comparison.

Closes #4538
---
Branch: https://github.com/tarantool/tarantool/tree/i.kosarev/gh-4538-key_def-extract-arrays
Issue: https://github.com/tarantool/tarantool/issues/4538

@ChangeLog:
 * Composite types extraction is now supported in key_def (gh-4538).

Changes in v2:
 - type check is now only performed on key_def creation
 - more tests added

Changes in v3:
 - to avoid saving incomparable type it is now being determined at the error path again
 - verbosed some tests peculiar properties

 src/box/key_def.h             | 20 +++++++++
 src/box/lua/key_def.c         | 26 ++++++------
 src/box/tuple_compare.cc      |  4 ++
 test/box-tap/key_def.test.lua | 78 ++++++++++++++++++++++++++++++++---
 4 files changed, 110 insertions(+), 18 deletions(-)

diff --git a/src/box/key_def.h b/src/box/key_def.h
index f4d9e76f2..9f358687a 100644
--- a/src/box/key_def.h
+++ b/src/box/key_def.h
@@ -515,6 +515,24 @@ key_def_has_collation(const struct key_def *key_def)
 	return false;
 }
 
+/**
+ * Return the first field type which can't be compared if @a key_def
+ * has such. Otherwise return field_type_MAX value.
+ */
+static inline enum field_type
+key_def_incomparable_type(const struct key_def *key_def)
+{
+	for (uint32_t i = 0; i < key_def->part_count; ++i) {
+		if (key_def->parts[i].type == FIELD_TYPE_ANY ||
+		    key_def->parts[i].type == FIELD_TYPE_ARRAY ||
+		    key_def->parts[i].type == FIELD_TYPE_MAP) {
+			/* Tuple comparators don't support these types. */
+			return key_def->parts[i].type;
+		}
+	}
+	return field_type_MAX;
+}
+
 /**
  * @brief Checks if \a field_type (MsgPack) is compatible \a type (KeyDef).
  * @param type KeyDef type
@@ -650,6 +668,7 @@ tuple_compare(struct tuple *tuple_a, hint_t tuple_a_hint,
 	      struct tuple *tuple_b, hint_t tuple_b_hint,
 	      struct key_def *key_def)
 {
+	assert(key_def->tuple_compare != NULL);
 	return key_def->tuple_compare(tuple_a, tuple_a_hint,
 				      tuple_b, tuple_b_hint, key_def);
 }
@@ -672,6 +691,7 @@ tuple_compare_with_key(struct tuple *tuple, hint_t tuple_hint,
 		       const char *key, uint32_t part_count,
 		       hint_t key_hint, struct key_def *key_def)
 {
+	assert(key_def->tuple_compare_with_key != NULL);
 	return key_def->tuple_compare_with_key(tuple, tuple_hint, key,
 					       part_count, key_hint, key_def);
 }
diff --git a/src/box/lua/key_def.c b/src/box/lua/key_def.c
index 1a99fab63..4d769f1f5 100644
--- a/src/box/lua/key_def.c
+++ b/src/box/lua/key_def.c
@@ -127,21 +127,9 @@ luaT_key_def_set_part(struct lua_State *L, struct key_part_def *part,
 	const char *type_name = lua_tolstring(L, -1, &type_len);
 	lua_pop(L, 1);
 	part->type = field_type_by_name(type_name, type_len);
-	switch (part->type) {
-	case FIELD_TYPE_ANY:
-	case FIELD_TYPE_VARBINARY:
-	case FIELD_TYPE_ARRAY:
-	case FIELD_TYPE_MAP:
-		/* Tuple comparators don't support these types. */
-		diag_set(IllegalParams, "Unsupported field type: %s",
-			 type_name);
-		return -1;
-	case field_type_MAX:
+	if (part->type == field_type_MAX) {
 		diag_set(IllegalParams, "Unknown field type: %s", type_name);
 		return -1;
-	default:
-		/* Pass though. */
-		break;
 	}
 
 	/* Set part->is_nullable and part->nullable_action. */
@@ -316,6 +304,12 @@ lbox_key_def_compare(struct lua_State *L)
 				     "compare(tuple_a, tuple_b)");
 	}
 
+	if (key_def->tuple_compare == NULL) {
+		diag_set(IllegalParams, "Unsupported field type: %s",
+			 field_type_strs[key_def_incomparable_type(key_def)]);
+		return luaT_error(L);
+	}
+
 	struct tuple *tuple_a, *tuple_b;
 	if ((tuple_a = luaT_key_def_check_tuple(L, key_def, 2)) == NULL)
 		return luaT_error(L);
@@ -349,6 +343,12 @@ lbox_key_def_compare_with_key(struct lua_State *L)
 				     "compare_with_key(tuple, key)");
 	}
 
+	if (key_def->tuple_compare_with_key == NULL) {
+		diag_set(IllegalParams, "Unsupported field type: %s",
+			 field_type_strs[key_def_incomparable_type(key_def)]);
+		return luaT_error(L);
+	}
+
 	struct tuple *tuple = luaT_key_def_check_tuple(L, key_def, 2);
 	if (tuple == NULL)
 		return luaT_error(L);
diff --git a/src/box/tuple_compare.cc b/src/box/tuple_compare.cc
index d059c709f..ad1ae784b 100644
--- a/src/box/tuple_compare.cc
+++ b/src/box/tuple_compare.cc
@@ -2081,5 +2081,9 @@ key_def_set_compare_func(struct key_def *def)
 			key_def_set_compare_func_json<false, false>(def);
 		}
 	}
+	if (key_def_incomparable_type(def) != field_type_MAX) {
+		def->tuple_compare = NULL;
+		def->tuple_compare_with_key = NULL;
+	}
 	key_def_set_hint_func(def);
 }
diff --git a/test/box-tap/key_def.test.lua b/test/box-tap/key_def.test.lua
index 3a4aad687..2a311b75e 100755
--- a/test/box-tap/key_def.test.lua
+++ b/test/box-tap/key_def.test.lua
@@ -215,7 +215,7 @@ end
 
 -- Case: extract_key().
 test:test('extract_key()', function(test)
-    test:plan(13)
+    test:plan(14)
 
     local key_def_a = key_def_lib.new({
         {type = 'unsigned', fieldno = 1},
@@ -245,6 +245,18 @@ test:test('extract_key()', function(test)
     }):extract_key({{a = {b = 'foo'}}}):totable()
     test:is_deeply(res, {'foo'}, 'JSON path (table argument)')
 
+    -- Composite types.
+    local tuple = {{{key = 'aa', data = 'aaaa'},
+                    {key = 'ab', data = 'aabb'},
+                    {key = 'bb', data = 'bbbb'}}, box.NULL,
+                    {a = 5, b = 6}}
+    local res = key_def_lib.new({
+        {type = 'array', fieldno = 1},
+        {type = 'map', fieldno = 2, is_nullable = true},
+        {type = 'any', fieldno = 3},
+    }):extract_key(tuple):totable()
+    test:is_deeply(res, tuple, 'Composite types')
+
     -- A key def has a **nullable** part with a field that is over
     -- a tuple size.
     --
@@ -335,7 +347,7 @@ end)
 
 -- Case: compare().
 test:test('compare()', function(test)
-    test:plan(8)
+    test:plan(9)
 
     local key_def_a = key_def_lib.new({
         {type = 'unsigned', fieldno = 1},
@@ -365,11 +377,19 @@ test:test('compare()', function(test)
             'case 3: less (table argument)')
     test:is(key_def_b:compare(tuple_a:totable(), tuple_c:totable()), 0,
             'case 4: equal (table argument)')
+
+    local cmp_err = 'Unsupported field type: array'
+    local key_def = key_def_lib.new({
+        {type = 'string', fieldno = 1},
+        {type = 'array', fieldno = 2, is_nullable = true},
+    })
+    local ok, err = pcall(key_def.compare, key_def, {'aa', {}}, {'bb', box.NULL})
+    test:is_deeply({ok, tostring(err)}, {false, cmp_err}, 'no composite comparison')
 end)
 
 -- Case: compare_with_key().
 test:test('compare_with_key()', function(test)
-    test:plan(2)
+    test:plan(3)
 
     local key_def_b = key_def_lib.new({
         {type = 'number', fieldno = 2},
@@ -382,11 +402,19 @@ test:test('compare_with_key()', function(test)
 
     local key = box.tuple.new({1, 22})
     test:is(key_def_b:compare_with_key(tuple_a, key), 0, 'tuple')
+
+    local cmp_err = 'Unsupported field type: map'
+    local key_def = key_def_lib.new({
+        {type = 'string', fieldno = 1},
+        {type = 'map', fieldno = 2},
+    })
+    local ok, err = pcall(key_def.compare_with_key, key_def, {'aa', {}}, {'bb', box.NULL})
+    test:is_deeply({ok, tostring(err)}, {false, cmp_err}, 'no composite comparison')
 end)
 
 -- Case: totable().
 test:test('totable()', function(test)
-    test:plan(2)
+    test:plan(3)
 
     local parts_a = {
         {type = 'unsigned', fieldno = 1}
@@ -395,14 +423,23 @@ test:test('totable()', function(test)
         {type = 'number', fieldno = 2},
         {type = 'number', fieldno = 3},
     }
+    local parts_composite = {
+        {type = 'number', fieldno = 2},
+        {type = 'array', fieldno = 3},
+        {type = 'map', fieldno = 4, is_nullable = true},
+    }
     local key_def_a = key_def_lib.new(parts_a)
     local key_def_b = key_def_lib.new(parts_b)
+    local key_def_composite = key_def_lib.new(parts_composite)
 
     local exp = set_key_part_defaults(parts_a)
     test:is_deeply(key_def_a:totable(), exp, 'case 1')
 
     local exp = set_key_part_defaults(parts_b)
     test:is_deeply(key_def_b:totable(), exp, 'case 2')
+
+    local exp = set_key_part_defaults(parts_composite)
+    test:is_deeply(key_def_composite:totable(), exp, 'composite case')
 end)
 
 -- Case: __serialize().
@@ -447,7 +484,7 @@ end)
 
 -- Case: merge().
 test:test('merge()', function(test)
-    test:plan(6)
+    test:plan(8)
 
     local key_def_a = key_def_lib.new({
         {type = 'unsigned', fieldno = 1},
@@ -488,6 +525,37 @@ test:test('merge()', function(test)
         'case 3: verify with :totable()')
     test:is_deeply(key_def_cb:extract_key(tuple_a):totable(),
         {1, 1, box.NULL, 22}, 'case 3: verify with :extract_key()')
+
+    local parts_unsigned = {
+        {type = 'unsigned', fieldno = 1, is_nullable = false},
+    }
+    local key_def_unsigned = key_def_lib.new(parts_unsigned)
+    local key_def_string = key_def_lib.new({
+        {type = 'string', fieldno = 1},
+    })
+    local key_def_array = key_def_lib.new({
+        {type = 'array', fieldno = 1},
+        {type = 'unsigned', fieldno = 2},
+    })
+    local key_def_map = key_def_lib.new({
+        {type = 'map', fieldno = 3, is_nullable = true},
+        {type = 'unsigned', fieldno = 2},
+    })
+
+    -- This test case just shows current behavior and it's presence does not
+    -- mean that the behaviour has to remain the same in future
+    local key_def_unsigned_string = key_def_unsigned:merge(key_def_string)
+    test:is_deeply(key_def_unsigned_string:totable(), parts_unsigned,
+        'in case of conflict we just get the field from the first key_def')
+
+    local key_def_array_map = key_def_array:merge(key_def_map)
+    local exp_parts = {
+        {type = 'array', fieldno = 1, is_nullable = false},
+        {type = 'unsigned', fieldno = 2, is_nullable = false},
+        {type = 'map', fieldno = 3, is_nullable = true},
+    }
+    test:is_deeply(key_def_array_map:totable(), exp_parts,
+        'composite case')
 end)
 
 os.exit(test:check() and 0 or 1)
-- 
2.17.1

Re: [Tarantool-patches] [PATCH v3] key_def: support composite types extraction

[Aleksandr Lyapunov]

Hi! thanks for the patch.

See 1 comment below.
If you add suggested line of code the patch will lgtm.

On 10/1/20 9:35 PM, Ilya Kosarev wrote:
> +/**
> + * Return the first field type which can't be compared if @a key_def
> + * has such. Otherwise return field_type_MAX value.
> + */
> +static inline enum field_type
> +key_def_incomparable_type(const struct key_def *key_def)
> +{
> +	for (uint32_t i = 0; i < key_def->part_count; ++i) {
> +		if (key_def->parts[i].type == FIELD_TYPE_ANY ||
What about FIELD_TYPE_VARBINARY? we cannot compare it either.
I think there must be added key_def->parts[i].type == 
FIELD_TYPE_VARBINARY ||
> +		    key_def->parts[i].type == FIELD_TYPE_ARRAY ||
> +		    key_def->parts[i].type == FIELD_TYPE_MAP) {
> +			/* Tuple comparators don't support these types. */
> +			return key_def->parts[i].type;
> +		}
> +	}
> +	return field_type_MAX;
> +}
> +

Re: [Tarantool-patches] [PATCH v3] key_def: support composite types extraction

[Aleksandr Lyapunov]

Oh, i've just found that FIELD_TYPE_VARBINARY is comparable.
Please forget my previous letter.
The patch LGTM, thanks.

On 10/9/20 12:18 PM, Aleksandr Lyapunov wrote:
> Hi! thanks for the patch.
>
> See 1 comment below.
> If you add suggested line of code the patch will lgtm.
>
> On 10/1/20 9:35 PM, Ilya Kosarev wrote:
>> +/**
>> + * Return the first field type which can't be compared if @a key_def
>> + * has such. Otherwise return field_type_MAX value.
>> + */
>> +static inline enum field_type
>> +key_def_incomparable_type(const struct key_def *key_def)
>> +{
>> +    for (uint32_t i = 0; i < key_def->part_count; ++i) {
>> +        if (key_def->parts[i].type == FIELD_TYPE_ANY ||
> What about FIELD_TYPE_VARBINARY? we cannot compare it either.
> I think there must be added key_def->parts[i].type == 
> FIELD_TYPE_VARBINARY ||
>> +            key_def->parts[i].type == FIELD_TYPE_ARRAY ||
>> +            key_def->parts[i].type == FIELD_TYPE_MAP) {
>> +            /* Tuple comparators don't support these types. */
>> +            return key_def->parts[i].type;
>> +        }
>> +    }
>> +    return field_type_MAX;
>> +}
>> +