[tarantool-patches] [PATCH v1 1/1] sql: make sql checks on server side
Kirill Shcherbatov
kshcherbatov at tarantool.org
Wed Sep 19 14:22:32 MSK 2018
To make SQL CHECKS on space insertion in LUA we have to
generate a temporal VDBE and execute it with a tuple
mapped in VDBE memory with new opcode OP_FieldMapMemory as
an argument. These actions are performed by a new special trigger
set for the space on server side.
The CHECKS are carried out twice for SQL requests because of
context and affinity changes to which the inserting tuple is
subjected.
At first, user could specify IGNORE keyword for SQL UPDATE that
is a part of PARSER CONTEXT and couldn't be handled on
server-side for now.
The other point is that tuple represented as msgpack on
server-side may have already changed field data type.
E.g.
CREATE TABLE t2(id primary key,
z TEXT CONSTRAINT three
CHECK(typeof(coalesce(z,'')) == 'text'));
INSERT INTO t2 VALUES(1, 3.14159);
Before the tuple for insertion into the server-side space was
generated, it's z field was of type REAL, which did not pass
the CHECK test. But then new compatible affinity BLOB is applied,
so making check on server-side no problems would be triggered
and tuple would be inserted successfully.
Closes #3691.
---
Branch: https://github.com/tarantool/tarantool/tree/kshch/gh-3691-checks-on-server-side
Issue: https://github.com/tarantool/tarantool/issues/3691
src/box/alter.cc | 37 +++++++++++++++++++++
src/box/sql.c | 67 +++++++++++++++++++++++++++++++++++++
src/box/sql.h | 15 +++++++++
src/box/sql/build.c | 2 +-
src/box/sql/insert.c | 86 +++++++++++++++++++++++++++---------------------
src/box/sql/sqliteInt.h | 27 ++++++++++++---
src/box/sql/vdbe.c | 20 ++++++++++-
test/sql/checks.result | 47 ++++++++++++++++++++++++++
test/sql/checks.test.lua | 17 ++++++++++
9 files changed, 275 insertions(+), 43 deletions(-)
diff --git a/src/box/alter.cc b/src/box/alter.cc
index 1757332..4684fc6 100644
--- a/src/box/alter.cc
+++ b/src/box/alter.cc
@@ -39,6 +39,7 @@
#include "coll_id_def.h"
#include "txn.h"
#include "tuple.h"
+#include "trigger.h"
#include "fiber.h" /* for gc_pool */
#include "scoped_guard.h"
#include "third_party/base64.h"
@@ -1425,6 +1426,27 @@ on_drop_space_rollback(struct trigger *trigger, void *event)
}
/**
+ * SQL-specific actions for space.
+ */
+static void
+on_replace_dd_space_sql(struct trigger *trigger, void *event)
+{
+ uint32_t space_id = (uint32_t)(uintptr_t)trigger->data;
+ struct space *space = space_cache_find(space_id);
+ assert(space != NULL);
+ struct txn *txn = (struct txn *) event;
+ struct txn_stmt *stmt = txn_current_stmt(txn);
+ if (stmt == NULL)
+ return;
+ struct tuple *new_tuple = stmt->new_tuple;
+ if (new_tuple == NULL)
+ return;
+ struct tuple *old_tuple = stmt->old_tuple;
+ if (sql_make_constraint_checks(space, new_tuple, old_tuple) != 0)
+ diag_raise();
+}
+
+/**
* Run the triggers registered on commit of a change in _space.
*/
static void
@@ -1747,6 +1769,21 @@ on_replace_dd_space(struct trigger * /* trigger */, void *event)
struct trigger *on_rollback =
txn_alter_trigger_new(on_create_space_rollback, space);
txn_on_rollback(txn, on_rollback);
+ /* Setup SQL-actions trigger if required. */
+ if (def->opts.checks != NULL) {
+ struct trigger *sql_actions_trigger =
+ (struct trigger *)malloc(sizeof(struct trigger));
+ if (sql_actions_trigger == NULL) {
+ tnt_raise(OutOfMemory, sizeof(struct trigger),
+ "calloc",
+ "sql_before_actions_trigger");
+ }
+ trigger_create(sql_actions_trigger,
+ on_replace_dd_space_sql,
+ (void *)(uintptr_t)space->def->id,
+ (trigger_f0)free);
+ trigger_add(&space->on_replace, sql_actions_trigger);
+ }
} else if (new_tuple == NULL) { /* DELETE */
access_check_ddl(old_space->def->name, old_space->def->id,
old_space->def->uid, SC_SPACE, PRIV_D, true);
diff --git a/src/box/sql.c b/src/box/sql.c
index d3f50c5..3ea4915 100644
--- a/src/box/sql.c
+++ b/src/box/sql.c
@@ -1763,3 +1763,70 @@ sql_checks_resolve_space_def_reference(ExprList *expr_list,
sql_parser_destroy(&parser);
return rc;
}
+
+int
+sql_make_constraint_checks(struct space *space, struct tuple *new_tuple,
+ struct tuple *old_tuple)
+{
+ assert(new_tuple != NULL);
+ struct ExprList *checks = space->def->opts.checks;
+ assert(checks != NULL);
+ struct session *user_session = current_session();
+ if ((user_session->sql_flags & SQLITE_IgnoreChecks) != 0)
+ return 0;
+
+ struct Parse parser;
+ sql_parser_create(&parser, sql_get());
+ struct Vdbe *v = sqlite3GetVdbe(&parser);
+ parser.pVdbe = v;
+ int reg = sqlite3GetTempRange(&parser, space->def->field_count);
+ struct region *region = &fiber()->gc;
+ uint32_t data_size = new_tuple->bsize + 1 +
+ old_tuple != NULL ?
+ sizeof(int)*space->def->field_count : 0;
+ char *data = region_alloc(region, data_size);
+ if (data == NULL) {
+ diag_set(OutOfMemory, data_size, "region_alloc", "data");
+ return -1;
+ }
+ memcpy(data, tuple_data(new_tuple), new_tuple->bsize);
+ data[new_tuple->bsize] = 0;
+ int *update_mask = NULL;
+ const char *old_tuple_raw = NULL;
+ if (old_tuple != NULL) {
+ update_mask = (int *)(data + old_tuple->bsize + 1);
+ memset(update_mask, -1, space->def->field_count);
+ old_tuple_raw = tuple_data(old_tuple);
+ mp_decode_array(&old_tuple_raw);
+ }
+ const char *new_tuple_raw = data;
+ mp_decode_array(&new_tuple_raw);
+ for (uint32_t i = 0; i < space->def->field_count; i++) {
+ const char *new_tuple_field = new_tuple_raw;
+ mp_next(&new_tuple_raw);
+ sqlite3VdbeAddOp4(v, OP_FieldMapMemory, reg + i, 0, 0,
+ (void *)new_tuple_field, P4_STATIC);
+ if (old_tuple_raw == NULL)
+ continue;
+ const char *old_tuple_field = old_tuple_raw;
+ mp_next(&old_tuple_raw);
+ update_mask[i] =
+ (old_tuple_raw - old_tuple_field !=
+ new_tuple_raw - new_tuple_field ||
+ memcmp(new_tuple_field, old_tuple_field,
+ old_tuple_raw - old_tuple_field) != 0) ? -1 : 0;
+ }
+ int skip_label = sqlite3VdbeMakeLabel(v);
+ vdbe_emit_checks_test(&parser, space->def->name, checks, reg,
+ ON_CONFLICT_ACTION_DEFAULT, skip_label,
+ update_mask);
+ sqlite3VdbeResolveLabel(v, skip_label);
+ sql_finish_coding(&parser);
+ struct sqlite3_stmt *stmt = (struct sqlite3_stmt *)v;
+ int rc = sqlite3_step(stmt);
+ if (rc != SQLITE_DONE)
+ diag_set(ClientError, ER_SQL_EXECUTE, v->zErrMsg);
+ sql_parser_destroy(&parser);
+ sqlite3_finalize(stmt);
+ return rc != SQLITE_DONE ? -1 : 0;
+}
diff --git a/src/box/sql.h b/src/box/sql.h
index 5350d39..14cc353 100644
--- a/src/box/sql.h
+++ b/src/box/sql.h
@@ -67,6 +67,8 @@ struct Parse;
struct Select;
struct Table;
struct sql_trigger;
+struct space;
+struct tuple;
/**
* Perform parsing of provided expression. This is done by
@@ -405,6 +407,19 @@ sql_select_from_table_count(const struct Select *select);
const char *
sql_select_from_table_name(const struct Select *select, int i);
+/**
+ * Make constrints checks for a tuple beeng inserted in space.
+ *
+ * @param space The space object to be inserted to.
+ * @param new_tuple New tuple to be inserted.
+ * @param old_tuple Old tuple to be replaced.
+ * @retval 0 On tuple may be inserted.
+ * @retval -1 Elsewhere. Diagnostic error is set.
+ */
+int
+sql_make_constraint_checks(struct space *space, struct tuple *new_tuple,
+ struct tuple *old_tuple);
+
#if defined(__cplusplus)
} /* extern "C" { */
#endif
diff --git a/src/box/sql/build.c b/src/box/sql/build.c
index a1e16b2..5faa5bc 100644
--- a/src/box/sql/build.c
+++ b/src/box/sql/build.c
@@ -999,7 +999,7 @@ struct ExprList *
space_checks_expr_list(uint32_t space_id)
{
struct space *space;
- space = space_by_id(space_id);
+ space = space_cache_find(space_id);
assert(space != NULL);
assert(space->def != NULL);
return space->def->opts.checks;
diff --git a/src/box/sql/insert.c b/src/box/sql/insert.c
index f7d489f..26a0faa 100644
--- a/src/box/sql/insert.c
+++ b/src/box/sql/insert.c
@@ -849,12 +849,54 @@ checkConstraintUnchanged(Expr * pExpr, int *aiChng)
}
void
+vdbe_emit_checks_test(struct Parse *parse_context, char *space_name,
+ struct ExprList *checks, int new_tuple_reg,
+ enum on_conflict_action on_conflict,
+ int ignore_label, int *upd_cols)
+{
+ assert(checks != NULL);
+ struct session *user_session = current_session();
+ assert((user_session->sql_flags & SQLITE_IgnoreChecks) == 0);
+ struct Vdbe *v = sqlite3GetVdbe(parse_context);
+ /*
+ * For CHECK constraint and for INSERT/UPDATE conflict
+ * action DEFAULT and ABORT in fact has the same meaning.
+ */
+ if (on_conflict == ON_CONFLICT_ACTION_DEFAULT)
+ on_conflict = ON_CONFLICT_ACTION_ABORT;
+ enum on_conflict_action on_conflict_check = on_conflict;
+ if (on_conflict == ON_CONFLICT_ACTION_REPLACE)
+ on_conflict_check = ON_CONFLICT_ACTION_ABORT;
+ parse_context->ckBase = new_tuple_reg;
+ for (int i = 0; i < checks->nExpr; i++) {
+ struct Expr *expr = checks->a[i].pExpr;
+ if (upd_cols != NULL &&
+ checkConstraintUnchanged(expr, upd_cols))
+ continue;
+ int all_ok = sqlite3VdbeMakeLabel(v);
+ sqlite3ExprIfTrue(parse_context, expr, all_ok,
+ SQLITE_JUMPIFNULL);
+ if (on_conflict == ON_CONFLICT_ACTION_IGNORE) {
+ sqlite3VdbeGoto(v, ignore_label);
+ } else {
+ char *name = checks->a[i].zName;
+ if (name == NULL)
+ name = space_name;
+ sqlite3HaltConstraint(parse_context,
+ SQLITE_CONSTRAINT_CHECK,
+ on_conflict_check, name,
+ P4_TRANSIENT, P5_ConstraintCheck);
+ }
+ sqlite3VdbeResolveLabel(v, all_ok);
+ }
+}
+
+void
vdbe_emit_constraint_checks(struct Parse *parse_context, struct Table *tab,
int new_tuple_reg,
enum on_conflict_action on_conflict,
int ignore_label, int *upd_cols)
{
- struct session *user_session = current_session();
struct sqlite3 *db = parse_context->db;
struct Vdbe *v = sqlite3GetVdbe(parse_context);
assert(v != NULL);
@@ -917,45 +959,15 @@ vdbe_emit_constraint_checks(struct Parse *parse_context, struct Table *tab,
unreachable();
}
}
- /*
- * For CHECK constraint and for INSERT/UPDATE conflict
- * action DEFAULT and ABORT in fact has the same meaning.
- */
- if (on_conflict == ON_CONFLICT_ACTION_DEFAULT)
- on_conflict = ON_CONFLICT_ACTION_ABORT;
- /* Test all CHECK constraints. */
+ struct session *user_session = current_session();
struct ExprList *checks = space_checks_expr_list(def->id);
- enum on_conflict_action on_conflict_check = on_conflict;
- if (on_conflict == ON_CONFLICT_ACTION_REPLACE)
- on_conflict_check = ON_CONFLICT_ACTION_ABORT;
if (checks != NULL &&
- (user_session->sql_flags & SQLITE_IgnoreChecks) == 0) {
- parse_context->ckBase = new_tuple_reg;
- for (int i = 0; i < checks->nExpr; i++) {
- struct Expr *expr = checks->a[i].pExpr;
- if (is_update &&
- checkConstraintUnchanged(expr, upd_cols))
- continue;
- int all_ok = sqlite3VdbeMakeLabel(v);
- sqlite3ExprIfTrue(parse_context, expr, all_ok,
- SQLITE_JUMPIFNULL);
- if (on_conflict == ON_CONFLICT_ACTION_IGNORE) {
- sqlite3VdbeGoto(v, ignore_label);
- } else {
- char *name = checks->a[i].zName;
- if (name == NULL)
- name = def->name;
- sqlite3HaltConstraint(parse_context,
- SQLITE_CONSTRAINT_CHECK,
- on_conflict_check, name,
- P4_TRANSIENT,
- P5_ConstraintCheck);
- }
- sqlite3VdbeResolveLabel(v, all_ok);
- }
+ (user_session->sql_flags & SQLITE_IgnoreChecks) == 0) {
+ vdbe_emit_checks_test(parse_context, def->name, checks,
+ new_tuple_reg, on_conflict, ignore_label,
+ upd_cols);
}
- sql_emit_table_affinity(v, tab->def, new_tuple_reg);
- /*
+ sql_emit_table_affinity(v, tab->def, new_tuple_reg); /*
* If PK is marked as INTEGER, use it as strict type,
* not as affinity. Emit code for type checking.
* FIXME: should be removed after introducing
diff --git a/src/box/sql/sqliteInt.h b/src/box/sql/sqliteInt.h
index 1d32c9a..4ea36be 100644
--- a/src/box/sql/sqliteInt.h
+++ b/src/box/sql/sqliteInt.h
@@ -3884,6 +3884,26 @@ vdbe_emit_constraint_checks(struct Parse *parse_context,
int ignore_label, int *upd_cols);
/**
+ * Generate code to do SQL Checks tests.
+ *
+ * @param parse_context Current parsing context.
+ * @param space_name The name of space being inserted or updated.
+ * @param checks The list of checks constraints.
+ * @param new_tuple_reg First register in a range holding values
+ * to insert.
+ * @param on_conflict On conflict error action of INSERT or
+ * UPDATE statement (for example INSERT OR REPLACE).
+ * @param ignore_label Jump to this label on an IGNORE resolution.
+ * @param upd_cols Columns to be updated with the size of table's
+ * field count. NULL for INSERT operation.
+ */
+void
+vdbe_emit_checks_test(struct Parse *parse_context, char *space_name,
+ struct ExprList *checks, int new_tuple_reg,
+ enum on_conflict_action on_conflict,
+ int ignore_label, int *upd_cols);
+
+ /**
* This routine generates code to finish the INSERT or UPDATE
* operation that was started by a prior call to
* vdbe_emit_constraint_checks. It encodes raw data which is held
@@ -3897,10 +3917,9 @@ vdbe_emit_constraint_checks(struct Parse *parse_context,
* @param tuple_len Number of registers to hold the tuple.
* @param on_conflict On conflict action.
*/
-void
-vdbe_emit_insertion_completion(struct Vdbe *v, int cursor_id, int raw_data_reg,
- uint32_t tuple_len,
- enum on_conflict_action on_conflict);
+ void vdbe_emit_insertion_completion(struct Vdbe *v, int cursor_id, int raw_data_reg,
+ uint32_t tuple_len,
+ enum on_conflict_action on_conflict);
void
sql_set_multi_write(Parse *, bool);
diff --git a/src/box/sql/vdbe.c b/src/box/sql/vdbe.c
index a5b907d..ac8f0da 100644
--- a/src/box/sql/vdbe.c
+++ b/src/box/sql/vdbe.c
@@ -49,6 +49,7 @@
#include "msgpuck/msgpuck.h"
+#include "box/tuple.h"
#include "box/schema.h"
#include "box/space.h"
#include "box/sequence.h"
@@ -2471,7 +2472,7 @@ case OP_IsNull: { /* same as TK_ISNULL, jump, in1 */
break;
}
-/* Opcode: NotNull P1 P2 * * *
+/* Opcode: NotNull P1 P2 * P4 *
* Synopsis: if r[P1]!=NULL goto P2
*
* Jump to P2 if the value in register P1 is not NULL.
@@ -2485,6 +2486,23 @@ case OP_NotNull: { /* same as TK_NOTNULL, jump, in1 */
break;
}
+/* Opcode: OP_FieldMapMemory P1 P2 * P4 *
+ * Synopsis: r[P1]=unpackrec(P4)
+ *
+ * Map msgpack from P4 register to VDBE memory register P1 and
+ * apply affinity P2 if specified.
+ */
+case OP_FieldMapMemory: {
+ assert(pOp->p4type == P4_STATIC);
+ char *raw = pOp->p4.z;
+ struct Mem *dest = &aMem[pOp->p1];
+ memAboutToChange(p, dest);
+ sqlite3VdbeMsgpackGet((const unsigned char *)raw, dest);
+ if (pOp->p2 != 0)
+ applyAffinity(dest, (char)pOp->p2);
+ break;
+}
+
/* Opcode: Column P1 P2 P3 P4 P5
* Synopsis: r[P3]=PX
*
diff --git a/test/sql/checks.result b/test/sql/checks.result
index 12a3aa1..2401735 100644
--- a/test/sql/checks.result
+++ b/test/sql/checks.result
@@ -139,6 +139,53 @@ s = box.space._space:insert(t)
- error: 'Wrong space options (field 5): invalid expression specified (SQL error:
bindings are not allowed in DDL)'
...
+--
+-- gh-3691: Do SQL checks on server side
+--
+box.sql.execute("CREATE TABLE t1(x INTEGER CONSTRAINT ONE CHECK( x<5 ), y REAL CONSTRAINT TWO CHECK( y>x ), z primary key);")
+---
+...
+box.space.T1:insert({7, 1, 1})
+---
+- error: 'Failed to execute SQL statement: CHECK constraint failed: ONE'
+...
+box.space.T1:insert({2, 1, 1})
+---
+- error: 'Failed to execute SQL statement: CHECK constraint failed: TWO'
+...
+box.space.T1:insert({2, 4, 1})
+---
+- [2, 4, 1]
+...
+box.sql.execute("DROP TABLE t1");
+---
+...
+opts = {checks = {{expr = 'X > 5', name = 'ONE'}}}
+---
+...
+format = {{name = 'X', type = 'unsigned'}}
+---
+...
+t = {513, 1, 'test', 'memtx', 0, opts, format}
+---
+...
+s = box.space._space:insert(t)
+---
+...
+_ = box.space.test:create_index('pk')
+---
+...
+box.space.test:insert({1})
+---
+- error: 'Failed to execute SQL statement: CHECK constraint failed: ONE'
+...
+box.space.test:insert({6})
+---
+- [6]
+...
+box.space.test:drop()
+---
+...
test_run:cmd("clear filter")
---
- true
diff --git a/test/sql/checks.test.lua b/test/sql/checks.test.lua
index 0582bbb..be59493 100644
--- a/test/sql/checks.test.lua
+++ b/test/sql/checks.test.lua
@@ -60,5 +60,22 @@ format = {{name = 'X', type = 'unsigned'}}
t = {513, 1, 'test', 'memtx', 0, opts, format}
s = box.space._space:insert(t)
+--
+-- gh-3691: Do SQL checks on server side
+--
+box.sql.execute("CREATE TABLE t1(x INTEGER CONSTRAINT ONE CHECK( x<5 ), y REAL CONSTRAINT TWO CHECK( y>x ), z primary key);")
+box.space.T1:insert({7, 1, 1})
+box.space.T1:insert({2, 1, 1})
+box.space.T1:insert({2, 4, 1})
+box.sql.execute("DROP TABLE t1");
+
+opts = {checks = {{expr = 'X > 5', name = 'ONE'}}}
+format = {{name = 'X', type = 'unsigned'}}
+t = {513, 1, 'test', 'memtx', 0, opts, format}
+s = box.space._space:insert(t)
+_ = box.space.test:create_index('pk')
+box.space.test:insert({1})
+box.space.test:insert({6})
+box.space.test:drop()
test_run:cmd("clear filter")
--
2.7.4
More information about the Tarantool-patches
mailing list