Remove ExitAfterRecords, always use WaitFor (#1206)

PeerDB-io · Feb 5, 2024 · e49b4c3 · e49b4c3
1 parent 92b752d
commit e49b4c3
Show file tree

Hide file tree

Showing 7 changed files with 284 additions and 395 deletions.
diff --git a/flow/cmd/handler.go b/flow/cmd/handler.go
@@ -142,8 +142,7 @@ func (h *FlowRequestHandler) CreateCDCFlow(
 	}
 
 	limits := &peerflow.CDCFlowLimits{
-		ExitAfterRecords: -1,
-		MaxBatchSize:     maxBatchSize,
+		MaxBatchSize: maxBatchSize,
 	}
 
 	if req.ConnectionConfigs.SoftDeleteColName == "" {

diff --git a/flow/e2e/bigquery/bigquery_helper.go b/flow/e2e/bigquery/bigquery_helper.go
@@ -415,10 +415,9 @@ func (b *BigQueryTestHelper) CheckNull(tableName string, colName []string) (bool
 }
 
 // check if NaN, Inf double values are null
-func (b *BigQueryTestHelper) CheckDoubleValues(tableName string, colName []string) (bool, error) {
-	csep := strings.Join(colName, ",")
-	command := fmt.Sprintf("SELECT %s FROM `%s.%s`",
-		csep, b.Config.DatasetId, tableName)
+func (b *BigQueryTestHelper) CheckDoubleValues(tableName string, c1 string, c2 string) (bool, error) {
+	command := fmt.Sprintf("SELECT %s, %s FROM `%s.%s`",
+		c1, c2, b.Config.DatasetId, tableName)
 	q := b.client.Query(command)
 	q.DisableQueryCache = true
 	it, err := q.Read(context.Background())
@@ -437,6 +436,10 @@ func (b *BigQueryTestHelper) CheckDoubleValues(tableName string, colName []strin
 		}
 	}
 
+	if len(row) == 0 {
+		return false, nil
+	}
+
 	floatArr, _ := row[1].([]float64)
 	if row[0] != nil || len(floatArr) > 0 {
 		return false, nil