minor pr fixes

pg-sharding · Aug 15, 2023 · ba3e09b · ba3e09b
1 parent 99d7727
commit ba3e09b
Show file tree

Hide file tree

Showing 2 changed files with 30 additions and 19 deletions.
diff --git a/coordinator/provider/coordinator.go b/coordinator/provider/coordinator.go
@@ -558,16 +558,18 @@ func (qc *qdbCoordinator) UnlockKeyRange(ctx context.Context, keyRangeID string)
 
 // Split TODO: check bounds and keyRangeID (sourceID)
 func (qc *qdbCoordinator) Split(ctx context.Context, req *kr.SplitKeyRange) error {
-	var krOld *qdb.KeyRange
-	var err error
-
 	spqrlog.Zero.Debug().
 		Str("krid", req.Krid).
 		Interface("bound", req.Bound).
 		Str("source-id", req.SourceID).
 		Msg("split request is")
 
-	if krOld, err = qc.db.LockKeyRange(ctx, req.SourceID); err != nil {
+	if _, err := qc.db.GetKeyRange(ctx, req.Krid); err == nil {
+		return fmt.Errorf("key range %v already present in qdb", req.Krid)
+	}
+
+	krOld, err := qc.db.LockKeyRange(ctx, req.SourceID)
+	if err != nil {
 		return err
 	}
 
@@ -580,9 +582,6 @@ func (qc *qdbCoordinator) Split(ctx context.Context, req *kr.SplitKeyRange) erro
 	if kr.CmpRangesLess(req.Bound, krOld.LowerBound) || !kr.CmpRangesLess(req.Bound, krOld.UpperBound) {
 		return fmt.Errorf("failed to split because bound is out of key range")
 	}
-	if _, err := qc.db.GetKeyRange(ctx, req.Krid); err == nil {
-		return fmt.Errorf("key range %v already present in qdb", req.Krid)
-	}
 
 	krNew := kr.KeyRangeFromDB(
 		&qdb.KeyRange{
@@ -816,6 +815,8 @@ func (qc *qdbCoordinator) Move(ctx context.Context, req *kr.MoveKeyRange) error
 	keyRange, _ := qc.db.GetKeyRange(ctx, req.Krid)
 	shardingRules, _ := qc.ListShardingRules(ctx)
 
+	// no need to move data to the same
+	// so pretend work is done
 	if keyRange.ShardID == req.ShardId {
 		return nil
 	}

diff --git a/qdb/ops/ops.go b/qdb/ops/ops.go
@@ -36,10 +36,9 @@ func AddShardingRuleWithChecks(ctx context.Context, qdb qdb.QDB, rule *shrule.Sh
 }
 
 func AddKeyRangeWithChecks(ctx context.Context, qdb qdb.QDB, keyRange *kr.KeyRange) error {
-	// TODO: do real validate
-	//if err := validateShard(ctx, qdb, keyRange.ShardID); err != nil {
-	//	return err
-	//}
+	if _, err := qdb.GetShard(ctx, keyRange.ShardID); err != nil {
+		return err
+	}
 
 	if _, err := qdb.GetKeyRange(ctx, keyRange.ID); err == nil {
 		return fmt.Errorf("key range %v already present in qdb", keyRange.ID)
@@ -51,10 +50,8 @@ func AddKeyRangeWithChecks(ctx context.Context, qdb qdb.QDB, keyRange *kr.KeyRan
 	}
 
 	for _, v := range existsKrids {
-		if kr.CmpRangesLessEqual(keyRange.LowerBound, v.LowerBound) && kr.CmpRangesLess(v.LowerBound, keyRange.UpperBound) ||
-			kr.CmpRangesLess(keyRange.LowerBound, v.UpperBound) && kr.CmpRangesLess(v.UpperBound, keyRange.UpperBound) ||
-			kr.CmpRangesLess(v.LowerBound, keyRange.UpperBound) && kr.CmpRangesLess(keyRange.UpperBound, v.UpperBound) {
-			return fmt.Errorf("key range %v intersects with %v present in qdb", keyRange.ID, v.KeyRangeID)
+		if doIntersect(keyRange, v) {
+			return fmt.Errorf("key range %v intersects with key range %v in QDB", keyRange.ID, v.KeyRangeID)
 		}
 	}
 
@@ -125,12 +122,25 @@ func ModifyKeyRangeWithChecks(ctx context.Context, qdb qdb.QDB, keyRange *kr.Key
 			// update req
 			continue
 		}
-		if kr.CmpRangesLessEqual(keyRange.LowerBound, v.LowerBound) && kr.CmpRangesLess(v.LowerBound, keyRange.UpperBound) ||
-			kr.CmpRangesLess(keyRange.LowerBound, v.UpperBound) && kr.CmpRangesLess(v.UpperBound, keyRange.UpperBound) ||
-			kr.CmpRangesLess(v.LowerBound, keyRange.UpperBound) && kr.CmpRangesLess(keyRange.UpperBound, v.UpperBound) {
-			return fmt.Errorf("key range %v intersects with %v present in qdb", keyRange.ID, v.KeyRangeID)
+		if doIntersect(keyRange, v) {
+			return fmt.Errorf("key range %v intersects with key range %v in QDB", keyRange.ID, v.KeyRangeID)
 		}
 	}
 
 	return qdb.UpdateKeyRange(ctx, keyRange.ToDB())
 }
+
+// This method checks if two key ranges intersect
+func doIntersect(l *kr.KeyRange, r *qdb.KeyRange) bool {
+	// l0     r0      l1      r1
+	// |------|-------|--------
+	//
+	// r0     l0      r1      l1
+	// -------|-------|-------|
+	//
+	// r0     l0      r1      l1
+	// |------|-------|--------
+	return kr.CmpRangesLessEqual(l.LowerBound, r.LowerBound) && kr.CmpRangesLess(r.LowerBound, l.UpperBound) ||
+		kr.CmpRangesLess(l.LowerBound, r.UpperBound) && kr.CmpRangesLess(r.UpperBound, l.UpperBound) ||
+		kr.CmpRangesLess(r.LowerBound, l.UpperBound) && kr.CmpRangesLess(l.UpperBound, r.UpperBound)
+}