implementing bree index: inserting multiple records and full scan wit…

…h BTreeIndex successed.

lib/execution/executors/executor_test/btree_index_executor_test.go

@@ -67,7 +67,9 @@ func testKeyDuplicateInsertDeleteWithBTreeIndex[T float32 | int32 | string](t *t
 
 	rangeScanP := createSpecifiedRangeScanPlanNode[T](c, tableMetadata, keyType, 0, nil, nil, index_constants.INDEX_KIND_BTREE)
 	results := executePlan(c, shi.GetBufferPoolManager(), txn, rangeScanP)
-	fmt.Println(results)
+	for _, foundVal := range results {
+		fmt.Println(foundVal.GetValue(tableMetadata.Schema(), 0).ToString())
+	}
 
 	//scanP := createSpecifiedPointScanPlanNode(accountId.(T), c, tableMetadata, keyType, index_constants.INDEX_KIND_BTREE)
 	//result = executePlan(c, shi.GetBufferPoolManager(), txn, scanP)

lib/execution/executors/range_scan_with_index_executor.go

@@ -80,13 +80,25 @@ func (e *RangeScanWithIndexExecutor) Next() (*tuple.Tuple, Done, error) {
 				e.txn.SetState(access.ABORTED)
 				return nil, true, errors.New("detect value update after iterator created. changes transaction state to aborted.")
 			}
-		case *index.BTreeIndex:
+		case *index.SkipListIndex:
 			orgKey := samehada_util.ExtractOrgKeyFromDicOrderComparableEncodedVarchar(key, orgKeyType)
+
 			if !curKeyVal.CompareEquals(*orgKey) {
 				// column value corresponding index key is updated
 				e.txn.SetState(access.ABORTED)
 				return nil, true, errors.New("detect value update after iterator created. changes transaction state to aborted.")
 			}
+		case *index.BTreeIndex:
+			//orgKey := samehada_util.ExtractOrgKeyFromDicOrderComparableEncodedVarchar(key, orgKeyType)
+
+			//if !curKeyVal.CompareEquals(*orgKey) {
+
+			// when BTreeIndex is used, key is original key
+			if !curKeyVal.CompareEquals(*key) {
+				// column value corresponding index key is updated
+				e.txn.SetState(access.ABORTED)
+				return nil, true, errors.New("detect value update after iterator created. changes transaction state to aborted.")
+			}
 		}
 
 		// check predicate

lib/samehada/samehada_util/samehada_util.go

@@ -339,6 +339,22 @@ func ExtractOrgKeyFromDicOrderComparableEncodedVarchar(encodedVal *types.Value,
 	}
 }
 
+func ExtractOrgKeyFromDicOrderComparableEncodedBytes(buf []byte, valType types.TypeID) *types.Value {
+	switch valType {
+	case types.Integer:
+		retVal := types.NewValue(decodeFromDicOrderComparableBytes(buf[3:len(buf)-8], valType).(int32))
+		return &retVal
+	case types.Float:
+		retVal := types.NewValue(decodeFromDicOrderComparableBytes(buf[3:len(buf)-8], valType).(float32))
+		return &retVal
+	case types.Varchar:
+		orgStr := string(buf[:len(buf)-(4+8)])
+		return GetPonterOfValue(types.NewVarchar(orgStr))
+	default:
+		panic("not supported type")
+	}
+}
+
 func SHAssert(cond bool, msg string) {
 	if !cond {
 		panic(msg)

lib/storage/index/btree_index.go

@@ -32,7 +32,19 @@ func (btreeItr *BtreeIndexIterator) Next() (done bool, err error, key *types.Val
 	}
 	uintRID := binary.BigEndian.Uint64(packedRID)
 	unpackedRID := samehada_util.UnpackUint64toRID(uintRID)
-	decodedKey := samehada_util.ExtractOrgKeyFromDicOrderComparableEncodedVarchar(types.NewValueFromBytes(keyBytes, types.Varchar), btreeItr.valType)
+
+	// attach isNull flag and length of value due to these info is not stored in BLTree
+	keyLen := uint16(len(keyBytes) - 8) // 8 is length of packedRID
+	keyLenBuf := make([]byte, 2)
+	binary.LittleEndian.PutUint16(keyLenBuf, keyLen)
+	newKeyBytes := make([]byte, 0, len(keyBytes)+3)
+	newKeyBytes = append(newKeyBytes, 0)
+	newKeyBytes = append(newKeyBytes, keyLenBuf...)
+	newKeyBytes = append(newKeyBytes, keyBytes...)
+
+	//decodedKey := samehada_util.ExtractOrgKeyFromDicOrderComparableEncodedVarchar(types.NewValueFromBytes(newKeyBytes, types.Varchar), btreeItr.valType)
+	//decodedKey := samehada_util.ExtractOrgKeyFromDicOrderComparableEncodedVarchar(samehada_util.GetPonterOfValue(types.NewVarchar(string(newKeyBytes))), btreeItr.valType)
+	decodedKey := samehada_util.ExtractOrgKeyFromDicOrderComparableEncodedBytes(newKeyBytes, btreeItr.valType)
 	return false, nil, decodedKey, &unpackedRID
 }
 
@@ -78,7 +90,7 @@ func (btidx *BTreeIndex) insertEntryInner(key *tuple.Tuple, rid page.RID, txn in
 	packedRID := samehada_util.PackRIDtoUint64(&rid)
 	var valBuf [8]byte
 	binary.BigEndian.PutUint64(valBuf[:], packedRID)
-	btidx.container.InsertKey(convedKeyVal.Serialize(), 0, valBuf, true)
+	btidx.container.InsertKey(convedKeyVal.SerializeOnlyVal(), 0, valBuf, true)
 }
 
 func (btidx *BTreeIndex) InsertEntry(key *tuple.Tuple, rid page.RID, txn interface{}) {
@@ -95,7 +107,7 @@ func (btidx *BTreeIndex) deleteEntryInner(key *tuple.Tuple, rid page.RID, txn in
 		btidx.updateMtx.RLock()
 		defer btidx.updateMtx.RUnlock()
 	}
-	btidx.container.DeleteKey(convedKeyVal.Serialize(), 0)
+	btidx.container.DeleteKey(convedKeyVal.SerializeOnlyVal(), 0)
 }
 
 func (btidx *BTreeIndex) DeleteEntry(key *tuple.Tuple, rid page.RID, txn interface{}) {
@@ -110,7 +122,7 @@ func (btidx *BTreeIndex) ScanKey(key *tuple.Tuple, txn interface{}) []page.RID {
 
 	btidx.updateMtx.RLock()
 	// Attention: returned itr's containing keys are string type Value which is constructed with byte arr of concatenated  original key and value
-	rangeItr := btidx.container.GetRangeItr(smallestKeyVal.Serialize(), biggestKeyVal.Serialize())
+	rangeItr := btidx.container.GetRangeItr(smallestKeyVal.SerializeOnlyVal(), biggestKeyVal.SerializeOnlyVal())
 
 	retArr := make([]page.RID, 0)
 	for ok, _, packedRID := rangeItr.Next(); ok; ok, _, packedRID = rangeItr.Next() {
@@ -153,10 +165,10 @@ func (btidx *BTreeIndex) GetRangeScanIterator(start_key *tuple.Tuple, end_key *t
 	var biggestKeyBytes []byte
 
 	if smallestKeyVal != nil {
-		smalledKeyBytes = smallestKeyVal.Serialize()
+		smalledKeyBytes = smallestKeyVal.SerializeOnlyVal()
 	}
 	if biggestKeyVal != nil {
-		biggestKeyBytes = biggestKeyVal.Serialize()
+		biggestKeyBytes = biggestKeyVal.SerializeOnlyVal()
 	}
 	return NewBtreeIndexIterator(btidx.container.GetRangeItr(smalledKeyBytes, biggestKeyBytes), btidx.metadata.tuple_schema.GetColumn(btidx.col_idx).GetType())
 }

lib/types/column_value.go

@@ -332,26 +332,26 @@ func (v Value) Serialize() []byte {
 	return []byte{}
 }
 
-//// no length info and isNull info
-//func (v Value) SerializeOnlyVal() []byte {
-//	switch v.valueType {
-//	case Integer:
-//		buf := new(bytes.Buffer)
-//		binary.Write(buf, binary.LittleEndian, v.ToInteger())
-//		return buf.Bytes()
-//	case Float:
-//		buf := new(bytes.Buffer)
-//		binary.Write(buf, binary.LittleEndian, v.ToFloat())
-//		return buf.Bytes()
-//	case Varchar:
-//		return []byte(v.ToVarchar())
-//	case Boolean:
-//		buf := new(bytes.Buffer)
-//		binary.Write(buf, binary.LittleEndian, v.ToBoolean())
-//		return buf.Bytes()
-//	}
-//	return []byte{}
-//}
+// no length info and isNull info
+func (v Value) SerializeOnlyVal() []byte {
+	switch v.valueType {
+	case Integer:
+		buf := new(bytes.Buffer)
+		binary.Write(buf, binary.LittleEndian, v.ToInteger())
+		return buf.Bytes()
+	case Float:
+		buf := new(bytes.Buffer)
+		binary.Write(buf, binary.LittleEndian, v.ToFloat())
+		return buf.Bytes()
+	case Varchar:
+		return []byte(v.ToVarchar())
+	case Boolean:
+		buf := new(bytes.Buffer)
+		binary.Write(buf, binary.LittleEndian, v.ToBoolean())
+		return buf.Bytes()
+	}
+	return []byte{}
+}
 
 // Size returns the size in bytes that the type will occupy inside the tuple
 func (v Value) Size() uint32 {