SQLite3 源码分析-数据插入

前记

在这一篇内容中,我们来看一下 insert into 语句是如何工作的,这里我们先给出了 insert into test values(88, 1.88, "Hello World") 的 vdbe 吗,因为我们在前一篇中已经讲述了关于 SQLite3 数据格式的相关内容,因此我们在这一章中讲述数据被插入之前是如何被组织的.

addr  opcode         p1    p2    p3    p4             p5  comment
----  -------------  ----  ----  ----  -------------  --  -------------
0     Init           0     9     0                    0   Start at 9
1     OpenWrite      0     2     0     3              0   root=2 iDb=0; test
2     Integer        1     2     0                    0   r[2]=1
3     Real           0     3     0     1.88           0   r[3]=1.88
4     String8        0     4     0     Hello World    0   r[4]='Hello World'
5     NewRowid       0     1     0                    0   r[1]=rowid
6     MakeRecord     2     3     5     DEB            0   r[5]=mkrec(r[2..4])
7     Insert         0     5     1     test           57  intkey=r[1] data=r[5]
8     Halt           0     0     0                    0
9     Transaction    0     1     1     0              1   usesStmtJournal=0
10    TableLock      0     2     1     test           0   iDb=0 root=2 write=1
11    Goto           0     1     0                    0

代码分析

insert into test values(88, 1.88, "Hello World")

cmd ::= with insert_cmd(R) INTO xfullname(X) idlist_opt(F) select(S)
        upsert(U). {
  sqlite3Insert(pParse, X, S, F, R, U);
}

select(A) ::= selectnowith(X). {
  Select *p = X;
  if( p ){
    parserDoubleLinkSelect(pParse, p);
  }
  A = p; /*A-overwrites-X*/
}

selectnowith(A) ::= oneselect(A).

oneselect(A) ::= values(A).

values(A) ::= VALUES LP nexprlist(X) RP. {
  A = sqlite3SelectNew(pParse,X,0,0,0,0,0,SF_Values,0);
}

nexprlist(A) ::= nexprlist(A) COMMA expr(Y).
    {A = sqlite3ExprListAppend(pParse,A,Y);}

我们可以看一下这个解析的过程，首先会进入到 INSERT 这个规则，xfullname 匹配的值是 ‘test’，接下来就是 SELECT 表达式,这个是一个非终止符，然后继续向下寻找，一直找到 values 这个规则，我们可以看到这里有一个 LP 和 RP 就是我们的左括号和右括号了，其中的 nextprlist 就是我们的实际数据了。

所以实际的的执行流程为：

1. sqlite3ExprListAppend
2. sqlite3SelectNew
3. sqlite3Insert
4. sqlite3VdbeExec

我们依次来分析一下。

sqlite3ExprListAppend

ExprList *sqlite3ExprListAppend(
  Parse *pParse,          /* Parsing context */
  ExprList *pList,        /* List to which to append. Might be NULL */
  Expr *pExpr             /* Expression to be appended. Might be NULL */
){
  struct ExprList_item *pItem;
  if( pList==0 ){
    return sqlite3ExprListAppendNew(pParse->db,pExpr);
  }
  if( pList->nAlloc<pList->nExpr+1 ){
    return sqlite3ExprListAppendGrow(pParse->db,pList,pExpr);
  }
  pItem = &pList->a[pList->nExpr++];
  *pItem = zeroItem;
  pItem->pExpr = pExpr;
  return pList;
}

这个函数的实际作用就是将我们的数据转换成一个表达式，表达式的实际数据就存储在 ExprList_item,这里一共解析 3 次，第一个的时候 pList 为空， 需要给他初始化依次，在初始化的过程中会预先分配 4 个 ExprList_item , 这就是下边的如果当前申请的空间能够放得下 pItem 就不再申请新空间。

最后我们将 pItem 放置到柔性数组 a 中。

sqlite3SelectNew

/*
** Allocate a new Select structure and return a pointer to that
** structure.
*/
Select *sqlite3SelectNew(
  Parse *pParse,        /* Parsing context */
  ExprList *pEList,     /* which columns to include in the result */
  SrcList *pSrc,        /* the FROM clause -- which tables to scan */
  Expr *pWhere,         /* the WHERE clause */
  ExprList *pGroupBy,   /* the GROUP BY clause */
  Expr *pHaving,        /* the HAVING clause */
  ExprList *pOrderBy,   /* the ORDER BY clause */
  u32 selFlags,         /* Flag parameters, such as SF_Distinct */
  Expr *pLimit          /* LIMIT value.  NULL means not used */
){
  Select *pNew, *pAllocated;
  Select standin;
  pAllocated = pNew = sqlite3DbMallocRawNN(pParse->db, sizeof(*pNew) );
  if( pNew==0 ){
    assert( pParse->db->mallocFailed );
    pNew = &standin;
  }
  if( pEList==0 ){
    pEList = sqlite3ExprListAppend(pParse, 0,
                                   sqlite3Expr(pParse->db,TK_ASTERISK,0));
  }
  pNew->pEList = pEList;
  pNew->op = TK_SELECT;
  pNew->selFlags = selFlags;
  pNew->iLimit = 0;
  pNew->iOffset = 0;
  pNew->selId = ++pParse->nSelect;
  pNew->addrOpenEphm[0] = -1;
  pNew->addrOpenEphm[1] = -1;
  pNew->nSelectRow = 0;
  if( pSrc==0 ) pSrc = sqlite3DbMallocZero(pParse->db, sizeof(*pSrc));
  pNew->pSrc = pSrc;
  pNew->pWhere = pWhere;
  pNew->pGroupBy = pGroupBy;
  pNew->pHaving = pHaving;
  pNew->pOrderBy = pOrderBy;
  pNew->pPrior = 0;
  pNew->pNext = 0;
  pNew->pLimit = pLimit;
  pNew->pWith = 0;
#ifndef SQLITE_OMIT_WINDOWFUNC
  pNew->pWin = 0;
  pNew->pWinDefn = 0;
#endif
  if( pParse->db->mallocFailed ) {
    clearSelect(pParse->db, pNew, pNew!=&standin);
    pAllocated = 0;
  }else{
    assert( pNew->pSrc!=0 || pParse->nErr>0 );
  }
  return pAllocated;
}

这个函数的作用就是将我们刚刚生成的 ExprList 对象放置到一个 Select 对象中，并且将其他的成员赋值，比如 WHERE 条件，ORDER BY 排序等等。

sqlite3Insert

void sqlite3Insert(
  Parse *pParse,        /* Parser context */
  SrcList *pTabList,    /* Name of table into which we are inserting */
  Select *pSelect,      /* A SELECT statement to use as the data source */
  IdList *pColumn,      /* Column names corresponding to IDLIST, or NULL. */
  int onError,          /* How to handle constraint errors */
  Upsert *pUpsert       /* ON CONFLICT clauses for upsert, or NULL */
){
  sqlite3 *db;          /* The main database structure */
  Table *pTab;          /* The table to insert into.  aka TABLE */
  int i, j;             /* Loop counters */
  Vdbe *v;              /* Generate code into this virtual machine */
  Index *pIdx;          /* For looping over indices of the table */
  int nColumn;          /* Number of columns in the data */
  int nHidden = 0;      /* Number of hidden columns if TABLE is virtual */
  int iDataCur = 0;     /* VDBE cursor that is the main data repository */
  int iIdxCur = 0;      /* First index cursor */
  int ipkColumn = -1;   /* Column that is the INTEGER PRIMARY KEY */
  int endOfLoop;        /* Label for the end of the insertion loop */
  int srcTab = 0;       /* Data comes from this temporary cursor if >=0 */
  int addrInsTop = 0;   /* Jump to label "D" */
  int addrCont = 0;     /* Top of insert loop. Label "C" in templates 3 and 4 */
  SelectDest dest;      /* Destination for SELECT on rhs of INSERT */
  int iDb;              /* Index of database holding TABLE */
  u8 useTempTable = 0;  /* Store SELECT results in intermediate table */
  u8 appendFlag = 0;    /* True if the insert is likely to be an append */
  u8 withoutRowid;      /* 0 for normal table.  1 for WITHOUT ROWID table */
  u8 bIdListInOrder;    /* True if IDLIST is in table order */
  ExprList *pList = 0;  /* List of VALUES() to be inserted  */
  int iRegStore;        /* Register in which to store next column */

  /* Register allocations */
  int regFromSelect = 0;/* Base register for data coming from SELECT */
  int regAutoinc = 0;   /* Register holding the AUTOINCREMENT counter */
  int regRowCount = 0;  /* Memory cell used for the row counter */
  int regIns;           /* Block of regs holding rowid+data being inserted */
  int regRowid;         /* registers holding insert rowid */
  int regData;          /* register holding first column to insert */
  int *aRegIdx = 0;     /* One register allocated to each index */


  db = pParse->db;
  
  if( pParse->nErr ){
    goto insert_cleanup;
  }
  
  dest.iSDParm = 0;  /* Suppress a harmless compiler warning */

  if( pSelect && (pSelect->selFlags & SF_Values)!=0 && pSelect->pPrior==0 ){
    pList = pSelect->pEList;
    pSelect->pEList = 0;
    sqlite3SelectDelete(db, pSelect);
    pSelect = 0;
  }
  
  pTab = sqlite3SrcListLookup(pParse, pTabList);
  if( pTab==0 ){
    goto insert_cleanup;
  }
  iDb = sqlite3SchemaToIndex(db, pTab->pSchema);
  
  if( sqlite3AuthCheck(pParse, SQLITE_INSERT, pTab->zName, 0,
                       db->aDb[iDb].zDbSName) ){
    goto insert_cleanup;
  }
  withoutRowid = !HasRowid(pTab);
  
  if( sqlite3IsReadOnly(pParse, pTab, tmask) ){
    goto insert_cleanup;
  }
  v = sqlite3GetVdbe(pParse);
  if( v==0 ) goto insert_cleanup;
  if( pParse->nested==0 ) sqlite3VdbeCountChanges(v);
  sqlite3BeginWriteOperation(pParse, pSelect || pTrigger, iDb);

  regAutoinc = autoIncBegin(pParse, iDb, pTab);
  regRowid = regIns = pParse->nMem+1;
  pParse->nMem += pTab->nCol + 1;
  if( IsVirtual(pTab) ){
    regRowid++;
    pParse->nMem++;
  }
  regData = regRowid+1;


  bIdListInOrder = (pTab->tabFlags & (TF_OOOHidden|TF_HasStored))==0;
  
  if( pSelect ){
   
  }else{
    NameContext sNC;
    memset(&sNC, 0, sizeof(sNC));
    sNC.pParse = pParse;
    srcTab = -1;
    
    if( pList ){
      nColumn = pList->nExpr;
      if( sqlite3ResolveExprListNames(&sNC, pList) ){
        goto insert_cleanup;
      }
    }else{
      nColumn = 0;
    }
  }

  /* If this is not a view, open the table and and all indices */
  if( !isView ){
    int nIdx;
    nIdx = sqlite3OpenTableAndIndices(pParse, pTab, OP_OpenWrite, 0, -1, 0,
                                      &iDataCur, &iIdxCur);
    aRegIdx = sqlite3DbMallocRawNN(db, sizeof(int)*(nIdx+2));
    if( aRegIdx==0 ){
      goto insert_cleanup;
    }
    for(i=0, pIdx=pTab->pIndex; i<nIdx; pIdx=pIdx->pNext, i++){
      
      aRegIdx[i] = ++pParse->nMem;
      pParse->nMem += pIdx->nColumn;
    }
    aRegIdx[i] = ++pParse->nMem;  /* Register to store the table record */
  }



  /* This is the top of the main insertion loop */
  if( useTempTable ){
    addrInsTop = sqlite3VdbeAddOp1(v, OP_Rewind, srcTab); VdbeCoverage(v);
    addrCont = sqlite3VdbeCurrentAddr(v);
  }else if( pSelect ){
    sqlite3VdbeReleaseRegisters(pParse, regData, pTab->nCol, 0, 0);
    addrInsTop = addrCont = sqlite3VdbeAddOp1(v, OP_Yield, dest.iSDParm);
    VdbeCoverage(v);
    if( ipkColumn>=0 ){
      sqlite3VdbeAddOp2(v, OP_Copy, regFromSelect+ipkColumn, regRowid);
    }
  }

  nHidden = 0;
  iRegStore = regData;  
  for(i=0; i<pTab->nCol; i++, iRegStore++){
    int k;
    u32 colFlags;
    
    if( i==pTab->iPKey ){
      sqlite3VdbeAddOp1(v, OP_SoftNull, iRegStore);
      continue;
    }
    if( ((colFlags = pTab->aCol[i].colFlags) & COLFLAG_NOINSERT)!=0 ){
      nHidden++;
      if( (colFlags & COLFLAG_VIRTUAL)!=0 ){
        /* Virtual columns do not participate in OP_MakeRecord.  So back up
        ** iRegStore by one slot to compensate for the iRegStore++ in the
        ** outer for() loop */
        iRegStore--;
        continue;
      }else if( (colFlags & COLFLAG_STORED)!=0 ){
        if( tmask & TRIGGER_BEFORE ){
          sqlite3VdbeAddOp1(v, OP_SoftNull, iRegStore);
        }
        continue;
      }else if( pColumn==0 ){
        /* Hidden columns that are not explicitly named in the INSERT
        ** get there default value */
        sqlite3ExprCodeFactorable(pParse, 
            sqlite3ColumnExpr(pTab, &pTab->aCol[i]),
            iRegStore);
        continue;
      }
    }

    if( useTempTable ){
      sqlite3VdbeAddOp3(v, OP_Column, srcTab, k, iRegStore); 
    }else if( pSelect ){
      if( regFromSelect!=regData ){
        sqlite3VdbeAddOp2(v, OP_SCopy, regFromSelect+k, iRegStore);
      }
    }else{
      Expr *pX = pList->a[k].pExpr;
      int y = sqlite3ExprCodeTarget(pParse, pX, iRegStore);
      if( y!=iRegStore ){
        sqlite3VdbeAddOp2(v,
          ExprHasProperty(pX, EP_Subquery) ? OP_Copy : OP_SCopy, y, iRegStore);
      }
    }
  }

  endOfLoop = sqlite3VdbeMakeLabel(pParse);
  if( tmask & TRIGGER_BEFORE ){
    int regCols = sqlite3GetTempRange(pParse, pTab->nCol+1);
    if( ipkColumn<0 ){
      sqlite3VdbeAddOp2(v, OP_Integer, -1, regCols);
    }else{
      int addr1;
      
      if( useTempTable ){
        sqlite3VdbeAddOp3(v, OP_Column, srcTab, ipkColumn, regCols);
      }else{
        
        sqlite3ExprCode(pParse, pList->a[ipkColumn].pExpr, regCols);
      }
      addr1 = sqlite3VdbeAddOp1(v, OP_NotNull, regCols); VdbeCoverage(v);
      sqlite3VdbeAddOp2(v, OP_Integer, -1, regCols);
      sqlite3VdbeJumpHere(v, addr1);
      sqlite3VdbeAddOp1(v, OP_MustBeInt, regCols); VdbeCoverage(v);
    }

    /* Copy the new data already generated. */
    
    sqlite3VdbeAddOp3(v, OP_Copy, regRowid+1, regCols+1, pTab->nNVCol-1);

    if( !isView ){
      sqlite3TableAffinity(v, pTab, regCols+1);
    }

    /* Fire BEFORE or INSTEAD OF triggers */
    sqlite3CodeRowTrigger(pParse, pTrigger, TK_INSERT, 0, TRIGGER_BEFORE, 
        pTab, regCols-pTab->nCol-1, onError, endOfLoop);

    sqlite3ReleaseTempRange(pParse, regCols, pTab->nCol+1);
  }

  if( !isView ){
    if( IsVirtual(pTab) ){
      /* The row that the VUpdate opcode will delete: none */
      sqlite3VdbeAddOp2(v, OP_Null, 0, regIns);
    }
    if( ipkColumn>=0 ){
      /* Compute the new rowid */
      if( useTempTable ){
        sqlite3VdbeAddOp3(v, OP_Column, srcTab, ipkColumn, regRowid);
      }else if( pSelect ){
        /* Rowid already initialized at tag-20191021-001 */
      }else{
        Expr *pIpk = pList->a[ipkColumn].pExpr;
        if( pIpk->op==TK_NULL && !IsVirtual(pTab) ){
          sqlite3VdbeAddOp3(v, OP_NewRowid, iDataCur, regRowid, regAutoinc);
          appendFlag = 1;
        }else{
          sqlite3ExprCode(pParse, pList->a[ipkColumn].pExpr, regRowid);
        }
      }
    }
  }
insert_cleanup:
  sqlite3SrcListDelete(db, pTabList);
  sqlite3ExprListDelete(db, pList);
  sqlite3UpsertDelete(db, pUpsert);
  sqlite3SelectDelete(db, pSelect);
  sqlite3IdListDelete(db, pColumn);
  if( aRegIdx ) sqlite3DbNNFreeNN(db, aRegIdx);
}

插入函数代码行数比较多，我们主要对我们的这个插入语句做一个说明：

在函数一开始的时候，pSelect 这个就是我们 sqlite3SelectNew 函数中生成的 Select 对象，因为我们这里仅仅是一个简单的 VALUES 这种形式的结构，所以这里直接将其中的 ExprList 获取出来，然后保存到 pList 中。

然后我们就使用 sqlite3SrcListLookup 来获取我们要插入数据的表。使用 sqlite3SchemaToIndex 来找到表所在的数据库。

从这里开始我们就获取了虚拟机 v 开始添加指令了。

首先我们使用 sqlite3OpenTableAndIndices 这个函数打开数据表机器索引。这里会添加一个 OpenWrite 的虚拟机指令。

接下来，我们进入一个 for 循环中，在这里我们会对我们数据添加虚拟机指令，在 sqlite3ExprCodeTarget 函数中，我们将我们实际的 Expr 转换成对应的 Integer,Real,String8 指令。

现在我们的数据已经准备好了，还差一个 rowid，那么就插入一个 NewRowid 指令，现在所有的数据都已具备，我们就可以使用 MakeRecord 指令来构造实际存储到文件中的二进制数据，这个后续可以参考 vdbe 执行的代码。

最后就是插入数据，结束虚拟机运行。

sqlite3VdbeExec

在这个函数中我们主要来看一下 MakeRecord 的相关代码：

  ** ------------------------------------------------------------------------
  ** | hdr-size | type 0 | type 1 | ... | type N-1 | data0 | ... | data N-1 | 
  ** ------------------------------------------------------------------------

这个函数的主要作用就是将我们的数据转换成可以存储到数据库中的二进制数据。他的格式可以参考一下后续的文章。

后记

慢慢的已经对sqlite3 的代码有一点点点简单的理解。