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

[Vladislav Shpilevoy <v.shpilevoy@tarantool.org>]

Hi! Thanks for the fixes! LGTM. Nikita,
please, review it.

> Patch with new fixes:
> 
> commit fe8415a79d401b741dcb565d34eb56495223f8b6
> Author: Mergen Imeev <imeevma@gmail.com>
> Date:   Fri Aug 31 15:50:17 2018 +0300
> 
>      sql: cleanup on failed creation operation
> 
>      Some creation operations create objects even on fail. This is
>      wrong and should be fixed. This patch adds destructors for such
>      objects.
> 
>      Closes #3592
> 
> diff --git a/src/box/sql/build.c b/src/box/sql/build.c
> index 60b49df..2ac86ab 100644
> --- a/src/box/sql/build.c
> +++ b/src/box/sql/build.c
> @@ -54,6 +54,54 @@
>   #include "box/schema.h"
>   #include "box/tuple_format.h"
>   #include "box/coll_id_cache.h"
> +#include "fiber.h"
> +
> +/**
> + * Structure that contains some information about record that was
> + * inserted into system space.
> + */
> +struct saved_record
> +{
> +    /** A link in a record list. */
> +    struct rlist link;
> +    /** Id of space in which the record was inserted. */
> +    uint32_t space_id;
> +    /** First register of the key of the record. */
> +    int reg_key;
> +    /** Number of registers the key consists of. */
> +    int reg_key_count;
> +    /** The number of the OP_SInsert operation. */
> +    int insertion_opcode;
> +};
> +
> +/**
> + * Save inserted in system space record in list.
> + *
> + * @param parser SQL Parser object.
> + * @param space_id Id of table in which record is inserted.
> + * @param reg_key Register that contains first field of the key.
> + * @param reg_key_count Exact number of fields of the key.
> + * @param insertion_opcode Number of OP_SInsert opcode.
> + */
> +static inline void
> +save_record(struct Parse *parser, uint32_t space_id, int reg_key,
> +        int reg_key_count, int insertion_opcode)
> +{
> +    struct saved_record *record =
> +        region_alloc(&fiber()->gc, sizeof(*record));
> +    if (record == NULL) {
> +        diag_set(OutOfMemory, sizeof(*record), "region_alloc",
> +             "record");
> +        parser->rc = SQL_TARANTOOL_ERROR;
> +        parser->nErr++;
> +        return;
> +    }
> +    record->space_id = space_id;
> +    record->reg_key = reg_key;
> +    record->reg_key_count = reg_key_count;
> +    record->insertion_opcode = insertion_opcode;
> +    rlist_add_entry(&parser->record_list, record, link);
> +}
> 
>   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) {
> +        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);
> +        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);
> +            /* Set P2 of SInsert. */
> +            sqlite3VdbeChangeP2(v, record->insertion_opcode,
> +                        v->nOp);
> +        }
> +        sqlite3VdbeAddOp2(v, OP_Halt, SQL_TARANTOOL_ERROR, 0);
> +    }
>       if (db->mallocFailed || parse_context->nErr != 0) {
>           if (parse_context->rc == SQLITE_OK)
>               parse_context->rc = SQLITE_ERROR;
> @@ -1101,13 +1172,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);
>       return;
>   error:
>       parse->rc = SQL_TARANTOOL_ERROR;
> @@ -1165,8 +1237,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 +1413,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 +1562,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);
>       }
>       /* Code creation of FK constraints, if any. */
>       struct fkey_parse *fk_parse;
> diff --git a/src/box/sql/prepare.c b/src/box/sql/prepare.c
> index e98e845..a4b65eb 100644
> --- a/src/box/sql/prepare.c
> +++ b/src/box/sql/prepare.c
> @@ -274,6 +274,7 @@ sql_parser_create(struct Parse *parser, sqlite3 *db)
>       memset(parser, 0, sizeof(struct Parse));
>       parser->db = db;
>       rlist_create(&parser->new_fkey);
> +    rlist_create(&parser->record_list);
>       region_create(&parser->region, &cord()->slabc);
>   }
> 
> diff --git a/src/box/sql/sqliteInt.h b/src/box/sql/sqliteInt.h
> index f56090d..c5becbd 100644
> --- a/src/box/sql/sqliteInt.h
> +++ b/src/box/sql/sqliteInt.h
> @@ -2764,6 +2764,11 @@ struct Parse {
>        * Foreign key constraint appeared in CREATE TABLE stmt.
>        */
>       struct rlist new_fkey;
> +    /**
> +     * List of all records that were inserted in system spaces
> +     * in current statement.
> +     */
> +    struct rlist record_list;
>       bool initiateTTrans;    /* Initiate Tarantool transaction */
>       /** True, if table to be created has AUTOINCREMENT PK. */
>       bool is_new_table_autoinc;
> diff --git a/src/box/sql/vdbe.c b/src/box/sql/vdbe.c
> index 0efc4dd..fc959bd 100644
> --- a/src/box/sql/vdbe.c
> +++ b/src/box/sql/vdbe.c
> @@ -1010,8 +1010,12 @@ case OP_Halt: {
>       p->pc = pcx;
>       if (p->rc) {
>           if (p->rc == SQL_TARANTOOL_ERROR) {
> -            assert(pOp->p4.z != NULL);
> -            box_error_set(__FILE__, __LINE__, pOp->p5, pOp->p4.z);
> +            if (pOp->p4.z == NULL) {
> +                assert(! diag_is_empty(diag_get()));
> +            } else {
> +                box_error_set(__FILE__, __LINE__, pOp->p5,
> +                          pOp->p4.z);
> +            }
>           } else if (pOp->p5 != 0) {
>               static const char * const azType[] = { "NOT NULL", "UNIQUE", "CHECK",
>                                      "FOREIGN KEY" };
> @@ -4318,8 +4322,8 @@ case OP_IdxInsert: {        /* in2 */
>       break;
>   }
> 
> -/* Opcode: SInsert P1 P2 * * P5
> - * Synopsis: space id = P1, key = r[P2]
> +/* Opcode: SInsert P1 P2 P3 * P5
> + * Synopsis: space id = P1, key = r[P3], on error goto P2
>    *
>    * This opcode is used only during DDL routine.
>    * In contrast to ordinary insertion, insertion to system spaces
> @@ -4332,15 +4336,15 @@ case OP_IdxInsert: {        /* in2 */
>    */
>   case OP_SInsert: {
>       assert(pOp->p1 > 0);
> -    assert(pOp->p2 >= 0);
> +    assert(pOp->p2 > 0);
> +    assert(pOp->p3 >= 0);
> 
> -    pIn2 = &aMem[pOp->p2];
> +    pIn3 = &aMem[pOp->p3];
>       struct space *space = space_by_id(pOp->p1);
>       assert(space != NULL);
>       assert(space_is_system(space));
> -    rc = tarantoolSqlite3Insert(space, pIn2->z, pIn2->z + pIn2->n);
> -    if (rc)
> -        goto abort_due_to_error;
> +    if (tarantoolSqlite3Insert(space, pIn3->z, pIn3->z + pIn3->n) != 0)
> +        goto jump_to_p2;
>       if (pOp->p5 & OPFLAG_NCHANGE)
>           p->nChange++;
>       break;
> diff --git a/test/sql/drop-table.result b/test/sql/drop-table.result
> index 08f2496..636f3e0 100644
> --- a/test/sql/drop-table.result
> +++ b/test/sql/drop-table.result
> @@ -33,3 +33,43 @@ box.sql.execute("INSERT INTO zzzoobar VALUES (111, 222, 'c3', 444)")
>   -- DROP TABLE should do the job
>   -- Debug
>   -- require("console").start()
> +--
> +-- gh-3592: segmentation fault when table with error during
> +-- creation is dropped.
> +-- We should grant user enough rights to create space, but not
> +-- enough to create index.
> +--
> +box.schema.user.create('tmp')
> +---
> +...
> +box.schema.user.grant('tmp', 'create', 'universe')
> +---
> +...
> +box.schema.user.grant('tmp', 'write', 'space', '_space')
> +---
> +...
> +box.schema.user.grant('tmp', 'write', 'space', '_schema')
> +---
> +...
> +box.session.su('tmp')
> +---
> +...
> +--
> +-- Error: user do not have rights to write in box.space._index.
> +-- Space that was already created should be automatically dropped.
> +--
> +box.sql.execute('create table t1 (id int primary key, a int)')
> +---
> +- error: Write access to space '_index' is denied for user 'tmp'
> +...
> +-- Error: no such table.
> +box.sql.execute('drop table t1')
> +---
> +- error: 'no such table: T1'
> +...
> +box.session.su('admin')
> +---
> +...
> +box.schema.user.drop('tmp')
> +---
> +...
> diff --git a/test/sql/drop-table.test.lua b/test/sql/drop-table.test.lua
> index 9663074..ba47716 100644
> --- a/test/sql/drop-table.test.lua
> +++ b/test/sql/drop-table.test.lua
> @@ -25,3 +25,24 @@ box.sql.execute("INSERT INTO zzzoobar VALUES (111, 222, 'c3', 444)")
> 
>   -- Debug
>   -- require("console").start()
> +
> +--
> +-- gh-3592: segmentation fault when table with error during
> +-- creation is dropped.
> +-- We should grant user enough rights to create space, but not
> +-- enough to create index.
> +--
> +box.schema.user.create('tmp')
> +box.schema.user.grant('tmp', 'create', 'universe')
> +box.schema.user.grant('tmp', 'write', 'space', '_space')
> +box.schema.user.grant('tmp', 'write', 'space', '_schema')
> +box.session.su('tmp')
> +--
> +-- Error: user do not have rights to write in box.space._index.
> +-- Space that was already created should be automatically dropped.
> +--
> +box.sql.execute('create table t1 (id int primary key, a int)')
> +-- Error: no such table.
> +box.sql.execute('drop table t1')
> +box.session.su('admin')
> +box.schema.user.drop('tmp')
>