{173778048} bdb_count should be able to count tables w no indices

Signed-off-by: Salil Chandra <schandra107@bloomberg.net>

Author: chands10

bdb/bdb_cursor.h

@@ -199,6 +199,7 @@ int bdb_get_lsn_context_from_timestamp(bdb_state_type *bdb_state,
 int bdb_get_context_from_lsn(bdb_state_type *bdb_state, void *lsnp,
                              unsigned long long *ret_context, int *bdberr);
 
-int bdb_direct_count(bdb_cursor_ifn_t *, int ixnum, int64_t *count, int is_snapcur, uint32_t last_commit_lsn_file, uint32_t last_commit_lsn_offset, uint32_t last_checkpoint_lsn_file, uint32_t last_checkpoint_lsn_offset);
+int bdb_direct_count_int(bdb_state_type *state, int ixnum, int64_t *rcnt, int is_snapcur, uint32_t modsnap_start_lsn_file, uint32_t modsnap_start_lsn_offset, uint32_t last_checkpoint_lsn_file, uint32_t last_checkpoint_lsn_offset, int parallel_count);
+int bdb_direct_count(bdb_cursor_ifn_t *cur, int ixnum, int64_t *rcnt, int is_snapcur, uint32_t modsnap_start_lsn_file, uint32_t modsnap_start_lsn_offset, uint32_t last_checkpoint_lsn_file, uint32_t last_checkpoint_lsn_offset);
 
 #endif

bdb/count.c

@@ -208,6 +208,7 @@ int bdb_count_int(bdb_state_type *bdb_state, int *bdberr)
         return -1;
     }
 
+    // TODO: We are not using the smallest index here, why? We are using ix0
     rc = bdb_state->dbp_ix[0]->cursor(bdb_state->dbp_ix[0], 0, &dbcp, 0);
     if (rc != 0) {
         myfree(buffer);
@@ -300,7 +301,13 @@ int bdb_count(bdb_state_type *bdb_state, int *bdberr)
     int ret;
     BDB_READLOCK("bdb_count");
 
-    ret = bdb_count_int(bdb_state, bdberr);
+    // ret = bdb_count_int(bdb_state, bdberr);
+    int64_t count;
+    // bdb_count_int would use ix 0 if available, do the same thing here
+    ret = bdb_direct_count_int(bdb_state, bdb_state->numix > 0 ? 0 : -1, &count, 0, 0, 0, 0, 0, 0);
     BDB_RELLOCK();
-    return ret;
+    if (ret == 0)
+        return count;
+    *bdberr = (ret == BDBERR_DEADLOCK ? ret : BDBERR_MISC);
+    return -1;
 }

bdb/cursor.c

@@ -7454,19 +7454,15 @@ static void *db_count(void *varg)
     return NULL;
 }
 
-int gbl_parallel_count = 0;
-int bdb_direct_count(bdb_cursor_ifn_t *cur, int ixnum, int64_t *rcnt, int is_snapcur, uint32_t modsnap_start_lsn_file, uint32_t modsnap_start_lsn_offset, uint32_t last_checkpoint_lsn_file, uint32_t last_checkpoint_lsn_offset)
+int bdb_direct_count_int(bdb_state_type *state, int ixnum, int64_t *rcnt, int is_snapcur, uint32_t modsnap_start_lsn_file, uint32_t modsnap_start_lsn_offset, uint32_t last_checkpoint_lsn_file, uint32_t last_checkpoint_lsn_offset, int parallel_count)
 {
     int64_t count = 0;
-    int parallel_count;
-    bdb_state_type *state = cur->impl->state;
     DB **db;
     int stripes;
     pthread_attr_t attr;
     if (ixnum < 0) { // data
         db = state->dbp_data[0];
         stripes = state->attr->dtastripe;
-        parallel_count = gbl_parallel_count;
         Pthread_attr_init(&attr);
 #ifdef PTHREAD_STACK_MIN
         Pthread_attr_setstacksize(&attr, PTHREAD_STACK_MIN + 512 * 1024);
@@ -7516,3 +7512,9 @@ int bdb_direct_count(bdb_cursor_ifn_t *cur, int ixnum, int64_t *rcnt, int is_sna
     if (rc == 0) *rcnt = count;
     return rc;
 }
+
+int gbl_parallel_count = 0;
+int bdb_direct_count(bdb_cursor_ifn_t *cur, int ixnum, int64_t *rcnt, int is_snapcur, uint32_t modsnap_start_lsn_file, uint32_t modsnap_start_lsn_offset, uint32_t last_checkpoint_lsn_file, uint32_t last_checkpoint_lsn_offset)
+{
+    return bdb_direct_count_int(cur->impl->state, ixnum, rcnt, is_snapcur, modsnap_start_lsn_file, modsnap_start_lsn_offset, last_checkpoint_lsn_file, last_checkpoint_lsn_offset, gbl_parallel_count);
+}