[Feature] Issue#26 REBASE TO TokuDB 7.5.6

Author: AliSQL

storage/tokudb/ft-index/buildheader/make_tdb.cc

@@ -339,7 +339,6 @@ static void print_defines (void) {
         dodefine_from_track(txn_flags, DB_INHERIT_ISOLATION);
         dodefine_from_track(txn_flags, DB_SERIALIZABLE);
         dodefine_from_track(txn_flags, DB_TXN_READ_ONLY);
-        dodefine_from_track(txn_flags, DB_READ_COMMITTED_ALWAYS);
     }
 
     /* TokuFT specific error codes*/
@@ -573,7 +572,7 @@ static void print_db_txn_struct (void) {
     STRUCT_SETUP(DB_TXN, abort,       "int (*%s) (DB_TXN *)");
     STRUCT_SETUP(DB_TXN, api_internal,"void *%s");
     STRUCT_SETUP(DB_TXN, commit,      "int (*%s) (DB_TXN*, uint32_t)");
-    STRUCT_SETUP(DB_TXN, prepare,     "int (*%s) (DB_TXN*, uint8_t gid[DB_GID_SIZE])");
+    STRUCT_SETUP(DB_TXN, prepare,     "int (*%s) (DB_TXN*, uint8_t gid[DB_GID_SIZE], uint32_t flags)");
     STRUCT_SETUP(DB_TXN, discard,     "int (*%s) (DB_TXN*, uint32_t)");
     STRUCT_SETUP(DB_TXN, id,          "uint32_t (*%s) (DB_TXN *)");
     STRUCT_SETUP(DB_TXN, mgrp,        "DB_ENV *%s /* In TokuFT, mgrp is a DB_ENV, not a DB_TXNMGR */");
@@ -582,10 +581,12 @@ static void print_db_txn_struct (void) {
 	"int (*txn_stat)(DB_TXN *, struct txn_stat **)", 
 	"int (*commit_with_progress)(DB_TXN*, uint32_t, TXN_PROGRESS_POLL_FUNCTION, void*)",
 	"int (*abort_with_progress)(DB_TXN*, TXN_PROGRESS_POLL_FUNCTION, void*)",
-	"int (*xa_prepare) (DB_TXN*, TOKU_XA_XID *)",
+	"int (*xa_prepare) (DB_TXN*, TOKU_XA_XID *, uint32_t flags)",
         "uint64_t (*id64) (DB_TXN*)",
         "void (*set_client_id)(DB_TXN *, uint64_t client_id)",
         "uint64_t (*get_client_id)(DB_TXN *)",
+        "bool (*is_prepared)(DB_TXN *)",
+        "DB_TXN *(*get_child)(DB_TXN *)",
 	NULL};
     sort_and_dump_fields("db_txn", false, extra);
 }
@@ -614,7 +615,7 @@ static void print_dbc_struct (void) {
 	"int (*c_getf_set_range_reverse)(DBC *, uint32_t, DBT *, YDB_CALLBACK_FUNCTION, void *)",
 	"int (*c_getf_set_range_with_bound)(DBC *, uint32_t, DBT *k, DBT *k_bound, YDB_CALLBACK_FUNCTION, void *)",
 	"int (*c_set_bounds)(DBC*, const DBT*, const DBT*, bool pre_acquire, int out_of_range_error)",
-        "void (*c_set_check_interrupt_callback)(DBC*, bool (*)(void*), void *)",
+        "void (*c_set_check_interrupt_callback)(DBC*, bool (*)(void*, uint64_t deleted_rows), void *)",
 	"void (*c_remove_restriction)(DBC*)",
         "char _internal[512]",
 	NULL};

storage/tokudb/ft-index/ft/cursor.cc

@@ -94,12 +94,12 @@ PATENT RIGHTS GRANT:
 #include "util/dbt.h"
 
 int toku_ft_cursor_create(FT_HANDLE ft_handle, FT_CURSOR cursor, TOKUTXN ttxn,
-                          enum cursor_read_type read_type,
+                          bool is_snapshot_read,
                           bool disable_prefetching,
                           bool is_temporary) {
-    if (read_type == C_READ_SNAPSHOT) {
+    if (is_snapshot_read) {
         invariant(ttxn != NULL);
-        int accepted = toku_txn_reads_txnid(ft_handle->ft->h->root_xid_that_created, ttxn, false); // last parameter is irrelevant
+        int accepted = toku_txn_reads_txnid(ft_handle->ft->h->root_xid_that_created, ttxn);
         if (accepted != TOKUDB_ACCEPT) {
             invariant(accepted == 0);
             return TOKUDB_MVCC_DICTIONARY_TOO_NEW;
@@ -109,7 +109,7 @@ int toku_ft_cursor_create(FT_HANDLE ft_handle, FT_CURSOR cursor, TOKUTXN ttxn,
     memset(cursor, 0, sizeof(*cursor));
     cursor->ft_handle = ft_handle;
     cursor->ttxn = ttxn;
-    cursor->read_type = read_type;
+    cursor->is_snapshot_read = is_snapshot_read;
     cursor->disable_prefetching = disable_prefetching;
     cursor->is_temporary = is_temporary;
     return 0;
@@ -126,8 +126,7 @@ void toku_ft_cursor_destroy(FT_CURSOR cursor) {
 int toku_ft_cursor(FT_HANDLE ft_handle, FT_CURSOR *cursorptr, TOKUTXN ttxn,
                    bool is_snapshot_read, bool disable_prefetching) {
     FT_CURSOR XCALLOC(cursor);
-    enum cursor_read_type read_type = is_snapshot_read ? C_READ_SNAPSHOT : C_READ_ANY;
-    int r = toku_ft_cursor_create(ft_handle, cursor, ttxn, read_type, disable_prefetching, false);
+    int r = toku_ft_cursor_create(ft_handle, cursor, ttxn, is_snapshot_read, disable_prefetching, false);
     if (r == 0) {
         *cursorptr = cursor;
     } else {
@@ -322,11 +321,11 @@ int toku_ft_cursor_shortcut(FT_CURSOR cursor, int direction, uint32_t index, bn_
         r = bd->fetch_klpair(index, &le, &foundkeylen, &foundkey);
         invariant_zero(r);
 
-        if (toku_ft_cursor_is_leaf_mode(cursor) || !le_val_is_del(le, cursor->read_type, cursor->ttxn)) {
+        if (toku_ft_cursor_is_leaf_mode(cursor) || !le_val_is_del(le, cursor->is_snapshot_read, cursor->ttxn)) {
             le_extract_val(
                 le,
                 toku_ft_cursor_is_leaf_mode(cursor),
-                cursor->read_type,
+                cursor->is_snapshot_read,
                 cursor->ttxn,
                 vallen,
                 val

storage/tokudb/ft-index/ft/cursor.h

@@ -99,7 +99,7 @@ struct ft_cursor {
     DBT range_lock_left_key, range_lock_right_key;
     bool prefetching;
     bool left_is_neg_infty, right_is_pos_infty;
-    enum cursor_read_type read_type; // true if query is reading from a snapshot, false otherwise
+    bool is_snapshot_read; // true if query is read_committed, false otherwise
     bool is_leaf_mode;
     bool disable_prefetching;
     bool is_temporary;
@@ -176,7 +176,7 @@ static inline void ft_search_finish(ft_search *search) {
 
 
 int toku_ft_cursor_create(FT_HANDLE ft_handle, FT_CURSOR cursor, TOKUTXN txn,
-                          enum cursor_read_type read_type,
+                          bool is_snapshot_read,
                           bool disable_prefetching,
                           bool is_temporary);
 

storage/tokudb/ft-index/ft/ft-internal.h

@@ -616,6 +616,7 @@ typedef enum {
     FT_PRO_RIGHTMOST_LEAF_SHORTCUT_SUCCESS,
     FT_PRO_RIGHTMOST_LEAF_SHORTCUT_FAIL_POS,
     FT_PRO_RIGHTMOST_LEAF_SHORTCUT_FAIL_REACTIVE,
+    FT_CURSOR_SKIP_DELETED_LEAF_ENTRY, // how many deleted leaf entries were skipped by a cursor
     FT_STATUS_NUM_ROWS
 } ft_status_entry;
 
@@ -654,7 +655,7 @@ int toku_upgrade_msn_from_root_to_header(int fd, FT ft) __attribute__((nonnull))
 // When lock_only is true, the callback only does optional lock tree locking.
 typedef int (*FT_GET_CALLBACK_FUNCTION)(uint32_t keylen, const void *key, uint32_t vallen, const void *val, void *extra, bool lock_only);
 
-typedef bool (*FT_CHECK_INTERRUPT_CALLBACK)(void *extra);
+typedef bool (*FT_CHECK_INTERRUPT_CALLBACK)(void *extra, uint64_t deleted_rows);
 
 struct ft_cursor;
 int toku_ft_search(FT_HANDLE ft_handle, ft_search *search, FT_GET_CALLBACK_FUNCTION getf, void *getf_v, struct ft_cursor *ftcursor, bool can_bulk_fetch);

storage/tokudb/ft-index/ft/ft-ops.cc

@@ -375,6 +375,8 @@ status_init(void)
     STATUS_INIT(FT_PRO_RIGHTMOST_LEAF_SHORTCUT_FAIL_POS,   nullptr, PARCOUNT, "promotion: tried the rightmost leaf shorcut but failed (out-of-bounds)", TOKU_ENGINE_STATUS);
     STATUS_INIT(FT_PRO_RIGHTMOST_LEAF_SHORTCUT_FAIL_REACTIVE,nullptr, PARCOUNT, "promotion: tried the rightmost leaf shorcut but failed (child reactive)", TOKU_ENGINE_STATUS);
 
+    STATUS_INIT(FT_CURSOR_SKIP_DELETED_LEAF_ENTRY,         CURSOR_SKIP_DELETED_LEAF_ENTRY, PARCOUNT, "cursor skipped deleted leaf entries", TOKU_ENGINE_STATUS|TOKU_GLOBAL_STATUS);
+
     ft_status.initialized = true;
 }
 static void status_destroy(void) {
@@ -3373,25 +3375,26 @@ ok: ;
 
     if (toku_ft_cursor_is_leaf_mode(ftcursor))
         goto got_a_good_value;        // leaf mode cursors see all leaf entries
-    if (le_val_is_del(le, ftcursor->read_type, ftcursor->ttxn)) {
+    if (le_val_is_del(le, ftcursor->is_snapshot_read, ftcursor->ttxn)) {
         // Provisionally deleted stuff is gone.
         // So we need to scan in the direction to see if we can find something.
-        // Every 100 deleted leaf entries check if the leaf's key is within the search bounds.
-        for (uint n_deleted = 1; ; n_deleted++) {
+        // Every 64 deleted leaf entries check if the leaf's key is within the search bounds.
+        for (uint64_t n_deleted = 1; ; n_deleted++) {
             switch (search->direction) {
             case FT_SEARCH_LEFT:
                 idx++;
-                if (idx >= bn->data_buffer.num_klpairs() ||
-                    ((n_deleted % 64) == 0 && !search_continue(search, key, keylen))) {
-                    if (ftcursor->interrupt_cb && ftcursor->interrupt_cb(ftcursor->interrupt_cb_extra)) {
+                if (idx >= bn->data_buffer.num_klpairs() || ((n_deleted % 64) == 0 && !search_continue(search, key, keylen))) {
+                    STATUS_INC(FT_CURSOR_SKIP_DELETED_LEAF_ENTRY, n_deleted);
+                    if (ftcursor->interrupt_cb && ftcursor->interrupt_cb(ftcursor->interrupt_cb_extra, n_deleted)) {
                         return TOKUDB_INTERRUPTED;
                     }
                     return DB_NOTFOUND;
                 }
                 break;
             case FT_SEARCH_RIGHT:
                 if (idx == 0) {
-                    if (ftcursor->interrupt_cb && ftcursor->interrupt_cb(ftcursor->interrupt_cb_extra)) {
+                    STATUS_INC(FT_CURSOR_SKIP_DELETED_LEAF_ENTRY, n_deleted);
+                    if (ftcursor->interrupt_cb && ftcursor->interrupt_cb(ftcursor->interrupt_cb_extra, n_deleted)) {
                         return TOKUDB_INTERRUPTED;
                     }
                     return DB_NOTFOUND;
@@ -3403,7 +3406,10 @@ ok: ;
             }
             r = bn->data_buffer.fetch_klpair(idx, &le, &keylen, &key);
             assert_zero(r); // we just validated the index
-            if (!le_val_is_del(le, ftcursor->read_type, ftcursor->ttxn)) {
+            if (!le_val_is_del(le, ftcursor->is_snapshot_read, ftcursor->ttxn)) {
+                STATUS_INC(FT_CURSOR_SKIP_DELETED_LEAF_ENTRY, n_deleted);
+                if (ftcursor->interrupt_cb)
+                    ftcursor->interrupt_cb(ftcursor->interrupt_cb_extra, n_deleted);
                 goto got_a_good_value;
             }
         }
@@ -3414,7 +3420,7 @@ ok: ;
         void *val;
 
         le_extract_val(le, toku_ft_cursor_is_leaf_mode(ftcursor),
-                       ftcursor->read_type, ftcursor->ttxn,
+                       ftcursor->is_snapshot_read, ftcursor->ttxn,
                        &vallen, &val);
         r = toku_ft_cursor_check_restricted_range(ftcursor, key, keylen);
         if (r == 0) {

storage/tokudb/ft-index/ft/ft.cc

@@ -1062,6 +1062,11 @@ garbage_helper(BLOCKNUM blocknum, int64_t UU(size), int64_t UU(address), void *e
             goto exit;
         }
     }
+    {
+        float a = info->used_space, b=info->total_space;
+        float percentage = (1 - (a / b)) * 100;
+        printf("LeafNode# %d has %d BasementNodes and %2.1f%% of the allocated space is garbage\n", (int)blocknum.b, node->n_children, percentage);
+    }
 exit:
     toku_ftnode_free(&node);
     toku_free(ndd);

storage/tokudb/ft-index/ft/leafentry.h

@@ -124,12 +124,6 @@ PATENT RIGHTS GRANT:
         val[]
 */
 
-enum cursor_read_type {
-    C_READ_ANY = 0,
-    C_READ_SNAPSHOT = 1,
-    C_READ_COMMITTED = 2
-};
-
 //
 // enum of possible values for LEAFENTRY->type field
 // LE_CLEAN means that there is a single committed value in a format that saves disk space
@@ -217,7 +211,7 @@ void wbuf_nocrc_LEAFENTRY(struct wbuf *w, LEAFENTRY le);
 int print_klpair (FILE *outf, const void* key, uint32_t keylen, LEAFENTRY v); // Print a leafentry out in human-readable form.
 
 int le_latest_is_del(LEAFENTRY le); // Return true if it is a provisional delete.
-int le_val_is_del(LEAFENTRY le, enum cursor_read_type read_type, TOKUTXN txn); // Returns true if the value that is to be read is empty
+int le_val_is_del(LEAFENTRY le, bool is_snapshot_read, TOKUTXN txn); // Returns true if the value that is to be read is empty
 bool le_is_clean(LEAFENTRY le); //Return how many xids exist (0 does not count)
 bool le_has_xids(LEAFENTRY le, XIDS xids); // Return true transaction represented by xids is still provisional in this leafentry (le's xid stack is a superset or equal to xids)
 void*     le_latest_val (LEAFENTRY le); // Return the latest val (return NULL for provisional deletes)
@@ -232,13 +226,13 @@ uint64_t le_outermost_uncommitted_xid (LEAFENTRY le);
 //      0:  context ignores this entry, id.
 //      TOKUDB_ACCEPT: context accepts id
 //      r|r!=0&&r!=TOKUDB_ACCEPT:  Quit early, return r, because something unexpected went wrong (error case)
-typedef int(*LE_ITERATE_CALLBACK)(TXNID id, TOKUTXN context, bool is_provisional);
+typedef int(*LE_ITERATE_CALLBACK)(TXNID id, TOKUTXN context);
 
 int le_iterate_val(LEAFENTRY le, LE_ITERATE_CALLBACK f, void** valpp, uint32_t *vallenp, TOKUTXN context);
 
 void le_extract_val(LEAFENTRY le,
                     // should we return the entire leafentry as the val?
-                    bool is_leaf_mode, enum cursor_read_type read_type,
+                    bool is_leaf_mode, bool is_snapshot_read,
                     TOKUTXN ttxn, uint32_t *vallen, void **val);
 
 size_t

storage/tokudb/ft-index/ft/logger/recover.cc

@@ -712,7 +712,7 @@ static int toku_recover_xstillopenprepared (struct logtype_xstillopenprepared *l
     }
     switch (renv->ss.ss) {
         case FORWARD_BETWEEN_CHECKPOINT_BEGIN_END: {
-            toku_txn_prepare_txn(txn, l->xa_xid);
+            toku_txn_prepare_txn(txn, l->xa_xid, 0);
             break;
         }
         case FORWARD_NEWER_CHECKPOINT_END: {
@@ -754,7 +754,7 @@ static int toku_recover_xcommit (struct logtype_xcommit *l, RECOVER_ENV renv) {
     assert(txn!=NULL);
 
     // commit the transaction
-    int r = toku_txn_commit_with_lsn(txn, true, l->lsn, false,
+    int r = toku_txn_commit_with_lsn(txn, true, l->lsn,
                                  NULL, NULL);
     assert(r == 0);
 
@@ -776,7 +776,7 @@ static int toku_recover_xprepare (struct logtype_xprepare *l, RECOVER_ENV renv)
     assert(txn!=NULL);
 
     // Save the transaction
-    toku_txn_prepare_txn(txn, l->xa_xid);
+    toku_txn_prepare_txn(txn, l->xa_xid, 0);
 
     return 0;
 }

storage/tokudb/ft-index/ft/tests/is_empty.cc

@@ -121,7 +121,7 @@ static void test_it (int N) {
 
     r = toku_open_ft_handle(FILENAME, 1, &ft, 1024, 256, TOKU_DEFAULT_COMPRESSION_METHOD, ct, txn, toku_builtin_compare_fun);                    CKERR(r);
 
-    r = toku_txn_commit_txn(txn, false, false, NULL, NULL);                                 CKERR(r);
+    r = toku_txn_commit_txn(txn, false, NULL, NULL);                                 CKERR(r);
     toku_txn_close_txn(txn);
     CHECKPOINTER cp = toku_cachetable_get_checkpointer(ct);
     r = toku_checkpoint(cp, logger, NULL, NULL, NULL, NULL, CLIENT_CHECKPOINT);                             CKERR(r);
@@ -131,7 +131,7 @@ static void test_it (int N) {
     for (int i=0; i<N; i++) {
 	r = toku_txn_begin_txn((DB_TXN*)NULL, (TOKUTXN)0, &txn, logger, TXN_SNAPSHOT_ROOT, false);                 CKERR(r);
 	r = toku_open_ft_handle(FILENAME, 0, &ft, 1024, 256, TOKU_DEFAULT_COMPRESSION_METHOD, ct, txn, toku_builtin_compare_fun);                CKERR(r);
-	r = toku_txn_commit_txn(txn, false, false, NULL, NULL);                             CKERR(r);
+	r = toku_txn_commit_txn(txn, false, NULL, NULL);                             CKERR(r);
 	toku_txn_close_txn(txn);
 
 	r = toku_txn_begin_txn((DB_TXN*)NULL, (TOKUTXN)0, &txn, logger, TXN_SNAPSHOT_ROOT, false);                 CKERR(r);
@@ -142,7 +142,7 @@ static void test_it (int N) {
 	memset(val, 'v', sizeof(val));
 	val[sizeof(val)-1]=0;
 	toku_ft_insert(ft, toku_fill_dbt(&k, key, 1+strlen(key)), toku_fill_dbt(&v, val, 1+strlen(val)), txn);
-	r = toku_txn_commit_txn(txn, false, false, NULL, NULL);                                 CKERR(r);
+	r = toku_txn_commit_txn(txn, false, NULL, NULL);                                 CKERR(r);
 	toku_txn_close_txn(txn);
 
 
@@ -154,7 +154,7 @@ static void test_it (int N) {
     for (int i=0; i<N; i++) {
 	r = toku_txn_begin_txn((DB_TXN*)NULL, (TOKUTXN)0, &txn, logger, TXN_SNAPSHOT_ROOT, false);                     CKERR(r);
 	r = toku_open_ft_handle(FILENAME, 0, &ft, 1024, 256, TOKU_DEFAULT_COMPRESSION_METHOD, ct, txn, toku_builtin_compare_fun);                CKERR(r);
-	r = toku_txn_commit_txn(txn, false, false, NULL, NULL);                                 CKERR(r);
+	r = toku_txn_commit_txn(txn, false, NULL, NULL);                                 CKERR(r);
 	toku_txn_close_txn(txn);
 
 	r = toku_txn_begin_txn((DB_TXN*)NULL, (TOKUTXN)0, &txn, logger, TXN_SNAPSHOT_ROOT, false);                     CKERR(r);
@@ -169,7 +169,7 @@ static void test_it (int N) {
 	assert(!is_empty);
 	}
 
-	r = toku_txn_commit_txn(txn, false, false, NULL, NULL);                                 CKERR(r);
+	r = toku_txn_commit_txn(txn, false, NULL, NULL);                                 CKERR(r);
 	toku_txn_close_txn(txn);
 
 	r = toku_checkpoint(cp, logger, NULL, NULL, NULL, NULL, CLIENT_CHECKPOINT);                             CKERR(r);
@@ -179,7 +179,7 @@ static void test_it (int N) {
     }
     r = toku_txn_begin_txn((DB_TXN*)NULL, (TOKUTXN)0, &txn, logger, TXN_SNAPSHOT_ROOT, false);                        CKERR(r);
     r = toku_open_ft_handle(FILENAME, 0, &ft, 1024, 256, TOKU_DEFAULT_COMPRESSION_METHOD, ct, txn, toku_builtin_compare_fun);                       CKERR(r);
-    r = toku_txn_commit_txn(txn, false, false, NULL, NULL);                                     CKERR(r);
+    r = toku_txn_commit_txn(txn, false, NULL, NULL);                                     CKERR(r);
     toku_txn_close_txn(txn);
 
     if (0) {

storage/tokudb/ft-index/ft/tests/le-cursor-provdel.cc

@@ -140,7 +140,7 @@ create_populate_tree(const char *logdir, const char *fname, int n) {
     error = toku_open_ft_handle(fname, 1, &ft, 1<<12, 1<<9, TOKU_DEFAULT_COMPRESSION_METHOD, ct, txn, test_ft_cursor_keycompare);
     assert(error == 0);
 
-    error = toku_txn_commit_txn(txn, true, false, NULL, NULL);
+    error = toku_txn_commit_txn(txn, true, NULL, NULL);
     assert(error == 0);
     toku_txn_close_txn(txn);
 
@@ -160,7 +160,7 @@ create_populate_tree(const char *logdir, const char *fname, int n) {
         assert(error == 0);
     }
 
-    error = toku_txn_commit_txn(txn, true, false, NULL, NULL);
+    error = toku_txn_commit_txn(txn, true, NULL, NULL);
     assert(error == 0);
     toku_txn_close_txn(txn);
 
@@ -210,7 +210,7 @@ test_provdel(const char *logdir, const char *fname, int n) {
     error = toku_open_ft_handle(fname, 1, &ft, 1<<12, 1<<9, TOKU_DEFAULT_COMPRESSION_METHOD, ct, txn, test_ft_cursor_keycompare);
     assert(error == 0);
 
-    error = toku_txn_commit_txn(txn, true, false, NULL, NULL);
+    error = toku_txn_commit_txn(txn, true, NULL, NULL);
     assert(error == 0);
     toku_txn_close_txn(txn);
 
@@ -257,11 +257,11 @@ test_provdel(const char *logdir, const char *fname, int n) {
 
     toku_le_cursor_close(cursor);
 
-    error = toku_txn_commit_txn(cursortxn, true, false, NULL, NULL);
+    error = toku_txn_commit_txn(cursortxn, true, NULL, NULL);
     assert(error == 0);
     toku_txn_close_txn(cursortxn);
 
-    error = toku_txn_commit_txn(txn, true, false, NULL, NULL);
+    error = toku_txn_commit_txn(txn, true, NULL, NULL);
     assert(error == 0);
     toku_txn_close_txn(txn);