Fix WHERE priority for Postgres (#59)

* Follow up to ba82ee1

Fix WHERE condition priority for PostgreSQL

* Fix lint errors

Author: m110

pkg/sql/adapters_pgx.go

@@ -66,25 +66,25 @@ func (c PgxBeginner) BeginTx(ctx context.Context, options *stdSQL.TxOptions) (Tx
 }
 
 func (c PgxBeginner) ExecContext(ctx context.Context, query string, args ...any) (Result, error) {
-	res, err := c.Conn.Exec(ctx, query, args...)
+	res, err := c.Exec(ctx, query, args...)
 
 	return PgxResult{res}, err
 }
 
 func (c PgxBeginner) QueryContext(ctx context.Context, query string, args ...any) (Rows, error) {
-	rows, err := c.Conn.Query(ctx, query, args...)
+	rows, err := c.Query(ctx, query, args...)
 
 	return PgxRows{rows}, err
 }
 
 func (t PgxTx) ExecContext(ctx context.Context, query string, args ...any) (Result, error) {
-	res, err := t.Tx.Exec(ctx, query, args...)
+	res, err := t.Exec(ctx, query, args...)
 
 	return PgxResult{res}, err
 }
 
 func (t PgxTx) QueryContext(ctx context.Context, query string, args ...any) (Rows, error) {
-	rows, err := t.Tx.Query(ctx, query, args...)
+	rows, err := t.Query(ctx, query, args...)
 
 	return PgxRows{rows}, err
 }

pkg/sql/queue_schema_adapter_mysql_test.go

@@ -56,14 +56,15 @@ func TestMySQLQueueSchemaAdapter(t *testing.T) {
 	}
 
 	var receivedMessages []*message.Message
+ReceiveLoop:
 	for i := 0; i < 5; i++ {
 		select {
 		case msg := <-messages:
 			receivedMessages = append(receivedMessages, msg)
 			msg.Ack()
 		case <-time.After(5 * time.Second):
 			t.Errorf("expected to receive message")
-			break
+			break ReceiveLoop
 		}
 	}
 

pkg/sql/queue_schema_adapter_postgresql.go

@@ -74,7 +74,7 @@ func queueInsertMarkers(count int) string {
 
 	index := 1
 	for i := 0; i < count; i++ {
-		result.WriteString(fmt.Sprintf("($%d,$%d,$%d),", index, index+1, index+2))
+		fmt.Fprintf(&result, "($%d,$%d,$%d),", index, index+1, index+2)
 		index += 3
 	}
 
@@ -104,7 +104,7 @@ func (s PostgreSQLQueueSchema) SelectQuery(params SelectQueryParams) (Query, err
 	if s.GenerateWhereClause != nil {
 		where, args = s.GenerateWhereClause(whereParams)
 		if where != "" {
-			where = "AND " + where
+			where = "AND (" + where + ")"
 		}
 	}
 

pkg/sql/queue_schema_adapter_postgresql_test.go

@@ -56,14 +56,15 @@ func TestPostgreSQLQueueSchemaAdapter(t *testing.T) {
 	}
 
 	var receivedMessages []*message.Message
+ReceiveLoop:
 	for i := 0; i < 5; i++ {
 		select {
 		case msg := <-messages:
 			receivedMessages = append(receivedMessages, msg)
 			msg.Ack()
 		case <-time.After(5 * time.Second):
 			t.Errorf("expected to receive message")
-			break
+			break ReceiveLoop
 		}
 	}
 

pkg/sql/schema_adapter_postgresql.go

@@ -95,7 +95,7 @@ func defaultInsertMarkers(count int) string {
 
 	index := 1
 	for i := 0; i < count; i++ {
-		result.WriteString(fmt.Sprintf("($%d,$%d,$%d,pg_current_xact_id()),", index, index+1, index+2))
+		fmt.Fprintf(&result, "($%d,$%d,$%d,pg_current_xact_id()),", index, index+1, index+2)
 		index += 3
 	}