3 files changed, 56 insertions, 12 deletions

diff --git a/ext/session/sessionchange.test b/ext/session/sessionchange.test
index c1c28622e..6e2cb0e82 100644
--- a/ext/session/sessionchange.test
+++ b/ext/session/sessionchange.test
@@ -96,6 +96,22 @@ do_test 1.5 {
 $iter finalize
 grp delete
 
+do_test 1.6 { 
+  set C [changeset_from_sql {
+    INSERT INTO t1 VALUES(1001, hex(randomblob(10000)), hex(randomblob(10000)));
+    INSERT INTO t1 VALUES(2001, hex(randomblob(10000)), hex(randomblob(10000)));
+    INSERT INTO t1 VALUES(3001, hex(randomblob(10000)), hex(randomblob(10000)));
+  }]
+
+  sqlite3changegroup grp
+  set iter [sqlite3changeset_start $C]
+  while {[$iter next]=="SQLITE_ROW"} {
+    grp add_change $iter
+  }
+  $iter finalize
+} SQLITE_OK
+
+grp delete
 
 
 finish_test
diff --git a/ext/session/sqlite3session.c b/ext/session/sqlite3session.c
index a8dd1d9be..175cacbe8 100644
--- a/ext/session/sqlite3session.c
+++ b/ext/session/sqlite3session.c
@@ -3396,14 +3396,15 @@ int sqlite3changeset_start_v2_strm(
 ** object and the buffer is full, discard some data to free up space.
 */
 static void sessionDiscardData(SessionInput *pIn){
-  if( pIn->xInput && pIn->iNext>=sessions_strm_chunk_size ){
-    int nMove = pIn->buf.nBuf - pIn->iNext;
+  if( pIn->xInput && pIn->iCurrent>=sessions_strm_chunk_size ){
+    int nMove = pIn->buf.nBuf - pIn->iCurrent;
     assert( nMove>=0 );
     if( nMove>0 ){
-      memmove(pIn->buf.aBuf, &pIn->buf.aBuf[pIn->iNext], nMove);
+      memmove(pIn->buf.aBuf, &pIn->buf.aBuf[pIn->iCurrent], nMove);
     }
-    pIn->buf.nBuf -= pIn->iNext;
-    pIn->iNext = 0;
+    pIn->buf.nBuf -= pIn->iCurrent;
+    pIn->iNext -= pIn->iCurrent;
+    pIn->iCurrent = 0;
     pIn->nData = pIn->buf.nBuf;
   }
 }
@@ -3757,8 +3758,8 @@ static int sessionChangesetNextOne(
   p->rc = sessionInputBuffer(&p->in, 2);
   if( p->rc!=SQLITE_OK ) return p->rc;
 
-  sessionDiscardData(&p->in);
   p->in.iCurrent = p->in.iNext;
+  sessionDiscardData(&p->in);
 
   /* If the iterator is already at the end of the changeset, return DONE. */
   if( p->in.iNext>=p->in.nData ){
@@ -6117,14 +6118,19 @@ int sqlite3changegroup_add_change(
   sqlite3_changegroup *pGrp,
   sqlite3_changeset_iter *pIter
 ){
+  int rc = SQLITE_OK;
+
   if( pIter->in.iCurrent==pIter->in.iNext 
    || pIter->rc!=SQLITE_OK 
    || pIter->bInvert
   ){
     /* Iterator does not point to any valid entry or is an INVERT iterator. */
-    return SQLITE_ERROR;
+    rc = SQLITE_ERROR;
+  }else{
+    pIter->in.bNoDiscard = 1;
+    rc = sessionOneChangeToHash(pGrp, pIter, 0);
   }
-  return sessionOneChangeToHash(pGrp, pIter, 0);
+  return rc;
 }
 
 /*
diff --git a/ext/session/test_session.c b/ext/session/test_session.c
index e167cb8fb..f28604abc 100644
--- a/ext/session/test_session.c
+++ b/ext/session/test_session.c
@@ -1459,6 +1459,9 @@ struct TestChangegroup {
 typedef struct TestChangeIter TestChangeIter;
 struct TestChangeIter {
   sqlite3_changeset_iter *pIter;
+
+  /* If this iter uses streaming. */
+  TestStreamInput in;
 };
 
 
@@ -1681,6 +1684,7 @@ static int SQLITE_TCLAPI test_sqlite3changeset_start(
   sqlite3_changeset_iter *pIter = 0;
   int flags = 0;
   int rc = SQLITE_OK;
+  int nAlloc = 0;                 /* Bytes of space to allocate */
 
   static int iCmd = 1;
   char zCmd[64];
@@ -1696,18 +1700,36 @@ static int SQLITE_TCLAPI test_sqlite3changeset_start(
     return TCL_ERROR;
   }
 
-  flags = isInvert ? SQLITE_CHANGESETSTART_INVERT : 0;
   pChangeset = (void *)Tcl_GetByteArrayFromObj(objv[objc-1], &nChangeset);
-  rc = sqlite3changeset_start_v2(&pIter, (int)nChangeset, pChangeset, flags);
+  flags = isInvert ? SQLITE_CHANGESETSTART_INVERT : 0;
+
+  nAlloc = sizeof(TestChangeIter);
+  if( test_tcl_integer(interp, SESSION_STREAM_TCL_VAR) ){
+    nAlloc += nChangeset;
+  }
+  pNew = (TestChangeIter*)ckalloc(nAlloc);
+  memset(pNew, 0, nAlloc);
+  if( test_tcl_integer(interp, SESSION_STREAM_TCL_VAR) ){
+    pNew->in.nStream = test_tcl_integer(interp, SESSION_STREAM_TCL_VAR);
+    pNew->in.nData = nChangeset;
+    pNew->in.aData = (unsigned char*)&pNew[1];
+    memcpy(pNew->in.aData, pChangeset, nChangeset);
+  }
+
+  if( pNew->in.nStream ){
+    void *pCtx = (void*)&pNew->in;
+    rc = sqlite3changeset_start_v2_strm(&pIter, testStreamInput, pCtx, flags);
+  }else{
+    rc = sqlite3changeset_start_v2(&pIter, (int)nChangeset, pChangeset, flags);
+  }
   if( rc!=SQLITE_OK ){
     char *zErr = sqlite3_mprintf(
         "error in sqlite3changeset_start_v2() - %d", rc
     );
     Tcl_AppendResult(interp, zErr, (char*)0);
+    ckfree(pNew);
     return TCL_ERROR;
   }
-
-  pNew = (TestChangeIter*)ckalloc(sizeof(TestChangeIter));
   pNew->pIter = pIter;
 
   sprintf(zCmd, "csiter%d", iCmd++);