<commit_before>package tdb

import (
	"bytes"
	"fmt"
	"time"

	"github.com/dustin/go-humanize"
	"github.com/oxtoacart/tdb/values"
	"github.com/spaolacci/murmur3"
	"github.com/tecbot/gorocksdb"
	"gopkg.in/vmihailenco/msgpack.v2"
)

type Point struct {
	Ts   time.Time
	Dims map[string]interface{}
	Vals map[string]float64
}

type Value interface {
	Val() float64

	Add(addend float64) Value
}

type FloatValue float64

func (v FloatValue) Val() float64 {
	return float64(v)
}

func (v FloatValue) Add(addend float64) Value {
	return FloatValue(v.Val() + addend)
}

type bucket struct {
	start time.Time
	val   values.Value
	prev  *bucket
}

type partition struct {
	t       *table
	inserts chan *insert
	tail    map[string]*bucket
}

type insert struct {
	ts  time.Time
	key []byte
	val values.Value
}

type archiveRequest struct {
	key string
	b   *bucket
}

func (db *DB) Insert(table string, point *Point) error {
	t := db.getTable(table)
	if t == nil {
		return fmt.Errorf("Unknown table %v", table)
	}

	return t.insert(point)
}

func (t *table) insert(point *Point) error {
	t.clock.Advance(point.Ts)
	vals := floatsToValues(point.Vals)
	// TODO: deal with situation where name of inserted field conflicts with
	// derived field
	for _, field := range t.derivedFields {
		vals[field.Name] = field.Expr(vals)
	}

	for field, val := range vals {
		key, err := point.keyFor(field)
		if err != nil {
			return err
		}
		h := int(murmur3.Sum32(key))
		p := h % len(t.partitions)
		select {
		case t.partitions[p].inserts <- &insert{point.Ts, key, val}:
			// ok
		default:
			log.Error("Partition full, dropping point")
		}
	}

	return nil
}

func (p *partition) processInserts() {
	// TODO: base this on the passage of fake time
	archivePeriod := p.t.hotPeriod / 10
	log.Debugf("Archiving every %v", archivePeriod)
	archiveTicker := p.t.clock.NewTicker(archivePeriod)
	for {
		select {
		case insert := <-p.inserts:
			p.insert(insert)
		case <-archiveTicker.C:
			p.requestArchiving()
		}
	}
}

func (p *partition) insert(insert *insert) {
	key := string(insert.key)
	now := p.t.clock.Now()
	start := roundTime(insert.ts, p.t.resolution)
	if now.Sub(start) > p.t.hotPeriod {
		log.Trace("Discarding insert outside of hot period")
		return
	}
	b := p.tail[key]
	if b == nil || b.start.Before(start) {
		p.t.statsMutex.Lock()
		if b == nil {
			p.t.stats.HotKeys++
		}
		p.t.statsMutex.Unlock()
		b = &bucket{start, insert.val, b}
		p.tail[key] = b
		return
	}
	for {
		if b.start == start {
			// Update existing bucket
			b.val = b.val.Plus(insert.val)
			return
		}
		if b.prev == nil || b.prev.start.Before(start) {
			// Insert new bucket
			p.t.statsMutex.Lock()
			p.t.statsMutex.Unlock()
			b.prev = &bucket{start, insert.val, b.prev}
			return
		}
		// Continue looking
		b = b.prev
	}
}

func (p *partition) requestArchiving() {
	now := p.t.clock.Now()
	log.Tracef("Requested archiving at %v", now)
	for key, b := range p.tail {
		if now.Sub(b.start) > p.t.hotPeriod {
			log.Tracef("Archiving full. %v / %v %v", b.start, now, b.prev != nil)
			delete(p.tail, key)
			p.t.statsMutex.Lock()
			p.t.stats.HotKeys--
			p.t.statsMutex.Unlock()
			p.t.toArchive <- &archiveRequest{key, b}
			continue
		}<commit_msg>Turned up log level on archive requests<commit_after>package tdb

import (
	"bytes"
	"fmt"
	"time"

	"github.com/dustin/go-humanize"
	"github.com/oxtoacart/tdb/values"
	"github.com/spaolacci/murmur3"
	"github.com/tecbot/gorocksdb"
	"gopkg.in/vmihailenco/msgpack.v2"
)

type Point struct {
	Ts   time.Time
	Dims map[string]interface{}
	Vals map[string]float64
}

type Value interface {
	Val() float64

	Add(addend float64) Value
}

type FloatValue float64

func (v FloatValue) Val() float64 {
	return float64(v)
}

func (v FloatValue) Add(addend float64) Value {
	return FloatValue(v.Val() + addend)
}

type bucket struct {
	start time.Time
	val   values.Value
	prev  *bucket
}

type partition struct {
	t       *table
	inserts chan *insert
	tail    map[string]*bucket
}

type insert struct {
	ts  time.Time
	key []byte
	val values.Value
}

type archiveRequest struct {
	key string
	b   *bucket
}

func (db *DB) Insert(table string, point *Point) error {
	t := db.getTable(table)
	if t == nil {
		return fmt.Errorf("Unknown table %v", table)
	}

	return t.insert(point)
}

func (t *table) insert(point *Point) error {
	t.clock.Advance(point.Ts)
	vals := floatsToValues(point.Vals)
	// TODO: deal with situation where name of inserted field conflicts with
	// derived field
	for _, field := range t.derivedFields {
		vals[field.Name] = field.Expr(vals)
	}

	for field, val := range vals {
		key, err := point.keyFor(field)
		if err != nil {
			return err
		}
		h := int(murmur3.Sum32(key))
		p := h % len(t.partitions)
		select {
		case t.partitions[p].inserts <- &insert{point.Ts, key, val}:
			// ok
		default:
			log.Error("Partition full, dropping point")
		}
	}

	return nil
}

func (p *partition) processInserts() {
	archivePeriod := p.t.hotPeriod / 10
	log.Debugf("Archiving every %v", archivePeriod)
	archiveTicker := p.t.clock.NewTicker(archivePeriod)
	for {
		select {
		case insert := <-p.inserts:
			p.insert(insert)
		case <-archiveTicker.C:
			p.requestArchiving()
		}
	}
}

func (p *partition) insert(insert *insert) {
	key := string(insert.key)
	now := p.t.clock.Now()
	start := roundTime(insert.ts, p.t.resolution)
	if now.Sub(start) > p.t.hotPeriod {
		log.Trace("Discarding insert outside of hot period")
		return
	}
	b := p.tail[key]
	if b == nil || b.start.Before(start) {
		p.t.statsMutex.Lock()
		if b == nil {
			p.t.stats.HotKeys++
		}
		p.t.statsMutex.Unlock()
		b = &bucket{start, insert.val, b}
		p.tail[key] = b
		return
	}
	for {
		if b.start == start {
			// Update existing bucket
			b.val = b.val.Plus(insert.val)
			return
		}
		if b.prev == nil || b.prev.start.Before(start) {
			// Insert new bucket
			p.t.statsMutex.Lock()
			p.t.statsMutex.Unlock()
			b.prev = &bucket{start, insert.val, b.prev}
			return
		}
		// Continue looking
		b = b.prev
	}
}

func (p *partition) requestArchiving() {
	now := p.t.clock.Now()
	log.Debugf("Requested archiving at %v", now)
	for key, b := range p.tail {
		if now.Sub(b.start) > p.t.hotPeriod {
			log.Tracef("Archiving full. %v / %v %v", b.start, now, b.prev != nil)
			delete(p.tail, key)
			p.t.statsMutex.Lock()
			p.t.stats.HotKeys--
			p.t.statsMutex.Unlock()
			p.t.toArchive <- &archiveRequest{key, b}
			continue
		}
{'fraction_non_alphanumeric': 0.14813084112149533, 'fraction_numerical': 0.0065420560747663555, 'mean_word_length': 3.213254593175853, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4681839', 'n_tokens_mistral': 2695, 'n_tokens_neox': 2491, 'n_words': 595}