added filter for vector lengths

Author: harshil-goel

dgraph/cmd/zero/oracle.go

@@ -229,7 +229,7 @@ func (o *Oracle) sendDeltasToSubscribers() {
 		newDelta := &pb.OracleDelta{}
 		useNewDelta := false
 		if o.doneUntil.DoneUntil() < waitFor() {
-			if len(delta.Txns) > 5 {
+			if len(delta.Txns) > 10 {
 				replacementTxn := []*pb.TxnStatus{}
 
 				ts := o.doneUntil.DoneUntil()

worker/draft.go

@@ -980,13 +980,15 @@ func (n *node) processApplyCh() {
 
 // TODO(Anurag - 4 May 2020): Are we using pkey? Remove if unused.
 func (n *node) commitOrAbort(pkey uint64, delta *pb.OracleDelta) error {
-	_, span := otel.Tracer("alpha.CommitLoop").Start(context.Background(), "commitOrAbort")
+	pctx, span := otel.Tracer("alpha.CommitLoop").Start(context.Background(), "alpha.commitOrAbort")
 	defer span.End()
 
 	x.PrintOracleDelta(delta)
 	// First let's commit all mutations to disk.
 	writer := posting.NewTxnWriter(pstore)
 	toDisk := func(start, commit uint64) {
+		_, tspan := otel.Tracer("alpha.CommitLoop").Start(pctx, "alpha.toDisk")
+		defer tspan.End()
 		txn := posting.Oracle().GetTxn(start)
 		if txn == nil || commit == 0 {
 			return
@@ -1013,17 +1015,11 @@ func (n *node) commitOrAbort(pkey uint64, delta *pb.OracleDelta) error {
 			panic(err)
 		}
 
-		span.AddEvent("Committed txn with start_ts: %d, commit_ts: %d", trace.WithAttributes(
+		tspan.AddEvent("Committed txn with start_ts: %d, commit_ts: %d", trace.WithAttributes(
 			attribute.Int64("start_ts", int64(start)),
 			attribute.Int64("commit_ts", int64(commit)),
 		))
-
-		if txn.Span != nil {
-			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)),
-			))
-		}
+		tspan.SetAttributes(attribute.Int64("start_ts", int64(start)), attribute.Int64("commit_ts", int64(commit)))
 	}
 
 	for _, status := range delta.Txns {
@@ -1036,7 +1032,7 @@ func (n *node) commitOrAbort(pkey uint64, delta *pb.OracleDelta) error {
 	span.AddEvent("Flushed to disk")
 	for _, status := range delta.Txns {
 		txn := posting.Oracle().GetTxn(status.StartTs)
-		if txn.Span != nil {
+		if txn != nil && txn.Span != nil {
 			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)),