mirror of
https://github.com/hanzoai/graph.git
synced 2026-10-30 18:35:37 +00:00
zeekay
f5aaeb712e
Derived from Dgraph (github.com/dgraph-io/dgraph) at commit 64804bd2, Apache-2.0. See NOTICE. WHAT THIS IS. The graph engine as a Go library: posting lists, the DQL parser, the type and tokenizer system, the index and schema layers, and the wire types. 39,170 lines of the upstream's 166,337. THE STORE is github.com/luxfi/zapdb, a Badger fork that kept the package name, the subpackages (options, pb, y) and the API — Open, OpenManaged, DB, Options — so the 74 import sites are aliased rather than rewritten. THE CODEC is ZAP. zapdb serializes with ZAP, not protobuf: its pb package builds the ZAP types by default and the protobuf ones only under `-tags grpc`, a path that does not compile (table/builder.go calls pb.Marshal, which that tag removes). One call site in posting needed converting; every other type here is Hanzo's own protobuf and is untouched. NO TRANSPORT. An embedded engine dials nothing, so nothing here imports google.golang.org/grpc. Removed: 20 transport functions in x, the generated service stubs for Raft, Zero and Worker (1,657 lines), and the JWT helper — identity belongs to IAM. The core called none of them. NOT KEPT: worker, conn, raftwal, the cluster commands, the GraphQL layer and the test harnesses. `query`, the DQL execution engine, went with them because it depends on worker; parsing DQL is available here, executing it is not. The history of the upstream project lives at github.com/dgraph-io/dgraph. This repository begins at the derivation because three quarters of that history describes code this fork does not carry.
119 lines
2.2 KiB
Go
119 lines
2.2 KiB
Go
/*
|
|
* SPDX-FileCopyrightText: © 2017-2026 Istari Digital, Inc.
|
|
* SPDX-License-Identifier: Apache-2.0
|
|
*/
|
|
|
|
package posting
|
|
|
|
import (
|
|
"math"
|
|
"sync"
|
|
|
|
"github.com/golang/glog"
|
|
|
|
badger "github.com/luxfi/zapdb"
|
|
"github.com/luxfi/zapdb/pb"
|
|
)
|
|
|
|
// TxnWriter is in charge or writing transactions to badger.
|
|
type TxnWriter struct {
|
|
db *badger.DB
|
|
wg sync.WaitGroup
|
|
che chan error
|
|
}
|
|
|
|
// NewTxnWriter returns a new TxnWriter instance.
|
|
func NewTxnWriter(db *badger.DB) *TxnWriter {
|
|
return &TxnWriter{
|
|
db: db,
|
|
che: make(chan error, 1),
|
|
}
|
|
}
|
|
|
|
func (w *TxnWriter) cb(err error) {
|
|
defer w.wg.Done()
|
|
if err == nil {
|
|
return
|
|
}
|
|
|
|
glog.Errorf("TxnWriter got error during callback: %v", err)
|
|
select {
|
|
case w.che <- err:
|
|
default:
|
|
}
|
|
}
|
|
|
|
// Write stores the given key-value pairs in badger.
|
|
func (w *TxnWriter) Write(kvs *pb.KVList) error {
|
|
for _, kv := range kvs.Kv {
|
|
var meta byte
|
|
if len(kv.UserMeta) > 0 {
|
|
meta = kv.UserMeta[0]
|
|
}
|
|
if err := w.SetAt(kv.Key, kv.Value, meta, kv.Version); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (w *TxnWriter) update(commitTs uint64, f func(txn *badger.Txn) error) error {
|
|
if commitTs == 0 {
|
|
return nil
|
|
}
|
|
txn := w.db.NewTransactionAt(math.MaxUint64, true)
|
|
defer txn.Discard()
|
|
|
|
err := f(txn)
|
|
if err == badger.ErrTxnTooBig {
|
|
// continue to commit.
|
|
} else if err != nil {
|
|
return err
|
|
}
|
|
w.wg.Add(1)
|
|
return txn.CommitAt(commitTs, w.cb)
|
|
}
|
|
|
|
// SetAt writes a key-value pair at the given timestamp.
|
|
func (w *TxnWriter) SetAt(key, val []byte, meta byte, ts uint64) error {
|
|
return w.update(ts, func(txn *badger.Txn) error {
|
|
switch meta {
|
|
case BitCompletePosting, BitEmptyPosting:
|
|
err := txn.SetEntry((&badger.Entry{
|
|
Key: key,
|
|
Value: val,
|
|
UserMeta: meta,
|
|
}).WithDiscard())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
default:
|
|
err := txn.SetEntry(&badger.Entry{
|
|
Key: key,
|
|
Value: val,
|
|
UserMeta: meta,
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
|
|
// Flush waits until all operations are done and all data is written to disk.
|
|
func (w *TxnWriter) Flush() error {
|
|
// No need to call Sync here.
|
|
return w.Wait()
|
|
}
|
|
|
|
func (w *TxnWriter) Wait() error {
|
|
w.wg.Wait()
|
|
select {
|
|
case err := <-w.che:
|
|
return err
|
|
default:
|
|
return nil
|
|
}
|
|
}
|