数据库内核月报 － 2016 / 12

typedef struct __toku_loader DB_LOADER;
struct __toku_loader {
  struct __toku_loader_internal *i;
  int (*set_error_callback)(DB_LOADER *loader, void (*error_cb)(DB *db, int i, int err, DBT *key, DBT *val, void *error_extra), void *error_extra);
  int (*set_poll_function)(DB_LOADER *loader, int (*poll_func)(void *extra, float progress), void *poll_extra);
  int (*put)(DB_LOADER *loader, DBT *key, DBT* val);
  int (*close)(DB_LOADER *loader);
  int (*abort)(DB_LOADER *loader);
};

struct __toku_loader_internal {
    DB_ENV *env;
    DB_TXN *txn;
    FTLOADER ft_loader;
    int N;
    DB **dbs; /* [N] */
    DB *src_db;
    uint32_t *db_flags;
    uint32_t *dbt_flags;
    uint32_t loader_flags;
    void (*error_callback)(DB *db, int i, int err, DBT *key, DBT *val, void *error_extra);
    void *error_extra;
    int  (*poll_func)(void *poll_extra, float progress);
    void *poll_extra;
    char *temp_file_template;
    DBT err_key;   /* error key */
    DBT err_val;   /* error val */
    int err_i;     /* error i   */
    int err_errno;
    char **inames_in_env; /* [N]  inames of new files to be created */

};

struct ft_loader_s {
    ...
    generate_row_for_put_func generate_row_for_put;
    CACHETABLE cachetable;
    bool did_reserve_memory;
    bool compress_intermediates;
    bool allow_puts;
    uint64_t reserved_memory;
    ...
}

struct ft_loader_s {
    ...
    DB *src_db;
    int N;
    DB **dbs; // N of these
    DESCRIPTOR *descriptors; // N of these.
    TXNID *root_xids_that_created; // N of these.
    const char **new_fnames_in_env; // N of these.
    ft_compare_func *bt_compare_funs; // N of these
    ...
}

struct ft_loader_s {
    ...
    uint64_t n_rows;
    struct rowset primary_rowset;
    struct rowset primary_rowset_temp;
    QUEUE primary_rowset_queue;
    toku_pthread_t extractor_thread;
    bool extractor_live;
    struct rowset *rows;// N of these.
    uint64_t *extracted_datasizes; // N of these.
    DBT  *last_key;// N of these.
    struct file_infos file_infos;
    ...
}

struct ft_loader_s {
    ...
    LSN load_lsn;
    TXNID load_root_xid;
    struct merge_fileset *fs;// N of these.
    QUEUE *fractal_queues; // N of these.
    toku_pthread_t *fractal_threads;// N of these.
    bool *fractal_threads_live; // N of these.
    unsigned fractal_workers;
    ...
};

int
toku_loader_create_loader(DB_ENV *env,
                          DB_TXN *txn,
                          DB_LOADER **blp,
                          DB *src_db,
                          int N,
                          DB *dbs[],
                          uint32_t db_flags[/*N*/],
                          uint32_t dbt_flags[/*N*/],
                          uint32_t loader_flags,
                          bool check_empty) {
    // lock tables and check empty
    for(int i=0;i<N;i++) {
        if (!(loader_flags&DB_PRELOCKED_WRITE)) {
            rval = toku_db_pre_acquire_table_lock(dbs[i], txn);
            if (rval!=0) {
                goto create_exit;
            }
        }
        if (check_empty) {
            bool empty = toku_ft_is_empty_fast(dbs[i]->i->ft_handle);
            if (!empty) {
                rval = ENOTEMPTY;
                goto create_exit;
            }
        }
    }

    {
        if (env->i->open_flags & DB_INIT_TXN) {
            rval = env->txn_begin(env, txn, &loader_txn, 0);
            if (rval) {
                goto create_exit;
            }
        }

        ft_compare_func compare_functions[N];
        for (int i=0; i<N; i++) {
            compare_functions[i] = env->i->bt_compare;
        }

        // time to open the big kahuna
        char **XMALLOC_N(N, new_inames_in_env);
        for (int i = 0; i < N; i++) {
            new_inames_in_env[i] = nullptr;
        }
        FT_HANDLE *XMALLOC_N(N, fts);
        for (int i=0; i<N; i++) {
            fts[i] = dbs[i]->i->ft_handle;
        }
        LSN load_lsn;
        rval = locked_load_inames(env, loader_txn, N, dbs, new_inames_in_env, &load_lsn, puts_allowed);
        if ( rval!=0 ) {
        }
        TOKUTXN ttxn = loader_txn ? db_txn_struct_i(loader_txn)->tokutxn : NULL;
        rval = toku_ft_loader_open(&loader->i->ft_loader,
                                   env->i->cachetable,
                                   env->i->generate_row_for_put,
                                   src_db,
                                   N,
                                   fts, 
                                   dbs,
                                   (const char **)new_inames_in_env,
                                   compare_functions,
                                   loader->i->temp_file_template,
                                   load_lsn,
                                   ttxn,
                                   puts_allowed,
                                   env->get_loader_memory_size(env),
                                   compress_intermediates,
                                   puts_allowed);
        if ( rval!=0 ) {
        }

        loader->i->inames_in_env = new_inames_in_env;
        toku_free(fts);

        if (!puts_allowed) {
            rval = ft_loader_close_and_redirect(loader);
            assert_zero(rval);
            loader->i->ft_loader = NULL;
            // close the ft_loader and skip to the redirection
            rval = 0;
        }

        rval = loader_txn->commit(loader_txn, 0);
        assert_zero(rval);
        loader_txn = nullptr;

        rval = 0;
    }
    *blp = loader;
 create_exit:
    if (loader_txn) {
        int r  = loader_txn->abort(loader_txn);
        assert_zero(r);
        loader_txn = nullptr;
    }
    if (rval == 0) {
    }
    else {
        free_loader(loader);
    }
    return rval;
}

int toku_ft_loader_open (FTLOADER *blp, /* out */
                         CACHETABLE cachetable,
                         generate_row_for_put_func g,
                         DB *src_db,
                         int N, FT_HANDLE fts[/*N*/], 
                         DB* dbs[/*N*/],
                         const char *new_fnames_in_env[/*N*/],
                         ft_compare_func bt_compare_functions[/*N*/],
                         const char *temp_file_template,
                         LSN load_lsn,
                         TOKUTXN txn,
                         bool reserve_memory,
                         uint64_t reserve_memory_size,
                         bool compress_intermediates,
                         bool allow_puts) {
    int result = 0;
    {
        int r = toku_ft_loader_internal_init(blp, cachetable, g, 
                                             src_db,
                                             N, fts, dbs,
                                             new_fnames_in_env,
                                             bt_compare_functions,
                                             temp_file_template,
                                             load_lsn,
                                             txn,
                                             reserve_memory,
                                             reserve_memory_size,
                                             compress_intermediates,
                                             allow_puts);
        if (r!=0) result = r;
    }
    if (result==0 && allow_puts) {
        FTLOADER bl = *blp;
        int r = toku_pthread_create(&bl->extractor_thread, NULL, extractor_thread, (void*)bl);
        if (r==0) {
            bl->extractor_live = true;
        } else  {
            result = r;
            (void) toku_ft_loader_internal_destroy(bl, true);
        }
    }
    return result;
}

static uint64_t memory_per_rowset_during_extract (FTLOADER bl)
// Return how much memory can be allocated for each rowset.
{
    if (size_factor==1) {
        return 16*1024;
    } else {
        // There is a primary rowset being maintained by the foreground thread.
        // There could be two more in the queue.
        // There is one rowset for each index (bl->N) being filled in.
        // Later we may have sort_and_write operations spawning in parallel, and will need to account for that.
        int n_copies = (1 // primary rowset
                        +EXTRACTOR_QUEUE_DEPTH  // the number of primaries in the queue
                        +bl->N // the N rowsets being constructed by the extractor thread.
                        +bl->N // the N sort buffers
                        +1     // Give the extractor thread one more so that it can have temporary space for sorting.  This is overkill.
                        );
        int64_t extra_reserved_memory = bl->N * FILE_BUFFER_SIZE;  // for each index we are writing to a file at any given time.
        int64_t tentative_rowset_size = ((int64_t)(bl->reserved_memory - extra_reserved_memory))/(n_copies);
        return MAX(tentative_rowset_size, (int64_t)MIN_ROWSET_MEMORY);
    }
}

int toku_loader_put(DB_LOADER *loader, DBT *key, DBT *val)
{
    int r = 0;
    int i = 0;

    if (loader->i->loader_flags & LOADER_DISALLOW_PUTS) {
        r = EINVAL;
        goto cleanup;
    }
    else {
        r = toku_ft_loader_put(loader->i->ft_loader, key, val);
    }
    if ( r != 0 ) {
    }
 cleanup:
    return r;
}

int toku_ft_loader_put (FTLOADER bl, DBT *key, DBT *val)
/* Effect: Put a key-value pair into the ft loader.  Called by DB_LOADER->put().
 * Return value: 0 on success, an error number otherwise.
 */
{
    if (!bl->allow_puts || ft_loader_get_error(&bl->error_callback))
        return EINVAL; // previous panic
    bl->n_rows++;
    return loader_do_put(bl, key, val);
}

struct row {
    size_t off; // the offset in the data array.
    int klen,vlen;
};

struct rowset {
    uint64_t memory_budget;
    size_t n_rows, n_rows_limit;
    struct row *rows;
    size_t n_bytes, n_bytes_limit;
    char *data;
};

static int row_wont_fit (struct rowset *rows, size_t size)
/* Effect: Return nonzero if adding a row of size SIZE would be too big (bigger than the buffer limit) */
{
    // Account for the memory used by the data and also the row structures.
    size_t memory_in_use = (rows->n_rows*sizeof(struct row)
                            + rows->n_bytes);
    return (rows->memory_budget <  memory_in_use + size);
}

static int loader_do_put(FTLOADER bl,
                         DBT *pkey,
                         DBT *pval)
{
    int result;
    result = add_row(&bl->primary_rowset, pkey, pval);
    if (result == 0 && row_wont_fit(&bl->primary_rowset, 0)) {
        // queue the rows for further processing by the extractor thread
        enqueue_for_extraction(bl);
        {
            int r = init_rowset(&bl->primary_rowset, memory_per_rowset_during_extract(bl));
            // bl->primary_rowset will get destroyed by toku_ft_loader_abort
            if (r != 0)
                result = r;
        }
    }
    return result;
}

static void enqueue_for_extraction (FTLOADER bl) {
    struct rowset *XMALLOC(enqueue_me);
    *enqueue_me = bl->primary_rowset;
    zero_rowset(&bl->primary_rowset);
    int r = toku_queue_enq(bl->primary_rowset_queue, (void*)enqueue_me, 1, NULL);
    resource_assert_zero(r);
}

static void* extractor_thread (void *blv) {
    FTLOADER bl = (FTLOADER)blv;
    int r = 0;
    while (1) {
        void *item;
        {
            int rq = toku_queue_deq(bl->primary_rowset_queue, &item, NULL, NULL);
            if (rq==EOF) break;
            invariant(rq==0); // other errors are arbitrarily bad.
        }
        struct rowset *primary_rowset = (struct rowset *)item;
        {
            r = process_primary_rows(bl, primary_rowset);
            if (r)
                ft_loader_set_panic(bl, r, false, 0, nullptr, nullptr);
        }
    }

    if (r == 0) {
        r = finish_primary_rows(bl);
        if (r)
            ft_loader_set_panic(bl, r, false, 0, nullptr, nullptr);

    }
    return NULL;
}

static int process_primary_rows_internal (FTLOADER bl, struct rowset *primary_rowset)
{
    int error_count = 0;
    int *XMALLOC_N(bl->N, error_codes);

    for (int i = 0; i < bl->N; i++) {
        error_codes[i] = 0;
        struct rowset *rows = &(bl->rows[i]);
        struct merge_fileset *fs = &(bl->fs[i]);
        ft_compare_func compare = bl->bt_compare_funs[i];

        for (size_t prownum=0; prownum<primary_rowset->n_rows; prownum++) {
            if (error_count) break;

            struct row *prow = &primary_rowset->rows[prownum];
            DBT pkey，pval；
            pkey.data = primary_rowset->data + prow->off;
            pkey.size = prow->klen;
            pval.data = primary_rowset->data + prow->off + prow->klen;
            pval.size = prow->vlen;

            DBT_ARRAY key_array;
            DBT_ARRAY val_array;
            if (bl->dbs[i] != bl->src_db) {
                int r = bl->generate_row_for_put(bl->dbs[i], bl->src_db, &dest_keys, &dest_vals, &pkey, &pval);
                if (r != 0) {
                    error_codes[i] = r;
                    inc_error_count();
                    break;
                }
                paranoid_invariant(dest_keys.size <= dest_keys.capacity);
                paranoid_invariant(dest_vals.size <= dest_vals.capacity);
                paranoid_invariant(dest_keys.size == dest_vals.size);

                key_array = dest_keys;
                val_array = dest_vals;
            } else {
                key_array.size = key_array.capacity = 1;
                key_array.dbts = &pkey;

                val_array.size = val_array.capacity = 1;
                val_array.dbts = &pval;
            }
            for (uint32_t row = 0; row < key_array.size; row++) {
                DBT *dest_key = &key_array.dbts[row];
                DBT *dest_val = &val_array.dbts[row];

                bl->extracted_datasizes[i] += ft_loader_leafentry_size(dest_key->size, dest_val->size, leafentry_xid(bl, i));

                if (row_wont_fit(rows, dest_key->size + dest_val->size)) {
                    int r = sort_and_write_rows(*rows, fs, bl, i, bl->dbs[i], compare);
                    init_rowset(rows, memory_per_rowset_during_extract(bl));
                    if (r != 0) {
                        error_codes[i] = r;
                        inc_error_count();
                        break;
                    }
                }
                int r = add_row(rows, dest_key, dest_val);
                if (r != 0) {
                    break;
                }
            }
        }
    }
    toku_dbt_array_destroy(&dest_keys);
    toku_dbt_array_destroy(&dest_vals);

    destroy_rowset(primary_rowset);
    toku_free(primary_rowset);
    int r = 0;
    if (error_count > 0) {
    }
    toku_free(error_codes);
    return r;
}

int sort_and_write_rows (struct rowset rows, struct merge_fileset *fs, FTLOADER bl, int which_db, DB *dest_db, ft_compare_func compare)
{
    int result;
    if (rows.n_rows == 0) {
        result = 0;
    } else {
        result = sort_rows(&rows, which_db, dest_db, compare, bl);

        if (result == 0) {
            DBT min_rowset_key = make_dbt(rows.data+rows.rows[0].off, rows.rows[0].klen);
            if (fs->have_sorted_output && compare(dest_db, &fs->prev_key, &min_rowset_key) < 0) {
                // write everything to the same output if the max key in the temp file (prev_key) is < min of the sorted rowset
                result = write_rowset_to_file(bl, fs->sorted_output, rows);
                if (result == 0) {
                    // set the max key in the temp file to the max key in the sorted rowset
                    result = toku_dbt_set(rows.rows[rows.n_rows-1].klen, rows.data + rows.rows[rows.n_rows-1].off, &fs->prev_key, NULL);
                }
            } else {
                // write the sorted rowset into a new temp file
                if (fs->have_sorted_output) {
                    fs->have_sorted_output = false;
                    result = ft_loader_fi_close(&bl->file_infos, fs->sorted_output, true);
                }
                if (result == 0) {
                    FIDX sfile = FIDX_NULL;
                    result = extend_fileset(bl, fs, &sfile);
                    if (result == 0) {
                        result = write_rowset_to_file(bl, sfile, rows);
                        if (result == 0) {
                            fs->have_sorted_output = true;
                            fs->sorted_output = sfile;
                            // set the max key in the temp file to the max key in the sorted rowset
                            result = toku_dbt_set(rows.rows[rows.n_rows-1].klen, rows.data + rows.rows[rows.n_rows-1].off, &fs->prev_key, NULL);
                        }
                    }
                }
            }
        }
    }

    destroy_rowset(&rows);

    return result;
}

int toku_loader_close(DB_LOADER *loader)
{
    if ( loader->i->err_errno != 0 ) {
        if ( loader->i->error_callback != NULL ) {
            loader->i->error_callback(loader->i->dbs[loader->i->err_i], loader->i->err_i, loader->i->err_errno, &loader->i->err_key, &loader->i->err_val, loader->i->error_extra);
        }
        if (!(loader->i->loader_flags & LOADER_DISALLOW_PUTS ) ) {
            r = toku_ft_loader_abort(loader->i->ft_loader, true);
            redirect_loader_to_empty_dictionaries(loader);
        }
        else {
            r = loader->i->err_errno;
        }
    }
    else { // no error outstanding
        if (!(loader->i->loader_flags & LOADER_DISALLOW_PUTS ) ) {
            r = ft_loader_close_and_redirect(loader);
            if (r) {
                redirect_loader_to_empty_dictionaries(loader);
            }
        }
    }
    free_loader(loader);
    return r;
}

static int ft_loader_close_and_redirect(DB_LOADER *loader) {
    int r;
    // use the bulk loader
    // in case you've been looking - here is where the real work is done!
    r = toku_ft_loader_close(loader->i->ft_loader, loader->i->error_callback,
                             loader->i->error_extra, loader->i->poll_func, loader->i->poll_extra);
    if ( r==0 ) {
        for (int i=0; i<loader->i->N; i++) {
            toku_multi_operation_client_lock(); //Must hold MO lock for dictionary_redirect.
            r = toku_dictionary_redirect(loader->i->inames_in_env[i],
                                         loader->i->dbs[i]->i->ft_handle,
                                         db_txn_struct_i(loader->i->txn)->tokutxn);
            toku_multi_operation_client_unlock();
            if ( r!=0 ) break;
        }
    }
    return r;
}

int toku_ft_loader_close (FTLOADER bl, ft_loader_error_func error_function, void *error_extra,
                          ft_loader_poll_func  poll_function,  void *poll_extra)
{
    int result = 0;
    int r;

    ft_loader_set_error_function(&bl->error_callback, error_function, error_extra);
    ft_loader_set_poll_function(&bl->poll_callback, poll_function, poll_extra);

    if (bl->extractor_live) {
        r = finish_extractor(bl);
        if (r)
            result = r;
        invariant(!bl->extractor_live);
    } else {
        r = finish_primary_rows(bl);
        if (r)
            result = r;
    }

    // check for an error during extraction
    if (result == 0) {
        r = ft_loader_call_error_function(&bl->error_callback);
        if (r)
            result = r;
    }

    if (result == 0) {
        r = toku_ft_loader_close_internal(bl);
        if (r && result == 0)
            result = r;
    } else
        toku_ft_loader_internal_destroy(bl, true);

    return result;
}

static int loader_do_i (FTLOADER bl,
                        int which_db,
                        DB *dest_db,
                        ft_compare_func compare,
                        const DESCRIPTOR descriptor,
                        const char *new_fname,
                        int progress_allocation // how much progress do I need to add into bl->progress by the end..
                        )
/* Effect: Handle the file creating for one particular DB in the bulk loader. */
/* Requires: The data is fully extracted, so we can do merges out of files and write the ft file. */
{
    struct merge_fileset *fs = &(bl->fs[which_db]);
    struct rowset *rows = &(bl->rows[which_db]);
    invariant(rows->data==NULL); // the rows should be all cleaned up already

    int r = toku_queue_create(&bl->fractal_queues[which_db], FRACTAL_WRITER_QUEUE_DEPTH);
    if (r) goto error;

    {
        mode_t mode = S_IRUSR+S_IWUSR + S_IRGRP+S_IWGRP;
        int fd = toku_os_open(new_fname, O_RDWR| O_CREAT | O_BINARY, mode); // #2621
        if (fd < 0) {
            r = get_error_errno(); goto error;
        }

        uint32_t target_nodesize, target_basementnodesize, target_fanout;
        enum toku_compression_method target_compression_method;
        r = dest_db->get_pagesize(dest_db, &target_nodesize);
        invariant_zero(r);
        r = dest_db->get_readpagesize(dest_db, &target_basementnodesize);
        invariant_zero(r);
        r = dest_db->get_compression_method(dest_db, &target_compression_method);
        invariant_zero(r);
        r = dest_db->get_fanout(dest_db, &target_fanout);
        invariant_zero(r);

        if (bl->allow_puts) {
            // a better allocation would be to figure out roughly how many merge passes we'll need.
            int allocation_for_merge = (2*progress_allocation)/3;
            progress_allocation -= allocation_for_merge;

            // This structure must stay live until the join below.
            struct fractal_thread_args fta = {
                bl,
                descriptor,
                fd,
                progress_allocation,
                bl->fractal_queues[which_db],
                bl->extracted_datasizes[which_db],
                0,
                which_db,
                target_nodesize,
                target_basementnodesize,
                target_compression_method,
                target_fanout
            };

            r = toku_pthread_create(bl->fractal_threads+which_db, NULL, fractal_thread, (void*)&fta);
            if (r) {
                int r2 __attribute__((__unused__)) = toku_queue_destroy(bl->fractal_queues[which_db]);
                // ignore r2, since we already have an error
                bl->fractal_queues[which_db] = nullptr;
                goto error;
            }
            invariant(bl->fractal_threads_live[which_db]==false);
            bl->fractal_threads_live[which_db] = true;

            r = merge_files(fs, bl, which_db, dest_db, compare, allocation_for_merge, bl->fractal_queues[which_db]);

            {
                void *toku_pthread_retval;
                int r2 = toku_pthread_join(bl->fractal_threads[which_db], &toku_pthread_retval);
                invariant(fta.bl==bl); // this is a gratuitous assertion to make sure that the fta struct is still live here.  A previous bug put that struct into a C block statement.
                resource_assert_zero(r2);
                invariant(toku_pthread_retval==NULL);
                invariant(bl->fractal_threads_live[which_db]);
                bl->fractal_threads_live[which_db] = false;
                if (r == 0) r = fta.errno_result;
            }
        } else {
            toku_queue_eof(bl->fractal_queues[which_db]);
            r = toku_loader_write_ft_from_q(bl, descriptor, fd, progress_allocation,
                                            bl->fractal_queues[which_db], bl->extracted_datasizes[which_db], which_db,
                                            target_nodesize, target_basementnodesize, target_compression_method, target_fanout);
        }
    }

 error: // this is the cleanup code.  Even if r==0 (no error) we fall through to here.
    if (bl->fractal_queues[which_db]) {
        int r2 = toku_queue_destroy(bl->fractal_queues[which_db]);
        invariant(r2==0);
        bl->fractal_queues[which_db] = nullptr;
    }

    // if we get here we need to free up the merge_fileset and the rowset, as well as the keys
    toku_free(rows->data); rows->data = NULL;
    toku_free(rows->rows); rows->rows = NULL;
    toku_free(fs->data_fidxs); fs->data_fidxs = NULL;
    return r;
}

// To compute a merge, we have a certain amount of memory to work with.
// We perform only one fanin at a time.
// If the fanout is F then we are using
//   F merges.  Each merge uses
//   DBUFIO_DEPTH buffers for double buffering.  Each buffer is of size at least MERGE_BUF_SIZE
// so the memory is
//   F*MERGE_BUF_SIZE*DBUFIO_DEPTH storage.
// We use some additional space to buffer the outputs.
//  That's FILE_BUFFER_SIZE for writing to a merge file if we are writing to a mergefile.
//  And we have FRACTAL_WRITER_ROWSETS*MERGE_BUF_SIZE per queue
//  And if we are doing a fractal, each worker could have have a fractal tree that it's working on.
//
// DBUFIO_DEPTH*F*MERGE_BUF_SIZE + FRACTAL_WRITER_ROWSETS*MERGE_BUF_SIZE + WORKERS*NODESIZE*2 <= RESERVED_MEMORY

static int64_t memory_avail_during_merge(FTLOADER bl, bool is_fractal_node) {
    // avail memory = reserved memory - WORKERS*NODESIZE*2 for the last merge stage only
    int64_t avail_memory = bl->reserved_memory;
    if (is_fractal_node) {
        // reserve space for the fractal writer thread buffers
        avail_memory -= (int64_t)ft_loader_get_fractal_workers_count(bl) * (int64_t)default_loader_nodesize * 2; // compressed and uncompressed buffers
    }
    return avail_memory;
}

static uint64_t memory_per_rowset_during_merge (FTLOADER bl, int merge_factor, bool is_fractal_node // if it is being sent to a q
                                                ) {
    int64_t memory_avail = memory_avail_during_merge(bl, is_fractal_node);
    int64_t nbuffers = DBUFIO_DEPTH * merge_factor;
    if (is_fractal_node)
        nbuffers += FRACTAL_WRITER_ROWSETS;
    return MAX(memory_avail / nbuffers, (int64_t)MIN_MERGE_BUF_SIZE);
}

static void redirect_loader_to_empty_dictionaries(DB_LOADER *loader) {
    DB_LOADER* tmp_loader = NULL;
    int r = toku_loader_create_loader(
        loader->i->env,
        loader->i->txn,
        &tmp_loader,
        loader->i->src_db,
        loader->i->N,
        loader->i->dbs,
        loader->i->db_flags,
        loader->i->dbt_flags,
        LOADER_DISALLOW_PUTS,
        false
        );
    lazy_assert_zero(r);
    r = toku_loader_close(tmp_loader);
}

int
toku_loader_create_loader(DB_ENV *env,
                          DB_TXN *txn,
                          DB_LOADER **blp,
                          DB *src_db,
                          int N,
                          DB *dbs[],
                          uint32_t db_flags[/*N*/],
                          uint32_t dbt_flags[/*N*/],
                          uint32_t loader_flags,
                          bool check_empty) {
        rval = locked_load_inames(env, loader_txn, N, dbs, new_inames_in_env, &load_lsn, puts_allowed);
        if ( rval!=0 ) {
            free_inames(new_inames_in_env, N);
            toku_free(fts);
            goto create_exit;
        }
        TOKUTXN ttxn = loader_txn ? db_txn_struct_i(loader_txn)->tokutxn : NULL;
        rval = toku_ft_loader_open(&loader->i->ft_loader,
                                   env->i->cachetable,
                                   env->i->generate_row_for_put,
                                   src_db,
                                   N,
                                   fts, dbs,
                                   (const char **)new_inames_in_env,
                                   compare_functions,
                                   loader->i->temp_file_template,
                                   load_lsn,
                                   ttxn,
                                   puts_allowed,
                                   env->get_loader_memory_size(env),
                                   compress_intermediates,
                                   puts_allowed);
        if ( rval!=0 ) {
            free_inames(new_inames_in_env, N);
            toku_free(fts);
            goto create_exit;
        }

        loader->i->inames_in_env = new_inames_in_env;
        toku_free(fts);

        if (!puts_allowed) {
            rval = ft_loader_close_and_redirect(loader);
            assert_zero(rval);
            loader->i->ft_loader = NULL;
            // close the ft_loader and skip to the redirection
            rval = 0;
        }