row.go

// Copyright 2022 Molecula Corp. (DBA FeatureBase).
// SPDX-License-Identifier: Apache-2.0
package rows

import (
	"encoding/json"
	"fmt"
	"sort"
	"time"

	"errors"

	"github.com/gernest/roaring"
	"github.com/gernest/roaring/shardwidth"
)

type DataType byte

const (
	String DataType = 1 + iota
	Int64
	Float64
	Timestamp
)

type ToRowser interface {
	ToRows(func(*RowResult) error) error
}

func RowsToTable(tr ToRowser, n int) (*Table, error) {
	var headers []*ColumnInfo
	rows := make([]RowValue, 0, n)

	// This callback gets called for every "row" in r.
	// Each row populates its position in the pre-allocated
	// `rows`. The headers get set based on those received
	// in the first row.
	cb := func(rr *RowResult) error {
		if len(rows) == 0 {
			headers = rr.Headers
		}
		rows = append(rows, rr.Columns)
		return nil
	}

	if err := tr.ToRows(cb); err != nil {
		return nil, fmt.Errorf("calling callback %w", err)
	}

	return &Table{
		Headers: headers,
		Rows:    rows,
	}, nil
}

type Table struct {
	Headers []*ColumnInfo
	Rows    []RowValue
}

type RowResult struct {
	Headers []*ColumnInfo
	Columns []Value
}

type ColumnInfo struct {
	Name     string
	DataType DataType
}

type RowValue []Value

type Value struct {
	string    *string
	int       *uint64
	double    *float64
	timestamp *uint64
}

func StringValue(s string) Value {
	return Value{
		string: &s,
	}
}
func IntValue(v uint64) Value {
	return Value{
		int: &v,
	}
}

func (v Value) String() string {
	if v.string != nil {
		return *v.string
	}
	return ""
}

func (v Value) Int() uint64 {
	if v.int != nil {
		return *v.int
	}
	return 0
}

func (v Value) Float64() float64 {
	if v.int != nil {
		return *v.double
	}
	return 0
}

func (v Value) Time() time.Time {
	if v.timestamp != nil {
		return time.UnixMilli(int64(*v.timestamp)).UTC()
	}
	return time.Time{}
}

// Row is a set of integers (the associated columns).
type Row struct {
	Segments []RowSegment

	// String keys translated to/from segment columns.
	Keys []string

	// Index tells what index this row is from - needed for key translation.
	Index string

	// Field tells what field this row is from if it's a "vertical"
	// row. It may be the result of a Distinct query or Rows
	// query. Knowing the index and field, we can figure out how to
	// interpret the row data.
	Field string

	// NoSplit indicates that this row may not be split.
	// This is used for `Rows` calls in a GroupBy.
	NoSplit bool
}

// NewRow returns a new instance of Row.
func NewRow(columns ...uint64) *Row {
	r := &Row{}
	for _, i := range columns {
		r.SetBit(i)
	}
	return r
}

func (r *Row) ToTable() (*Table, error) {
	var n int
	if len(r.Keys) > 0 {
		n = len(r.Keys)
	} else {
		n = len(r.Columns())
	}
	return RowsToTable(r, n)
}

func (r *Row) ToRows(callback func(*RowResult) error) error {
	if len(r.Keys) > 0 {
		// Column keys
		ci := []*ColumnInfo{
			{Name: "_id", DataType: String},
		}
		for _, x := range r.Keys {
			if err := callback(&RowResult{
				Headers: ci,
				Columns: []Value{StringValue(x)},
			}); err != nil {
				return fmt.Errorf("calling callback %w", err)
			}
			ci = nil // only send on the first
		}
	} else {
		// Column IDs
		ci := []*ColumnInfo{
			{Name: "_id", DataType: Int64},
		}
		for _, x := range r.Columns() {
			if err := callback(&RowResult{
				Headers: ci,
				Columns: []Value{IntValue(x)},
			}); err != nil {
				return fmt.Errorf("calling callback %w", err)
			}
			ci = nil // only send on the first
		}
	}
	return nil
}

func (r *Row) Clone() (clone *Row) {
	if r == nil {
		return nil
	}
	var keyClone []string
	if len(r.Keys) > 0 {
		keyClone = make([]string, len(r.Keys))
		copy(keyClone, r.Keys)
	}

	clone = &Row{
		Keys:  keyClone,
		Index: r.Index,
		Field: r.Field,
	}

	for _, seg := range r.Segments {
		segClone := RowSegment{
			shard:    seg.shard,
			writable: true, // we know it is safe; it is a copy.
			n:        seg.n,
		}
		if seg.data != nil {
			segClone.data = seg.data.Clone() // *roaring.Bitmap
		}
		// segClone.InvalidateCount() // not needed?
		clone.Segments = append(clone.Segments, segClone)
	}
	return clone
}

// NewRowFromBitmap divides a bitmap into rows, which it now calls shards. This
// transposes; data that was in any shard for Row 0 is now considered shard 0,
// etcetera.
func NewRowFromBitmap(b *roaring.Bitmap) *Row {
	r := &Row{}
	if b == nil {
		return r
	}
	rowNum := uint64(0)
	for col, ok := b.MinAt(rowNum * shardwidth.ShardWidth); ok; col, ok = b.MinAt(rowNum * shardwidth.ShardWidth) {
		rowNum = col / shardwidth.ShardWidth
		seg := RowSegment{
			shard:    rowNum,
			data:     b.OffsetRange(rowNum*shardwidth.ShardWidth, rowNum*shardwidth.ShardWidth, (rowNum+1)*shardwidth.ShardWidth),
			writable: true,
		}
		seg.n = seg.data.Count()
		r.Segments = append(r.Segments, seg)
		rowNum++
	}
	return r
}

// NewRowFromRoaring parses a roaring data file as a row, dividing it into
// bitmaps and rowSegments based on shard width.
func NewRowFromRoaring(data []byte) *Row {
	bitmaps, shards := roaring.RoaringToBitmaps(data, shardwidth.ShardWidth)
	r := &Row{Segments: make([]RowSegment, len(bitmaps))}
	for i := range bitmaps {
		segment := RowSegment{
			shard:    shards[i],
			data:     bitmaps[i],
			writable: false,
			n:        bitmaps[i].Count(),
		}
		r.Segments[i] = segment
	}
	return r
}

// Hash calculate checksum code be useful in block hash join
func (r *Row) Hash() uint64 {
	hash := uint64(0)
	for i := range r.Segments {
		hash = r.Segments[i].data.Hash(hash)
	}
	return hash
}

// Roaring returns the row treated as a unified roaring bitmap.
func (r *Row) Roaring() []byte {
	bitmaps := make([]*roaring.Bitmap, len(r.Segments))
	for i := range r.Segments {
		bitmaps[i] = r.Segments[i].data
	}
	return roaring.BitmapsToRoaring(bitmaps)
}

// IsEmpty returns true if the row doesn't contain any set bits.
func (r *Row) IsEmpty() bool {
	if len(r.Segments) == 0 {
		return true
	}
	for i := range r.Segments {
		if r.Segments[i].n > 0 {
			return false
		}
	}
	return true
}

func (r *Row) Freeze() {
	for _, s := range r.Segments {
		s.Freeze()
	}
}

// Merge merges data from other into r.
func (r *Row) Merge(other *Row) {
	var segments []RowSegment

	itr := newMergeSegmentIterator(r.Segments, other.Segments)
	for s0, s1 := itr.next(); s0 != nil || s1 != nil; s0, s1 = itr.next() {
		// Use the other row's data if segment is missing.
		if s0 == nil {
			segments = append(segments, *s1)
			continue
		} else if s1 == nil {
			segments = append(segments, *s0)
			continue
		}

		// Otherwise merge.
		s0.Merge(s1)
		segments = append(segments, *s0)
	}

	r.Segments = segments
	r.invalidateCount()
}

// intersectionCount returns the number of intersections between r and other.
func (r *Row) intersectionCount(other *Row) uint64 {
	var n uint64

	itr := newMergeSegmentIterator(r.Segments, other.Segments)
	for s0, s1 := itr.next(); s0 != nil || s1 != nil; s0, s1 = itr.next() {
		// Ignore non-overlapping segments.
		if s0 == nil || s1 == nil {
			continue
		}

		n += s0.IntersectionCount(s1)
	}
	return n
}

// Intersect returns the itersection of r and other.
func (r *Row) Intersect(other *Row) *Row {
	var segments []RowSegment

	itr := newMergeSegmentIterator(r.Segments, other.Segments)
	for s0, s1 := itr.next(); s0 != nil || s1 != nil; s0, s1 = itr.next() {
		// Ignore non-overlapping segments.
		if s0 == nil || s1 == nil {
			continue
		}
		segments = append(segments, *s0.Intersect(s1))
	}

	return &Row{Segments: segments}
}

// Any returns true if row contains any bits.
func (r *Row) Any() bool {
	for _, s := range r.Segments {
		if s.data.Any() {
			return true
		}
	}
	return false
}

// Xor returns the xor of r and other.
func (r *Row) Xor(other *Row) *Row {
	var segments []RowSegment

	itr := newMergeSegmentIterator(r.Segments, other.Segments)
	for s0, s1 := itr.next(); s0 != nil || s1 != nil; s0, s1 = itr.next() {
		if s1 == nil {
			segments = append(segments, *s0)
			continue
		} else if s0 == nil {
			segments = append(segments, *s1)
			continue
		}

		segments = append(segments, *s0.Xor(s1))
	}

	return &Row{Segments: segments}
}

// Union returns the bitwise union of r and other.
func (r *Row) Union(others ...*Row) *Row {
	segments := make([][]RowSegment, 0, len(others)+1)
	if len(r.Segments) > 0 {
		segments = append(segments, r.Segments)
	}
	nextSegs := make([][]RowSegment, 0, len(others)+1)
	toProcess := make([]*RowSegment, 0, len(others)+1)
	var output []RowSegment
	for _, other := range others {
		if len(other.Segments) > 0 {
			segments = append(segments, other.Segments)
		}
	}
	for len(segments) > 0 {
		shard := segments[0][0].shard
		for _, segs := range segments {
			if segs[0].shard < shard {
				shard = segs[0].shard
			}
		}
		nextSegs = nextSegs[:0]
		toProcess = toProcess[:0]
		for _, segs := range segments {
			if segs[0].shard == shard {
				toProcess = append(toProcess, &segs[0])
				segs = segs[1:]
			}
			if len(segs) > 0 {
				nextSegs = append(nextSegs, segs)
			}
		}
		// at this point, "toProcess" is a list of all the segments
		// sharing the lowest ID, and nextSegs is a list of all the others.
		// Swap the segment lists (so we don't have to reallocate it)
		segments, nextSegs = nextSegs, segments
		if len(toProcess) == 1 {
			output = append(output, *toProcess[0])
		} else {
			output = append(output, *toProcess[0].Union(toProcess[1:]...))
		}
	}
	return &Row{Index: r.Index, Field: r.Field, Segments: output}
}

// Difference returns the diff of r and other.
func (r *Row) Difference(others ...*Row) *Row {
	var output []RowSegment
	o := make(map[uint64][]*RowSegment)

	for x := range others {
		for y := range others[x].Segments {
			segment := others[x].Segments[y]
			o[segment.shard] = append(o[segment.shard], &segment)
		}
	}
	for _, segment := range r.Segments {

		dest, ok := o[segment.shard]
		if ok {
			output = append(output, *segment.Difference(dest...))
		} else {
			output = append(output, segment)
		}
	}
	return &Row{Segments: output}
}

// Shift returns the bitwise shift of r by n bits.
// Currently only positive shift values are supported.
//
// NOTE: the Shift method is currently unsupported, and
// is considerred to be incorrect. Please DO NOT use it.
// We are leaving it here in case someone internally wants
// to use it with the understanding that the results may
// be incorrect.
//
// Why unsupported? For a full description, see:
// https://github.com/featurebasedb/pilosa/issues/403.
// In short, the current implementation will shift a bit
// at the edge of a shard out of the shard and into a
// container which is assumed to be an invalid container
// for the shard. So for example, shifting the last bit
// of shard 0 (containers 0-15) will shift that bit out
// to container 16. While this "sort of" works, it
// breaks an assumption about containers, and might stop
// working in the future if that assumption is enforced.
func (r *Row) Shift(n int64) (*Row, error) {
	if n < 0 {
		return nil, errors.New("cannot shift by negative values")
	} else if n == 0 {
		return r, nil
	}

	work := r
	var segments []RowSegment
	for i := int64(0); i < n; i++ {
		segments = segments[:0]
		for _, segment := range work.Segments {
			shifted, err := segment.Shift()
			if err != nil {
				return nil, fmt.Errorf("shifting row segment %w", err)
			}
			segments = append(segments, *shifted)
		}
		work = &Row{Segments: segments}
	}

	return work, nil
}

// SetBit sets the i-th column of the row.
func (r *Row) SetBit(i uint64) (changed bool) {
	return r.createSegmentIfNotExists(i / shardwidth.ShardWidth).SetBit(i)
}

// segment returns a segment for a given shard.
// Returns nil if segment does not exist.
func (r *Row) segment(shard uint64) *RowSegment {
	if i := sort.Search(len(r.Segments), func(i int) bool {
		return r.Segments[i].shard >= shard
	}); i < len(r.Segments) && r.Segments[i].shard == shard {
		return &r.Segments[i]
	}
	return nil
}

func (r *Row) createSegmentIfNotExists(shard uint64) *RowSegment {
	i := sort.Search(len(r.Segments), func(i int) bool {
		return r.Segments[i].shard >= shard
	})

	// Return exact match.
	if i < len(r.Segments) && r.Segments[i].shard == shard {
		return &r.Segments[i]
	}

	// Insert new segment.
	r.Segments = append(r.Segments, RowSegment{data: roaring.NewSliceBitmap()})
	if i < len(r.Segments) {
		copy(r.Segments[i+1:], r.Segments[i:])
	}
	r.Segments[i] = RowSegment{
		data:     roaring.NewSliceBitmap(),
		shard:    shard,
		writable: true,
	}

	return &r.Segments[i]
}

// invalidateCount updates the cached count in the row.
func (r *Row) invalidateCount() {
	for i := range r.Segments {
		r.Segments[i].InvalidateCount()
	}
}

// Count returns the number of columns in the row.
func (r *Row) Count() uint64 {
	var n uint64
	if r == nil {
		// Count(Distinct()) on an empty field panics here
		return n
	}
	for i := range r.Segments {
		n += r.Segments[i].Count()
	}
	return n
}

// MarshalJSON returns a JSON-encoded byte slice of r.
func (r *Row) MarshalJSON() ([]byte, error) {
	var o struct {
		Columns []uint64 `json:"columns"`
		Keys    []string `json:"keys,omitempty"`
	}
	o.Columns = r.Columns()
	o.Keys = r.Keys

	return json.Marshal(&o)
}

// Columns returns the columns in r as a slice of ints.
func (r *Row) Columns() []uint64 {
	// We occasionally hit cases where we want to call Columns on something
	// that might not exist, but a nil slice would be fine.
	if r == nil {
		return nil
	}
	a := make([]uint64, 0, r.Count())
	for i := range r.Segments {
		a = append(a, r.Segments[i].Columns()...)
	}
	return a
}

func (r *Row) ShardColumns() []int64 {
	// We occasionally hit cases where we want to call Columns on something
	// that might not exist, but a nil slice would be fine.
	if r == nil {
		return nil
	}
	a := make([]int64, 0, r.Count())
	for i := range r.Segments {
		a = append(a, r.Segments[i].ShardColumns()...)
	}
	return a
}

// Includes returns true if the row contains the given column.
func (r *Row) Includes(col uint64) bool {
	shard := col / shardwidth.ShardWidth
	for i := range r.Segments {
		if r.Segments[i].shard == shard {
			return r.Segments[i].data.Contains(col)
		}
	}
	return false
}

// RowSegment holds a subset of a row.
// This could point to a mmapped roaring bitmap or an in-memory bitmap. The
// width of the segment will always match the shard width.
type RowSegment struct {
	// Shard this segment belongs to
	shard uint64

	// Underlying raw bitmap implementation.
	// This is an mmapped bitmap if writable is false. Otherwise
	// it is a heap allocated bitmap which can be manipulated.
	data     *roaring.Bitmap
	writable bool

	// Bit count
	n uint64
}

func (s *RowSegment) Shard() uint64 {
	return s.shard
}

func (s *RowSegment) Freeze() {
	s.data = s.data.Freeze()
}

/*
// Raw returns the row segment as a byte slice.
// It may be used by the gRPC server to deliver results
// as a roaring bitmap instead of a stream of RowResults.
func (s *rowSegment) Raw() (uint64, []byte) {
	var buf bytes.Buffer
	s.data.WriteTo(&buf)
	return s.shard, buf.Bytes()
}
*/

// Merge adds chunks from other to s.
// Chunks in s are overwritten if they exist in other.
func (s *RowSegment) Merge(other *RowSegment) {
	s.ensureWritable()

	itr := other.data.Iterator()
	for v, eof := itr.Next(); !eof; v, eof = itr.Next() {
		s.SetBit(v)
	}
}

// IntersectionCount returns the number of intersections between s and other.
func (s *RowSegment) IntersectionCount(other *RowSegment) uint64 {
	return s.data.IntersectionCount(other.data)
}

// Intersect returns the itersection of s and other.
func (s *RowSegment) Intersect(other *RowSegment) *RowSegment {
	data := s.data.Intersect(other.data)

	return &RowSegment{
		data:  data,
		shard: s.shard,
		n:     data.Count(),
	}
}

// Union returns the bitwise union of s and other.
func (s *RowSegment) Union(others ...*RowSegment) *RowSegment {
	datas := make([]*roaring.Bitmap, len(others))
	for i, other := range others {
		datas[i] = other.data
	}
	data := s.data.Union(datas...)

	return &RowSegment{
		data:  data,
		shard: s.shard,
		n:     data.Count(),
	}
}

// Difference returns the diff of s and other.
func (s *RowSegment) Difference(others ...*RowSegment) *RowSegment {
	datas := make([]*roaring.Bitmap, len(others))
	for i, other := range others {
		datas[i] = other.data
	}
	data := s.data.Difference(datas...)

	return &RowSegment{
		data:  data,
		shard: s.shard,
		n:     data.Count(),
	}
}

// Xor returns the xor of s and other.
func (s *RowSegment) Xor(other *RowSegment) *RowSegment {
	data := s.data.Xor(other.data)

	return &RowSegment{
		data:  data,
		shard: s.shard,
		n:     data.Count(),
	}
}

// Shift returns s shifted by 1 bit.
func (s *RowSegment) Shift() (*RowSegment, error) {
	// TODO: deal with overflow
	// See issue: https://github.com/featurebasedb/pilosa/issues/403
	data, err := s.data.Shift(1)
	if err != nil {
		return nil, fmt.Errorf("shifting roaring data %w", err)
	}

	return &RowSegment{
		data:  data,
		shard: s.shard,
		n:     data.Count(),
	}, nil
}

// SetBit sets the i-th column of the row.
func (s *RowSegment) SetBit(i uint64) (changed bool) {
	s.ensureWritable()
	changed, _ = s.data.Add(i)
	if changed {
		s.n++
	}
	return changed
}

// ClearBit clears the i-th column of the row.
func (s *RowSegment) ClearBit(i uint64) (changed bool) {
	s.ensureWritable()

	changed, _ = s.data.Remove(i)
	if changed {
		s.n--
	}
	return changed
}

// InvalidateCount updates the cached count in the row.
func (s *RowSegment) InvalidateCount() {
	s.n = s.data.Count()
}

// Columns returns a list of all columns set in the segment.
func (s *RowSegment) Columns() []uint64 {
	a := make([]uint64, 0, s.Count())
	itr := s.data.Iterator()
	for v, eof := itr.Next(); !eof; v, eof = itr.Next() {
		a = append(a, v)
	}
	return a
}

// Columns returns a list of all columns set in the segment, normalized from 0-shardwidth-1
func (s *RowSegment) ShardColumns() []int64 {
	a := make([]int64, 0, s.Count())
	itr := s.data.Iterator()
	mask := uint64(shardwidth.ShardWidth - 1)
	for v, eof := itr.Next(); !eof; v, eof = itr.Next() {
		a = append(a, int64(mask&v))
	}
	return a
}

// Count returns the number of set columns in the row.
func (s *RowSegment) Count() uint64 { return s.n }

// ensureWritable clones the segment if it is pointing to non-writable data.
func (s *RowSegment) ensureWritable() {
	if s.writable {
		return
	}

	// This doesn't actually clone all the containers, but does clone
	// the bitmap itself -- we get a new bitmap, but it just marks the
	// containers as frozen and shares them. It's now safe to write to
	// this bitmap, but the actual containers are copy-on-write.
	s.data = s.data.Freeze()
	s.writable = true
}

// mergeSegmentIterator produces an iterator that loops through two sets of segments.
type mergeSegmentIterator struct {
	a0, a1 []RowSegment
}

// newMergeSegmentIterator returns a new instance of mergeSegmentIterator.
func newMergeSegmentIterator(a0, a1 []RowSegment) mergeSegmentIterator {
	return mergeSegmentIterator{a0: a0, a1: a1}
}

// next returns the next set of segments.
func (itr *mergeSegmentIterator) next() (s0, s1 *RowSegment) {
	// Find current segments.
	if len(itr.a0) > 0 {
		s0 = &itr.a0[0]
	}
	if len(itr.a1) > 0 {
		s1 = &itr.a1[0]
	}

	// Return if either or both are nil.
	if s0 == nil && s1 == nil {
		return
	} else if s0 == nil {
		itr.a1 = itr.a1[1:]
		return
	} else if s1 == nil {
		itr.a0 = itr.a0[1:]
		return
	}

	// Otherwise determine which is first.
	if s0.shard < s1.shard {
		itr.a0 = itr.a0[1:]
		return s0, nil
	} else if s0.shard > s1.shard {
		itr.a1 = itr.a1[1:]
		return s1, nil
	}

	// Return both if shards are equal.
	itr.a0, itr.a1 = itr.a0[1:], itr.a1[1:]
	return s0, s1
}