[tarantool-patches] Re: [PATCH v2 1/1] sql: cleanup on failed creation operation

["n.pettik" <korablev@tarantool.org>]

Attach please next time diff between two versions,
so as I can avoid to review whole patch again.

Also, check Travis before you send patch:

https://travis-ci.org/tarantool/tarantool/jobs/438801852
https://travis-ci.org/tarantool/tarantool/jobs/438801839

Build fails.

>> 
>>>  void
>>>  sql_finish_coding(struct Parse *parse_context)
>>> @@ -62,6 +110,29 @@ sql_finish_coding(struct Parse *parse_context)
>>>      struct sqlite3 *db = parse_context->db;
>>>      struct Vdbe *v = sqlite3GetVdbe(parse_context);
>>>      sqlite3VdbeAddOp0(v, OP_Halt);
>>> +    /*
>>> +     * Destructors for all created in current statement
>>> +     * spaces, indexes, sequences etc. There is no need to
>>> +     * create destructor for last SInsert.
>>> +     */
>>> +    if (rlist_empty(&parse_context->record_list) == 0) {
>> 
>> Nit: if(! rlist_empty()) 
> Fixed.

No, it is not fixed.

> diff --git a/src/box/sql/build.c b/src/box/sql/build.c
> index a806fb4..6a97ff9 100644
> --- a/src/box/sql/build.c
> +++ b/src/box/sql/build.c
>  void
>  sql_finish_coding(struct Parse *parse_context)
>  {
> @@ -62,6 +109,40 @@ sql_finish_coding(struct Parse *parse_context)
>      struct sqlite3 *db = parse_context->db;
>      struct Vdbe *v = sqlite3GetVdbe(parse_context);
>      sqlite3VdbeAddOp0(v, OP_Halt);
> +    /*
> +     * In case statement "CREATE TABLE ..." fails it can
> +     * left some records in system spaces that shouldn't be
> +     * there. To clean-up properly this code is added. Last
> +     * record isn't deleted because if statement fails than
> +     * it won't be created. This code works the same way for
> +     * other "CREATE ..." statements but it won't delete
> +     * anything as these statements create no more than one
> +     * record.
> +     */
> +    if (rlist_empty(&parse_context->record_list) == 0) {
> +        struct saved_record *record =
> +            rlist_shift_entry(&parse_context->record_list,
> +                      struct saved_record, link);
> +        /* Set P2 of SInsert. */
> +        sqlite3VdbeChangeP2(v, record->insertion_opcode, v->nOp);
> +        char *comment = "Delete entry from %s if CREATE TABLE fails”;

Nit: const char *.

> +        rlist_foreach_entry(record, &parse_context->record_list, link) {
> +            int record_reg = ++parse_context->nMem;
> +            sqlite3VdbeAddOp3(v, OP_MakeRecord, record->reg_key,
> +                      record->reg_key_count, record_reg);
> +            sqlite3VdbeAddOp2(v, OP_SDelete, record->space_id,
> +                      record_reg);
> +            struct space *space = space_by_id(record->space_id);
> +            VdbeComment((v, comment, space_name(space)));
> +            /* Set P2 of SInsert. */
> +            sqlite3VdbeChangeP2(v, record->insertion_opcode,
> +                        v->nOp);
> +        }
> +        sqlite3VdbeAddOp2(v, OP_Halt, SQL_TARANTOOL_ERROR, 0);
> +        VdbeComment((v,
> +                 "Exit with an error if CREATE statement fails"));
> +    }
> +
>      if (db->mallocFailed || parse_context->nErr != 0) {
>          if (parse_context->rc == SQLITE_OK)
>              parse_context->rc = SQLITE_ERROR;
> @@ -1101,13 +1182,14 @@ vdbe_emit_create_index(struct Parse *parse, struct space_def *def,
>      sqlite3VdbeAddOp4(v, OP_Blob, index_parts_sz, entry_reg + 5,
>                SQL_SUBTYPE_MSGPACK, index_parts, P4_STATIC);
>      sqlite3VdbeAddOp3(v, OP_MakeRecord, entry_reg, 6, tuple_reg);
> -    sqlite3VdbeAddOp2(v, OP_SInsert, BOX_INDEX_ID, tuple_reg);
> +    sqlite3VdbeAddOp3(v, OP_SInsert, BOX_INDEX_ID, 0, tuple_reg);
>      /*
>       * Non-NULL value means that index has been created via
>       * separate CREATE INDEX statement.
>       */
>      if (idx_def->opts.sql != NULL)
>          sqlite3VdbeChangeP5(v, OPFLAG_NCHANGE);
> +    save_record(parse, BOX_INDEX_ID, entry_reg, 2, v->nOp - 1);

As I sad you don’t need to add to this list index entries when it comes
to CREATE INDEX. You can tell index from unique constraint by
existence of opts.sql string:

+++ b/src/box/sql/build.c
@@ -1185,11 +1185,15 @@ vdbe_emit_create_index(struct Parse *parse, struct space_def *def,
        sqlite3VdbeAddOp3(v, OP_SInsert, BOX_INDEX_ID, 0, tuple_reg);
        /*
         * Non-NULL value means that index has been created via
-        * separate CREATE INDEX statement.
+        * separate CREATE INDEX statement. On the other hand,
+        * we need to register all indexes incoming in
+        * CREATE TABLE in order to remove them on table creation
+        * fail.
         */
        if (idx_def->opts.sql != NULL)
                sqlite3VdbeChangeP5(v, OPFLAG_NCHANGE);
-       save_record(parse, BOX_INDEX_ID, entry_reg, 2, v->nOp - 1);
+       else
+               save_record(parse, BOX_INDEX_ID, entry_reg, 2, v->nOp - 1);
        return;

However, in this case we need to jump to abort_due_to_error label
during processing of OP_SInsert. I don’t insist on this change tho,
it is up to you. 

>      return;
>  error:
>      parse->rc = SQL_TARANTOOL_ERROR;
> @@ -1165,8 +1247,9 @@ createSpace(Parse * pParse, int iSpaceId, char *zStmt)
>      sqlite3VdbeAddOp4(v, OP_Blob, table_stmt_sz, iFirstCol + 6,
>                SQL_SUBTYPE_MSGPACK, table_stmt, P4_STATIC);
>      sqlite3VdbeAddOp3(v, OP_MakeRecord, iFirstCol, 7, iRecord);
> -    sqlite3VdbeAddOp2(v, OP_SInsert, BOX_SPACE_ID, iRecord);
> +    sqlite3VdbeAddOp3(v, OP_SInsert, BOX_SPACE_ID, 0, iRecord);
>      sqlite3VdbeChangeP5(v, OPFLAG_NCHANGE);
> +    save_record(pParse, BOX_SPACE_ID, iFirstCol, 1, v->nOp - 1);
>      return;
>  error:
>      pParse->nErr++;
> @@ -1340,9 +1423,11 @@ vdbe_emit_fkey_create(struct Parse *parse_context, const struct fkey_def *fk)
>                parent_links, P4_DYNAMIC);
>      sqlite3VdbeAddOp3(vdbe, OP_MakeRecord, constr_tuple_reg, 9,
>                constr_tuple_reg + 9);
> -    sqlite3VdbeAddOp2(vdbe, OP_SInsert, BOX_FK_CONSTRAINT_ID,
> +    sqlite3VdbeAddOp3(vdbe, OP_SInsert, BOX_FK_CONSTRAINT_ID, 0,
>                constr_tuple_reg + 9);
>      sqlite3VdbeChangeP5(vdbe, OPFLAG_NCHANGE);
> +    save_record(parse_context, BOX_FK_CONSTRAINT_ID, constr_tuple_reg, 2,
> +            vdbe->nOp - 1);
>      sqlite3ReleaseTempRange(parse_context, constr_tuple_reg, 10);
>      return;
>  error:
> @@ -1487,14 +1572,18 @@ sqlite3EndTable(Parse * pParse,    /* Parse context */
>          int reg_seq_record =
>              emitNewSysSequenceRecord(pParse, reg_seq_id,
>                           p->def->name);
> -        sqlite3VdbeAddOp2(v, OP_SInsert, BOX_SEQUENCE_ID,
> +        sqlite3VdbeAddOp3(v, OP_SInsert, BOX_SEQUENCE_ID, 0,
>                    reg_seq_record);
> +        save_record(pParse, BOX_SEQUENCE_ID, reg_seq_record + 1, 1,
> +                v->nOp - 1);
>          /* Do an insertion into _space_sequence. */
>          int reg_space_seq_record =
>              emitNewSysSpaceSequenceRecord(pParse, reg_space_id,
>                                reg_seq_id);
> -        sqlite3VdbeAddOp2(v, OP_SInsert, BOX_SPACE_SEQUENCE_ID,
> +        sqlite3VdbeAddOp3(v, OP_SInsert, BOX_SPACE_SEQUENCE_ID, 0,
>                    reg_space_seq_record);
> +        save_record(pParse, BOX_SPACE_SEQUENCE_ID,
> +                reg_space_seq_record + 1, 1, v->nOp - 1);

If we are creating FK constraint with ALTER TABLE ADD CONSTRAINT,
we don’t need to add it to list of things to be deleted (the same as for index).

@@ -1426,8 +1430,10 @@ vdbe_emit_fkey_create(struct Parse *parse_context, const struct fkey_def *fk)
        sqlite3VdbeAddOp3(vdbe, OP_SInsert, BOX_FK_CONSTRAINT_ID, 0,
                          constr_tuple_reg + 9);
        sqlite3VdbeChangeP5(vdbe, OPFLAG_NCHANGE);
-       save_record(parse_context, BOX_FK_CONSTRAINT_ID, constr_tuple_reg, 2,
-                   vdbe->nOp - 1);
+       if (parse_context->pNewTable != NULL) {
+               save_record(parse_context, BOX_FK_CONSTRAINT_ID,
+                           constr_tuple_reg, 2, vdbe->nOp - 1);
+       }

> diff --git a/test/sql/drop-table.test.lua b/test/sql/drop-table.test.lua
> index 9663074..1bc8894 100644
> --- a/test/sql/drop-table.test.lua
> +++ b/test/sql/drop-table.test.lua
> @@ -25,3 +25,62 @@ box.sql.execute("INSERT INTO zzzoobar VALUES (111, 222, 'c3', 444)")
> +
> +box.session.su('admin')
> +
> +--
> +-- Check that _space, _index and _sequence have the same number of
> +-- records.
> +--
> +space_count == #box.space._space:select()
> +index_count == #box.space._index:select()
> +sequence_count == #box.space._sequence:select()
> +
> +--
> +-- Give user right to write in _index. Still have not enough
> +-- rights to write in _sequence.
> +--
> +box.schema.user.grant('tmp', 'write', 'space', '_index')
> +box.session.su('tmp')
> +
> +--
> +-- Error: user do not have rights to write in _sequence.
> +--
> +box.sql.execute('create table t2 (id int primary key autoincrement, a unique, b unique, c unique, d unique)’)

Nit: for SQL statements please use uppercase.

> +
> +box.session.su('admin')
> +
> +--
> +-- Check that _space, _index and _sequence have the same number of
> +-- records.
> +--
> +space_count == #box.space._space:select()
> +index_count == #box.space._index:select()
> +sequence_count == #box.space._sequence:select()
> +
> +box.schema.user.drop('tmp’)

I see no tests involving FK constraints. Add them as well.