added filter for vector lengths

Author: darkcoderrises

posting/oracle.go

@@ -16,6 +16,7 @@ import (
 	"github.com/dgraph-io/badger/v4"
 	"github.com/golang/glog"
 	ostats "go.opencensus.io/stats"
+	"go.opentelemetry.io/otel/trace"
 
 	"github.com/hypermodeinc/dgraph/v25/protos/pb"
 	"github.com/hypermodeinc/dgraph/v25/tok/index"
@@ -54,6 +55,8 @@ type Txn struct {
 	lastUpdate time.Time
 
 	cache *LocalCache // This pointer does not get modified.
+
+	Span trace.Span
 }
 
 // struct to implement Txn interface from vector-indexer

worker/draft.go

@@ -68,8 +68,6 @@ type node struct {
 	opsLock     sync.Mutex
 	cdcTracker  *CDC
 	canCampaign bool
-
-	startTsKey map[uint64]uint64
 }
 
 type op int
@@ -279,7 +277,6 @@ func newNode(store *raftwal.DiskStorage, gid uint32, id uint64, myAddr string) *
 		closer:     z.NewCloser(4), // Matches CLOSER:1
 		ops:        make(map[op]operation),
 		cdcTracker: newCDC(),
-		startTsKey: make(map[uint64]uint64),
 	}
 	return n
 }
@@ -939,8 +936,8 @@ func (n *node) processApplyCh() {
 				case proposal.Mutations != nil:
 					tags = append(tags, tag.Upsert(x.KeyMethod, "apply.Mutations"))
 					span.SetAttributes(attribute.Int64("start_ts", int64(proposal.Mutations.StartTs)))
-					n.startTsKey[proposal.Mutations.StartTs] = key
-					fmt.Println("Setting proposal key", proposal.Mutations.StartTs, key)
+					posting.Oracle().GetTxn(proposal.Mutations.StartTs).Span = span
+
 				case proposal.Delta != nil:
 					tags = append(tags, tag.Upsert(x.KeyMethod, "apply.Delta"))
 				}
@@ -1021,9 +1018,7 @@ func (n *node) commitOrAbort(pkey uint64, delta *pb.OracleDelta) error {
 			attribute.Int64("commit_ts", int64(commit)),
 		))
 
-		spani := trace.SpanFromContext(n.Ctx(n.startTsKey[start]))
-		fmt.Println("FIRST", spani, n.startTsKey[start], start)
-		spani.AddEvent("Committed txn with start_ts: %d, commit_ts: %d", trace.WithAttributes(
+		txn.Span.AddEvent("Committed txn with start_ts: %d, commit_ts: %d", trace.WithAttributes(
 			attribute.Int64("start_ts", int64(start)),
 			attribute.Int64("commit_ts", int64(commit)),
 		))
@@ -1038,13 +1033,11 @@ func (n *node) commitOrAbort(pkey uint64, delta *pb.OracleDelta) error {
 
 	span.AddEvent("Flushed to disk")
 	for _, status := range delta.Txns {
-		spani := trace.SpanFromContext(n.Ctx(n.startTsKey[status.StartTs]))
-		fmt.Println("Second", spani)
-		spani.AddEvent("Flushed txn with start_ts: %d, commit_ts: %d", trace.WithAttributes(
+		txn := posting.Oracle().GetTxn(status.StartTs)
+		txn.Span.AddEvent("Flushed txn with start_ts: %d, commit_ts: %d", trace.WithAttributes(
 			attribute.Int64("start_ts", int64(status.StartTs)),
 			attribute.Int64("commit_ts", int64(status.CommitTs)),
 		))
-		delete(n.startTsKey, status.StartTs)
 	}
 
 	if x.WorkerConfig.HardSync {

worker/proposal.go

@@ -221,6 +221,8 @@ func (n *node) proposeAndWait(ctx context.Context, proposal *pb.Proposal) (perr
 		cctx, cancel := context.WithCancel(context.Background())
 		defer cancel()
 
+		cctx = trace.ContextWithSpan(cctx, span)
+
 		errCh := make(chan error, 1)
 		pctx := &conn.ProposalCtx{
 			ErrCh: errCh,