split a pma into 2, cursors on the pma not yet implemented

git-svn-id: file:///svn/tokudb@63 c7de825b-a66e-492c-adef-691d508d4ae1

newbrt/pma-test.c

@@ -6,6 +6,7 @@
 #include <string.h>
 #include <stdlib.h>
 #include <stdio.h>
+#include <arpa/inet.h>
 
 static void test_make_space_at (void) {
     PMA pma;
@@ -534,6 +535,89 @@ void test_pma_compare_fun (int wrong_endian_p) {
     r=pma_free(&pma); assert(r==0);
 }
 
+void test_pma_split(int n) {
+    PMA pmaa, pmab, pmac;
+    int error;
+    int i;
+
+    printf("test_pma_split:%d\n", n);
+
+    error = pma_create(&pmaa, default_compare_fun);
+    assert(error == 0);
+
+    /* insert some kv pairs */
+    for (i=0; i<n; i++) {
+        DBT dbtk, dbtv;
+        char k[5]; int v;
+
+        sprintf(k, "%4.4d", i);
+        fill_dbt(&dbtk, &k, strlen(k)+1);
+        v = i;
+        fill_dbt(&dbtv, &v, sizeof v);
+
+        error = pma_insert(pmaa, &dbtk, &dbtv, 0); 
+        assert(error == BRT_OK);
+    }
+
+    printf("a:"); print_pma(pmaa);
+
+    error = pma_split(pmaa, &pmab, &pmac, 0, 0);
+    assert(error == 0);
+
+    printf("a:"); print_pma(pmaa);
+    printf("b:"); print_pma(pmab);
+    printf("c:"); print_pma(pmac);
+
+    error = pma_free(&pmaa);
+    assert(error == 0);
+    error = pma_free(&pmab);
+    assert(error == 0);
+    error = pma_free(&pmac);
+    assert(error == 0);
+}
+
+void test_pma_split_varkey() {
+    char *keys[] = {
+        "this", "is", "a", "key", "this is a really really big key", "zz", 0 };
+    PMA pmaa, pmab, pmac;
+    int error;
+    int i;
+
+    printf("test_pma_split_varkey\n");
+
+    error = pma_create(&pmaa, default_compare_fun);
+    assert(error == 0);
+
+    /* insert some kv pairs */
+    for (i=0; keys[i]; i++) {
+        DBT dbtk, dbtv;
+        char v;
+
+        fill_dbt(&dbtk, keys[i], strlen(keys[i])+1);
+        v = i;
+        fill_dbt(&dbtv, &v, sizeof v);
+
+        error = pma_insert(pmaa, &dbtk, &dbtv, 0); 
+        assert(error == BRT_OK);
+    }
+
+    printf("a:"); print_pma(pmaa);
+
+    error = pma_split(pmaa, &pmab, &pmac, 0, 0);
+    assert(error == 0);
+
+    printf("a:"); print_pma(pmaa);
+    printf("b:"); print_pma(pmab);
+    printf("c:"); print_pma(pmac);
+
+    error = pma_free(&pmaa);
+    assert(error == 0);
+    error = pma_free(&pmab);
+    assert(error == 0);
+    error = pma_free(&pmac);
+    assert(error == 0);
+}
+
 void pma_tests (void) {
     memory_check=1;
     test_pma_compare_fun(0);      memory_check_all_free();
@@ -549,6 +633,12 @@ void pma_tests (void) {
     test_keycompare();            memory_check_all_free();
     test_pma_random_pick();       memory_check_all_free();
     test_pma_cursor();            memory_check_all_free();
+    test_pma_split(0);            memory_check_all_free();
+    test_pma_split(1);            memory_check_all_free();
+    test_pma_split(2);            memory_check_all_free();
+    test_pma_split(8);            memory_check_all_free();
+    test_pma_split(32);           memory_check_all_free();
+    test_pma_split_varkey();      memory_check_all_free();
 }
 
 int main (int argc __attribute__((__unused__)), char *argv[] __attribute__((__unused__))) {

newbrt/pma.c

@@ -292,6 +292,34 @@ int pma_create (PMA *pma, int (*compare_fun)(DB*,const DBT*,const DBT*)) {
     return 0;
 }
 
+int pma_init_array(PMA pma, int n) {
+    int i;
+    int twor;
+
+    if (pma->pairs) {
+        toku_free(pma->pairs);
+        pma->pairs = 0;
+    }
+
+    /* find the smallest power of 2 >= n */
+    twor = 4;
+    while (twor < n)
+        twor *= 2;
+
+    pma->N = twor;
+    MALLOC_N(1+pma->N, pma->pairs);
+    pma->pairs[pma->N].key = (void *) 0xdeadbeef;
+
+    for (i=0; i<pma->N; i++) {
+        pma->pairs[i].key = 0;
+        pma->pairs[i].keylen = 0;
+        pma->pairs[i].val = 0;
+        pma->pairs[i].vallen = 0;
+    }
+    pmainternal_calculate_parameters(pma);
+    return 0;
+}
+
 
 int pma_cursor (PMA pma, PMA_CURSOR *cursp) {
     PMA_CURSOR MALLOC(curs);
@@ -538,3 +566,81 @@ void pma_iterate (PMA pma, void(*f)(bytevec,ITEMLEN,bytevec,ITEMLEN, void*), voi
 	}
     }
 }
+
+struct pair *pma_extract_pairs(PMA pma) {
+    int npairs;
+    struct pair *pairs;
+    int i;
+    int lastpair;
+
+    npairs = pma_n_entries(pma);
+    pairs = toku_malloc(npairs * sizeof (struct pair));
+    lastpair = 0;
+    for (i=0; i<pma_index_limit(pma); i++) {
+        if (pma->pairs[i].key != 0) {
+            pairs[lastpair] = pma->pairs[i];
+            pma->pairs[i].key = 0;
+            pma->pairs[i].val = 0;
+            lastpair += 1;
+        }
+    }
+    assert(lastpair == npairs);
+    return pairs;
+}
+
+int pma_split(PMA old, PMA *newa, PMA *newb, 
+        PMA_CURSOR *cursors, int ncursors) {
+    int error;
+    int npairs;
+    struct pair *pairs;
+    int sumlen;
+    int runlen;
+    int len;
+    int i;
+    int spliti;
+
+    assert(cursors == 0 && ncursors == 0);
+
+    /* create the new pma's */
+    error = pma_create(newa, old->compare_fun);
+    if (error != 0)
+        return error;
+    error = pma_create(newb, old->compare_fun);
+    if (error != 0) {
+        pma_free(newb);
+        return error;
+    }
+
+    /* extract the pairs */
+    npairs = pma_n_entries(old);
+    pairs = pma_extract_pairs(old);
+    old->n_pairs_present = 0;
+
+    /* split the pairs in half by length */
+    sumlen = 0;
+    for (i=0; i<npairs; i++)
+        sumlen += 4 + pairs[i].keylen + 4 + pairs[i].vallen;
+
+    runlen = 0;
+    for (i=0; i < npairs; i++) {
+        len = 4 + pairs[i].keylen + 4 + pairs[i].vallen;
+        if (runlen + len > sumlen/2)
+            break;
+        runlen += len;
+    }
+    spliti = i;
+
+    /* put the first 1/2 of pairs into newa */
+    pma_init_array(*newa, 2 * spliti);
+    distribute_data((*newa)->pairs, pma_index_limit(*newa), &pairs[0], spliti);
+    (*newa)->n_pairs_present = spliti;
+
+    /* put the second 1/2 of pairs into newb */
+    pma_init_array(*newb, 2 * (npairs-spliti));
+    distribute_data((*newb)->pairs, pma_index_limit(*newb), &pairs[spliti], npairs-spliti);
+    (*newb)->n_pairs_present = npairs-spliti;
+
+    toku_free(pairs);
+
+    return 0;
+}

newbrt/pma.h

@@ -36,6 +36,8 @@ int pma_delete (PMA, DBT *, DB*);
  * Don't modify the returned data.  Don't free it. */
 enum pma_errors pma_lookup (PMA, DBT*, DBT*, DB*);
 
+int pma_split(PMA old, PMA *newa, PMA *newb, PMA_CURSOR *cursors, int ncursors);
+
 /* Move the cursor to the beginning or the end or to a key */
 int pma_cursor (PMA, PMA_CURSOR *);
 int pma_cursor_free (PMA_CURSOR*);