qb/storage/navigation.go

package storage

import (
	bin "gordenko.dev/dima/bin/little"
)

const (
	PageNoSize = 4
)

type KeyComparator interface {
	CompareTo(int) int
}

type ValueAtComparator struct {
	buf       []byte
	timestamp uint32
}

func (s ValueAtComparator) CompareTo(elemIdx int) int {
	var (
		pos     = elemIdx * IndexRecordSize
		elem, _ = bin.GetUint32(s.buf[pos:])
	)

	if s.timestamp < elem {
		return -1
	} else if s.timestamp > elem {
		return 1
	} else {
		return 0
	}
}

func BinarySearch(qty int, keyComparator KeyComparator) (elemIdx int, isFound bool) {
	if qty == 0 {
		return
	}
	a := 0
	b := qty - 1
	for {
		var (
			elemIdx = (b-a)/2 + a
			code    = keyComparator.CompareTo(elemIdx)
		)
		if code == 1 {
			a = elemIdx + 1
			if a > b {
				return elemIdx, false // +1
			}
		} else if code == -1 {
			b = elemIdx - 1
			if b < a {
				if elemIdx == 0 {
					return 0, false
				} else {
					return elemIdx - 1, false
				}
			}
		} else {
			return elemIdx, true
		}
	}
}

func FindPageOnIndexLevelTail(level IndexLevelTail, timestamp uint32) (pageNo uint32) {
	comparator := ValueAtComparator{
		buf:       level.Buffer,
		timestamp: timestamp,
	}
	elemIdx, _ := BinarySearch(level.RecordsCount, comparator)
	pos := elemIdx*IndexRecordSize + 4 // timestamp size
	pageNo, _ = bin.GetUint32(level.Buffer[pos:])
	return
}

func FindPageOnIndexPage(page []byte, timestamp uint32) (pageNo uint32) {
	comparator := ValueAtComparator{
		buf:       page,
		timestamp: timestamp,
	}
	count, _ := bin.GetUint16(page[indexRecordsCountIdx:])
	elemIdx, _ := BinarySearch(int(count), comparator)
	pos := elemIdx*IndexRecordSize + 4 // timestamp size
	pageNo, _ = bin.GetUint32(page[pos:])
	return
}

func IsZeroLevelPage(buf []byte) bool {
	return buf[isZeroLevelIdx] == 1
}

// можна перевірити хвости від 0 до ... Перевіряю since кожного хвоста.
// Якщо timestamp >=, шукаю pageNo бінарним пошуком. І сторінка мені однозначно підходить.
// Якщо timestamp <, піднімаюсь вище. Якщо рівнів більше немає - until вказано за межами Range показань.
func FindPageOnIndexLevelTails(levels []IndexLevelTail, timestamp uint32) (pageNo uint32, isDataPage bool) {
	for i, level := range levels {
		tailSince, _ := bin.GetUint32(level.Buffer)
		if timestamp >= tailSince {
			pageNo = FindPageOnIndexLevelTail(level, timestamp)
			return pageNo, i == 0
		}
	}
	return
}

type DeleteSinceOnIndexPageResult struct {
	Buffer       []byte
	RecordsCount int
	PageNumbers  []uint32
	//Idx          int
}

func DeleteSinceOnIndexPage(buf []byte, since uint32) DeleteSinceOnIndexPageResult {
	comparator := ValueAtComparator{
		buf:       buf,
		timestamp: since,
	}
	count, _ := bin.GetUint16(buf[indexRecordsCountIdx:])
	elemIdx, _ := BinarySearch(int(count), comparator)
	newbuf := make([]byte, IndexPageSize)
	copy(newbuf, buf[:elemIdx*IndexRecordSize]) // [0..idx)
	pos := elemIdx*IndexRecordSize + 4          // timestamp size
	deleteCount := int(count) - elemIdx
	var pageNumbers []uint32
	for range deleteCount {
		pageNo, _ := bin.GetUint32(buf[pos:])
		pageNumbers = append(pageNumbers, pageNo)
		pos += IndexRecordSize
	}

	return DeleteSinceOnIndexPageResult{
		Buffer:       newbuf,
		RecordsCount: elemIdx,
		PageNumbers:  pageNumbers,
		//Idx:          elemIdx,
	}
}

type DeleteOnIndexTailResult struct {
	RecordsCount int
	PageNumbers  []uint32
}

func DeleteSinceOnIndexTail(level IndexLevelTail, since uint32) DeleteOnIndexTailResult {
	comparator := ValueAtComparator{
		buf:       level.Buffer,
		timestamp: since,
	}
	elemIdx, _ := BinarySearch(level.RecordsCount, comparator)
	pos := elemIdx*IndexRecordSize + 4 // timestamp size
	deleteCount := level.RecordsCount - elemIdx
	var pageNumbers []uint32
	for range deleteCount {
		pageNo, _ := bin.GetUint32(level.Buffer[pos:])
		pageNumbers = append(pageNumbers, pageNo)
		pos += IndexRecordSize
	}
	return DeleteOnIndexTailResult{
		RecordsCount: elemIdx,
		PageNumbers:  pageNumbers,
		//Idx:          elemIdx,
	}
}

func DeleteUntilOnIndexTail(level IndexLevelTail, until uint32) DeleteOnIndexTailResult {
	comparator := ValueAtComparator{
		buf:       level.Buffer,
		timestamp: until,
	}
	elemIdx, _ := BinarySearch(level.RecordsCount, comparator)
	pos := 4 // timestamp size
	deleteCount := elemIdx + 1
	var pageNumbers []uint32
	for range deleteCount {
		pageNo, _ := bin.GetUint32(level.Buffer[pos:])
		pageNumbers = append(pageNumbers, pageNo)
		pos += IndexRecordSize
	}
	copy(level.Buffer, level.Buffer[:])
	return DeleteOnIndexTailResult{
		RecordsCount: elemIdx,
		PageNumbers:  pageNumbers,
		//Idx:          elemIdx,
	}
}

// func getPrevPageNo(buf []byte) uint32 {
// 	pageNo, _ := bin.GetUint32(buf[prevPageIdx:])
// 	return pageNo
// }

// func GetIndexRecordsSince(buf []byte) (pageNo uint32) {
// 	pageNo, _ = bin.GetUint32(buf)
// 	return
// }

// func FindPageOnRecords(records []byte, timestamp uint32) (pageNo uint32) {
// 	comparator := ValueAtComparator{
// 		buf:       records,
// 		timestamp: timestamp,
// 	}
// 	count := len(records) / IndexRecordSize
// 	elemIdx, _ := BinarySearch(count, comparator)
// 	pageNo, _ = bin.GetUint32(records[elemIdx*IndexRecordSize:])
// 	return
// }

// func findPageNoIdx(buf []byte, timestamp uint32) (idx int) {
// 	comparator := ValueAtComparator{
// 		buf:       buf,
// 		timestamp: timestamp,
// 	}
// 	qty, _ := bin.GetUint16(buf[indexRecordsQtyIdx:])
// 	elemIdx, _ := BinarySearch(int(qty), comparator)
// 	return elemIdx
// }

// func findPageNoForDeleteSince(buf []byte, since uint32) (uint32, int, int) {
// 	comparator := ValueAtComparator{
// 		buf:       buf,
// 		timestamp: since,
// 	}
// 	qty, _ := bin.GetUint16(buf[indexRecordsQtyIdx:])
// 	elemIdx, _ := BinarySearch(int(qty), comparator)
// 	pos := elemIdx * timestampSize
// 	timestamp, _ := bin.GetUint32(buf[pos:])

// 	if timestamp == since {
// 		if elemIdx == 0 {
// 			return 0, int(qty), -1
// 		}
// 		elemIdx--
// 	}
// 	pos = indexFooterIdx - (elemIdx+1)*PageNoSize
// 	pageNo, _ := bin.GetUint32(buf[pos:])
// 	return pageNo, int(qty), elemIdx
// }

// func getLastPageNo(buf []byte) (pageNo uint32) {
// 	qty, _ := bin.GetUint16(buf[indexRecordsQtyIdx:])
// 	pos := indexFooterIdx - qty*PageNoSize
// 	pageNo, _ = bin.GetUint32(buf[pos:])
// 	return
// }

// func getPageNo(buf []byte, idx int) (pageNo uint32) {
// 	pos := indexFooterIdx - (idx+1)*PageNoSize
// 	pageNo, _ = bin.GetUint32(buf[pos:])
// 	return
// }

// from index page
func ListPageNumbers(buf []byte) (pageNumbers []uint32) {
	count, _ := bin.GetUint16(buf[indexRecordsCountIdx:])
	pos := timestampSize
	for range count {
		pageNo, _ := bin.GetUint32(buf[pos:])
		pageNumbers = append(pageNumbers, pageNo)
		pos += IndexRecordSize
	}
	return
}

// // include since timestamp
// func listPageNumbersSince(buf []byte, timestamp uint32) (pageNumbers []uint32) {
// 	comparator := ValueAtComparator{
// 		buf:       buf,
// 		timestamp: timestamp,
// 	}
// 	qty, _ := bin.GetUint16(buf[indexRecordsQtyIdx:])
// 	elemIdx, _ := BinarySearch(int(qty), comparator)
// 	pos := indexFooterIdx - (elemIdx+1)*PageNoSize
// 	for range qty {
// 		pageNo, _ := bin.GetUint32(buf[pos:])
// 		pageNumbers = append(pageNumbers, pageNo)
// 		pos -= PageNoSize
// 	}
// 	return
// }

// func getSince(buf []byte) uint32 {
// 	return bin.GetUint32(buf[0:])
// }

// func appendPair(buf []byte, timestamp uint32, pageNo uint32) bool {
// 	qty := bin.GetUint16AsInt(buf[indexRecordsQtyIdx:])
// 	free := indexFooterIdx - qty*pairSize
// 	if free < pairSize {
// 		return false
// 	}
// 	pos := qty * timestampSize
// 	bin.PutUint32(buf[pos:], timestamp)
// 	pos = indexFooterIdx - (qty+1)*PageNoSize
// 	bin.PutUint32(buf[pos:], pageNo)
// 	bin.PutIntAsUint16(buf[indexRecordsQtyIdx:], qty+1)
// 	return true
// }