[FAB-16302] Update historydb key format

The historydb code assumes that the keys do not contain
a nil byte. In the presence of such keys, a historydb query
may include wrong results due to key collision.

This CR fixes this issue. However, it makes a slight change
in the format of the keys that are persisted and hence will
require dropping and rebuilding of existing historydb.

Change-Id: I8b451fff10e15e22948b01f18fddb16911866d76
Signed-off-by: manish <manish.sethi@gmail.com>
Signed-off-by: Wenjian Qiao <wenjianq@gmail.com>

Author: manish-sethi

core/ledger/kvledger/history/historydb/histmgr_helper.go

@@ -1,17 +1,7 @@
 /*
-Copyright IBM Corp. 2016 All Rights Reserved.
+Copyright IBM Corp. All Rights Reserved.
 
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-		 http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
+SPDX-License-Identifier: Apache-2.0
 */
 
 package historydb
@@ -25,13 +15,15 @@ import (
 // CompositeKeySep is a nil byte used as a separator between different components of a composite key
 var CompositeKeySep = []byte{0x00}
 
-//ConstructCompositeHistoryKey builds the History Key of namespace~key~blocknum~trannum
+// ConstructCompositeHistoryKey builds the History Key of namespace~len(key)~key~blocknum~trannum
 // using an order preserving encoding so that history query results are ordered by height
+// Note: this key format is different than the format in pre-v2.0 releases and requires
+//       a historydb rebuild when upgrading an older version to v2.0.
 func ConstructCompositeHistoryKey(ns string, key string, blocknum uint64, trannum uint64) []byte {
-
 	var compositeKey []byte
 	compositeKey = append(compositeKey, []byte(ns)...)
 	compositeKey = append(compositeKey, CompositeKeySep...)
+	compositeKey = append(compositeKey, util.EncodeOrderPreservingVarUint64(uint64(len(key)))...)
 	compositeKey = append(compositeKey, []byte(key)...)
 	compositeKey = append(compositeKey, CompositeKeySep...)
 	compositeKey = append(compositeKey, util.EncodeOrderPreservingVarUint64(blocknum)...)
@@ -40,12 +32,13 @@ func ConstructCompositeHistoryKey(ns string, key string, blocknum uint64, trannu
 	return compositeKey
 }
 
-//ConstructPartialCompositeHistoryKey builds a partial History Key namespace~key~
+// ConstructPartialCompositeHistoryKey builds a partial History Key namespace~len(key)~key~
 // for use in history key range queries
 func ConstructPartialCompositeHistoryKey(ns string, key string, endkey bool) []byte {
 	var compositeKey []byte
 	compositeKey = append(compositeKey, []byte(ns)...)
 	compositeKey = append(compositeKey, CompositeKeySep...)
+	compositeKey = append(compositeKey, util.EncodeOrderPreservingVarUint64(uint64(len(key)))...)
 	compositeKey = append(compositeKey, []byte(key)...)
 	compositeKey = append(compositeKey, CompositeKeySep...)
 	if endkey {

core/ledger/kvledger/history/historydb/histmgr_helper_test.go

@@ -1,48 +1,66 @@
 /*
-Copyright IBM Corp. 2016 All Rights Reserved.
+Copyright IBM Corp. All Rights Reserved.
 
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-		 http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
+SPDX-License-Identifier: Apache-2.0
 */
 
 package historydb
 
 import (
+	"bytes"
 	"testing"
 
+	"github.com/hyperledger/fabric/common/ledger/util"
 	"github.com/stretchr/testify/assert"
 )
 
 var strKeySep = string(CompositeKeySep)
 
-func TestConstructCompositeKey(t *testing.T) {
-	compositeKey := ConstructCompositeHistoryKey("ns1", "key1", 1, 1)
-	assert.NotNil(t, compositeKey)
-	//historyleveldb_test.go tests the actual output
-}
+func TestCompositeKeyConstruction(t *testing.T) {
+	type keyComponents struct {
+		ns, key         string
+		blkNum, tranNum uint64
+	}
 
-func TestConstructPartialCompositeKey(t *testing.T) {
-	compositeStartKey := ConstructPartialCompositeHistoryKey("ns1", "key1", false)
-	compositeEndKey := ConstructPartialCompositeHistoryKey("ns1", "key1", true)
+	testData := []*keyComponents{
+		{"ns1", "key1", 1, 0},
+		{"ns1", "key1\x00", 1, 5},
+		{"ns1", "key1\x00\x00", 100, 100},
+		{"ns1", "key\x00\x001", 100, 100},
+		{"ns1", "\x00key\x00\x001", 100, 100},
+	}
 
-	assert.Equal(t, []byte("ns1"+strKeySep+"key1"+strKeySep), compositeStartKey)
-	assert.Equal(t, []byte("ns1"+strKeySep+"key1"+strKeySep+string([]byte{0xff})), compositeEndKey)
+	for _, testDatum := range testData {
+		key := ConstructCompositeHistoryKey(testDatum.ns, testDatum.key, testDatum.blkNum, testDatum.tranNum)
+		startKey := ConstructPartialCompositeHistoryKey(testDatum.ns, testDatum.key, false)
+		endKey := ConstructPartialCompositeHistoryKey(testDatum.ns, testDatum.key, true)
+		assert.Equal(t, bytes.Compare(startKey, key), -1) //startKey should be smaller than key
+		assert.Equal(t, bytes.Compare(endKey, key), 1)    //endKey should be greater than key
+	}
+
+	for i, testDatum := range testData {
+		for j, another := range testData {
+			if i == j {
+				continue
+			}
+			startKey := ConstructPartialCompositeHistoryKey(testDatum.ns, testDatum.key, false)
+			endKey := ConstructPartialCompositeHistoryKey(testDatum.ns, testDatum.key, true)
+
+			anotherKey := ConstructCompositeHistoryKey(another.ns, another.key, another.blkNum, another.tranNum)
+			assert.False(t, bytes.Compare(anotherKey, startKey) == 1 && bytes.Compare(anotherKey, endKey) == -1) //any key should not fall in the range of start/end key range query for any other key
+		}
+	}
 }
 
 func TestSplitCompositeKey(t *testing.T) {
-	compositeFullKey := []byte("ns1" + strKeySep + "key1" + strKeySep + "extra bytes to split")
+	compositeFullKey := ConstructCompositeHistoryKey("ns1", "key1", 20, 200)
 	compositePartialKey := ConstructPartialCompositeHistoryKey("ns1", "key1", false)
-
 	_, extraBytes := SplitCompositeHistoryKey(compositeFullKey, compositePartialKey)
+	blkNum, bytesConsumed, err := util.DecodeOrderPreservingVarUint64(extraBytes)
+	assert.NoError(t, err)
+	txNum, _, err := util.DecodeOrderPreservingVarUint64(extraBytes[bytesConsumed:])
+	assert.NoError(t, err)
 	// second position should hold the extra bytes that were split off
-	assert.Equal(t, []byte("extra bytes to split"), extraBytes)
+	assert.Equal(t, blkNum, uint64(20))
+	assert.Equal(t, txNum, uint64(200))
 }