odh2: second on-disk-header format (insert/update timestamps, larger blobs)

bbinc/cdb2_constants.h

@@ -19,7 +19,14 @@
 
 #define COMDB2_MAX_RECORD_SIZE 16384
 #define LONG_REQMS 2000
-#define MAXBLOBLENGTH ((1 << 28) - 1)   /* (1 << ODH_LENGTH_BITS) - 1 */
+#define MAXBLOBLENGTH                                                                                                  \
+    ((1 << 28) - 1) /* (1 << ODH_LENGTH_BITS) - 1; the odh1                                                            \
+                       limit -- odh1 packs length into 28                                                              \
+                       bits so it can never exceed this */
+#define MAXBLOBLENGTH2                                                                                                 \
+    0x7fffffff                          /* odh2 stores a full 32-bit length;                                           \
+                                          cap at INT_MAX to stay safe in the                                           \
+                                          signed-int paths above bdb */
 #define MAXBLOBS 15                     /* Should be bdb's MAXDTAFILES - 1 */
 #define MAXCOLNAME 99                   /* not incl. \0 */
 #define MAXCOLUMNS 1024

bdb/bdb_api.h

@@ -1934,6 +1934,7 @@ int bdb_user_get_all_tran(tran_type *tran, char ***users, int *num);
 
 void bdb_set_instant_schema_change(bdb_state_type *bdb_state, int isc);
 void bdb_set_inplace_updates(bdb_state_type *bdb_state, int ipu);
+void bdb_set_odh2(bdb_state_type *bdb_state, int odh2);
 void bdb_set_csc2_version(bdb_state_type *bdb_state, uint8_t version);
 
 int bdb_get_active_stripe(bdb_state_type *bdb_state);

bdb/bdb_cursor.h

@@ -80,6 +80,8 @@ typedef struct bdb_cursor_ifn {
     int (*datalen)(struct bdb_cursor_ifn *cur);
     int (*rrn)(struct bdb_cursor_ifn *cur);
     unsigned long long (*genid)(struct bdb_cursor_ifn *cur);
+    uint32_t (*insert_secs)(struct bdb_cursor_ifn *cur);
+    uint32_t (*update_secs)(struct bdb_cursor_ifn *cur);
     int (*dbnum)(struct bdb_cursor_ifn *cur);
     void *(*datacopy)(struct bdb_cursor_ifn *cur);
     uint8_t (*ver)(struct bdb_cursor_ifn *cur);

bdb/bdb_fetch.h

@@ -72,6 +72,8 @@ typedef struct {
     uint8_t for_write;
     void *(*fn_malloc)(size_t); /* user-specified malloc function */
     void (*fn_free)(void *); /* user-specified free function */
+    uint32_t insert_secs;    /* out: odh2 data-record insert time (0 if not odh2) */
+    uint32_t update_secs;    /* out: odh2 data-record last-update time (0 if not) */
 } bdb_fetch_args_t;
 
 int bdb_fetch(bdb_state_type *bdb_handle, void *ix, int ixnum, int ixlen,

bdb/bdb_int.h

@@ -62,33 +62,56 @@ enum {
     ODH_UPDATEID_BITS = 12,
     ODH_LENGTH_BITS = 28,
 
-    ODH_SIZE = 7, /* We may extend for larger headers in the future,
-                     but the minimum size shall always be 7 bytes. */
-
-    ODH_SIZE_RESERVE = 7, /* Callers wishing to provide a buffer into which
-                             a record will be packed should allow this many
-                             bytes on top of the record size for the ODH.
-                             Right now this is the same as ODH_SIZE - one
-                             day it may be the max possible ODH size if we
-                             start adding fields. */
-
-    ODH_FLAG_COMPR_MASK = 0x7
+    ODH_SIZE = 7, /* Size of the original (odh1) on-disk header.  This is the
+                     minimum ODH size and shall always be 7 bytes. */
+
+    ODH2_SIZE = 16, /* Size of the version-2 (odh2) on-disk header.  odh2 is a
+                       strict superset of odh1: flags/csc2vers/updateid occupy
+                       the same bytes, length becomes a clean 32-bit field, and
+                       32-bit insert/update timestamps are appended.  odh2
+                       records are flagged by ODH2_FLAG in the flags byte. */
+
+    ODH_SIZE_RESERVE = ODH2_SIZE, /* Callers wishing to provide a buffer into
+                             which a record will be packed should allow this
+                             many bytes on top of the record size for the ODH.
+                             This must be the MAXIMUM possible header size so a
+                             buffer fits either an odh1 or an odh2 header. */
+
+    ODH_FLAG_COMPR_MASK = 0x7, /* flags bits 0-2: compression algorithm */
+
+    ODH2_FLAG = 0x80 /* flags bit 7: set on odh2 records.  odh1 records only
+                        ever set the compression bits (0-2), so this bit is an
+                        unambiguous odh1/odh2 discriminator.  bit 3 is reserved
+                        for future compression-mask growth. */
 };
 
+/* Actual on-disk header size implied by a record's flags byte.  Only valid
+ * when ondisk_header is enabled for the table. */
+static inline int odh_size_from_flags(uint8_t flags)
+{
+    return (flags & ODH2_FLAG) ? ODH2_SIZE : ODH_SIZE;
+}
+
 /* snapisol log ops */
 typedef enum log_ops { LOG_APPLY = 0, LOG_PRESCAN = 1, LOG_BACKFILL = 2 } log_ops_t;
 
 /* These are the fields of the ondisk header.  This is not the ondisk
  * representation but a convenient format for passing the header around in
  * our code. */
 struct odh {
-    uint32_t length;   /* actually only 28 bits of this can be used leading to
-                          a max value of (1<<ODH_LENGTH_BITS)-1 */
+    uint32_t length;   /* For odh1 only 28 bits are usable (max
+                          (1<<ODH_LENGTH_BITS)-1).  For odh2 the full 32 bits
+                          are stored, though writes are capped at INT_MAX. */
     uint16_t updateid; /* actually only 12 bits of this can be used leading to
                           a max value of (1<<ODH_UPDATEID_BITS)-1 */
     uint8_t csc2vers;
     uint8_t flags;
 
+    uint32_t insert_secs; /* odh2 only: unsigned seconds since the 1970 epoch
+                             when the record was first inserted (0 for odh1). */
+    uint32_t update_secs; /* odh2 only: unsigned seconds since the 1970 epoch
+                             of the most recent update (0 for odh1). */
+
     void *recptr; /* Some functions set this to point to the
                      decompressed record data. */
 };
@@ -411,6 +434,9 @@ struct bdb_cursor_impl_tag {
     /* cursor position */
     int rrn;                  /* == 2 (don't need this) */
     unsigned long long genid; /* genid of current entry */
+    uint32_t insert_secs;     /* odh2 insert time of current entry (0 if the
+                                 row is not odh2 / no odh was decoded) */
+    uint32_t update_secs;     /* odh2 update time of current entry (0 if none) */
     void *data;               /* points inside one of  bdb_berkdb_t if valid */
     int datalen;              /* size of payload */
 
@@ -871,6 +897,12 @@ struct bdb_state_tag {
 
     signed char instant_schema_change;
 
+    /* odh2: write the version-2 on-disk header (insert/update timestamps,
+     * 32-bit length).  Requires ondisk_header.  Also implicitly forced at
+     * write time when the database is in genid48 format (a genid48 record must
+     * never be written as odh1, or it would carry no insert timestamp). */
+    signed char odh2;
+
     signed char rep_handle_dead;
 
     /* keep this as an int, it's read locklessly */
@@ -1629,6 +1661,7 @@ uint8_t *rep_udp_filepage_type_put(const filepage_type *p_filepage_type,
 const uint8_t *db_lsn_type_put(const DB_LSN *p_db_lsn, uint8_t *p_buf,
                                const uint8_t *p_buf_end);
 void poke_updateid(void *buf, int updateid);
+void poke_update_secs(void *buf, uint32_t secs);
 
 void bdb_genid_sanity_check(bdb_state_type *bdb_state, unsigned long long genid,
                             int stripe);
@@ -1719,9 +1752,9 @@ int bdb_committed_durable(bdb_state_type *bdb_state);
 
 int bdb_list_all_fileids_for_newsi(bdb_state_type *, hash_t *);
 
-int bdb_prepare_put_pack_updateid(bdb_state_type *bdb_state, int is_blob,
-                                  DBT *data, DBT *data2, int updateid,
-                                  void **freeptr, void *stackbuf, int odhready);
+int bdb_prepare_put_pack_updateid(bdb_state_type *bdb_state, int is_blob, DBT *data, DBT *data2, int updateid,
+                                  void **freeptr, void *stackbuf, int odhready, uint32_t preserve_insert_secs);
+int peek_odh2_insert_secs(const void *buf, size_t buflen, uint32_t *insert_secs);
 
 int net_get_lsn_rectype(const void *buf, int buflen, DB_LSN *lsn, int *myrectype);
 void pstack_self(void);

bdb/cursor.c

@@ -178,6 +178,8 @@ static int bdb_cursor_pause(bdb_cursor_ifn_t *pcur_ifn, int *bdberr);
 static void *bdb_cursor_data(bdb_cursor_ifn_t *cur);
 static int bdb_cursor_datalen(bdb_cursor_ifn_t *cur);
 static unsigned long long bdb_cursor_genid(bdb_cursor_ifn_t *cur);
+static uint32_t bdb_cursor_insert_secs(bdb_cursor_ifn_t *cur);
+static uint32_t bdb_cursor_update_secs(bdb_cursor_ifn_t *cur);
 static int bdb_cursor_rrn(bdb_cursor_ifn_t *cur);
 static int bdb_cursor_dbnum(bdb_cursor_ifn_t *cur);
 static void *bdb_cursor_datacopy(bdb_cursor_ifn_t *cur);
@@ -630,6 +632,8 @@ bdb_cursor_ifn_t *bdb_cursor_open(
     pcur_ifn->data = bdb_cursor_data;
     pcur_ifn->datalen = bdb_cursor_datalen;
     pcur_ifn->genid = bdb_cursor_genid;
+    pcur_ifn->insert_secs = bdb_cursor_insert_secs;
+    pcur_ifn->update_secs = bdb_cursor_update_secs;
     pcur_ifn->rrn = bdb_cursor_rrn;
     pcur_ifn->dbnum = bdb_cursor_dbnum;
     pcur_ifn->datacopy = bdb_cursor_datacopy;
@@ -3691,6 +3695,16 @@ static unsigned long long bdb_cursor_genid(bdb_cursor_ifn_t *cur)
     return cur->impl->genid;
 }
 
+static uint32_t bdb_cursor_insert_secs(bdb_cursor_ifn_t *cur)
+{
+    return cur->impl->insert_secs;
+}
+
+static uint32_t bdb_cursor_update_secs(bdb_cursor_ifn_t *cur)
+{
+    return cur->impl->update_secs;
+}
+
 static int bdb_cursor_rrn(bdb_cursor_ifn_t *cur) { return cur->impl->rrn; }
 
 static int serial_update_lastkey(bdb_cursor_impl_t *cur, char *key, int keylen)
@@ -3838,6 +3852,14 @@ static int bdb_btree_merge(bdb_cursor_impl_t *cur, int stripe_rl, int page_rl,
         cur->lastpage = page_rl;
         cur->lastindex = index_rl;
         cur->ver = ver_rl;
+        /* cur->rl is a live, callable cursor only when the merged record came
+         * from it.  Page-order / add-cursor paths supply the record from a temp
+         * table and leave cur->rl NULL; those rows carry synthetic genids and
+         * read back NULL timestamps regardless, so 0 is correct there. */
+        if (cur->rl)
+            cur->rl->odh2_times(cur->rl, &cur->insert_secs, &cur->update_secs);
+        else
+            cur->insert_secs = cur->update_secs = 0;
         if (cur->type == BDBC_IX && bdb_keycontainsgenid(cur->state, cur->idx))
             cur->datalen -= sizeof(unsigned long long);
         cur->genid = genid_rl;
@@ -3884,6 +3906,10 @@ static int bdb_btree_merge(bdb_cursor_impl_t *cur, int stripe_rl, int page_rl,
         /* This is a synthetic row- it's version will be the 'current' version.
          */
         cur->ver = bdb_state->version;
+        /* synthetic (shadow) row: no odh2 timestamps, and no time in the genid
+         * either -- it is transient; readers report NULL for it. */
+        cur->insert_secs = 0;
+        cur->update_secs = 0;
 
         if (cur->type == BDBC_IX && !cur->state->ixdta[cur->idx] &&
             pdatalen_sd > sizeof(unsigned long long)) {
@@ -3976,6 +4002,14 @@ static int bdb_btree_merge(bdb_cursor_impl_t *cur, int stripe_rl, int page_rl,
         cur->lastpage = page_rl;
         cur->lastindex = index_rl;
         cur->ver = ver_rl;
+        /* cur->rl is a live, callable cursor only when the merged record came
+         * from it.  Page-order / add-cursor paths supply the record from a temp
+         * table and leave cur->rl NULL; those rows carry synthetic genids and
+         * read back NULL timestamps regardless, so 0 is correct there. */
+        if (cur->rl)
+            cur->rl->odh2_times(cur->rl, &cur->insert_secs, &cur->update_secs);
+        else
+            cur->insert_secs = cur->update_secs = 0;
 
         if (cur->type == BDBC_IX && bdb_keycontainsgenid(cur->state, cur->idx))
             cur->datalen -= sizeof(unsigned long long);
@@ -4021,6 +4055,10 @@ static int bdb_btree_merge(bdb_cursor_impl_t *cur, int stripe_rl, int page_rl,
         cur->data = data_sd;
         cur->datalen = datalen_sd;
         cur->ver = bdb_state->version;
+        /* synthetic (shadow) row: no odh2 timestamps, and no time in the genid
+         * either -- it is transient; readers report NULL for it. */
+        cur->insert_secs = 0;
+        cur->update_secs = 0;
         if (cur->type == BDBC_IX)
             cur->datalen -= sizeof(unsigned long long);
         cur->genid = genid_sd;

bdb/cursor_ll.c

@@ -104,6 +104,7 @@ static int bdb_berkdb_dtasize(bdb_berkdb_t *pberkdb, int *dtasize, int *bdberr);
 static int bdb_berkdb_key(bdb_berkdb_t *pberkdb, char **key, int *bdberr);
 static int bdb_berkdb_keysize(bdb_berkdb_t *pberkdb, int *keysize, int *bdberr);
 static int bdb_berkdb_ver(bdb_berkdb_t *pberkdb, uint8_t *ver, int *bdberr);
+static int bdb_berkdb_odh2_times(bdb_berkdb_t *pberkdb, uint32_t *insert_secs, uint32_t *update_secs);
 static int bdb_berkdb_insert(bdb_berkdb_t *pberkdb, char *key, int keylen,
                              char *dta, int dtalen, int *bdberr);
 static int bdb_berkdb_delete(bdb_berkdb_t *pberkdb, int *bdberr);
@@ -503,6 +504,7 @@ bdb_berkdb_t *bdb_berkdb_open(bdb_cursor_impl_t *cur, int type, int maxdata,
     pberkdb->is_at_eof = bdb_berkdb_is_at_eof;
 
     pberkdb->ver = bdb_berkdb_ver;
+    pberkdb->odh2_times = bdb_berkdb_odh2_times;
 
     berkdb->cur = cur;
 
@@ -738,6 +740,8 @@ static int process_bulk_odh(bdb_berkdb_t *pberkdb, int *bdberr)
 
     bt->odh.size = odh.length;
     bt->ver = odh.csc2vers;
+    bt->insert_secs = odh.insert_secs;
+    bt->update_secs = odh.update_secs;
     if (ip_updates_enabled(berkdb->cur->state)) {
         genptr = (unsigned long long *)bt->lastkey;
 #ifdef _SUN_SOURCE
@@ -1176,6 +1180,22 @@ static int bdb_berkdb_ver(bdb_berkdb_t *pberkdb, uint8_t *ver, int *bdberr)
     return 0;
 }
 
+/* odh2 insert/update timestamps of the current real-stream payload.  Only the
+ * real (committed) stream decodes and carries these; other stream types return
+ * 0 so callers fall back to the genid-based time. */
+static int bdb_berkdb_odh2_times(bdb_berkdb_t *pberkdb, uint32_t *insert_secs, uint32_t *update_secs)
+{
+    bdb_berkdb_impl_t *berkdb = pberkdb->impl;
+    if (berkdb->type == BERKDB_REAL && berkdb->u.rl.use_odh) {
+        *insert_secs = pberkdb->impl->u.rl.insert_secs;
+        *update_secs = pberkdb->impl->u.rl.update_secs;
+    } else {
+        *insert_secs = 0;
+        *update_secs = 0;
+    }
+    return 0;
+}
+
 static int bdb_berkdb_is_at_eof(struct bdb_berkdb *pberkdb)
 {
     return pberkdb->impl->at_eof;

bdb/cursor_ll.h

@@ -42,6 +42,8 @@ typedef struct bdb_realdb_tag {
     DBT data; /* owns the buffer for data & key */
     DBT key;
     uint8_t ver;
+    uint32_t insert_secs; /* odh2 insert time of current payload (0 if none) */
+    uint32_t update_secs; /* odh2 update time of current payload (0 if none) */
 
     /* bulk api requirements; enabled only if tmpbulklen>0 */
     DBT bulk; /* owns the buffer for itself */
@@ -184,6 +186,7 @@ typedef struct bdb_berkdb {
     int (*key)(struct bdb_berkdb *berkdb, char **key, int *bdberr);
     int (*keysize)(struct bdb_berkdb *berkdb, int *keysize, int *bdberr);
     int (*ver)(struct bdb_berkdb *berkdb, uint8_t *ver, int *bdberr);
+    int (*odh2_times)(struct bdb_berkdb *berkdb, uint32_t *insert_secs, uint32_t *update_secs);
     int (*find)(struct bdb_berkdb *berkdb, void *key, int keysize, int how,
                 int *bdberr);
     int (*insert)(struct bdb_berkdb *berkdb, char *key, int keylen, char *dta,

bdb/fetch.c

@@ -1076,6 +1076,11 @@ static int bdb_fetch_int_ll(
 
                     *reqdtalen = odh.length;
                     *ver = odh.csc2vers;
+                    /* Surface the data record's odh2 timestamps so callers that
+                     * fetch by genid (e.g. comdb2_*_timestamp read via a
+                     * secondary index) get the row's real insert/update time. */
+                    args->insert_secs = odh.insert_secs;
+                    args->update_secs = odh.update_secs;
                 }
             } else if (bdb_state->ondisk_header && bdb_state->ixdta[ixnum] &&
                        bdb_state->datacopy_odh) {
@@ -1644,6 +1649,11 @@ static int bdb_fetch_int_ll(
 
                     *reqdtalen = odh.length;
                     *ver = odh.csc2vers;
+                    /* Surface the data record's odh2 timestamps so callers that
+                     * fetch by genid (e.g. comdb2_*_timestamp read via a
+                     * secondary index) get the row's real insert/update time. */
+                    args->insert_secs = odh.insert_secs;
+                    args->update_secs = odh.update_secs;
                 }
             } else if (bdb_state->ondisk_header && bdb_state->ixdta[ixnum] &&
                        bdb_state->datacopy_odh) {

bdb/genid.c

@@ -402,6 +402,12 @@ int get_epoch_plusplus(bdb_state_type *bdb_state)
 
 int genid_contains_time(bdb_state_type *bdb_state)
 {
+    /* The genid format lives only on the parent (env) handle; a table (child)
+     * handle's genid_format is never set (always 0).  Normalise to the parent so
+     * callers passing a table handle (e.g. init_odh, max_blob_length_for_table)
+     * see the real format instead of a spurious "time-based". */
+    if (bdb_state->parent)
+        bdb_state = bdb_state->parent;
     return bdb_state->genid_format == LLMETA_GENID_ORIGINAL;
 }
 

bdb/ll.c

@@ -1097,11 +1097,21 @@ static int ll_dta_upd_int(bdb_state_type *bdb_state, int rrn,
              * Otherwise there could be splits in the middle of the btree,
              * which we can't handle under page-order tablescan.  */
 
+            /* For odh2 records, carry the original insert time forward so an
+             * update does not reset it (update_secs is refreshed to "now" by
+             * init_odh).  Source it from the old record's odh2 insert_secs, or
+             * from the old (time-based) genid for an odh1 record being
+             * upgraded.  Only the data record (dtafile 0) carries this. */
+            uint32_t preserve_insert_secs = 0;
+            if (dtafile == 0 && malloceddta) {
+                if (!peek_odh2_insert_secs(old_dta_out_lcl.data, old_dta_out_lcl.size, &preserve_insert_secs))
+                    preserve_insert_secs = (uint32_t)bdb_genid_timestamp(oldgenid);
+            }
+
             /* Format the payload. */
             DBT packeddta;
-            rc = bdb_prepare_put_pack_updateid(bdb_state, is_blob, dta,
-                                               &packeddta, -1, &freedtaptr,
-                                               formatted_record, odhready);
+            rc = bdb_prepare_put_pack_updateid(bdb_state, is_blob, dta, &packeddta, -1, &freedtaptr, formatted_record,
+                                               odhready, preserve_insert_secs);
             recptr = packeddta.data;
             formatted_record_len = packeddta.size;