paulhorn/sqlite
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix rebasing of UPDATE changes against a set of remote changesets that feature

Browse Source 
both OMIT and REPLACE conflict resolution on different fields of the same row.

FossilOrigin-Name: d8bc3fdb6ba165ca8d7cab857ede8e7e6e2fac24ad59580c5e1db1a4942d295c
This commit is contained in:
dan
2018-03-21 17:29:53 +00:00
parent bd45374cc8
commit 24a0c4534a
5 changed files with 107 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ext/session/sessionrebase.test
-2
View File
@@ -120,10 +120,8 @@ proc do_rebase_test {tn sql1 sql2 conflict_handler {testsql ""} {testres ""}} {
#puts [changeset_to_list [lindex $rebase 0]] ; breakpoint
#puts [llength $rebase]
if {$i==2 && $tn=="3.3.1"} breakpoint
sqlite3rebaser_create R
foreach r $rebase {
puts [changeset_to_list $r]
R configure $r
}
set c1r [R rebase $c1]
ext/session/sqlite3session.c
+76 -33
View File
@@ -516,7 +516,7 @@ static int sessionPreupdateHash(
static int sessionSerialLen(u8 *a){
int e = *a;
int n;
if( e==0 ) return 1;
if( e==0 || e==0xFF ) return 1;
if( e==SQLITE_NULL ) return 1;
if( e==SQLITE_INTEGER || e==SQLITE_FLOAT ) return 9;
return sessionVarintGet(&a[1], &n) + 1 + n;
@@ -4527,6 +4527,7 @@ static int sessionChangeMerge(
SessionChange **ppNew /* OUT: Merged change */
){
SessionChange *pNew = 0;
int rc = SQLITE_OK;
if( !pExist ){
pNew = (SessionChange *)sqlite3_malloc(sizeof(SessionChange) + nRec);
@@ -4536,16 +4537,66 @@ static int sessionChangeMerge(
memset(pNew, 0, sizeof(SessionChange));
pNew->op = op2;
pNew->bIndirect = bIndirect;
pNew->nRecord = nRec;
pNew->aRecord = (u8*)&pNew[1];
memcpy(pNew->aRecord, aRec, nRec);
}else if( bRebase){
/*
** op1=INSERT/R, op2=INSERT/R ->
** op1=INSERT/R, op2=INSERT/O ->
** op1=INSERT/O, op2=INSERT/R ->
** op1=INSERT/O, op2=INSERT/O ->
*/
if( bIndirect==0 || bRebase==0 ){
pNew->nRecord = nRec;
memcpy(pNew->aRecord, aRec, nRec);
}else{
int i;
u8 *pIn = aRec;
u8 *pOut = pNew->aRecord;
for(i=0; i<pTab->nCol; i++){
int nIn = sessionSerialLen(pIn);
if( *pIn==0 ){
*pOut++ = 0;
}else if( pTab->abPK[i]==0 ){
*pOut++ = 0xFF;
}else{
memcpy(pOut, pIn, nIn);
pOut += nIn;
}
pIn += nIn;
}
pNew->nRecord = pOut - pNew->aRecord;
}
}else if( bRebase ){
if( pExist->op==SQLITE_DELETE && pExist->bIndirect ){
*ppNew = pExist;
}else{
int nByte = nRec + pExist->nRecord + sizeof(SessionChange);
pNew = (SessionChange*)sqlite3_malloc(nByte);
if( pNew==0 ){
rc = SQLITE_NOMEM;
}else{
int i;
u8 *a1 = pExist->aRecord;
u8 *a2 = aRec;
u8 *pOut;
memset(pNew, 0, nByte);
pNew->bIndirect = bIndirect || pExist->bIndirect;
pNew->op = op2;
pOut = pNew->aRecord = (u8*)&pNew[1];
for(i=0; i<pTab->nCol; i++){
int n1 = sessionSerialLen(a1);
int n2 = sessionSerialLen(a2);
if( *a1==0xFF || *a2==0xFF ){
*pOut++ = 0xFF;
}else if( *a2==0 ){
memcpy(pOut, a1, n1);
pOut += n1;
}else{
memcpy(pOut, a2, n2);
pOut += n2;
}
a1 += n1;
a2 += n2;
}
pNew->nRecord = pOut - pNew->aRecord;
}
sqlite3_free(pExist);
}
}else{
int op1 = pExist->op;
@@ -4639,7 +4690,7 @@ static int sessionChangeMerge(
}
*ppNew = pNew;
return SQLITE_OK;
return rc;
}
/*
@@ -4999,7 +5050,7 @@ static void sessionAppendRecordMerge(
pOut += nn1;
}
}else{
if( *a1==0 ){
if( *a1==0 || *a1==0xFF ){
memcpy(pOut, a2, nn2);
pOut += nn2;
}else{
@@ -5017,11 +5068,11 @@ static void sessionAppendRecordMerge(
}
static void sessionAppendPartialUpdate(
SessionBuffer *pBuf,
sqlite3_changeset_iter *pIter,
u8 *aRec, int nRec,
u8 *aChange, int nChange,
int *pRc
SessionBuffer *pBuf, /* Append record here */
sqlite3_changeset_iter *pIter, /* Iterator pointed at local change */
u8 *aRec, int nRec, /* Local change */
u8 *aChange, int nChange, /* Record to rebase against */
int *pRc /* IN/OUT: Return Code */
){
sessionBufferGrow(pBuf, 2+nRec+nChange, pRc);
if( *pRc==SQLITE_OK ){
@@ -5040,6 +5091,10 @@ static void sessionAppendPartialUpdate(
if( !pIter->abPK[i] ) bData = 1;
memcpy(pOut, a1, n1);
pOut += n1;
}else if( a2[0]!=0xFF ){
bData = 1;
memcpy(pOut, a2, n2);
pOut += n2;
}else{
*pOut++ = '\0';
}
@@ -5051,7 +5106,7 @@ static void sessionAppendPartialUpdate(
for(i=0; i<pIter->nCol; i++){
int n1 = sessionSerialLen(a1);
int n2 = sessionSerialLen(a2);
if( pIter->abPK[i] || a2[0]==0 ){
if( pIter->abPK[i] || a2[0]!=0xFF ){
memcpy(pOut, a1, n1);
pOut += n1;
}else{
@@ -5065,7 +5120,6 @@ static void sessionAppendPartialUpdate(
}
}
static int sessionRebase(
sqlite3_rebaser *p, /* Rebaser hash table */
sqlite3_changeset_iter *pIter, /* Input data */
@@ -5139,20 +5193,9 @@ static int sessionRebase(
);
}
}else{
if( pChange->bIndirect==0 ){
u8 *pCsr = aRec;
sessionAppendByte(&sOut, SQLITE_UPDATE, &rc);
sessionAppendByte(&sOut, pIter->bIndirect, &rc);
sessionAppendRecordMerge(&sOut, pIter->nCol, 0,
aRec, nRec, pChange->aRecord, pChange->nRecord, &rc
);
sessionSkipRecord(&pCsr, pIter->nCol);
sessionAppendBlob(&sOut, pCsr, nRec - (pCsr-aRec), &rc);
}else{
sessionAppendPartialUpdate(&sOut, pIter,
aRec, nRec, pChange->aRecord, pChange->nRecord, &rc
);
}
sessionAppendPartialUpdate(&sOut, pIter,
aRec, nRec, pChange->aRecord, pChange->nRecord, &rc
);
}
break;
ext/session/sqlite3session.h
+22 -5
View File
@@ -1256,14 +1256,14 @@ int sqlite3changeset_apply_v2(
** This may conflict with a remote UPDATE or DELETE. In both cases the
** only possible resolution is OMIT. If the remote operation was a
** DELETE, then add no change to the rebased changeset. If the remote
** operation was an UPDATE, then the old.* fields of change are updated
** operation was an UPDATE, then the old.* fields of change are updated
** to reflect the new.* values in the UPDATE.
**
** <dt>Local UPDATE<dd>
** This may conflict with a remote UPDATE or DELETE. If it conflicts
** with a DELETE, and the conflict resolution was OMIT, then the update
** is changed into an INSERT. Any undefined values in the new.* record
** from the update change are filled in using the old.* values from
** from the update change are filled in using the old.* values from
** the conflicting DELETE. Or, if the conflict resolution was REPLACE,
** the UPDATE change is simply omitted from the rebased changeset.
**
@@ -1271,12 +1271,29 @@ int sqlite3changeset_apply_v2(
** the old.* values are rebased using the new.* values in the remote
** change. Or, if the resolution is REPLACE, then the change is copied
** into the rebased changeset with updates to columns also updated by
** the conflicting UPDATE removed. If this means no columns would be
** updated, the change is omitted.
** the conflicting remote UPDATE removed. If this means no columns would
** be updated, the change is omitted.
** </dl>
**
** A local change may be rebased against multiple remote changes
** simultaneously.
** simultaneously. If a single key is modified by multiple remote
** changesets, they are combined as follows before the local changeset
** is rebased:
**
** <ul>
** <li> If there has been one or more REPLACE resolutions on a
** key, it is rebased according to a REPLACE.
**
** <li> If there have been no REPLACE resolutions on a key, then
** the local changeset is rebased according to the most recent
** of the OMIT resolutions.
** </ul>
**
** Note that conflict resolutions from multiple remote changesets are
** combined on a per-field basis, not per-row. This means that in the
** case of multiple remote UPDATE operations, some fields of a single
** local change may be rebased for REPLACE while others are rebased for
** OMIT.
*/
typedef struct sqlite3_rebaser sqlite3_rebaser;