Use pq.Array to avoid hitting parameter limits

atc/db/pipeline.go

@@ -4,8 +4,7 @@ import (
 	"database/sql"
 	"encoding/json"
 	"fmt"
-	"strconv"
-	"strings"
+	"github.com/lib/pq"
 	"time"
 
 	"code.cloudfoundry.org/lager"
@@ -993,36 +992,28 @@ func (p *pipeline) DeleteBuildEventsByBuildIDs(buildIDs []int) error {
 		return nil
 	}
 
-	interfaceBuildIDs := make([]interface{}, len(buildIDs))
-	for i, buildID := range buildIDs {
-		interfaceBuildIDs[i] = buildID
-	}
-
-	indexStrings := make([]string, len(buildIDs))
-	for i := range indexStrings {
-		indexStrings[i] = "$" + strconv.Itoa(i+1)
-	}
-
 	tx, err := p.conn.Begin()
 	if err != nil {
 		return err
 	}
 
 	defer Rollback(tx)
 
+	a := pq.Array(buildIDs)
+
 	_, err = tx.Exec(`
    DELETE FROM `+p.eventsTable()+`
-	 WHERE build_id IN (`+strings.Join(indexStrings, ",")+`)
-	 `, interfaceBuildIDs...)
+	 WHERE build_id = ANY($1)
+	 `, a)
 	if err != nil {
 		return err
 	}
 
 	_, err = tx.Exec(`
 		UPDATE builds
 		SET reap_time = now()
-		WHERE id IN (`+strings.Join(indexStrings, ",")+`)
-	`, interfaceBuildIDs...)
+		WHERE id = ANY($1)
+	`, a)
 	if err != nil {
 		return err
 	}

atc/db/pipeline_test.go

@@ -1,15 +1,15 @@
 package db_test
 
 import (
-	"fmt"
-	"strconv"
-	"time"
-
 	"code.cloudfoundry.org/clock"
+	"fmt"
 	"github.com/concourse/concourse/atc/creds"
 	"github.com/concourse/concourse/atc/creds/credsfakes"
 	"github.com/concourse/concourse/atc/db/dbtest"
 	"github.com/concourse/concourse/vars"
+	"github.com/lib/pq"
+	"strconv"
+	"time"
 
 	"github.com/concourse/concourse/atc"
 	"github.com/concourse/concourse/atc/db"
@@ -1452,6 +1452,44 @@ var _ = Describe("Pipeline", func() {
 			// Not required behavior, just a sanity check for what I think will happen
 			Expect(build4DB.ReapTime()).To(Equal(build1DB.ReapTime()))
 		})
+		It("deletes all build logs when there are more than 65_536", func() {
+			txn, err := dbConn.Begin()
+			Expect(err).ToNot(HaveOccurred())
+
+			// we break the abstractions so that we can efficiently bulk insert a heap of build events
+			// if we did this the "right" way (like the test above) it's excruciatingly slow.
+			stmt, err := txn.Prepare(pq.CopyIn(fmt.Sprintf("pipeline_build_events_%d", pipeline.ID()), "event_id", "build_id", "type", "version", "payload"))
+			Expect(err).ToNot(HaveOccurred())
+
+			numToInsert := 250_000
+			ids := make([]int, numToInsert)
+			for i := 0; i < numToInsert; i++ {
+				_, err = stmt.Exec(i, i, i, i, "")
+				Expect(err).ToNot(HaveOccurred())
+				ids[i] = i
+			}
+
+			_, err = stmt.Exec()
+			Expect(err).ToNot(HaveOccurred())
+
+			err = stmt.Close()
+			Expect(err).ToNot(HaveOccurred())
+
+			err = txn.Commit()
+			Expect(err).ToNot(HaveOccurred())
+
+			var count int
+			err = dbConn.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM pipeline_build_events_%d", pipeline.ID())).Scan(&count)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(count).To(Equal(numToInsert))
+
+			err = pipeline.DeleteBuildEventsByBuildIDs(ids)
+			Expect(err).ToNot(HaveOccurred())
+
+			err = dbConn.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM pipeline_build_events_%d", pipeline.ID())).Scan(&count)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(count).To(Equal(0))
+		})
 	})
 
 	Describe("Jobs", func() {