ADDED ext/session/sessionrebase.test Index: ext/session/sessionrebase.test ================================================================== --- ext/session/sessionrebase.test +++ ext/session/sessionrebase.test @@ -0,0 +1,125 @@ +# 2018 March 14 +# +# The author disclaims copyright to this source code. In place of +# a legal notice, here is a blessing: +# +# May you do good and not evil. +# May you find forgiveness for yourself and forgive others. +# May you share freely, never taking more than you give. +# +#*********************************************************************** +# This file implements regression tests for SQLite library. +# + +if {![info exists testdir]} { + set testdir [file join [file dirname [info script]] .. .. test] +} +source [file join [file dirname [info script]] session_common.tcl] +source $testdir/tester.tcl +ifcapable !session {finish_test; return} + +set testprefix sessionrebase + +set ::lConflict [list] +proc xConflict {args} { + set res [lindex $::lConflict 0] + set ::lConflict [lrange $::lConflict 1 end] + return $res +} + +#------------------------------------------------------------------------- +# The following test cases - 1.* - test that the rebase blobs output by +# sqlite3_changeset_apply_v2 look correct in some simple cases. The blob +# is itself a changeset, containing records determined as follows: +# +# * For each conflict resolved with REPLACE, the rebase blob contains +# a DELETE record. All fields other than the PK fields are undefined. +# +# * For each conflict resolved with OMIT, the rebase blob contains an +# INSERT record. For an INSERT or UPDATE operation, the indirect flag +# is clear and all updated fields are defined. For a DELETE operation, +# the indirect flag is set and all non-PK fields left undefined. +# +proc do_apply_v2_test {tn sql modsql conflict_handler res} { + + execsql BEGIN + sqlite3session S db main + S attach * + execsql $sql + set changeset [S changeset] + S delete + execsql ROLLBACK + + execsql BEGIN + execsql $modsql + set ::lConflict $conflict_handler + set blob [sqlite3changeset_apply_v2 db $changeset xConflict] + execsql ROLLBACK + + uplevel [list do_test $tn [list changeset_to_list $blob] [list {*}$res]] +} + +do_execsql_test 1.0 { + CREATE TABLE t1(a INTEGER PRIMARY KEY, b); + INSERT INTO t1 VALUES(1, 'value A'); +} + +do_apply_v2_test 1.1.1 { + UPDATE t1 SET b = 'value B' WHERE a=1; +} { + UPDATE t1 SET b = 'value C' WHERE a=1; +} { + OMIT +} { + {INSERT t1 0 X. {} {i 1 t {value B}}} +} +do_apply_v2_test 1.1.2 { + UPDATE t1 SET b = 'value B' WHERE a=1; +} { + UPDATE t1 SET b = 'value C' WHERE a=1; +} { + REPLACE +} { + {DELETE t1 0 X. {i 1 {} {}} {}} +} + +do_apply_v2_test 1.2.1 { + INSERT INTO t1 VALUES(2, 'first'); +} { + INSERT INTO t1 VALUES(2, 'second'); +} { + OMIT +} { + {INSERT t1 0 X. {} {i 2 t first}} +} +do_apply_v2_test 1.2.2 { + INSERT INTO t1 VALUES(2, 'first'); +} { + INSERT INTO t1 VALUES(2, 'second'); +} { + REPLACE +} { + {DELETE t1 0 X. {i 2 {} {}} {}} +} + +do_apply_v2_test 1.3.1 { + DELETE FROM t1 WHERE a=1; +} { + UPDATE t1 SET b='value D' WHERE a=1; +} { + OMIT +} { + {INSERT t1 1 X. {} {i 1 {} {}}} +} +do_apply_v2_test 1.3.2 { + DELETE FROM t1 WHERE a=1; +} { + UPDATE t1 SET b='value D' WHERE a=1; +} { + REPLACE +} { + {DELETE t1 0 X. {i 1 {} {}} {}} +} + + +finish_test Index: ext/session/sqlite3session.c ================================================================== --- ext/session/sqlite3session.c +++ ext/session/sqlite3session.c @@ -3408,10 +3408,12 @@ const char **azCol; /* Array of column names */ u8 *abPK; /* Boolean array - true if column is in PK */ int bStat1; /* True if table is sqlite_stat1 */ int bDeferConstraints; /* True to defer constraints */ SessionBuffer constraints; /* Deferred constraints are stored here */ + SessionBuffer rebase; /* Rebase information (if any) here */ + int bRebaseStarted; /* If table header is already in rebase */ }; /* ** Formulate a statement to DELETE a row from database db. Assuming a table ** structure like this: @@ -3788,10 +3790,64 @@ if( rc!=SQLITE_ROW ) rc = sqlite3_reset(pSelect); } return rc; } + +static int sessionRebaseAdd( + SessionApplyCtx *p, + int eType, + sqlite3_changeset_iter *pIter +){ + int rc = SQLITE_OK; + int i; + int eOp = pIter->op; + if( p->bRebaseStarted==0 ){ + /* Append a table-header to the rebase buffer */ + const char *zTab = pIter->zTab; + sessionAppendByte(&p->rebase, 'T', &rc); + sessionAppendVarint(&p->rebase, p->nCol, &rc); + sessionAppendBlob(&p->rebase, p->abPK, p->nCol, &rc); + sessionAppendBlob(&p->rebase, (u8*)zTab, (int)strlen(zTab)+1, &rc); + p->bRebaseStarted = 1; + } + + assert( eType==SQLITE_CHANGESET_REPLACE||eType==SQLITE_CHANGESET_OMIT ); + assert( eOp==SQLITE_DELETE || eOp==SQLITE_INSERT || eOp==SQLITE_UPDATE ); + + if( eType==SQLITE_CHANGESET_REPLACE ){ + sessionAppendByte(&p->rebase, SQLITE_DELETE, &rc); + sessionAppendByte(&p->rebase, 0, &rc); + for(i=0; inCol; i++){ + if( p->abPK[i]==0 ){ + sessionAppendByte(&p->rebase, 0, &rc); + }else{ + sqlite3_value *pVal = 0; + if( eOp==SQLITE_INSERT ){ + sqlite3changeset_new(pIter, i, &pVal); + }else{ + sqlite3changeset_old(pIter, i, &pVal); + } + sessionAppendValue(&p->rebase, pVal, &rc); + } + } + }else{ + sessionAppendByte(&p->rebase, SQLITE_INSERT, &rc); + sessionAppendByte(&p->rebase, eOp==SQLITE_DELETE, &rc); + for(i=0; inCol; i++){ + sqlite3_value *pVal = 0; + if( eOp!=SQLITE_INSERT && p->abPK[i] ){ + sqlite3changeset_old(pIter, i, &pVal); + }else{ + sqlite3changeset_new(pIter, i, &pVal); + } + sessionAppendValue(&p->rebase, pVal, &rc); + } + } + + return rc; +} /* ** Invoke the conflict handler for the change that the changeset iterator ** currently points to. ** @@ -3864,11 +3920,11 @@ /* Instead of invoking the conflict handler, append the change blob ** to the SessionApplyCtx.constraints buffer. */ u8 *aBlob = &pIter->in.aData[pIter->in.iCurrent]; int nBlob = pIter->in.iNext - pIter->in.iCurrent; sessionAppendBlob(&p->constraints, aBlob, nBlob, &rc); - res = SQLITE_CHANGESET_OMIT; + return SQLITE_OK; }else{ /* No other row with the new.* primary key. */ res = xConflict(pCtx, eType+1, pIter); if( res==SQLITE_CHANGESET_REPLACE ) rc = SQLITE_MISUSE; } @@ -3890,10 +3946,13 @@ default: rc = SQLITE_MISUSE; break; } + if( rc==SQLITE_OK ){ + rc = sessionRebaseAdd(p, res, pIter); + } } return rc; } @@ -4065,46 +4124,46 @@ int bReplace = 0; int bRetry = 0; int rc; rc = sessionApplyOneOp(pIter, pApply, xConflict, pCtx, &bReplace, &bRetry); - assert( rc==SQLITE_OK || (bRetry==0 && bReplace==0) ); - - /* If the bRetry flag is set, the change has not been applied due to an - ** SQLITE_CHANGESET_DATA problem (i.e. this is an UPDATE or DELETE and - ** a row with the correct PK is present in the db, but one or more other - ** fields do not contain the expected values) and the conflict handler - ** returned SQLITE_CHANGESET_REPLACE. In this case retry the operation, - ** but pass NULL as the final argument so that sessionApplyOneOp() ignores - ** the SQLITE_CHANGESET_DATA problem. */ - if( bRetry ){ - assert( pIter->op==SQLITE_UPDATE || pIter->op==SQLITE_DELETE ); - rc = sessionApplyOneOp(pIter, pApply, xConflict, pCtx, 0, 0); - } - - /* If the bReplace flag is set, the change is an INSERT that has not - ** been performed because the database already contains a row with the - ** specified primary key and the conflict handler returned - ** SQLITE_CHANGESET_REPLACE. In this case remove the conflicting row - ** before reattempting the INSERT. */ - else if( bReplace ){ - assert( pIter->op==SQLITE_INSERT ); - rc = sqlite3_exec(db, "SAVEPOINT replace_op", 0, 0, 0); - if( rc==SQLITE_OK ){ - rc = sessionBindRow(pIter, - sqlite3changeset_new, pApply->nCol, pApply->abPK, pApply->pDelete); - sqlite3_bind_int(pApply->pDelete, pApply->nCol+1, 1); - } - if( rc==SQLITE_OK ){ - sqlite3_step(pApply->pDelete); - rc = sqlite3_reset(pApply->pDelete); - } - if( rc==SQLITE_OK ){ + if( rc==SQLITE_OK ){ + /* If the bRetry flag is set, the change has not been applied due to an + ** SQLITE_CHANGESET_DATA problem (i.e. this is an UPDATE or DELETE and + ** a row with the correct PK is present in the db, but one or more other + ** fields do not contain the expected values) and the conflict handler + ** returned SQLITE_CHANGESET_REPLACE. In this case retry the operation, + ** but pass NULL as the final argument so that sessionApplyOneOp() ignores + ** the SQLITE_CHANGESET_DATA problem. */ + if( bRetry ){ + assert( pIter->op==SQLITE_UPDATE || pIter->op==SQLITE_DELETE ); rc = sessionApplyOneOp(pIter, pApply, xConflict, pCtx, 0, 0); } - if( rc==SQLITE_OK ){ - rc = sqlite3_exec(db, "RELEASE replace_op", 0, 0, 0); + + /* If the bReplace flag is set, the change is an INSERT that has not + ** been performed because the database already contains a row with the + ** specified primary key and the conflict handler returned + ** SQLITE_CHANGESET_REPLACE. In this case remove the conflicting row + ** before reattempting the INSERT. */ + else if( bReplace ){ + assert( pIter->op==SQLITE_INSERT ); + rc = sqlite3_exec(db, "SAVEPOINT replace_op", 0, 0, 0); + if( rc==SQLITE_OK ){ + rc = sessionBindRow(pIter, + sqlite3changeset_new, pApply->nCol, pApply->abPK, pApply->pDelete); + sqlite3_bind_int(pApply->pDelete, pApply->nCol+1, 1); + } + if( rc==SQLITE_OK ){ + sqlite3_step(pApply->pDelete); + rc = sqlite3_reset(pApply->pDelete); + } + if( rc==SQLITE_OK ){ + rc = sessionApplyOneOp(pIter, pApply, xConflict, pCtx, 0, 0); + } + if( rc==SQLITE_OK ){ + rc = sqlite3_exec(db, "RELEASE replace_op", 0, 0, 0); + } } } return rc; } @@ -4176,11 +4235,12 @@ int(*xConflict)( void *pCtx, /* Copy of fifth arg to _apply() */ int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */ sqlite3_changeset_iter *p /* Handle describing change and conflict */ ), - void *pCtx /* First argument passed to xConflict */ + void *pCtx, /* First argument passed to xConflict */ + void **ppRebase, int *pnRebase /* OUT: Rebase information */ ){ int schemaMismatch = 0; int rc; /* Return code */ const char *zTab = 0; /* Name of current table */ int nTab = 0; /* Result of sqlite3Strlen30(zTab) */ @@ -4217,10 +4277,11 @@ sqlite3_finalize(sApply.pInsert); sqlite3_finalize(sApply.pSelect); memset(&sApply, 0, sizeof(sApply)); sApply.db = db; sApply.bDeferConstraints = 1; + sApply.bRebaseStarted = 0; /* If an xFilter() callback was specified, invoke it now. If the ** xFilter callback returns zero, skip this table. If it returns ** non-zero, proceed. */ schemaMismatch = (xFilter && (0==xFilter(pCtx, zNew))); @@ -4326,19 +4387,51 @@ }else{ sqlite3_exec(db, "ROLLBACK TO changeset_apply", 0, 0, 0); sqlite3_exec(db, "RELEASE changeset_apply", 0, 0, 0); } + if( rc==SQLITE_OK && ppRebase && pnRebase ){ + *ppRebase = (void*)sApply.rebase.aBuf; + *pnRebase = sApply.rebase.nBuf; + sApply.rebase.aBuf = 0; + } sqlite3_finalize(sApply.pInsert); sqlite3_finalize(sApply.pDelete); sqlite3_finalize(sApply.pUpdate); sqlite3_finalize(sApply.pSelect); sqlite3_free((char*)sApply.azCol); /* cast works around VC++ bug */ sqlite3_free((char*)sApply.constraints.aBuf); + sqlite3_free((char*)sApply.rebase.aBuf); sqlite3_mutex_leave(sqlite3_db_mutex(db)); return rc; } + +int sqlite3changeset_apply_v2( + sqlite3 *db, /* Apply change to "main" db of this handle */ + int nChangeset, /* Size of changeset in bytes */ + void *pChangeset, /* Changeset blob */ + int(*xFilter)( + void *pCtx, /* Copy of sixth arg to _apply() */ + const char *zTab /* Table name */ + ), + int(*xConflict)( + void *pCtx, /* Copy of sixth arg to _apply() */ + int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */ + sqlite3_changeset_iter *p /* Handle describing change and conflict */ + ), + void *pCtx, /* First argument passed to xConflict */ + void **ppRebase, int *pnRebase +){ + sqlite3_changeset_iter *pIter; /* Iterator to skip through changeset */ + int rc = sqlite3changeset_start(&pIter, nChangeset, pChangeset); + if( rc==SQLITE_OK ){ + rc = sessionChangesetApply( + db, pIter, xFilter, xConflict, pCtx, ppRebase, pnRebase + ); + } + return rc; +} /* ** Apply the changeset passed via pChangeset/nChangeset to the main database ** attached to handle "db". Invoke the supplied conflict handler callback ** to resolve any conflicts encountered while applying the change. @@ -4356,23 +4449,45 @@ int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */ sqlite3_changeset_iter *p /* Handle describing change and conflict */ ), void *pCtx /* First argument passed to xConflict */ ){ - sqlite3_changeset_iter *pIter; /* Iterator to skip through changeset */ - int rc = sqlite3changeset_start(&pIter, nChangeset, pChangeset); - if( rc==SQLITE_OK ){ - rc = sessionChangesetApply(db, pIter, xFilter, xConflict, pCtx); - } - return rc; + return sqlite3changeset_apply_v2( + db, nChangeset, pChangeset, xFilter, xConflict, pCtx, 0, 0 + ); } /* ** Apply the changeset passed via xInput/pIn to the main database ** attached to handle "db". Invoke the supplied conflict handler callback ** to resolve any conflicts encountered while applying the change. */ +int sqlite3changeset_apply_v2_strm( + sqlite3 *db, /* Apply change to "main" db of this handle */ + int (*xInput)(void *pIn, void *pData, int *pnData), /* Input function */ + void *pIn, /* First arg for xInput */ + int(*xFilter)( + void *pCtx, /* Copy of sixth arg to _apply() */ + const char *zTab /* Table name */ + ), + int(*xConflict)( + void *pCtx, /* Copy of sixth arg to _apply() */ + int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */ + sqlite3_changeset_iter *p /* Handle describing change and conflict */ + ), + void *pCtx, /* First argument passed to xConflict */ + void **ppRebase, int *pnRebase +){ + sqlite3_changeset_iter *pIter; /* Iterator to skip through changeset */ + int rc = sqlite3changeset_start_strm(&pIter, xInput, pIn); + if( rc==SQLITE_OK ){ + rc = sessionChangesetApply( + db, pIter, xFilter, xConflict, pCtx, ppRebase, pnRebase + ); + } + return rc; +} int sqlite3changeset_apply_strm( sqlite3 *db, /* Apply change to "main" db of this handle */ int (*xInput)(void *pIn, void *pData, int *pnData), /* Input function */ void *pIn, /* First arg for xInput */ int(*xFilter)( @@ -4384,16 +4499,13 @@ int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */ sqlite3_changeset_iter *p /* Handle describing change and conflict */ ), void *pCtx /* First argument passed to xConflict */ ){ - sqlite3_changeset_iter *pIter; /* Iterator to skip through changeset */ - int rc = sqlite3changeset_start_strm(&pIter, xInput, pIn); - if( rc==SQLITE_OK ){ - rc = sessionChangesetApply(db, pIter, xFilter, xConflict, pCtx); - } - return rc; + return sqlite3changeset_apply_v2_strm( + db, xInput, pIn, xFilter, xConflict, pCtx, 0, 0 + ); } /* ** sqlite3_changegroup handle. */ Index: ext/session/sqlite3session.h ================================================================== --- ext/session/sqlite3session.h +++ ext/session/sqlite3session.h @@ -1100,10 +1100,27 @@ int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */ sqlite3_changeset_iter *p /* Handle describing change and conflict */ ), void *pCtx /* First argument passed to xConflict */ ); + +int sqlite3changeset_apply_v2( + sqlite3 *db, /* Apply change to "main" db of this handle */ + int nChangeset, /* Size of changeset in bytes */ + void *pChangeset, /* Changeset blob */ + int(*xFilter)( + void *pCtx, /* Copy of sixth arg to _apply() */ + const char *zTab /* Table name */ + ), + int(*xConflict)( + void *pCtx, /* Copy of sixth arg to _apply() */ + int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */ + sqlite3_changeset_iter *p /* Handle describing change and conflict */ + ), + void *pCtx, /* First argument passed to xConflict */ + void **ppRebase, int *pnRebase +); /* ** CAPI3REF: Constants Passed To The Conflict Handler ** ** Values that may be passed as the second argument to a conflict-handler. @@ -1300,10 +1317,26 @@ void *pCtx, /* Copy of sixth arg to _apply() */ int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */ sqlite3_changeset_iter *p /* Handle describing change and conflict */ ), void *pCtx /* First argument passed to xConflict */ +); +int sqlite3changeset_apply_v2_strm( + sqlite3 *db, /* Apply change to "main" db of this handle */ + int (*xInput)(void *pIn, void *pData, int *pnData), /* Input function */ + void *pIn, /* First arg for xInput */ + int(*xFilter)( + void *pCtx, /* Copy of sixth arg to _apply() */ + const char *zTab /* Table name */ + ), + int(*xConflict)( + void *pCtx, /* Copy of sixth arg to _apply() */ + int eConflict, /* DATA, MISSING, CONFLICT, CONSTRAINT */ + sqlite3_changeset_iter *p /* Handle describing change and conflict */ + ), + void *pCtx, /* First argument passed to xConflict */ + void **ppRebase, int *pnRebase ); int sqlite3changeset_concat_strm( int (*xInputA)(void *pIn, void *pData, int *pnData), void *pInA, int (*xInputB)(void *pIn, void *pData, int *pnData), Index: ext/session/test_session.c ================================================================== --- ext/session/test_session.c +++ ext/session/test_session.c @@ -709,14 +709,12 @@ *pnData = nRet; return SQLITE_OK; } -/* -** sqlite3changeset_apply DB CHANGESET CONFLICT-SCRIPT ?FILTER-SCRIPT? -*/ -static int SQLITE_TCLAPI test_sqlite3changeset_apply( +static int SQLITE_TCLAPI testSqlite3changesetApply( + int bV2, void * clientData, Tcl_Interp *interp, int objc, Tcl_Obj *CONST objv[] ){ @@ -725,10 +723,12 @@ int rc; /* Return code from changeset_invert() */ void *pChangeset; /* Buffer containing changeset */ int nChangeset; /* Size of buffer aChangeset in bytes */ TestConflictHandler ctx; TestStreamInput sStr; + void *pRebase = 0; + int nRebase = 0; memset(&sStr, 0, sizeof(sStr)); sStr.nStream = test_tcl_integer(interp, SESSION_STREAM_TCL_VAR); if( objc!=4 && objc!=5 ){ @@ -746,13 +746,20 @@ ctx.pConflictScript = objv[3]; ctx.pFilterScript = objc==5 ? objv[4] : 0; ctx.interp = interp; if( sStr.nStream==0 ){ - rc = sqlite3changeset_apply(db, nChangeset, pChangeset, - (objc==5) ? test_filter_handler : 0, test_conflict_handler, (void *)&ctx - ); + if( bV2==0 ){ + rc = sqlite3changeset_apply(db, nChangeset, pChangeset, + (objc==5)?test_filter_handler:0, test_conflict_handler, (void *)&ctx + ); + }else{ + rc = sqlite3changeset_apply_v2(db, nChangeset, pChangeset, + (objc==5)?test_filter_handler:0, test_conflict_handler, (void *)&ctx, + &pRebase, &nRebase + ); + } }else{ sStr.aData = (unsigned char*)pChangeset; sStr.nData = nChangeset; rc = sqlite3changeset_apply_strm(db, testStreamInput, (void*)&sStr, (objc==5) ? test_filter_handler : 0, test_conflict_handler, (void *)&ctx @@ -759,14 +766,42 @@ ); } if( rc!=SQLITE_OK ){ return test_session_error(interp, rc, 0); + }else{ + Tcl_ResetResult(interp); + if( bV2 && pRebase ){ + Tcl_SetObjResult(interp, Tcl_NewByteArrayObj(pRebase, nRebase)); + } } - Tcl_ResetResult(interp); + sqlite3_free(pRebase); return TCL_OK; } + +/* +** sqlite3changeset_apply DB CHANGESET CONFLICT-SCRIPT ?FILTER-SCRIPT? +*/ +static int SQLITE_TCLAPI test_sqlite3changeset_apply( + void * clientData, + Tcl_Interp *interp, + int objc, + Tcl_Obj *CONST objv[] +){ + return testSqlite3changesetApply(0, clientData, interp, objc, objv); +} +/* +** sqlite3changeset_apply_v2 DB CHANGESET CONFLICT-SCRIPT ?FILTER-SCRIPT? +*/ +static int SQLITE_TCLAPI test_sqlite3changeset_apply_v2( + void * clientData, + Tcl_Interp *interp, + int objc, + Tcl_Obj *CONST objv[] +){ + return testSqlite3changesetApply(1, clientData, interp, objc, objv); +} /* ** sqlite3changeset_apply_replace_all DB CHANGESET */ static int SQLITE_TCLAPI test_sqlite3changeset_apply_replace_all( @@ -1027,10 +1062,11 @@ { "sqlite3session", test_sqlite3session }, { "sqlite3session_foreach", test_sqlite3session_foreach }, { "sqlite3changeset_invert", test_sqlite3changeset_invert }, { "sqlite3changeset_concat", test_sqlite3changeset_concat }, { "sqlite3changeset_apply", test_sqlite3changeset_apply }, + { "sqlite3changeset_apply_v2", test_sqlite3changeset_apply_v2 }, { "sqlite3changeset_apply_replace_all", test_sqlite3changeset_apply_replace_all }, { "sql_exec_changeset", test_sql_exec_changeset }, }; int i;