ensure endTs greater than startTs (#21194)

ensure endTs greater than startTs Approved by: @daviszhen
matrixorigin · Jan 13, 2025 · 0e291d2 · 0e291d2
1 parent 3b2d1e5
commit 0e291d2
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 6 deletions.
diff --git a/pkg/frontend/cdc.go b/pkg/frontend/cdc.go
@@ -317,21 +317,25 @@ func doCreateCdc(ctx context.Context, ses *Session, create *tree.CreateCDC) (err
 		return moerr.NewInternalErrorf(ctx, "invalid exclude expression: %s, err: %v", exclude, err)
 	}
 
-	var ts time.Time
+	var ts0, ts1 time.Time
 	startTs := cdcTaskOptionsMap[cdc2.StartTs]
 	if startTs != "" {
-		if ts, err = parseTimestamp(startTs, ses.timeZone); err != nil {
+		if ts0, err = parseTimestamp(startTs, ses.timeZone); err != nil {
 			return moerr.NewInternalErrorf(ctx, "invalid startTs: %s, supported timestamp format: `%s`, or `%s`", startTs, time.DateTime, time.RFC3339)
 		}
-		startTs = ts.Format(time.RFC3339)
+		startTs = ts0.Format(time.RFC3339)
 	}
 
 	endTs := cdcTaskOptionsMap[cdc2.EndTs]
 	if endTs != "" {
-		if ts, err = parseTimestamp(endTs, ses.timeZone); err != nil {
+		if ts1, err = parseTimestamp(endTs, ses.timeZone); err != nil {
 			return moerr.NewInternalErrorf(ctx, "invalid endTs: %s, supported timestamp format: `%s`, or `%s`", endTs, time.DateTime, time.RFC3339)
 		}
-		endTs = ts.Format(time.RFC3339)
+		endTs = ts1.Format(time.RFC3339)
+	}
+
+	if startTs != "" && endTs != "" && !ts1.After(ts0) {
+		return moerr.NewInternalErrorf(ctx, "startTs: %s should be less than endTs: %s", startTs, endTs)
 	}
 
 	//step 4: check source uri format and strip password

diff --git a/pkg/frontend/cdc_test.go b/pkg/frontend/cdc_test.go
@@ -493,9 +493,19 @@ func Test_doCreateCdc_invalidStartTs(t *testing.T) {
 			"123456",
 		},
 	}
-
 	err := doCreateCdc(context.Background(), ses, create)
 	assert.Error(t, err)
+
+	create.Option = []string{
+		"Level",
+		cdc2.AccountLevel,
+		cdc2.StartTs,
+		"2025-01-03 15:20:00",
+		cdc2.EndTs,
+		"2025-01-03 14:20:00",
+	}
+	err = doCreateCdc(context.Background(), ses, create)
+	assert.Error(t, err)
 }
 
 type testTaskData struct {