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

[Imeev Mergen <imeevma@tarantool.org>]

Hello! Thanks for review!


On 09/18/2018 07:18 PM, Vladislav Shpilevoy wrote:
> Hi! Thanks for the patch! See 12 comments, my review
> fixes on the branch and at the end of the email.
>
> On 07/09/2018 21:04, imeevma@tarantool.org wrote:
>> Some creation operations create objects even on fail. This is
>> wrong and should be fixed. This patch adds destructors for such
>> objects.
>>
>> Closes #3592
>> ---
>> Branch: 
>> https://github.com/tarantool/tarantool/tree/imeevma/gh-3592-cleanup-on-failed-create
>> Issue: https://github.com/tarantool/tarantool/issues/3592
>>
>>   src/box/sql/build.c          | 109 
>> ++++++++++++++++++++++++++++++++++++++++++-
>>   src/box/sql/sqliteInt.h      |   4 ++
>>   src/box/sql/vdbe.c           |  55 +++++++++++++++++++---
>>   test/sql/drop-table.result   |  38 +++++++++++++++
>>   test/sql/drop-table.test.lua |  19 ++++++++
>>   5 files changed, 218 insertions(+), 7 deletions(-)
>>
>> diff --git a/src/box/sql/build.c b/src/box/sql/build.c
>> index a1e16b2..96d0624 100644
>> --- a/src/box/sql/build.c
>> +++ b/src/box/sql/build.c
>> @@ -54,6 +54,78 @@
>>   #include "box/schema.h"
>>   #include "box/tuple_format.h"
>>   #include "box/coll_id_cache.h"
>> +#include "fiber.h"
>> +
>> +/**
>> + * Element of list that contains some information about records
>> + * which were inserted into system spaces.
>> + */
>> +struct saved_records
>
> 1. One struct contains info about one record, it is not? So why
> 'saved_records' instead of 'saved_record'?
Fixed.
>
>> +{
>> +    /* A link in record list. */
>
> 2. Double ** for out of function comments.
Squashed.
>
>> +    struct rlist link;
>> +    /* Id of space in which record were inserted. */
>> +    int space_id;
>
> 3. Space id is uint32.
Squashed.
>
>> +    /* First register of key of record. */
>> +    int record_key;
>> +    /* Length of key of record. */
>> +    int record_key_len;
>
> 4. What is len? Strlen? Here it is number of registers as
> I understand.
Fixed.
>
>> +    /* Register with table name.*/
>> +    int record_name;
>> +    /* Flags of this element of list. */
>> +    int flag;
>> +};
>> +
>> +/**
>> + * List of all records that were inserted in system spaces in
>> + * current statement.
>> + */
>> +static struct rlist record_list;
>
> 5. Please, do not use any global variables to store query-specific
> information. It is a part of parser.
Done.
>
>> +/**
>> + * Flag to show if there were any saved records in current
>> + * statement.
>> + */
>> +static bool is_records_saved = false;
>
> 6. rlist_empty().
Done.
>
>> +
>> +/**
>> + * Save inserted in system space record in list saved_records.
>> + *
>> + * @param parser SQL Parser object.
>> + * @param space_id Id of table in which record is inserted.
>> + * @param record_key Register that contains first field of key.
>> + * @param record_key_len Exact number of fields of key.
>> + * @param record_name Register contains name of table which were
>> + *        created when record was inserted into box.space._space.
>> + * @param flag OPFLAG_NCHANGE or 0.
>
> 7. Why do you need any flags for rollback of a statement? As I
> understand, if a DDL statement fails, a user will receive only
> error. No meta is supposed to be in a response, so NCHANGE (flag
> to increment meta rowcount) is not needed, it is not?
>
> If you bother that this counter will affect next requests - it
> will not. The counter is reset on each new statement.
Done.
>
>> + * @retval -1 memory error.
>> + * @retval 0 success.
>> + */
>> +static inline void
>> +save_record(Parse *parser, int space_id, int record_key, int 
>> record_key_len,
>> +        int record_name, int flag)
>> +{
>> +    if (!is_records_saved) {
>> +        rlist_create(&record_list);
>
> 8. Just create it always together with parser.
Done.
>
>> +        is_records_saved = true;
>> +    }
>> +    struct saved_records *saved_records =
>> +        region_alloc(&fiber()->gc, sizeof(*saved_records));
>> +    if (saved_records == NULL) {
>> +        diag_set(OutOfMemory, sizeof(*saved_records), "region_alloc",
>> +             "saved_records");
>> +        parser->rc = SQL_TARANTOOL_ERROR;
>> +        parser->nErr++;
>> +        return;
>> +    }
>> +    saved_records->space_id = space_id;
>> +    saved_records->record_key = record_key;
>> +    saved_records->record_key_len = record_key_len;
>> +    saved_records->record_name = record_name;
>> +    saved_records->flag = flag | OPFLAG_DESTRUCTOR;
>> +    if (record_name != 0)
>> +        saved_records->flag |= OPFLAG_CLEAR_HASH;
>> +    rlist_add_entry(&record_list, saved_records, link);
>> +}
>>     void
>>   sql_finish_coding(struct Parse *parse_context)
>> diff --git a/src/box/sql/sqliteInt.h b/src/box/sql/sqliteInt.h
>> index 1d32c9a..fc479b1 100644
>> --- a/src/box/sql/sqliteInt.h
>> +++ b/src/box/sql/sqliteInt.h
>> @@ -2843,6 +2843,10 @@ struct Parse {
>>   #define OPFLAG_PERMUTE       0x01    /* OP_Compare: use the 
>> permutation */
>>   #define OPFLAG_SAVEPOSITION  0x02    /* OP_Delete: keep cursor 
>> position */
>>   #define OPFLAG_AUXDELETE     0x04    /* OP_Delete: index in a 
>> DELETE op */
>> +/* OP_SDelete: work as destructor. */
>> +#define OPFLAG_DESTRUCTOR    0x02
>> +/* OP_SDelete: remove table from tblHash. */
>
> 9. Forget about this flag. It is not needed. You can do
> whatever you want with rowcount if a result is error anyway.
Done.
>
>> +#define OPFLAG_CLEAR_HASH    0x04
>>     #define OPFLAG_SAME_FRAME    0x01    /* OP_FCopy: use same frame 
>> for source
>>                        * register
>> diff --git a/src/box/sql/vdbe.c b/src/box/sql/vdbe.c
>> index a5b907d..595d7d4 100644
>> --- a/src/box/sql/vdbe.c
>> +++ b/src/box/sql/vdbe.c
>> @@ -610,6 +610,17 @@ int sqlite3VdbeExec(Vdbe *p)
>>       u64 start;                 /* CPU clock count at start of 
>> opcode */
>>   #endif
>>       struct session *user_session = current_session();
>> +    /*
>> +     * Opcode to where we should jump on error in OP_SInsert.
>> +     */
>> +    int op_clean_on_error = p->nOp - 1;
>> +    /*
>> +     * On error in OP_SInsert rc is saved here before
>> +     * clean-up. After clean-up this values is retrieved and
>> +     * rc set. If error happened during clean-up this value
>> +     * neglected.
>> +     */
>> +    int saved_rc = 0;
>
> 10. As I remember, we've decided to use jump (like goto jump_to_p2)
> from sinsert to start iteration through destructors and OP_Halt to
> finish them. Please, do not change or use OP_Gosub and do not
> introduce OP_Error. Use OP_Halt only and jumps. The vdbe code should
> look like:
>
>     insert into _space, on error jump to p3/p2 = ::error_N::
>     insert into _index, on error jump to p3/p2 = ::error_N-1::
>     ...
>     success op_halt
> ::error1::
>     process error 1 (sdelete, etc)
> ::error2::
>     process error 2
>     ...
> ::errorN::
>     op_halt with error
>
> Also I see that you add OP_Error always, even if there is nothing
> to delete (in sql_finish_coding) - why?
>
> You can argue that you do not know error opcode numbers before you
> have created them to do jumps to p2/p3 but you could generate
> dummy jump targets during op_sinsert creation, remember those
> opcode numbers in saved_records list and change their p3/p2 during
> error section generation when you already know error opcode numbers.
Done. To make it work this way I added some functionality on
register P3 of OP_Halt operation. I think it would be too complicated
to do it another way.
>
> 11. You do not need to save rc. It is always SQL_TARANTOOL_ERROR
> (if it is not, then we have another bug that should be fixed).
>
>>       /*** INSERT STACK UNION HERE ***/
>>         assert(p->magic==VDBE_MAGIC_RUN);  /* sqlite3_step() verifies 
>> this */> diff --git a/test/sql/drop-table.result 
>> b/test/sql/drop-table.result
>> index 08f2496..e0349cc 100644
>> --- a/test/sql/drop-table.result
>> +++ b/test/sql/drop-table.result
>> @@ -33,3 +33,41 @@ 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.
>> +--
>> +box.schema.user.create('tmp')
>> +---
>> +...
>> +box.schema.user.grant('tmp', 'create', 'space')
>> +---
>> +...
>> +box.schema.user.grant('tmp', 'write', 'space', '_space')
>> +---
>> +...
>> +box.schema.user.grant('tmp', 'write', 'space', '_schema')
>> +---
>> +...
>> +box.session.su('tmp')
Fixed.
>
> 12. What is going on here? Why it is so complex? Why the previous
> lines in the same file work without a special user and rights?
>
>> +---
>> +...
>> +--
>> +-- 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')
>> +---
>> +...
Discussed, comment added.
>
> =====================================================================
>
> diff --git a/src/box/sql/build.c b/src/box/sql/build.c
> index 96d062440..1db65dcfe 100644
> --- a/src/box/sql/build.c
> +++ b/src/box/sql/build.c
> @@ -62,17 +62,17 @@
>   */
>  struct saved_records
>  {
> -    /* A link in record list. */
> +    /** A link in a record list. */
>      struct rlist link;
> -    /* Id of space in which record were inserted. */
> -    int space_id;
> -    /* First register of key of record. */
> +    /** Id of space in which the record was inserted. */
> +    uint32_t space_id;
> +    /** First register of key of the record. */
>      int record_key;
> -    /* Length of key of record. */
> +    /** Length of key of the record. */
>      int record_key_len;
> -    /* Register with table name.*/
> +    /** Register with table name.*/
>      int record_name;
> -    /* Flags of this element of list. */
> +    /** Flags of this element of list. */
>      int flag;
>  };
>
> @@ -101,8 +101,8 @@ static bool is_records_saved = false;
>   * @retval 0 success.
>   */
>  static inline void
> -save_record(Parse *parser, int space_id, int record_key, int 
> record_key_len,
> -        int record_name, int flag)
> +save_record(struct Parse *parser, uint32_t space_id, int record_key,
> +        int record_key_len, int record_name, int flag)
>  {
>      if (!is_records_saved) {
>          rlist_create(&record_list);

commit 1f18fa692305f28891f0dbb87e74e863e0c002c3
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..028a1c5 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 SInsert operation. */
+    int nOp;
+};
+
+/**
+ * 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 nOp_after The number of operation that follows SInsert.
+ */
+static inline void
+save_record(struct Parse *parser, uint32_t space_id, int reg_key,
+        int reg_key_count, int nOp_after)
+{
+    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->nOp = nOp_after - 1;
+    rlist_add_entry(&parser->record_list, record, link);
+}

  void
  sql_finish_coding(struct Parse *parse_context)
@@ -62,6 +110,22 @@ 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.
+     */
+    while (rlist_empty(&parse_context->record_list) == 0) {
+        int record_reg = ++parse_context->nMem;
+        struct saved_record *record =
+            rlist_shift_entry(&parse_context->record_list,
+                      struct saved_record, link);
+        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->nOp, v->nOp);
+    }
+    sqlite3VdbeAddOp3(v, OP_Halt, SQL_TARANTOOL_ERROR, 0, 1);
      if (db->mallocFailed || parse_context->nErr != 0) {
          if (parse_context->rc == SQLITE_OK)
              parse_context->rc = SQLITE_ERROR;
@@ -1101,13 +1165,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);
      return;
  error:
      parse->rc = SQL_TARANTOOL_ERROR;
@@ -1165,8 +1230,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);
      return;
  error:
      pParse->nErr++;
@@ -1340,9 +1406,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);
      sqlite3ReleaseTempRange(parse_context, constr_tuple_reg, 10);
      return;
  error:
@@ -1487,14 +1555,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);
          /* 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);
      }
      /* 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..c90cf1c 100644
--- a/src/box/sql/vdbe.c
+++ b/src/box/sql/vdbe.c
@@ -943,7 +943,7 @@ case OP_HaltIfNull: {      /* in3 */
      FALLTHROUGH;
  }

-/* Opcode:  Halt P1 P2 * P4 P5
+/* Opcode:  Halt P1 P2 P3 P4 P5
   *
   * Exit immediately.  All open cursors, etc are closed
   * automatically.
@@ -951,7 +951,9 @@ case OP_HaltIfNull: {      /* in3 */
   * P1 is the result code returned by sqlite3_exec(),
   * sqlite3_reset(), or sqlite3_finalize().  For a normal halt,
   * this should be SQLITE_OK (0).
- * For errors, it can be some other value.  If P1!=0 then P2 will
+ * For errors, it can be some other value.
+ * If P1 == SQL_TARANTOOL_ERROR and P3 != 0 then just go to
+ * abort_due_to_error. If P1!=0 and P3 == 0 then P2 will
   * determine whether or not to rollback the current transaction.
   * Do not rollback if P2==ON_CONFLICT_ACTION_FAIL. Do the rollback
   * if P2==ON_CONFLICT_ACTION_ROLLBACK.  If
@@ -1005,6 +1007,10 @@ case OP_Halt: {
          pOp = &aOp[pcx];
          break;
      }
+    if (pOp->p1 == SQL_TARANTOOL_ERROR && pOp->p3 != 0) {
+        rc = SQL_TARANTOOL_ERROR;
+        goto abort_due_to_error;
+    }
      p->rc = pOp->p1;
      p->errorAction = (u8)pOp->p2;
      p->pc = pcx;
@@ -4318,8 +4324,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 +4338,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')