[tarantool-patches] [PATCH v4 3/4] box: introduce column_mask for ck constraint

[Kirill Shcherbatov <kshcherbatov@tarantool.org>]

This patch introduces a smart bitmask of fields are mentioned in
ck constraint. This allows to do not bind useless fields and
do not parse more fields that is required.

Needed for #3691
---
 src/box/ck_constraint.c  | 66 ++++++++++++++++++++++++++++++++++------
 src/box/ck_constraint.h  |  6 ++++
 src/box/sql.h            |  5 ++-
 src/box/sql/build.c      |  2 +-
 src/box/sql/resolve.c    | 19 +++++++++---
 src/box/sql/sqlInt.h     |  6 ++++
 test/sql/checks.result   | 59 +++++++++++++++++++++++++++++++++++
 test/sql/checks.test.lua | 20 ++++++++++++
 8 files changed, 167 insertions(+), 16 deletions(-)

diff --git a/src/box/ck_constraint.c b/src/box/ck_constraint.c
index 43798be76..4756f7970 100644
--- a/src/box/ck_constraint.c
+++ b/src/box/ck_constraint.c
@@ -46,17 +46,21 @@ const char *ck_constraint_language_strs[] = {"SQL"};
  * tree traversal.
  * @param ck_constraint Check constraint object to update.
  * @param space_def Space definition to use.
+ * @param column_mask[out] The "smart" column mask of fields are
+ *                         referenced by AST.
  * @retval 0 on success.
  * @retval -1 on error.
  */
 static int
 ck_constraint_resolve_field_names(struct Expr *expr,
-				  struct space_def *space_def)
+				  struct space_def *space_def,
+				  uint64_t *column_mask)
 {
 	struct Parse parser;
 	sql_parser_create(&parser, sql_get(), default_flags);
 	parser.parse_only = true;
-	sql_resolve_self_reference(&parser, space_def, NC_IsCheck, expr, NULL);
+	sql_resolve_self_reference(&parser, space_def, NC_IsCheck, expr, NULL,
+				   column_mask);
 	int rc = parser.is_aborted ? -1 : 0;
 	sql_parser_destroy(&parser);
 	return rc;
@@ -75,6 +79,8 @@ ck_constraint_resolve_field_names(struct Expr *expr,
  *             given @ck_constraint_def, see for
  *             (sql_expr_compile +
  *              ck_constraint_resolve_space_def) implementation.
+ * @param column_mask The "smart" column mask of fields are
+ *                    referenced by @a expr.
  * @param space_def The space definition of the space this check
  *                  constraint is constructed for.
  * @retval not NULL sql_stmt program pointer on success.
@@ -82,7 +88,8 @@ ck_constraint_resolve_field_names(struct Expr *expr,
  */
 static struct sql_stmt *
 ck_constraint_program_compile(struct ck_constraint_def *ck_constraint_def,
-			      struct Expr *expr, struct space_def *space_def)
+			      struct Expr *expr, uint64_t column_mask,
+			      struct space_def *space_def)
 {
 	struct sql *db = sql_get();
 	struct Parse parser;
@@ -98,10 +105,24 @@ ck_constraint_program_compile(struct ck_constraint_def *ck_constraint_def,
 	 * bind tuple fields there before execution.
 	 */
 	uint32_t field_count = space_def->field_count;
+	/*
+	 * Use column mask to prepare binding only for
+	 * referenced fields.
+	 */
 	int bind_tuple_reg = sqlGetTempRange(&parser, field_count);
-	for (uint32_t i = 0; i < field_count; i++) {
+	struct bit_iterator it;
+	bit_iterator_init(&it, &column_mask, sizeof(uint64_t), true);
+	size_t used_fieldno = bit_iterator_next(&it);
+	for (; used_fieldno < SIZE_MAX; used_fieldno = bit_iterator_next(&it)) {
 		sqlVdbeAddOp2(v, OP_Variable, ++parser.nVar,
-			      bind_tuple_reg + i);
+			      bind_tuple_reg + used_fieldno);
+	}
+	if (column_mask_fieldno_is_set(column_mask, 63)) {
+		used_fieldno = 64;
+		for (; used_fieldno < (size_t)field_count; used_fieldno++) {
+			sqlVdbeAddOp2(v, OP_Variable, ++parser.nVar,
+				      bind_tuple_reg + used_fieldno);
+		}
 	}
 	/* Generate ck constraint test code. */
 	vdbe_emit_ck_constraint(&parser, expr, ck_constraint_def->name,
@@ -142,6 +163,13 @@ ck_constraint_program_run(struct ck_constraint *ck_constraint,
 	 */
 	struct space *space = space_by_id(ck_constraint->space_id);
 	assert(space != NULL);
+	/* Use column mask to bind only referenced fields. */
+	struct bit_iterator it;
+	bit_iterator_init(&it, &ck_constraint->column_mask,
+			  sizeof(uint64_t), true);
+	size_t used_fieldno = bit_iterator_next(&it);
+	if (used_fieldno == SIZE_MAX)
+		return 0;
 	/*
 	 * When last format fields are nullable, they are
 	 * 'optional' i.e. they may not be present in the tuple.
@@ -149,15 +177,33 @@ ck_constraint_program_run(struct ck_constraint *ck_constraint,
 	uint32_t tuple_field_count = mp_decode_array(&new_tuple);
 	uint32_t field_count =
 		MIN(tuple_field_count, space->def->field_count);
-	for (uint32_t i = 0; i < field_count; i++) {
+	uint32_t bind_pos = 1;
+	for (uint32_t fieldno = 0; fieldno < field_count; fieldno++) {
+		if (used_fieldno == SIZE_MAX &&
+		    !column_mask_fieldno_is_set(ck_constraint->column_mask,
+						63)) {
+			/* No more required fields are left. */
+			break;
+		}
+		if (used_fieldno != SIZE_MAX &&
+		    (size_t)fieldno < used_fieldno) {
+			/*
+			 * Skip unused fields is mentioned in
+			 * column mask.
+			 */
+			mp_next(&new_tuple);
+			continue;
+		}
 		struct sql_bind bind;
-		if (sql_bind_decode(&bind, i + 1, &new_tuple) != 0 ||
-		    sql_bind_column(ck_constraint->stmt, &bind, i + 1) != 0) {
+		if (sql_bind_decode(&bind, bind_pos, &new_tuple) != 0 ||
+		    sql_bind_column(ck_constraint->stmt, &bind, bind_pos) != 0) {
 			diag_set(ClientError, ER_CK_CONSTRAINT_FAILED,
 				 ck_constraint->def->name,
 				 ck_constraint->def->expr_str);
 			return -1;
 		}
+		bind_pos++;
+		used_fieldno = bit_iterator_next(&it);
 	}
 	/* Checks VDBE can't expire, reset expired flag and go. */
 	struct Vdbe *v = (struct Vdbe *) ck_constraint->stmt;
@@ -220,7 +266,8 @@ ck_constraint_new(struct ck_constraint_def *ck_constraint_def,
 		sql_expr_compile(sql_get(), ck_constraint_def->expr_str,
 				 strlen(ck_constraint_def->expr_str));
 	if (expr == NULL ||
-	    ck_constraint_resolve_field_names(expr, space_def) != 0) {
+	    ck_constraint_resolve_field_names(expr, space_def,
+					&ck_constraint->column_mask) != 0) {
 		diag_set(ClientError, ER_CREATE_CK_CONSTRAINT,
 			 ck_constraint_def->name,
 			 box_error_message(box_error_last()));
@@ -228,6 +275,7 @@ ck_constraint_new(struct ck_constraint_def *ck_constraint_def,
 	}
 	ck_constraint->stmt =
 		ck_constraint_program_compile(ck_constraint_def, expr,
+					      ck_constraint->column_mask,
 					      space_def);
 	if (ck_constraint->stmt == NULL)
 		goto error;
diff --git a/src/box/ck_constraint.h b/src/box/ck_constraint.h
index a948a074e..141bdc11d 100644
--- a/src/box/ck_constraint.h
+++ b/src/box/ck_constraint.h
@@ -88,6 +88,12 @@ struct ck_constraint {
 	 * message when ck condition unsatisfied.
 	 */
 	struct sql_stmt *stmt;
+	/**
+	 * The "smart" column mask of fields are referenced by
+	 * AST. Then the last bit of the mask is set when field
+	 * #63 and greater has referenced.
+	 */
+	uint64_t column_mask;
 	/**
 	 * Trigger object executing check constraint before
 	 * insert and replace operations.
diff --git a/src/box/sql.h b/src/box/sql.h
index 9c8a96a75..13f5984f0 100644
--- a/src/box/sql.h
+++ b/src/box/sql.h
@@ -279,11 +279,14 @@ sql_expr_list_append(struct sql *db, struct ExprList *expr_list,
  * @param type NC_IsCheck or NC_IdxExpr.
  * @param expr Expression to resolve.  May be NULL.
  * @param expr_list Expression list to resolve.  May be NUL.
+ * @param[out] column_mask The "smart" column mask of fields are
+ *                         referenced by AST.
  */
 void
 sql_resolve_self_reference(struct Parse *parser, struct space_def *def,
 			   int type, struct Expr *expr,
-			   struct ExprList *expr_list);
+			   struct ExprList *expr_list,
+			   uint64_t *column_mask);
 
 /**
  * Initialize a new parser object.
diff --git a/src/box/sql/build.c b/src/box/sql/build.c
index f2bbbb9c4..5eac4ee40 100644
--- a/src/box/sql/build.c
+++ b/src/box/sql/build.c
@@ -2198,7 +2198,7 @@ index_fill_def(struct Parse *parse, struct index *index,
 	for (int i = 0; i < expr_list->nExpr; i++) {
 		struct Expr *expr = expr_list->a[i].pExpr;
 		sql_resolve_self_reference(parse, space_def, NC_IdxExpr,
-					   expr, 0);
+					   expr, NULL, NULL);
 		if (parse->is_aborted)
 			goto cleanup;
 
diff --git a/src/box/sql/resolve.c b/src/box/sql/resolve.c
index 7cf79692c..cbd5cd724 100644
--- a/src/box/sql/resolve.c
+++ b/src/box/sql/resolve.c
@@ -556,8 +556,11 @@ resolveExprStep(Walker * pWalker, Expr * pExpr)
 	case TK_ID:{
 			if ((pNC->ncFlags & NC_AllowAgg) != 0)
 				pNC->ncFlags |= NC_HasUnaggregatedId;
-			return lookupName(pParse, 0, pExpr->u.zToken, pNC,
-					  pExpr);
+			int rc = lookupName(pParse, 0, pExpr->u.zToken, pNC,
+					    pExpr);
+			column_mask_set_fieldno(&pNC->column_mask,
+						pExpr->iColumn);
+			return rc;
 		}
 
 		/* A table name and column name:     ID.ID
@@ -583,8 +586,11 @@ resolveExprStep(Walker * pWalker, Expr * pExpr)
 				zTable = pRight->pLeft->u.zToken;
 				zColumn = pRight->pRight->u.zToken;
 			}
-			return lookupName(pParse, zTable, zColumn, pNC,
-					  pExpr);
+			int rc = lookupName(pParse, zTable, zColumn, pNC,
+					    pExpr);
+			column_mask_set_fieldno(&pNC->column_mask,
+						pExpr->iColumn);
+			return rc;
 		}
 
 		/* Resolve function names
@@ -1612,7 +1618,7 @@ sqlResolveSelectNames(Parse * pParse,	/* The parser context */
 void
 sql_resolve_self_reference(struct Parse *parser, struct space_def *def,
 			   int type, struct Expr *expr,
-			   struct ExprList *expr_list)
+			   struct ExprList *expr_list, uint64_t *column_mask)
 {
 	/* Fake SrcList for parser->create_table_def */
 	SrcList sSrc;
@@ -1632,8 +1638,11 @@ sql_resolve_self_reference(struct Parse *parser, struct space_def *def,
 	sNC.pParse = parser;
 	sNC.pSrcList = &sSrc;
 	sNC.ncFlags = type;
+	sNC.column_mask = 0;
 	if (sqlResolveExprNames(&sNC, expr) != 0)
 		return;
 	if (expr_list != NULL)
 		sqlResolveExprListNames(&sNC, expr_list);
+	if (column_mask != NULL)
+		*column_mask = sNC.column_mask;
 }
diff --git a/src/box/sql/sqlInt.h b/src/box/sql/sqlInt.h
index d353d7906..0eca37653 100644
--- a/src/box/sql/sqlInt.h
+++ b/src/box/sql/sqlInt.h
@@ -2349,6 +2349,12 @@ struct NameContext {
 	int nRef;		/* Number of names resolved by this context */
 	int nErr;		/* Number of errors encountered while resolving names */
 	u16 ncFlags;		/* Zero or more NC_* flags defined below */
+	/**
+	 * The "smart" column mask of fields are referenced
+	 * by AST. Then the last bit of the mask is set when
+	 * field #63 and greater has referenced.
+	 */
+	uint64_t column_mask;
 };
 
 /*
diff --git a/test/sql/checks.result b/test/sql/checks.result
index b74572e3e..04f58c90d 100644
--- a/test/sql/checks.result
+++ b/test/sql/checks.result
@@ -454,6 +454,65 @@ s:insert({1, 666})
 s:drop()
 ---
 ...
+--
+-- Test binding optimisation.
+--
+s = box.schema.create_space('test')
+---
+...
+_ = s:create_index('pk', {parts = {1, 'integer'}})
+---
+...
+format65 = {}
+---
+...
+test_run:cmd("setopt delimiter ';'")
+---
+- true
+...
+for i = 1,66 do
+        table.insert(format65, {name='X'..i, type='integer', is_nullable = true})
+end
+test_run:cmd("setopt delimiter ''");
+---
+...
+s:format(format65)
+---
+...
+_ = box.space._ck_constraint:insert({'X1is666andX65is666', s.id, false, 'X1 == 666 and X65 == 666 and X63 IS NOT NULL', 'SQL'})
+---
+...
+s:insert(s:frommap({X1 = 1, X65 = 1}))
+---
+- error: 'Check constraint failed ''X1is666andX65is666'': X1 == 666 and X65 == 666
+    and X63 IS NOT NULL'
+...
+s:insert(s:frommap({X1 = 666, X65 = 1}))
+---
+- error: 'Check constraint failed ''X1is666andX65is666'': X1 == 666 and X65 == 666
+    and X63 IS NOT NULL'
+...
+s:insert(s:frommap({X1 = 1, X65 = 666}))
+---
+- error: 'Check constraint failed ''X1is666andX65is666'': X1 == 666 and X65 == 666
+    and X63 IS NOT NULL'
+...
+s:insert(s:frommap({X1 = 666, X65 = 666}))
+---
+- error: 'Check constraint failed ''X1is666andX65is666'': X1 == 666 and X65 == 666
+    and X63 IS NOT NULL'
+...
+s:insert(s:frommap({X1 = 666, X65 = 666, X63 = 1}))
+---
+- [666, null, null, null, null, null, null, null, null, null, null, null, null, null,
+  null, null, null, null, null, null, null, null, null, null, null, null, null, null,
+  null, null, null, null, null, null, null, null, null, null, null, null, null, null,
+  null, null, null, null, null, null, null, null, null, null, null, null, null, null,
+  null, null, null, null, null, null, 1, null, 666]
+...
+s:drop()
+---
+...
 test_run:cmd("clear filter")
 ---
 - true
diff --git a/test/sql/checks.test.lua b/test/sql/checks.test.lua
index a284edf52..a23dcce51 100644
--- a/test/sql/checks.test.lua
+++ b/test/sql/checks.test.lua
@@ -156,4 +156,24 @@ s:insert({1, 3.14})
 s:insert({1, 666})
 s:drop()
 
+--
+-- Test binding optimisation.
+--
+s = box.schema.create_space('test')
+_ = s:create_index('pk', {parts = {1, 'integer'}})
+format65 = {}
+test_run:cmd("setopt delimiter ';'")
+for i = 1,66 do
+        table.insert(format65, {name='X'..i, type='integer', is_nullable = true})
+end
+test_run:cmd("setopt delimiter ''");
+s:format(format65)
+_ = box.space._ck_constraint:insert({'X1is666andX65is666', s.id, false, 'X1 == 666 and X65 == 666 and X63 IS NOT NULL', 'SQL'})
+s:insert(s:frommap({X1 = 1, X65 = 1}))
+s:insert(s:frommap({X1 = 666, X65 = 1}))
+s:insert(s:frommap({X1 = 1, X65 = 666}))
+s:insert(s:frommap({X1 = 666, X65 = 666}))
+s:insert(s:frommap({X1 = 666, X65 = 666, X63 = 1}))
+s:drop()
+
 test_run:cmd("clear filter")
-- 
2.21.0