<commit_before>package influxdb

import (
	"encoding/binary"
	"sort"
	"strings"
	"time"
	"unsafe"

	"github.com/boltdb/bolt"
)

// metastore represents the low-level data store for metadata.
type metastore struct {
	db *bolt.DB
}

// open initializes the metastore.
func (m *metastore) open(path string) error {
	// Open the bolt-backed database.
	db, err := bolt.Open(path, 0600, &bolt.Options{Timeout: 1 * time.Second})
	if err != nil {
		return err
	}
	m.db = db

	// Initialize the metastore.
	if err := m.init(); err != nil {
		return err
	}

	return nil
}

// close closes the store.
func (m *metastore) close() error {
	if m.db != nil {
		return m.db.Close()
	}
	return nil
}

// init initializes the metastore to ensure all top-level buckets are created.
func (m *metastore) init() error {
	return m.db.Update(func(tx *bolt.Tx) error {
		_, _ = tx.CreateBucketIfNotExists([]byte("Server"))
		_, _ = tx.CreateBucketIfNotExists([]byte("DataNodes"))
		_, _ = tx.CreateBucketIfNotExists([]byte("Databases"))
		_, _ = tx.CreateBucketIfNotExists([]byte("Users"))
		return nil
	})
}

// view executes a function in the context of a read-only transaction.
func (m *metastore) view(fn func(*metatx) error) error {
	return m.db.View(func(tx *bolt.Tx) error { return fn(&metatx{tx}) })
}

// update executes a function in the context of a read-write transaction.
func (m *metastore) update(fn func(*metatx) error) error {
	return m.db.Update(func(tx *bolt.Tx) error { return fn(&metatx{tx}) })
}

// mustView executes a function in the context of a read-only transaction.
// Panics if system error occurs. Return error from the fn for validation errors.
func (m *metastore) mustView(fn func(*metatx) error) (err error) {
	if e := m.view(func(tx *metatx) error {
		err = fn(tx)
		return nil
	}); e != nil {
		panic("metastore view: " + err.Error())
	}
	return
}

// mustUpdate executes a function in the context of a read-write transaction.
// Panics if a disk or system error occurs. Return error from the fn for validation errors.
func (m *metastore) mustUpdate(fn func(*metatx) error) (err error) {
	if e := m.update(func(tx *metatx) error {
		err = fn(tx)
		return nil
	}); e != nil {
		panic("metastore update: " + err.Error())
	}
	return
}

// metatx represents a metastore transaction.
type metatx struct {
	*bolt.Tx
}

// id returns the server id.
func (tx *metatx) id() (id uint64) {
	if v := tx.Bucket([]byte("Server")).Get([]byte("id")); v != nil {
		id = btou64(v)
	}
	return
}

// setID sets the server id.
func (tx *metatx) setID(v uint64) error {
	return tx.Bucket([]byte("Server")).Put([]byte("id"), u64tob(v))
}

// databases returns a list of all data nodes from the metastore.
func (tx *metatx) dataNodes() (a []*DataNode) {
	c := tx.Bucket([]byte("DataNodes")).Cursor()
	for k, v := c.First(); k != nil; k, v = c.Next() {
		n := newDataNode()
		mustUnmarshalJSON(v, &n)
		a = append(a, n)
	}
	sort.Sort(dataNodes(a))
	return
}

// saveDataNode persists a data node to the metastore.
func (tx *metatx) saveDataNode(n *DataNode) error {
	return tx.Bucket([]byte("DataNodes")).Put(u64tob(n.ID), mustMarshalJSON(n))
}

// deleteDataNode removes data node from the metastore.
func (tx *metatx) deleteDataNode(id uint64) error {
	return tx.Bucket([]byte("DataNodes")).Delete(u64tob(id))
}

// databases returns a list of all databases from the metastore.
func (tx *metatx) databases() (a []*database) {
	c := tx.Bucket([]byte("Databases")).Cursor()
	for k, _ := c.First(); k != nil; k, _ = c.Next() {
		b := c.Bucket().Bucket(k)
		v := b.Get([]byte("meta"))
		db := newDatabase()
		mustUnmarshalJSON(v, &db)<commit_msg>Remove data node sort in the metastore.<commit_after>package influxdb

import (
	"encoding/binary"
	"sort"
	"strings"
	"time"
	"unsafe"

	"github.com/boltdb/bolt"
)

// metastore represents the low-level data store for metadata.
type metastore struct {
	db *bolt.DB
}

// open initializes the metastore.
func (m *metastore) open(path string) error {
	// Open the bolt-backed database.
	db, err := bolt.Open(path, 0600, &bolt.Options{Timeout: 1 * time.Second})
	if err != nil {
		return err
	}
	m.db = db

	// Initialize the metastore.
	if err := m.init(); err != nil {
		return err
	}

	return nil
}

// close closes the store.
func (m *metastore) close() error {
	if m.db != nil {
		return m.db.Close()
	}
	return nil
}

// init initializes the metastore to ensure all top-level buckets are created.
func (m *metastore) init() error {
	return m.db.Update(func(tx *bolt.Tx) error {
		_, _ = tx.CreateBucketIfNotExists([]byte("Server"))
		_, _ = tx.CreateBucketIfNotExists([]byte("DataNodes"))
		_, _ = tx.CreateBucketIfNotExists([]byte("Databases"))
		_, _ = tx.CreateBucketIfNotExists([]byte("Users"))
		return nil
	})
}

// view executes a function in the context of a read-only transaction.
func (m *metastore) view(fn func(*metatx) error) error {
	return m.db.View(func(tx *bolt.Tx) error { return fn(&metatx{tx}) })
}

// update executes a function in the context of a read-write transaction.
func (m *metastore) update(fn func(*metatx) error) error {
	return m.db.Update(func(tx *bolt.Tx) error { return fn(&metatx{tx}) })
}

// mustView executes a function in the context of a read-only transaction.
// Panics if system error occurs. Return error from the fn for validation errors.
func (m *metastore) mustView(fn func(*metatx) error) (err error) {
	if e := m.view(func(tx *metatx) error {
		err = fn(tx)
		return nil
	}); e != nil {
		panic("metastore view: " + err.Error())
	}
	return
}

// mustUpdate executes a function in the context of a read-write transaction.
// Panics if a disk or system error occurs. Return error from the fn for validation errors.
func (m *metastore) mustUpdate(fn func(*metatx) error) (err error) {
	if e := m.update(func(tx *metatx) error {
		err = fn(tx)
		return nil
	}); e != nil {
		panic("metastore update: " + err.Error())
	}
	return
}

// metatx represents a metastore transaction.
type metatx struct {
	*bolt.Tx
}

// id returns the server id.
func (tx *metatx) id() (id uint64) {
	if v := tx.Bucket([]byte("Server")).Get([]byte("id")); v != nil {
		id = btou64(v)
	}
	return
}

// setID sets the server id.
func (tx *metatx) setID(v uint64) error {
	return tx.Bucket([]byte("Server")).Put([]byte("id"), u64tob(v))
}

// dataNodes returns a list of all data nodes from the metastore.
func (tx *metatx) dataNodes() (a []*DataNode) {
	c := tx.Bucket([]byte("DataNodes")).Cursor()
	for k, v := c.First(); k != nil; k, v = c.Next() {
		n := newDataNode()
		mustUnmarshalJSON(v, &n)
		a = append(a, n)
	}
	return
}

// saveDataNode persists a data node to the metastore.
func (tx *metatx) saveDataNode(n *DataNode) error {
	return tx.Bucket([]byte("DataNodes")).Put(u64tob(n.ID), mustMarshalJSON(n))
}

// deleteDataNode removes data node from the metastore.
func (tx *metatx) deleteDataNode(id uint64) error {
	return tx.Bucket([]byte("DataNodes")).Delete(u64tob(id))
}

// databases returns a list of all databases from the metastore.
func (tx *metatx) databases() (a []*database) {
	c := tx.Bucket([]byte("Databases")).Cursor()
	for k, _ := c.First(); k != nil; k, _ = c.Next() {
		b := c.Bucket().Bucket(k)
		v := b.Get([]byte("meta"))
		db := newDatabase()
		mustUnmarshalJSON(v, &db)
{'fraction_non_alphanumeric': 0.15869954539192727, 'fraction_numerical': 0.005234880837580934, 'mean_word_length': 3.888888888888889, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '1296142', 'n_tokens_mistral': 2710, 'n_tokens_neox': 2568, 'n_words': 790}