fixes lexico ordering

Author: nickzelei

worker/pkg/benthos/sql/input_sql_raw.go

@@ -172,7 +172,15 @@ func (s *pooledInput) Connect(ctx context.Context) error {
 		}
 		s.logger.Debug("using paged query")
 		query = *s.pagedQueryStatic
-		args = append(args, s.continuationToken.Contents.LastReadOrderValues...)
+
+		// Build arguments for lexicographical ordering
+		lastValues := s.continuationToken.Contents.LastReadOrderValues
+		for i := 0; i < len(s.orderByColumns); i++ {
+			// For each OR condition, add values up to and including current position
+			for j := 0; j <= i; j++ {
+				args = append(args, lastValues[j])
+			}
+		}
 		args = append(args, *s.expectedTotalRows)
 	}
 

worker/pkg/query-builder2/querybuilder.go

@@ -165,12 +165,19 @@ func (qb *QueryBuilder) buildPageQuery(schema, tableName string, query *goqu.Sel
 	key := qb.getTableKey(schema, tableName)
 	orderBy := qb.orderBy[key]
 	if len(orderBy) > 0 {
-		// Add where clause using order by columns directly with goqu
-		conditions := make([]exp.Expression, len(orderBy))
-		for i, col := range orderBy {
-			conditions[i] = goqu.T(rootAlias).Col(col).Gt(0)
+		// Build lexicographical ordering conditions
+		var conditions []exp.Expression
+		for i := 0; i < len(orderBy); i++ {
+			var subConditions []exp.Expression
+			// Add equality conditions for all columns before current
+			for j := 0; j < i; j++ {
+				subConditions = append(subConditions, goqu.T(rootAlias).Col(orderBy[j]).Eq(goqu.L("?", 0)))
+			}
+			// Add greater than condition for current column
+			subConditions = append(subConditions, goqu.T(rootAlias).Col(orderBy[i]).Gt(goqu.L("?", 0)))
+			conditions = append(conditions, goqu.And(subConditions...))
 		}
-		query = query.Where(goqu.And(conditions...))
+		query = query.Where(goqu.Or(conditions...))
 	}
 	return query.Prepared(true)
 }

worker/pkg/workflows/datasync/workflow/workflow.go

@@ -536,6 +536,9 @@ func invokeSync(
 		err := workflow.ExecuteChildWorkflow(workflow.WithChildOptions(ctx, workflow.ChildWorkflowOptions{
 			WorkflowID:    getChildWorkflowId(info.WorkflowExecution.ID, config.Name, workflow.Now(ctx)),
 			StaticSummary: fmt.Sprintf("Syncing %s.%s", config.TableSchema, config.TableName),
+			RetryPolicy: &temporal.RetryPolicy{
+				MaximumAttempts: 1,
+			},
 		}), tsWf.TableSync, &tablesync_workflow.TableSyncRequest{
 			AccountId:           accId,
 			Id:                  config.Name,