Many hyperlinks are disabled.
Use anonymous login
to enable hyperlinks.
Overview
Comment: | Add streaming version of sqlite3changeset_invert() to sessions module. |
---|---|
Downloads: | Tarball | ZIP archive |
Timelines: | family | ancestors | descendants | both | sessions |
Files: | files | file ages | folders |
SHA1: |
8ded6a46794c7bff1c8b790c662ba7e9 |
User & Date: | dan 2014-09-25 14:54:20.019 |
Context
2014-09-25
| ||
20:43 | Add streaming version of sqlite3changeset_concat(). (check-in: 88eb6656bd user: dan tags: sessions) | |
14:54 | Add streaming version of sqlite3changeset_invert() to sessions module. (check-in: 8ded6a4679 user: dan tags: sessions) | |
2014-09-24
| ||
17:13 | Add streaming version of sqlite3changeset_apply(). Tests and fixes for the same and sqlite3changeset_start_str(). (check-in: b917fc1468 user: dan tags: sessions) | |
Changes
Changes to ext/session/sqlite3session.c.
︙ | ︙ | |||
330 331 332 333 334 335 336 | } } }else{ nByte = 1; if( aBuf ) aBuf[0] = '\0'; } | | > | 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 | } } }else{ nByte = 1; if( aBuf ) aBuf[0] = '\0'; } if( pnWrite ) *pnWrite += nByte; return SQLITE_OK; } /* ** This macro is used to calculate hash key values for data structures. In ** order to use this macro, the entire data structure must be represented ** as a series of unsigned integers. In order to calculate a hash-key value ** for a data structure represented as three such integers, the macro may ** then be used as follows: |
︙ | ︙ | |||
1325 1326 1327 1328 1329 1330 1331 1332 1333 1334 1335 1336 1337 1338 | }else{ p->aBuf = aNew; p->nAlloc = nNew; } } return (*pRc!=SQLITE_OK); } /* ** This function is a no-op if *pRc is other than SQLITE_OK when it is ** called. Otherwise, append a single byte to the buffer. ** ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before ** returning. | > > > > > > > > > > > > > > > > > > > > > > > | 1326 1327 1328 1329 1330 1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348 1349 1350 1351 1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 | }else{ p->aBuf = aNew; p->nAlloc = nNew; } } return (*pRc!=SQLITE_OK); } /* ** Append the value passed as the second argument to the buffer passed ** as the first. ** ** This function is a no-op if *pRc is non-zero when it is called. ** Otherwise, if an error occurs, *pRc is set to an SQLite error code ** before returning. */ static void sessionAppendValue(SessionBuffer *p, sqlite3_value *pVal, int *pRc){ int rc = *pRc; if( rc==SQLITE_OK ){ int nByte = 0; sessionSerializeValue(0, pVal, &nByte); sessionBufferGrow(p, nByte, &rc); if( rc==SQLITE_OK ){ rc = sessionSerializeValue(&p->aBuf[p->nBuf], pVal, 0); p->nBuf += nByte; }else{ *pRc = rc; } } } /* ** This function is a no-op if *pRc is other than SQLITE_OK when it is ** called. Otherwise, append a single byte to the buffer. ** ** If an OOM condition is encountered, set *pRc to SQLITE_NOMEM before ** returning. |
︙ | ︙ | |||
2263 2264 2265 2266 2267 2268 2269 2270 2271 2272 2273 2274 2275 2276 | } if( pIn->aData[pIn->iNext + nRead]==0 ) break; rc = sessionInputBuffer(pIn, nRead + 100); } if( pnByte ) *pnByte = nRead+1; return rc; } /* ** The input pointer currently points to the second byte of a table-header. ** Specifically, to the following: ** ** + number of columns in table (varint) ** + array of PK flags (1 byte per column), | > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > | 2287 2288 2289 2290 2291 2292 2293 2294 2295 2296 2297 2298 2299 2300 2301 2302 2303 2304 2305 2306 2307 2308 2309 2310 2311 2312 2313 2314 2315 2316 2317 2318 2319 2320 2321 2322 2323 2324 2325 2326 2327 2328 2329 2330 2331 2332 | } if( pIn->aData[pIn->iNext + nRead]==0 ) break; rc = sessionInputBuffer(pIn, nRead + 100); } if( pnByte ) *pnByte = nRead+1; return rc; } /* ** The input pointer currently points to the first byte of the first field ** of a record consisting of nCol columns. This function ensures the entire ** record is buffered. */ static int sessionChangesetBufferRecord( SessionInput *pIn, int nCol, int *pnByte ){ int rc = SQLITE_OK; int nByte = 0; int i; for(i=0; rc==SQLITE_OK && i<nCol; i++){ int eType; rc = sessionInputBuffer(pIn, nByte + 10); if( rc==SQLITE_OK ){ eType = pIn->aData[pIn->iNext + nByte++]; if( eType==SQLITE_TEXT || eType==SQLITE_BLOB ){ int n; nByte += sessionVarintGet(&pIn->aData[pIn->iNext+nByte], &n); nByte += n; rc = sessionInputBuffer(pIn, nByte); }else if( eType==SQLITE_INTEGER || eType==SQLITE_FLOAT ){ nByte += 8; } } } *pnByte = nByte; return rc; } /* ** The input pointer currently points to the second byte of a table-header. ** Specifically, to the following: ** ** + number of columns in table (varint) ** + array of PK flags (1 byte per column), |
︙ | ︙ | |||
2578 2579 2580 2581 2582 2583 2584 | } sqlite3_free(p->tblhdr.aBuf); sqlite3_free(p->in.buf.aBuf); sqlite3_free(p); return rc; } | < | < | < | > < < < | > > > > | | < | < < < < < < < < < | > > | > > | | < | | | | > > | < | < | < < | < | > > > > | | > < | | < | | | | | < < > > > > > > > > | | > > > | > < | > > > | | > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > | | 2634 2635 2636 2637 2638 2639 2640 2641 2642 2643 2644 2645 2646 2647 2648 2649 2650 2651 2652 2653 2654 2655 2656 2657 2658 2659 2660 2661 2662 2663 2664 2665 2666 2667 2668 2669 2670 2671 2672 2673 2674 2675 2676 2677 2678 2679 2680 2681 2682 2683 2684 2685 2686 2687 2688 2689 2690 2691 2692 2693 2694 2695 2696 2697 2698 2699 2700 2701 2702 2703 2704 2705 2706 2707 2708 2709 2710 2711 2712 2713 2714 2715 2716 2717 2718 2719 2720 2721 2722 2723 2724 2725 2726 2727 2728 2729 2730 2731 2732 2733 2734 2735 2736 2737 2738 2739 2740 2741 2742 2743 2744 2745 2746 2747 2748 2749 2750 2751 2752 2753 2754 2755 2756 2757 2758 2759 2760 2761 2762 2763 2764 2765 2766 2767 2768 2769 2770 2771 2772 2773 2774 2775 2776 2777 2778 2779 2780 2781 2782 2783 2784 2785 2786 2787 2788 2789 2790 2791 2792 2793 2794 2795 2796 2797 2798 2799 2800 2801 2802 2803 2804 2805 2806 2807 2808 2809 2810 2811 2812 2813 2814 2815 2816 2817 2818 2819 2820 2821 2822 2823 2824 2825 2826 2827 2828 2829 2830 2831 2832 2833 2834 2835 2836 2837 2838 2839 2840 2841 2842 2843 2844 | } sqlite3_free(p->tblhdr.aBuf); sqlite3_free(p->in.buf.aBuf); sqlite3_free(p); return rc; } static int sessionChangesetInvert( SessionInput *pInput, /* Input changeset */ int (*xOutput)(void *pOut, const void *pData, int nData), void *pOut, int *pnInverted, /* OUT: Number of bytes in output changeset */ void **ppInverted /* OUT: Inverse of pChangeset */ ){ int rc = SQLITE_OK; /* Return value */ SessionBuffer sOut; /* Output buffer */ int nCol = 0; /* Number of cols in current table */ u8 *abPK = 0; /* PK array for current table */ sqlite3_value **apVal = 0; /* Space for values for UPDATE inversion */ SessionBuffer sPK = {0, 0, 0}; /* PK array for current table */ /* Initialize the output buffer */ memset(&sOut, 0, sizeof(SessionBuffer)); /* Zero the output variables in case an error occurs. */ if( ppInverted ){ *ppInverted = 0; *pnInverted = 0; } while( 1 ){ u8 eType; /* Test for EOF. */ if( (rc = sessionInputBuffer(pInput, 2)) ) goto finished_invert; if( pInput->iNext>=pInput->nData ) break; eType = pInput->aData[pInput->iNext]; switch( eType ){ case 'T': { /* A 'table' record consists of: ** ** * A constant 'T' character, ** * Number of columns in said table (a varint), ** * An array of nCol bytes (sPK), ** * A nul-terminated table name. */ int nByte; int nVar; pInput->iNext++; if( (rc = sessionChangesetBufferTblhdr(pInput, &nByte)) ){ goto finished_invert; } nVar = sessionVarintGet(&pInput->aData[pInput->iNext], &nCol); sPK.nBuf = 0; sessionAppendBlob(&sPK, &pInput->aData[pInput->iNext+nVar], nCol, &rc); sessionAppendByte(&sOut, eType, &rc); sessionAppendBlob(&sOut, &pInput->aData[pInput->iNext], nByte, &rc); if( rc ) goto finished_invert; pInput->iNext += nByte; sqlite3_free(apVal); apVal = 0; abPK = sPK.aBuf; break; } case SQLITE_INSERT: case SQLITE_DELETE: { int nByte; int bIndirect = pInput->aData[pInput->iNext+1]; int eType2 = (eType==SQLITE_DELETE ? SQLITE_INSERT : SQLITE_DELETE); pInput->iNext += 2; assert( rc==SQLITE_OK ); rc = sessionChangesetBufferRecord(pInput, nCol, &nByte); sessionAppendByte(&sOut, eType2, &rc); sessionAppendByte(&sOut, bIndirect, &rc); sessionAppendBlob(&sOut, &pInput->aData[pInput->iNext], nByte, &rc); pInput->iNext += nByte; if( rc ) goto finished_invert; break; } case SQLITE_UPDATE: { int iCol; if( 0==apVal ){ apVal = (sqlite3_value **)sqlite3_malloc(sizeof(apVal[0])*nCol*2); if( 0==apVal ){ rc = SQLITE_NOMEM; goto finished_invert; } memset(apVal, 0, sizeof(apVal[0])*nCol*2); } /* Write the header for the new UPDATE change. Same as the original. */ sessionAppendByte(&sOut, eType, &rc); sessionAppendByte(&sOut, pInput->aData[pInput->iNext+1], &rc); /* Read the old.* and new.* records for the update change. */ pInput->iNext += 2; rc = sessionReadRecord(pInput, nCol, 0, &apVal[0]); if( rc==SQLITE_OK ){ rc = sessionReadRecord(pInput, nCol, 0, &apVal[nCol]); } /* Write the new old.* record. Consists of the PK columns from the ** original old.* record, and the other values from the original ** new.* record. */ for(iCol=0; rc==SQLITE_OK && iCol<nCol; iCol++){ sqlite3_value *pVal = apVal[iCol + (abPK[iCol] ? 0 : nCol)]; sessionAppendValue(&sOut, pVal, &rc); } /* Write the new new.* record. Consists of a copy of all values ** from the original old.* record, except for the PK columns, which ** are set to "undefined". */ for(iCol=0; rc==SQLITE_OK && iCol<nCol; iCol++){ sqlite3_value *pVal = (abPK[iCol] ? 0 : apVal[iCol]); sessionAppendValue(&sOut, pVal, &rc); } for(iCol=0; iCol<nCol*2; iCol++){ sqlite3ValueFree(apVal[iCol]); } memset(apVal, 0, sizeof(apVal[0])*nCol*2); if( rc!=SQLITE_OK ){ goto finished_invert; } break; } default: rc = SQLITE_CORRUPT_BKPT; goto finished_invert; } assert( rc==SQLITE_OK ); if( xOutput && sOut.nBuf>=SESSIONS_STR_CHUNK_SIZE ){ rc = xOutput(pOut, sOut.aBuf, sOut.nBuf); sOut.nBuf = 0; if( rc!=SQLITE_OK ) goto finished_invert; } } assert( rc==SQLITE_OK ); if( pnInverted ){ *pnInverted = sOut.nBuf; *ppInverted = sOut.aBuf; sOut.aBuf = 0; }else if( sOut.nBuf>0 ){ rc = xOutput(pOut, sOut.aBuf, sOut.nBuf); } finished_invert: sqlite3_free(sOut.aBuf); sqlite3_free(apVal); sqlite3_free(sPK.aBuf); return rc; } /* ** Invert a changeset object. */ int sqlite3changeset_invert( int nChangeset, /* Number of bytes in input */ const void *pChangeset, /* Input changeset */ int *pnInverted, /* OUT: Number of bytes in output changeset */ void **ppInverted /* OUT: Inverse of pChangeset */ ){ SessionInput sInput; /* Set up the input stream */ memset(&sInput, 0, sizeof(SessionInput)); sInput.nData = nChangeset; sInput.aData = (u8*)pChangeset; return sessionChangesetInvert(&sInput, 0, 0, pnInverted, ppInverted); } /* ** Streaming version of sqlite3changeset_invert(). */ int sqlite3changeset_invert_str( int (*xInput)(void *pIn, void *pData, int *pnData), void *pIn, int (*xOutput)(void *pOut, const void *pData, int nData), void *pOut ){ SessionInput sInput; int rc; /* Set up the input stream */ memset(&sInput, 0, sizeof(SessionInput)); sInput.xInput = xInput; sInput.pIn = pIn; rc = sessionChangesetInvert(&sInput, xOutput, pOut, 0, 0); sqlite3_free(sInput.buf.aBuf); return rc; } typedef struct SessionApplyCtx SessionApplyCtx; struct SessionApplyCtx { sqlite3 *db; sqlite3_stmt *pDelete; /* DELETE statement */ sqlite3_stmt *pUpdate; /* UPDATE statement */ sqlite3_stmt *pInsert; /* INSERT statement */ sqlite3_stmt *pSelect; /* SELECT statement */ |
︙ | ︙ |
Changes to ext/session/sqlite3session.h.
︙ | ︙ | |||
665 666 667 668 669 670 671 672 673 674 675 676 677 678 | ** WARNING/TODO: This function currently assumes that the input is a valid ** changeset. If it is not, the results are undefined. */ int sqlite3changeset_invert( int nIn, const void *pIn, /* Input changeset */ int *pnOut, void **ppOut /* OUT: Inverse of input */ ); /* ** CAPI3REF: Concatenate Two Changeset Objects ** ** This function is used to concatenate two changesets, A and B, into a ** single changeset. The result is a changeset equivalent to applying ** changeset A followed by changeset B. | > > > > > > > > > > | 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 | ** WARNING/TODO: This function currently assumes that the input is a valid ** changeset. If it is not, the results are undefined. */ int sqlite3changeset_invert( int nIn, const void *pIn, /* Input changeset */ int *pnOut, void **ppOut /* OUT: Inverse of input */ ); /* ** Streaming version of sqlite3changeset_invert(). */ int sqlite3changeset_invert_str( int (*xInput)(void *pIn, void *pData, int *pnData), void *pIn, int (*xOutput)(void *pOut, const void *pData, int nData), void *pOut ); /* ** CAPI3REF: Concatenate Two Changeset Objects ** ** This function is used to concatenate two changesets, A and B, into a ** single changeset. The result is a changeset equivalent to applying ** changeset A followed by changeset B. |
︙ | ︙ |
Changes to ext/session/test_session.c.
︙ | ︙ | |||
672 673 674 675 676 677 678 | static int test_sqlite3changeset_invert( void * clientData, Tcl_Interp *interp, int objc, Tcl_Obj *CONST objv[] ){ int rc; /* Return code from changeset_invert() */ | | < | < > > > > | > | > > > > > | > > < | | | 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 | static int test_sqlite3changeset_invert( void * clientData, Tcl_Interp *interp, int objc, Tcl_Obj *CONST objv[] ){ int rc; /* Return code from changeset_invert() */ TestStreamInput sIn; /* Input stream */ TestSessionsBlob sOut; /* Output blob */ if( objc!=2 ){ Tcl_WrongNumArgs(interp, 1, objv, "CHANGESET"); return TCL_ERROR; } memset(&sIn, 0, sizeof(sIn)); memset(&sOut, 0, sizeof(sOut)); sIn.nStream = test_tcl_integer(interp, SESSION_STREAM_TCL_VAR); sIn.aData = Tcl_GetByteArrayFromObj(objv[1], &sIn.nData); if( sIn.nStream ){ rc = sqlite3changeset_invert_str( testStreamInput, (void*)&sIn, testSessionsOutput, (void*)&sOut ); }else{ rc = sqlite3changeset_invert(sIn.nData, sIn.aData, &sOut.n, &sOut.p); } if( rc!=SQLITE_OK ){ rc = test_session_error(interp, rc); }else{ Tcl_SetObjResult(interp,Tcl_NewByteArrayObj((unsigned char*)sOut.p,sOut.n)); } sqlite3_free(sOut.p); return rc; } /* ** sqlite3changeset_concat LEFT RIGHT */ static int test_sqlite3changeset_concat( void * clientData, |
︙ | ︙ |