INTEGRATION [PR#1031 > development/8.1] bugfix: S3C-2899 implement v1 format for DelimiterVersions listing

lib/algos/list/basic.js

@@ -28,14 +28,14 @@ class List extends Extension {
     }
 
     genMDParams() {
-        const params = {
+        const params = this.parameters ? {
             gt: this.parameters.gt,
             gte: this.parameters.gte || this.parameters.start,
             lt: this.parameters.lt,
             lte: this.parameters.lte || this.parameters.end,
             keys: this.parameters.keys,
             values: this.parameters.values,
-        };
+        } : {};
         Object.keys(params).forEach(key => {
             if (params[key] === null || params[key] === undefined) {
                 delete params[key];

lib/algos/list/delimiterVersions.js

@@ -1,14 +1,13 @@
 'use strict'; // eslint-disable-line strict
 
-const errors = require('../../errors');
 const Delimiter = require('./delimiter').Delimiter;
 const Version = require('../../versioning/Version').Version;
 const VSConst = require('../../versioning/constants').VersioningConstants;
 const { inc, FILTER_END, FILTER_ACCEPT, FILTER_SKIP, SKIP_NONE } =
     require('./tools');
 
 const VID_SEP = VSConst.VersionId.Separator;
-const { BucketVersioningKeyFormat } = VSConst;
+const { DbPrefixes, BucketVersioningKeyFormat } = VSConst;
 
 /**
  * Handle object listing with parameters
@@ -34,13 +33,19 @@ class DelimiterVersions extends Delimiter {
         // listing results
         this.NextMarker = parameters.keyMarker;
         this.NextVersionIdMarker = undefined;
-    }
 
-    genMDParams() {
-        if (this.vFormat === BucketVersioningKeyFormat.v0) {
-            return this.genMDParamsV0();
-        }
-        throw errors.NotImplemented;
+        Object.assign(this, {
+            [BucketVersioningKeyFormat.v0]: {
+                genMDParams: this.genMDParamsV0,
+                filter: this.filterV0,
+                skipping: this.skippingV0,
+            },
+            [BucketVersioningKeyFormat.v1]: {
+                genMDParams: this.genMDParamsV1,
+                filter: this.filterV1,
+                skipping: this.skippingV1,
+            },
+        }[this.vFormat]);
     }
 
     genMDParamsV0() {
@@ -57,15 +62,69 @@ class DelimiterVersions extends Delimiter {
             if (this.parameters.versionIdMarker) {
                 // versionIdMarker should always come with keyMarker
                 // but may not be the other way around
-                params.gt = `${this.parameters.keyMarker}` +
-                    `${VID_SEP}${this.parameters.versionIdMarker}`;
+                params.gt = this.parameters.keyMarker
+                    + VID_SEP
+                    + this.parameters.versionIdMarker;
             } else {
                 params.gt = inc(this.parameters.keyMarker + VID_SEP);
             }
         }
         return params;
     }
 
+    genMDParamsV1() {
+        // return an array of two listing params sets to ask for
+        // synchronized listing of M and V ranges
+        const params = [{}, {}];
+        if (this.parameters.prefix) {
+            params[0].gte = DbPrefixes.Master + this.parameters.prefix;
+            params[0].lt = DbPrefixes.Master + inc(this.parameters.prefix);
+            params[1].gte = DbPrefixes.Version + this.parameters.prefix;
+            params[1].lt = DbPrefixes.Version + inc(this.parameters.prefix);
+        } else {
+            params[0].gte = DbPrefixes.Master;
+            params[0].lt = inc(DbPrefixes.Master); // stop after the last master key
+            params[1].gte = DbPrefixes.Version;
+            params[1].lt = inc(DbPrefixes.Version); // stop after the last version key
+        }
+        if (this.parameters.keyMarker) {
+            if (params[1].gte <= DbPrefixes.Version + this.parameters.keyMarker) {
+                delete params[0].gte;
+                delete params[1].gte;
+                params[0].gt = DbPrefixes.Master + inc(this.parameters.keyMarker + VID_SEP);
+                if (this.parameters.versionIdMarker) {
+                    // versionIdMarker should always come with keyMarker
+                    // but may not be the other way around
+                    params[1].gt = DbPrefixes.Version
+                        + this.parameters.keyMarker
+                        + VID_SEP
+                        + this.parameters.versionIdMarker;
+                } else {
+                    params[1].gt = DbPrefixes.Version
+                        + inc(this.parameters.keyMarker + VID_SEP);
+                }
+            }
+        }
+        return params;
+    }
+
+    /**
+     * Used to synchronize listing of M and V prefixes by object key
+     *
+     * @param {object} masterObj object listed from first range
+     * returned by genMDParamsV1() (the master keys range)
+     * @param {object} versionObj object listed from second range
+     * returned by genMDParamsV1() (the version keys range)
+     * @return {number} comparison result:
+     *   * -1 if master key < version key
+     *   * 1 if master key > version key
+     */
+    compareObjects(masterObj, versionObj) {
+        const masterKey = masterObj.key.slice(DbPrefixes.Master.length);
+        const versionKey = versionObj.key.slice(DbPrefixes.Version.length);
+        return masterKey < versionKey ? -1 : 1;
+    }
+
     /**
      *  Add a (key, versionId, value) tuple to the listing.
      *  Set the NextMarker to the current key
@@ -92,7 +151,8 @@ class DelimiterVersions extends Delimiter {
     }
 
     /**
-     *  Filter to apply on each iteration, based on:
+     *  Filter to apply on each iteration if bucket is in v0
+     *  versioning key format, based on:
      *  - prefix
      *  - delimiter
      *  - maxKeys
@@ -102,14 +162,31 @@ class DelimiterVersions extends Delimiter {
      *  @param {String} obj.value - The value of the element
      *  @return {number}          - indicates if iteration should continue
      */
-    filter(obj) {
-        if (this.vFormat === BucketVersioningKeyFormat.v0) {
-            if (Version.isPHD(obj.value)) {
-                return FILTER_ACCEPT; // trick repd to not increase its streak
-            }
-            return this.filterCommon(obj.key, obj.value);
+    filterV0(obj) {
+        if (Version.isPHD(obj.value)) {
+            return FILTER_ACCEPT; // trick repd to not increase its streak
         }
-        throw errors.NotImplemented;
+        return this.filterCommon(obj.key, obj.value);
+    }
+
+    /**
+     *  Filter to apply on each iteration if bucket is in v1
+     *  versioning key format, based on:
+     *  - prefix
+     *  - delimiter
+     *  - maxKeys
+     *  The marker is being handled directly by levelDB
+     *  @param {Object} obj       - The key and value of the element
+     *  @param {String} obj.key   - The key of the element
+     *  @param {String} obj.value - The value of the element
+     *  @return {number}          - indicates if iteration should continue
+     */
+    filterV1(obj) {
+        // this function receives both M and V keys, but their prefix
+        // length is the same so we can remove their prefix without
+        // looking at the type of key
+        return this.filterCommon(obj.key.slice(DbPrefixes.Master.length),
+                                 obj.value);
     }
 
     filterCommon(key, value) {
@@ -144,13 +221,6 @@ class DelimiterVersions extends Delimiter {
         return this.addContents({ key: nonversionedKey, value, versionId });
     }
 
-    skipping() {
-        if (this.vFormat === BucketVersioningKeyFormat.v0) {
-            return this.skippingV0();
-        }
-        throw errors.NotImplemented;
-    }
-
     skippingV0() {
         if (this.NextMarker) {
             const index = this.NextMarker.lastIndexOf(this.delimiter);
@@ -161,6 +231,16 @@ class DelimiterVersions extends Delimiter {
         return SKIP_NONE;
     }
 
+    skippingV1() {
+        const skipV0 = this.skippingV0();
+        if (skipV0 === SKIP_NONE) {
+            return SKIP_NONE;
+        }
+        // skip to the same object key in both M and V range listings
+        return [DbPrefixes.Master + skipV0,
+                DbPrefixes.Version + skipV0];
+    }
+
     /**
      *  Return an object containing all mandatory fields to use once the
      *  iteration is done, doesn't show a NextMarker field if the output

lib/algos/list/tools.js

@@ -1,3 +1,5 @@
+const { DbPrefixes } = require('../../versioning/constants').VersioningConstants;
+
 // constants for extensions
 const SKIP_NONE = undefined; // to be inline with the values of NextMarker
 const FILTER_ACCEPT = 1;
@@ -31,9 +33,36 @@ function inc(str) {
             String.fromCharCode(str.charCodeAt(str.length - 1) + 1)) : str;
 }
 
+/**
+ * Transform listing parameters for v0 versioning key format to make
+ * it compatible with v1 format
+ *
+ * @param {object} v0params - listing parameters for v0 format
+ * @return {object} - listing parameters for v1 format
+ */
+function listingParamsMasterKeysV0ToV1(v0params) {
+    const v1params = Object.assign({}, v0params);
+    if (v0params.gt !== undefined) {
+        v1params.gt = `${DbPrefixes.Master}${v0params.gt}`;
+    } else if (v0params.gte !== undefined) {
+        v1params.gte = `${DbPrefixes.Master}${v0params.gte}`;
+    } else {
+        v1params.gte = DbPrefixes.Master;
+    }
+    if (v0params.lt !== undefined) {
+        v1params.lt = `${DbPrefixes.Master}${v0params.lt}`;
+    } else if (v0params.lte !== undefined) {
+        v1params.lte = `${DbPrefixes.Master}${v0params.lte}`;
+    } else {
+        v1params.lt = inc(DbPrefixes.Master); // stop after the last master key
+    }
+    return v1params;
+}
+
 module.exports = {
     checkLimit,
     inc,
+    listingParamsMasterKeysV0ToV1,
     SKIP_NONE,
     FILTER_END,
     FILTER_SKIP,