path: root/src/third_party/wiredtiger/src/cursor/cur_bulk.c

/*-
 * Copyright (c) 2014-2019 MongoDB, Inc.
 * Copyright (c) 2008-2014 WiredTiger, Inc.
 *	All rights reserved.
 *
 * See the file LICENSE for redistribution information.
 */

#include "wt_internal.h"

/*
 * __bulk_col_keycmp_err --
 *     Error routine when column-store keys inserted out-of-order.
 */
static int
__bulk_col_keycmp_err(WT_CURSOR_BULK *cbulk)
{
    WT_CURSOR *cursor;
    WT_SESSION_IMPL *session;

    session = (WT_SESSION_IMPL *)cbulk->cbt.iface.session;
    cursor = &cbulk->cbt.iface;

    WT_RET_MSG(session, EINVAL, "bulk-load presented with out-of-order keys: %" PRIu64
                                " is less "
                                "than previously inserted key %" PRIu64,
      cursor->recno, cbulk->recno);
}

/*
 * __curbulk_insert_fix --
 *     Fixed-length column-store bulk cursor insert.
 */
static int
__curbulk_insert_fix(WT_CURSOR *cursor)
{
    WT_BTREE *btree;
    WT_CURSOR_BULK *cbulk;
    WT_DECL_RET;
    WT_SESSION_IMPL *session;
    uint64_t recno;

    cbulk = (WT_CURSOR_BULK *)cursor;
    btree = cbulk->cbt.btree;

    /*
     * Bulk cursor inserts are updates, but don't need auto-commit transactions because they are
     * single-threaded and not visible until the bulk cursor is closed.
     */
    CURSOR_API_CALL(cursor, session, insert, btree);
    WT_STAT_CONN_INCR(session, cursor_insert_bulk);
    WT_STAT_DATA_INCR(session, cursor_insert_bulk);

    /*
     * If the "append" flag was configured, the application doesn't have to supply a key, else
     * require a key.
     */
    if (F_ISSET(cursor, WT_CURSTD_APPEND))
        recno = cbulk->recno + 1;
    else {
        WT_ERR(__cursor_checkkey(cursor));
        if ((recno = cursor->recno) <= cbulk->recno)
            WT_ERR(__bulk_col_keycmp_err(cbulk));
    }
    WT_ERR(__cursor_checkvalue(cursor));

    /*
     * Insert any skipped records as deleted records, update the current record count.
     */
    for (; recno != cbulk->recno + 1; ++cbulk->recno)
        WT_ERR(__wt_bulk_insert_fix(session, cbulk, true));
    cbulk->recno = recno;

    /* Insert the current record. */
    ret = __wt_bulk_insert_fix(session, cbulk, false);

err:
    API_END_RET(session, ret);
}

/*
 * __curbulk_insert_fix_bitmap --
 *     Fixed-length column-store bulk cursor insert for bitmaps.
 */
static int
__curbulk_insert_fix_bitmap(WT_CURSOR *cursor)
{
    WT_BTREE *btree;
    WT_CURSOR_BULK *cbulk;
    WT_DECL_RET;
    WT_SESSION_IMPL *session;

    cbulk = (WT_CURSOR_BULK *)cursor;
    btree = cbulk->cbt.btree;

    /*
     * Bulk cursor inserts are updates, but don't need auto-commit transactions because they are
     * single-threaded and not visible until the bulk cursor is closed.
     */
    CURSOR_API_CALL(cursor, session, insert, btree);
    WT_STAT_CONN_INCR(session, cursor_insert_bulk);
    WT_STAT_DATA_INCR(session, cursor_insert_bulk);

    WT_ERR(__cursor_checkvalue(cursor));

    /* Insert the current record. */
    ret = __wt_bulk_insert_fix_bitmap(session, cbulk);

err:
    API_END_RET(session, ret);
}

/*
 * __curbulk_insert_var --
 *     Variable-length column-store bulk cursor insert.
 */
static int
__curbulk_insert_var(WT_CURSOR *cursor)
{
    WT_BTREE *btree;
    WT_CURSOR_BULK *cbulk;
    WT_DECL_RET;
    WT_SESSION_IMPL *session;
    uint64_t recno;

    cbulk = (WT_CURSOR_BULK *)cursor;
    btree = cbulk->cbt.btree;

    /*
     * Bulk cursor inserts are updates, but don't need auto-commit transactions because they are
     * single-threaded and not visible until the bulk cursor is closed.
     */
    CURSOR_API_CALL(cursor, session, insert, btree);
    WT_STAT_CONN_INCR(session, cursor_insert_bulk);
    WT_STAT_DATA_INCR(session, cursor_insert_bulk);

    /*
     * If the "append" flag was configured, the application doesn't have to supply a key, else
     * require a key.
     */
    if (F_ISSET(cursor, WT_CURSTD_APPEND))
        recno = cbulk->recno + 1;
    else {
        WT_ERR(__cursor_checkkey(cursor));
        if ((recno = cursor->recno) <= cbulk->recno)
            WT_ERR(__bulk_col_keycmp_err(cbulk));
    }
    WT_ERR(__cursor_checkvalue(cursor));

    if (!cbulk->first_insert) {
        /*
         * If not the first insert and the key space is sequential, compare the current value
         * against the last value; if the same, just increment the RLE count.
         */
        if (recno == cbulk->recno + 1 && cbulk->last.size == cursor->value.size &&
          memcmp(cbulk->last.data, cursor->value.data, cursor->value.size) == 0) {
            ++cbulk->rle;
            ++cbulk->recno;
            goto duplicate;
        }

        /* Insert the previous key/value pair. */
        WT_ERR(__wt_bulk_insert_var(session, cbulk, false));
    } else
        cbulk->first_insert = false;

    /*
     * Insert any skipped records as deleted records, update the current record count and RLE
     * counter.
     */
    if (recno != cbulk->recno + 1) {
        cbulk->rle = (recno - cbulk->recno) - 1;
        WT_ERR(__wt_bulk_insert_var(session, cbulk, true));
    }
    cbulk->rle = 1;
    cbulk->recno = recno;

    /* Save a copy of the value for the next comparison. */
    ret = __wt_buf_set(session, &cbulk->last, cursor->value.data, cursor->value.size);

duplicate:
err:
    API_END_RET(session, ret);
}

/*
 * __bulk_row_keycmp_err --
 *     Error routine when row-store keys inserted out-of-order.
 */
static int
__bulk_row_keycmp_err(WT_CURSOR_BULK *cbulk)
{
    WT_CURSOR *cursor;
    WT_DECL_ITEM(a);
    WT_DECL_ITEM(b);
    WT_DECL_RET;
    WT_SESSION_IMPL *session;

    session = (WT_SESSION_IMPL *)cbulk->cbt.iface.session;
    cursor = &cbulk->cbt.iface;

    WT_ERR(__wt_scr_alloc(session, 512, &a));
    WT_ERR(__wt_scr_alloc(session, 512, &b));

    WT_ERR_MSG(session, EINVAL,
      "bulk-load presented with out-of-order keys: %s compares smaller "
      "than previously inserted key %s",
      __wt_buf_set_printable(session, cursor->key.data, cursor->key.size, a),
      __wt_buf_set_printable(session, cbulk->last.data, cbulk->last.size, b));

err:
    __wt_scr_free(session, &a);
    __wt_scr_free(session, &b);
    return (ret);
}

/*
 * __curbulk_insert_row --
 *     Row-store bulk cursor insert, with key-sort checks.
 */
static int
__curbulk_insert_row(WT_CURSOR *cursor)
{
    WT_BTREE *btree;
    WT_CURSOR_BULK *cbulk;
    WT_DECL_RET;
    WT_SESSION_IMPL *session;
    int cmp;

    cbulk = (WT_CURSOR_BULK *)cursor;
    btree = cbulk->cbt.btree;

    /*
     * Bulk cursor inserts are updates, but don't need auto-commit transactions because they are
     * single-threaded and not visible until the bulk cursor is closed.
     */
    CURSOR_API_CALL(cursor, session, insert, btree);
    WT_STAT_CONN_INCR(session, cursor_insert_bulk);
    WT_STAT_DATA_INCR(session, cursor_insert_bulk);

    WT_ERR(__cursor_checkkey(cursor));
    WT_ERR(__cursor_checkvalue(cursor));

    /*
     * If this isn't the first key inserted, compare it against the last key to ensure the
     * application doesn't accidentally corrupt the table.
     */
    if (!cbulk->first_insert) {
        WT_ERR(__wt_compare(session, btree->collator, &cursor->key, &cbulk->last, &cmp));
        if (cmp <= 0)
            WT_ERR(__bulk_row_keycmp_err(cbulk));
    } else
        cbulk->first_insert = false;

    /* Save a copy of the key for the next comparison. */
    WT_ERR(__wt_buf_set(session, &cbulk->last, cursor->key.data, cursor->key.size));

    ret = __wt_bulk_insert_row(session, cbulk);

err:
    API_END_RET(session, ret);
}

/*
 * __curbulk_insert_row_skip_check --
 *     Row-store bulk cursor insert, without key-sort checks.
 */
static int
__curbulk_insert_row_skip_check(WT_CURSOR *cursor)
{
    WT_BTREE *btree;
    WT_CURSOR_BULK *cbulk;
    WT_DECL_RET;
    WT_SESSION_IMPL *session;

    cbulk = (WT_CURSOR_BULK *)cursor;
    btree = cbulk->cbt.btree;

    /*
     * Bulk cursor inserts are updates, but don't need auto-commit transactions because they are
     * single-threaded and not visible until the bulk cursor is closed.
     */
    CURSOR_API_CALL(cursor, session, insert, btree);
    WT_STAT_CONN_INCR(session, cursor_insert_bulk);
    WT_STAT_DATA_INCR(session, cursor_insert_bulk);

    WT_ERR(__cursor_checkkey(cursor));
    WT_ERR(__cursor_checkvalue(cursor));

    ret = __wt_bulk_insert_row(session, cbulk);

err:
    API_END_RET(session, ret);
}

/*
 * __wt_curbulk_init --
 *     Initialize a bulk cursor.
 */
int
__wt_curbulk_init(
  WT_SESSION_IMPL *session, WT_CURSOR_BULK *cbulk, bool bitmap, bool skip_sort_check)
{
    WT_CURSOR *c;
    WT_CURSOR_BTREE *cbt;

    c = &cbulk->cbt.iface;
    cbt = &cbulk->cbt;

    /* Bulk cursors only support insert and close (reset is a no-op). */
    __wt_cursor_set_notsup(c);
    switch (cbt->btree->type) {
    case BTREE_COL_FIX:
        c->insert = bitmap ? __curbulk_insert_fix_bitmap : __curbulk_insert_fix;
        break;
    case BTREE_COL_VAR:
        c->insert = __curbulk_insert_var;
        break;
    case BTREE_ROW:
        /*
         * Row-store order comparisons are expensive, so we optionally skip them when we know the
         * input is correct.
         */
        c->insert = skip_sort_check ? __curbulk_insert_row_skip_check : __curbulk_insert_row;
        break;
    }

    cbulk->first_insert = true;
    cbulk->recno = 0;
    cbulk->bitmap = bitmap;
    if (bitmap)
        F_SET(c, WT_CURSTD_RAW);

    return (__wt_bulk_init(session, cbulk));
}