<commit_before>package gostore

import (
	"fmt"
	"github.com/asaskevich/govalidator"
	r "github.com/dancannon/gorethink"
	"github.com/dustin/gojson"
	"github.com/jinzhu/now"
	"github.com/mgutz/logxi/v1"
	"strings"
	"time"
)

var logger = log.New("gostore.rethink")

func NewRethinkObjectStore(session *r.Session, database string) RethinkStore {
	s := RethinkStore{session, database}
	s.CreateDatabase()
	return s
}

type RethinkStore struct {
	Session  *r.Session
	Database string
}

type RethinkRows struct {
	cursor *r.Cursor
}

func (s RethinkRows) LastError() error {
	return nil
}
func NewRethinkRows(cursor *r.Cursor) RethinkRows {
	return RethinkRows{cursor}
}

func (s RethinkRows) Next(dst interface{}) (bool, error) {
	if !s.cursor.Next(dst) {
		//		logger.Debug("Error getting next", "err", s.cursor.Err(), "isNil", s.cursor.IsNil())
		return false, s.cursor.Err()
	}
	return true, nil
}

func (s RethinkRows) Close() {
	s.cursor.Close()
}

func (s RethinkStore) CreateDatabase() (err error) {
	return r.DBCreate(s.Database).Exec(s.Session)
}

func (s RethinkStore) GetStore() interface{} {
	return s.Session
}

func hasIndex(name string, indexes []interface{}) bool {
	for _, v := range indexes {
		if v.(string) == name {
			return true
		}
	}
	return false
}

//TODO: fix index creation, indexes are not created properly
func (rs RethinkStore) CreateTable(store string, schema interface{}) (err error) {
	logger.Info("creating table " + store)
	var res []interface{}
	_ = r.DB(rs.Database).TableCreate(store).Exec(rs.Session)
	// if err != nil{
	// 	// logger.Warn("unable to create table ", "table", store)
	//
	// }
	result, err := r.DB(rs.Database).Table(store).IndexList().Run(rs.Session)
	if err != nil {
		panic(err)
	}
	if result.Err() != nil {
		panic(err)
	}
	result.All(&res)
	logger.Debug(store+" indexList", res)
	//also create indexes
	if schema != nil {
		s := schema.(map[string]interface{})
		if indexes, ok := s["index"].(map[string]interface{}); ok {
			for name, _vals := range indexes {
				if !hasIndex(name, res) {
					logger.Info("creating index", "name", name, "val", _vals)
					if vals, ok := _vals.([]interface{}); ok && len(vals) > 0 {
						logger.Info("creating compound index", "name", name, "vals", vals)
						if _, err = r.DB(rs.Database).Table(store).IndexCreateFunc(name, func(row r.Term) interface{} {
							index_fields := []interface{}{}
							for _, v := range vals {
								index_fields = append(index_fields, row.Field(v.(string)))
							}
							return index_fields
						}).RunWrite(rs.Session); err != nil {
							return
						}
					} else {
						logger.Info("creating index", "name", name)
						if err = r.DB(rs.Database).Table(store).IndexCreate(name).Exec(rs.Session); err != nil {
							// logger.Warn("cannot create index [" + name + "] in " + store)
							// logger.Warn("cannot create index")
							println(err.Error())

						} else {
							if err = r.DB(rs.Database).Table(store).IndexWait(name).Exec(rs.Session); err == nil {
								logger.Info("created index [" + name + "] in " + store)
							}
						}
					}
				}
			}
		}
	}

	return
}

type TermOperators map[string]func(args ...interface{}) interface{}

func parseFilter(args string) (dtarg interface{}) {
	//this also handles args
	vals := strings.Split(args, "|")
	if len(vals) > 1 {
		//check type
		if vals[1] == "dt" {
			if it, err := govalidator.ToInt(vals[0]); err == nil {
				logger.Info("< op date", "time", it)
				return r.EpochTime(it)
			}
			//				t, err := now.Parse(vals[0])
			if t, err := time.Parse(
				time.RFC3339,
				vals[0]); err == nil {
				return r.EpochTime(t.Unix())
			}
			if t, err := now.Parse(vals[0]); err == nil {
				return r.EpochTime(t.Unix())
			}
		}
	}
	return vals[0]
}

var filterOps TermOperators = TermOperators{
	"~": func(args ...interface{}) interface{} {
		vals := strings.Split(args[1].(string), "|")
		baseTerm := args[0].(r.Term)
		if len(vals) > 0 {

		}
		return baseTerm.Match(vals[0])
	},
	">": func(args ...interface{}) interface{} {
		baseTerm := args[0].(r.Term)
		return baseTerm.Gt(parseFilter(args[1].(string)))
	},
	"<": func(args ...interface{}) interface{} {
		baseTerm := args[0].(r.Term)
		return baseTerm.Lt(parseFilter(args[1].(string)))
	},
}

// TODO: Transforms filter args into rethinkdb filter args
func (s RethinkStore) ParseFilterArgs(filter map[string]interface{}, indexes []string, opts ObjectStoreOptions) r.Term {
	var (
		t r.Term = r.And(1)
	)
	logger.Debug("filterArgs", "filter", filter, "indexes", indexes, "opts", opts)
	//TODO: Optimize this by passing indexes as well as field types
	for k, v := range filter {
		val := v.(string)
		val_rune := []rune(val)
		if len(val_rune) > 0 {
			first := string(val_rune[0])
			if op, ok := filterOps[first]; ok {
				tv := string([]rune(val)[1:])
				t = t.And(op(r.Row.Field(k), tv))
			} else {
				t = t.And(r.Row.Field(k).Eq(v))
			}
		}
	}
	// logger.Debug(t.String())
	return t
}

func (s RethinkStore) All(count int, skip int, store string) (rrows ObjectRows, err error) {
	result, err := r.DB(s.Database).Table(store).OrderBy(r.OrderByOpts{Index: r.Desc("id")}).Run(s.Session)
	if err != nil {
		return
	}
	rrows = RethinkRows{result}
	return
}

func (s RethinkStore) AllCursor(store string) (ObjectRows, error) {
	result, err := r.DB(s.Database).Table(store).Run(s.Session)
	if err != nil {<commit_msg>handle or operator for fields<commit_after>package gostore

import (
	"fmt"
	"github.com/asaskevich/govalidator"
	r "github.com/dancannon/gorethink"
	"github.com/dustin/gojson"
	"github.com/jinzhu/now"
	"github.com/mgutz/logxi/v1"
	"strings"
	"time"
)

var logger = log.New("gostore.rethink")

func NewRethinkObjectStore(session *r.Session, database string) RethinkStore {
	s := RethinkStore{session, database}
	s.CreateDatabase()
	return s
}

type RethinkStore struct {
	Session  *r.Session
	Database string
}

type RethinkRows struct {
	cursor *r.Cursor
}

func (s RethinkRows) LastError() error {
	return nil
}
func NewRethinkRows(cursor *r.Cursor) RethinkRows {
	return RethinkRows{cursor}
}

func (s RethinkRows) Next(dst interface{}) (bool, error) {
	if !s.cursor.Next(dst) {
		//		logger.Debug("Error getting next", "err", s.cursor.Err(), "isNil", s.cursor.IsNil())
		return false, s.cursor.Err()
	}
	return true, nil
}

func (s RethinkRows) Close() {
	s.cursor.Close()
}

func (s RethinkStore) CreateDatabase() (err error) {
	return r.DBCreate(s.Database).Exec(s.Session)
}

func (s RethinkStore) GetStore() interface{} {
	return s.Session
}

func hasIndex(name string, indexes []interface{}) bool {
	for _, v := range indexes {
		if v.(string) == name {
			return true
		}
	}
	return false
}

//TODO: fix index creation, indexes are not created properly
func (rs RethinkStore) CreateTable(store string, schema interface{}) (err error) {
	logger.Info("creating table " + store)
	var res []interface{}
	_ = r.DB(rs.Database).TableCreate(store).Exec(rs.Session)
	// if err != nil{
	// 	// logger.Warn("unable to create table ", "table", store)
	//
	// }
	result, err := r.DB(rs.Database).Table(store).IndexList().Run(rs.Session)
	if err != nil {
		panic(err)
	}
	if result.Err() != nil {
		panic(err)
	}
	result.All(&res)
	logger.Debug(store+" indexList", res)
	//also create indexes
	if schema != nil {
		s := schema.(map[string]interface{})
		if indexes, ok := s["index"].(map[string]interface{}); ok {
			for name, _vals := range indexes {
				if !hasIndex(name, res) {
					logger.Info("creating index", "name", name, "val", _vals)
					if vals, ok := _vals.([]interface{}); ok && len(vals) > 0 {
						logger.Info("creating compound index", "name", name, "vals", vals)
						if _, err = r.DB(rs.Database).Table(store).IndexCreateFunc(name, func(row r.Term) interface{} {
							index_fields := []interface{}{}
							for _, v := range vals {
								index_fields = append(index_fields, row.Field(v.(string)))
							}
							return index_fields
						}).RunWrite(rs.Session); err != nil {
							return
						}
					} else {
						logger.Info("creating index", "name", name)
						if err = r.DB(rs.Database).Table(store).IndexCreate(name).Exec(rs.Session); err != nil {
							// logger.Warn("cannot create index [" + name + "] in " + store)
							// logger.Warn("cannot create index")
							println(err.Error())

						} else {
							if err = r.DB(rs.Database).Table(store).IndexWait(name).Exec(rs.Session); err == nil {
								logger.Info("created index [" + name + "] in " + store)
							}
						}
					}
				}
			}
		}
	}

	return
}

type TermOperators map[string]func(args ...interface{}) r.Term

func parseFilter(args string) (dtarg interface{}) {
	//this also handles args
	vals := strings.Split(args, "|")
	if len(vals) > 1 {
		//check type
		if vals[1] == "dt" {
			if it, err := govalidator.ToInt(vals[0]); err == nil {
				logger.Info("< op date", "time", it)
				return r.EpochTime(it)
			}
			//				t, err := now.Parse(vals[0])
			if t, err := time.Parse(
				time.RFC3339,
				vals[0]); err == nil {
				return r.EpochTime(t.Unix())
			}
			if t, err := now.Parse(vals[0]); err == nil {
				return r.EpochTime(t.Unix())
			}
		}
	}
	return vals[0]
}

var filterOps TermOperators = TermOperators{
	"~": func(args ...interface{}) r.Term {
		vals := strings.Split(args[1].(string), "|")
		baseTerm := args[0].(r.Term)
		if len(vals) > 0 {

		}
		return baseTerm.Match(vals[0])
	},
	">": func(args ...interface{}) r.Term {
		baseTerm := args[0].(r.Term)
		return baseTerm.Gt(parseFilter(args[1].(string)))
	},
	"<": func(args ...interface{}) r.Term {
		baseTerm := args[0].(r.Term)
		return baseTerm.Lt(parseFilter(args[1].(string)))
	},
}

func orTerm(baseTerm r.Term, to r.Term) r.Term {
	return baseTerm.Or(to)
}

func andTerm(baseTerm r.Term, to r.Term) r.Term {
	return baseTerm.And(to)
}

// TODO: Transforms filter args into rethinkdb filter args
func (s RethinkStore) ParseFilterArgs(filter map[string]interface{}, indexes []string, opts ObjectStoreOptions) r.Term {
	var (
		t r.Term = r.And(1)
	)
	logger.Debug("filterArgs", "filter", filter, "indexes", indexes, "opts", opts)
	//TODO: Optimize this by passing indexes as well as field types
	orGroup := map[string]r.Term{}
	for k, v := range filter {
		val := v.(string)
		key_rune := []rune(k)
		val_rune := []rune(val)

		if len(val_rune) > 0 {
			var start int = 1
			// joinMethod := andTerm
			first := string(val_rune[0])
			//Handle `or` grouping by key
			if string(key_rune[0]) == "|" {
				//split |groupName|field_name
				orGroupName := ""
				k_index := 0
				rune_val := ""
				for _rune_index, v := range key_rune[1:] {
					rune_val = string(v)
					if rune_val == "|" {
						k_index = _rune_index + 2
						break
					}
					orGroupName += rune_val
				}
				k = string(key_rune[k_index:])
				if op, ok := filterOps[first]; ok {
					tv := string([]rune(val)[start:])
					if orGroupVal, ok := orGroup[orGroupName]; !ok {
						orGroup[orGroupName] = op(r.Row.Field(k), tv)
					} else {
						orGroup[orGroupName] = orGroupVal.Or(op(r.Row.Field(k), tv))
					}
				} else {
					if orGroupVal, ok := orGroup[orGroupName]; !ok {
						orGroup[orGroupName] = r.Row.Field(k).Eq(v)
					} else {
						orGroup[orGroupName] = orGroupVal.Or(r.Row.Field(k).Eq(v))
					}
				}
			} else {
				if op, ok := filterOps[first]; ok {
					tv := string([]rune(val)[start:])
					t = t.And(op(r.Row.Field(k), tv))
				} else {
					t = t.And(r.Row.Field(k).Eq(v))
				}
			}

		}
	}
	for _, v := range orGroup {
		t = t.And(v)
	}
	// logger.Debug(t.String())
	return t
}

func (s RethinkStore) All(count int, skip int, store string) (rrows ObjectRows, err error) {
	result, err := r.DB(s.Database).Table(store).OrderBy(r.OrderByOpts{Index: r.Desc("id")}).Run(s.Session)
	if err != nil {
		return
	}
	rrows = RethinkRows{result}
	return
}

func (s RethinkStore) AllCursor(store string) (ObjectRows, error) {
	result, err := r.DB(s.Database).Table(store).Run(s.Session)
	if err != nil {
{'fraction_non_alphanumeric': 0.17515257921578464, 'fraction_numerical': 0.004514672686230248, 'mean_word_length': 3.307526107310047, 'pattern_counts': {'":': 6, '<': 7, '<?xml version=': 0, '>': 13, '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': '3563989', 'n_tokens_mistral': 5017, 'n_tokens_neox': 4417, 'n_words': 999}