Set batch priority on dedup and join (#412)

* Set batch priority on dedup and join
* Check that job was run with batch priority
* Restore dryrun
* Fix comments

cloud/bq/ops.go

@@ -113,10 +113,15 @@ func (to TableOps) Dedup(ctx context.Context, dryRun bool) (bqiface.Job, error)
 	if q == nil {
 		return nil, dataset.ErrNilQuery
 	}
-	if dryRun {
-		qc := bqiface.QueryConfig{QueryConfig: bigquery.QueryConfig{DryRun: dryRun, Q: qs}}
-		q.SetQueryConfig(qc)
+	qc := bqiface.QueryConfig{
+		QueryConfig: bigquery.QueryConfig{
+			Q:      qs,
+			DryRun: dryRun,
+			// Schedule as batch job to avoid quota limits for interactive jobs.
+			Priority: bigquery.BatchPriority,
+		},
 	}
+	q.SetQueryConfig(qc)
 	return q.Run(ctx)
 }
 
@@ -281,6 +286,8 @@ func (to TableOps) Join(ctx context.Context, dryRun bool) (bqiface.Job, error) {
 				Field:                  "date",
 				RequirePartitionFilter: true,
 			},
+			// Schedule as batch job to avoid quota limits for interactive jobs.
+			Priority: bigquery.BatchPriority,
 		},
 		Dst: dest,
 	}

cloud/bq/ops_test.go

@@ -9,6 +9,7 @@ import (
 	"testing"
 	"time"
 
+	"cloud.google.com/go/bigquery"
 	"github.com/m-lab/etl-gardener/cloud/bq"
 	"github.com/m-lab/etl-gardener/ops"
 	"github.com/m-lab/etl-gardener/tracker"
@@ -90,6 +91,13 @@ func TestValidateQueries(t *testing.T) {
 			if status.Err() != nil {
 				t.Fatal(t.Name(), err, bq.DedupQuery(*qp))
 			}
+			cfg, err := j.Config()
+			if err != nil {
+				t.Fatalf("TestValidateQueries() Dedup job Config failed: %v", err)
+			}
+			if cfg.(*bigquery.QueryConfig).Priority != bigquery.BatchPriority {
+				t.Errorf("TestValidateQueries() Dedup job priority is not Batch: %v", bigquery.BatchPriority)
+			}
 
 			if !ops.JoinableDatatypes[qp.Job.Datatype] {
 				return
@@ -103,6 +111,13 @@ func TestValidateQueries(t *testing.T) {
 			if status.Err() != nil {
 				t.Fatal(t.Name(), err, bq.JoinQuery(*qp))
 			}
+			cfg, err = j.Config()
+			if err != nil {
+				t.Fatalf("TestValidateQueries() Join job config failed: %v", err)
+			}
+			if cfg.(*bigquery.QueryConfig).Priority != bigquery.BatchPriority {
+				t.Errorf("TestValidateQueries() Join job priority is not Batch: %v", bigquery.BatchPriority)
+			}
 		})
 	}
 }