3286

Committed 02 Mar 2022 02:09PM UTC coverage: 61.13%. First build

Build # 3286

Build Type

push

travis-ci-com

Committed by

web-flow

Commit Message

<a href="https://github.com/m-lab/etl-gardener/commit/<a class=hub.com/m-lab/etl-gardener/commit/<a class="double-link" href="https://git"><a class=hub.com/m-lab/etl-gardener/commit/<a class="double-link" href="https://git"><a class=hub.com/m-lab/etl-gardener/commit/<a class="double-link" href="https://git"><a class=hub.com/m-lab/etl-gardener/commit/882f4499405af22aa2a3423503e91195d86dc9e0">882f44994">&lt;a href=&quot;https://github.com/m-lab/etl-gardener/commit/</a><a class="double-link" href="https://github.com/m-lab/etl-gardener/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/m-lab/etl-gardener/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/m-lab/etl-gardener/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/m-lab/etl-gardener/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/m-lab/etl-gardener/commit/882f4499405af22aa2a3423503e91195d86dc9e0">882f44994</a><a href="https://github.com/m-lab/etl-gardener/commit/882f4499405af22aa2a3423503e91195d86dc9e0">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/m-lab/etl-gardener/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/m-lab/etl-gardener/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/m-lab/etl-gardener/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/m-lab/etl-gardener/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/m-lab/etl-gardener/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/m-lab/etl-gardener/commit/882f4499405af22aa2a3423503e91195d86dc9e0&quot;&gt;882f44994&lt;/a&gt;&lt;a href=&quot;https://github.com/m-lab/etl-gardener/commit/882f4499405af22aa2a3423503e91195d86dc9e0&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/m-lab/etl-gardener/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/m-lab/etl-gardener/comm... (continued)

Run Details

2 of 2 new or added lines in 1 file covered. (100.0%)

1958 of 3203 relevant lines covered (61.13%)

0.65 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

52.05

/cloud/bq/ops.go

package bq

import (
        "bytes"
        "context"
        "errors"
        "fmt"
        "html/template"
        "log"

        "cloud.google.com/go/bigquery"
        "github.com/googleapis/google-cloud-go-testing/bigquery/bqiface"

        "github.com/m-lab/go/dataset"

        "github.com/m-lab/etl-gardener/tracker"
)

// TableOps is used to construct and execute table partition operations.
type TableOps struct {
        client     bqiface.Client
        LoadSource string // The bucket/path to load from.
        Project    string
        Date       string // Name of the partition field
        Job        tracker.Job
        // map key is the single field name, value is fully qualified name
        PartitionKeys map[string]string
        OrderKeys     string
}

// ErrDatatypeNotSupported is returned by Query for unsupported datatypes.
var ErrDatatypeNotSupported = errors.New("Datatype not supported")

// NewTableOps creates a suitable QueryParams for a Job.
// The context is used to create a bigquery client, and should be kept alive while
// the querier is in use.
func NewTableOps(ctx context.Context, job tracker.Job, project string, loadSource string) (*TableOps, error) {
        c, err := bigquery.NewClient(ctx, project)
        if err != nil {
                return nil, err
        }
        bqClient := bqiface.AdaptClient(c)
        return NewTableOpsWithClient(bqClient, job, project, loadSource)
}

// NewTableOpsWithClient creates a suitable QueryParams for a Job.
func NewTableOpsWithClient(client bqiface.Client, job tracker.Job, project string, loadSource string) (*TableOps, error) {
        switch job.Datatype {
        case "switch":
                fallthrough
        case "annotation":
                fallthrough
        case "hopannotation1":
                fallthrough
        case "pcap":
                fallthrough
        case "scamper1":
                fallthrough
        case "ndt5":
                fallthrough
        case "ndt7":
                return &TableOps{
                        client:        client,
                        LoadSource:    loadSource,
                        Project:       project,
                        Date:          "date",
                        Job:           job,
                        PartitionKeys: map[string]string{"id": "id"},
                        OrderKeys:     "",
                }, nil

        default:
                return nil, ErrDatatypeNotSupported
        }
}

var queryTemplates = map[string]*template.Template{
        "dedup": dedupTemplate,
}

// makeQuery creates a query from a template.
func (to TableOps) makeQuery(t *template.Template) string {
        out := bytes.NewBuffer(nil)
        err := t.Execute(out, to)
        if err != nil {
                log.Println(err)
        }
        return out.String()
}

// dedupQuery returns the appropriate query in string form.
func dedupQuery(to TableOps) string {
        return to.makeQuery(dedupTemplate)
}

// Dedup initiates a deduplication query, and returns the bqiface.Job.
func (to TableOps) Dedup(ctx context.Context, dryRun bool) (bqiface.Job, error) {
        qs := dedupQuery(to)
        if len(qs) == 0 {
                return nil, dataset.ErrNilQuery
        }
        if to.client == nil {
                return nil, dataset.ErrNilBqClient
        }
        q := to.client.Query(qs)
        if q == nil {
                return nil, dataset.ErrNilQuery
        }
        if dryRun {
                qc := bqiface.QueryConfig{QueryConfig: bigquery.QueryConfig{DryRun: dryRun, Q: qs}}
                q.SetQueryConfig(qc)
        }
        return q.Run(ctx)
}

// LoadToTmp loads the tmp_ exp table from GCS files.
func (to TableOps) LoadToTmp(ctx context.Context, dryRun bool) (bqiface.Job, error) {
        if dryRun {
                return nil, errors.New("dryrun not implemented")
        }
        if to.client == nil {
                return nil, dataset.ErrNilBqClient
        }

        gcsRef := bigquery.NewGCSReference(to.LoadSource)
        gcsRef.SourceFormat = bigquery.JSON

        dest := to.client.
                Dataset("tmp_" + to.Job.Experiment).
                Table(to.Job.Datatype)
        if dest == nil {
                return nil, ErrTableNotFound
        }
        loader := dest.LoaderFrom(gcsRef)
        loadConfig := bqiface.LoadConfig{}
        loadConfig.WriteDisposition = bigquery.WriteAppend
        loadConfig.Dst = dest
        loadConfig.Src = gcsRef
        loader.SetLoadConfig(loadConfig)

        return loader.Run(ctx)
}

// CopyToRaw copies the tmp_ job partition to the raw_ job partition.
func (to TableOps) CopyToRaw(ctx context.Context, dryRun bool) (bqiface.Job, error) {
        if dryRun {
                return nil, errors.New("dryrun not implemented")
        }
        if to.client == nil {
                return nil, dataset.ErrNilBqClient
        }
        tableName := to.Job.Datatype + "$" + to.Job.Date.Format("20060102")
        src := to.client.Dataset("tmp_" + to.Job.Experiment).Table(tableName)
        dest := to.client.Dataset("raw_" + to.Job.Experiment).Table(tableName)

        copier := dest.CopierFrom(src)
        config := bqiface.CopyConfig{}
        config.WriteDisposition = bigquery.WriteTruncate
        config.Dst = dest
        config.Srcs = append(config.Srcs, src)
        copier.SetCopyConfig(config)
        return copier.Run(ctx)
}

// TODO get the tmp_ and raw_ from the job Target?
const tmpTable = "`{{.Project}}.tmp_{{.Job.Experiment}}.{{.Job.Datatype}}`"
const rawTable = "`{{.Project}}.raw_{{.Job.Experiment}}.{{.Job.Datatype}}`"
const joinedTable = "`{{.Project}}.{{.Job.Experiment}}.{{.Job.Datatype}}`"

var dedupTemplate = template.Must(template.New("").Parse(`
#standardSQL
# Delete all duplicate rows based on key and prefered priority ordering.
# This is resource intensive for tcpinfo - 20 slot hours for 12M rows with 250M snapshots,
# roughly proportional to the memory footprint of the table partition.
# The query is very cheap if there are no duplicates.
DELETE
FROM ` + tmpTable + ` AS target
WHERE {{.Date}} = "{{.Job.Date.Format "2006-01-02"}}"
# This identifies all rows that don't match rows to preserve.
AND NOT EXISTS (
  # This creates list of rows to preserve, based on key and priority.
  WITH keep AS (
  SELECT * EXCEPT(row_number) FROM (
    SELECT
      {{range $k, $v := .PartitionKeys}}{{$v}}, {{end}}
          parser.Time,
      ROW_NUMBER() OVER (
        PARTITION BY {{range $k, $v := .PartitionKeys}}{{$v}}, {{end}}date
        ORDER BY {{.OrderKeys}} parser.Time DESC
      ) row_number
      FROM (
        SELECT * FROM ` + tmpTable + `
        WHERE {{.Date}} = "{{.Job.Date.Format "2006-01-02"}}"
      )
    )
    WHERE row_number = 1
  )
  SELECT * FROM keep
  # This matches against the keep table based on keys.  Sufficient select keys must be
  # used to distinguish the preferred row from the others.
  WHERE
    {{range $k, $v := .PartitionKeys}}target.{{$v}} = keep.{{$k}} AND {{end}}
    target.parser.Time = keep.Time
)`))

// DeleteTmp deletes the tmp table partition.
func (to TableOps) DeleteTmp(ctx context.Context) error {
        if to.client == nil {
                return dataset.ErrNilBqClient
        }
        // TODO - name should be field in queryer.
        tmp := to.client.Dataset("tmp_" + to.Job.Experiment).Table(
                fmt.Sprintf("%s$%s", to.Job.Datatype, to.Job.Date.Format("20060102")))
        log.Println("Deleting", tmp.FullyQualifiedName())
        return tmp.Delete(ctx)
}

// joinTemplate is used to create join queries, based on Job details.
// TODO - explore whether using query parameters would improve readability
// instead of executing this template.  Query params cannot be used for
// table names, but would work for most other variables.
var joinTemplate = template.Must(template.New("").Parse(`
#standardSQL
# Join the ndt7 data with server and client annotation.
WITH {{.Job.Datatype}} AS (
SELECT *
FROM ` + rawTable + `
WHERE {{.Date}} = "{{.Job.Date.Format "2006-01-02"}}"
),

# Need to remove dups?
ann AS (
SELECT *
FROM ` + "`{{.Project}}.raw_{{.Job.Experiment}}.annotation`" + `
WHERE {{.Date}} BETWEEN DATE_SUB("{{.Job.Date.Format "2006-01-02"}}", INTERVAL 1 DAY) AND "{{.Job.Date.Format "2006-01-02"}}"
)

SELECT {{.Job.Datatype}}.id, {{.Job.Datatype}}.date, {{.Job.Datatype}}.parser,
       ann.* EXCEPT(id, date, parser), {{.Job.Datatype}}.* EXCEPT(id, date, parser)
FROM {{.Job.Datatype}} LEFT JOIN ann USING (id)
`))

// Join joins the raw tables into annotated tables.
func (to TableOps) Join(ctx context.Context, dryRun bool) (bqiface.Job, error) {
        qs := to.makeQuery(joinTemplate)

        if len(qs) == 0 {
                return nil, dataset.ErrNilQuery
        }
        if to.client == nil {
                return nil, dataset.ErrNilBqClient
        }
        q := to.client.Query(qs)
        if q == nil {
                return nil, dataset.ErrNilQuery
        }
        // The destintation is a partition in a table based on the job
        // type and date.  Initially, this will only be ndt7.
        dest := to.client.Dataset(to.Job.Experiment).Table(
                to.Job.Datatype + "$" + to.Job.Date.Format("20060102"))
        qc := bqiface.QueryConfig{
                QueryConfig: bigquery.QueryConfig{
                        DryRun: dryRun,
                        Q:      qs,
                        // We want to replace the whole partition
                        WriteDisposition: bigquery.WriteTruncate,
                        // Create the table if it doesn't exist
                        CreateDisposition: bigquery.CreateIfNeeded,
                        // Allow additional fields introduced by the raw tables to be automatically
                        // added to the joined, materialized output table.
                        SchemaUpdateOptions: []string{"ALLOW_FIELD_ADDITION", "ALLOW_FIELD_RELAXATION"},
                        // Partitioning spec, in event we have to create the table.
                        TimePartitioning: &bigquery.TimePartitioning{
                                Field:                  "date",
                                RequirePartitionFilter: true,
                        },
                },
                Dst: dest,
        }
        q.SetQueryConfig(qc)
        return q.Run(ctx)
}

1	package bq
2
3	import (
4	"bytes"
5	"context"
6	"errors"
7	"fmt"
8	"html/template"
9	"log"
10
11	"cloud.google.com/go/bigquery"
12	"github.com/googleapis/google-cloud-go-testing/bigquery/bqiface"
13
14	"github.com/m-lab/go/dataset"
15
16	"github.com/m-lab/etl-gardener/tracker"
17	)
18
19	// TableOps is used to construct and execute table partition operations.
20	type TableOps struct {
21	client bqiface.Client
22	LoadSource string // The bucket/path to load from.
23	Project string
24	Date string // Name of the partition field
25	Job tracker.Job
26	// map key is the single field name, value is fully qualified name
27	PartitionKeys map[string]string
28	OrderKeys string
29	}
30
31	// ErrDatatypeNotSupported is returned by Query for unsupported datatypes.
32	var ErrDatatypeNotSupported = errors.New("Datatype not supported")
33
34	// NewTableOps creates a suitable QueryParams for a Job.
35	// The context is used to create a bigquery client, and should be kept alive while
36	// the querier is in use.
37	func NewTableOps(ctx context.Context, job tracker.Job, project string, loadSource string) (*TableOps, error) {	1✔
38	c, err := bigquery.NewClient(ctx, project)	1✔
39	if err != nil {	1✔
40	return nil, err	×
41	}	×
42	bqClient := bqiface.AdaptClient(c)	1✔
43	return NewTableOpsWithClient(bqClient, job, project, loadSource)	1✔
44	}
45
46	// NewTableOpsWithClient creates a suitable QueryParams for a Job.
47	func NewTableOpsWithClient(client bqiface.Client, job tracker.Job, project string, loadSource string) (*TableOps, error) {	1✔
48	switch job.Datatype {	1✔
49	case "switch":	×
50	fallthrough	×
51	case "annotation":	1✔
52	fallthrough	1✔
53	case "hopannotation1":	1✔
54	fallthrough	1✔
55	case "pcap":	1✔
56	fallthrough	1✔
57	case "scamper1":	1✔
58	fallthrough	1✔
59	case "ndt5":	1✔
60	fallthrough	1✔
61	case "ndt7":	1✔
62	return &TableOps{	1✔
63	client: client,	1✔
64	LoadSource: loadSource,	1✔
65	Project: project,	1✔
66	Date: "date",	1✔
67	Job: job,	1✔
68	PartitionKeys: map[string]string{"id": "id"},	1✔
69	OrderKeys: "",	1✔
70	}, nil	1✔
71
72	default:	×
73	return nil, ErrDatatypeNotSupported	×
74	}
75	}
76
77	var queryTemplates = map[string]*template.Template{
78	"dedup": dedupTemplate,
79	}
80
81	// makeQuery creates a query from a template.
82	func (to TableOps) makeQuery(t *template.Template) string {	1✔
83	out := bytes.NewBuffer(nil)	1✔
84	err := t.Execute(out, to)	1✔
85	if err != nil {	1✔
86	log.Println(err)	×
87	}	×
88	return out.String()	1✔
89	}
90
91	// dedupQuery returns the appropriate query in string form.
92	func dedupQuery(to TableOps) string {	1✔
93	return to.makeQuery(dedupTemplate)	1✔
94	}	1✔
95
96	// Dedup initiates a deduplication query, and returns the bqiface.Job.
97	func (to TableOps) Dedup(ctx context.Context, dryRun bool) (bqiface.Job, error) {	1✔
98	qs := dedupQuery(to)	1✔
99	if len(qs) == 0 {	1✔
100	return nil, dataset.ErrNilQuery	×
101	}	×
102	if to.client == nil {	1✔
103	return nil, dataset.ErrNilBqClient	×
104	}	×
105	q := to.client.Query(qs)	1✔
106	if q == nil {	1✔
107	return nil, dataset.ErrNilQuery	×
108	}	×
109	if dryRun {	2✔
110	qc := bqiface.QueryConfig{QueryConfig: bigquery.QueryConfig{DryRun: dryRun, Q: qs}}	1✔
111	q.SetQueryConfig(qc)	1✔
112	}	1✔
113	return q.Run(ctx)	1✔
114	}
115
116	// LoadToTmp loads the tmp_ exp table from GCS files.
117	func (to TableOps) LoadToTmp(ctx context.Context, dryRun bool) (bqiface.Job, error) {	×
118	if dryRun {	×
119	return nil, errors.New("dryrun not implemented")	×
120	}	×
121	if to.client == nil {	×
122	return nil, dataset.ErrNilBqClient	×
123	}	×
124
125	gcsRef := bigquery.NewGCSReference(to.LoadSource)	×
126	gcsRef.SourceFormat = bigquery.JSON	×
127		×
128	dest := to.client.	×
129	Dataset("tmp_" + to.Job.Experiment).	×
130	Table(to.Job.Datatype)	×
131	if dest == nil {	×
132	return nil, ErrTableNotFound	×
133	}	×
134	loader := dest.LoaderFrom(gcsRef)	×
135	loadConfig := bqiface.LoadConfig{}	×
136	loadConfig.WriteDisposition = bigquery.WriteAppend	×
137	loadConfig.Dst = dest	×
138	loadConfig.Src = gcsRef	×
139	loader.SetLoadConfig(loadConfig)	×
140		×
141	return loader.Run(ctx)	×
142	}
143
144	// CopyToRaw copies the tmp_ job partition to the raw_ job partition.
145	func (to TableOps) CopyToRaw(ctx context.Context, dryRun bool) (bqiface.Job, error) {	×
146	if dryRun {	×
147	return nil, errors.New("dryrun not implemented")	×
148	}	×
149	if to.client == nil {	×
150	return nil, dataset.ErrNilBqClient	×
151	}	×
152	tableName := to.Job.Datatype + "$" + to.Job.Date.Format("20060102")	×
153	src := to.client.Dataset("tmp_" + to.Job.Experiment).Table(tableName)	×
154	dest := to.client.Dataset("raw_" + to.Job.Experiment).Table(tableName)	×
155		×
156	copier := dest.CopierFrom(src)	×
157	config := bqiface.CopyConfig{}	×
158	config.WriteDisposition = bigquery.WriteTruncate	×
159	config.Dst = dest	×
160	config.Srcs = append(config.Srcs, src)	×
161	copier.SetCopyConfig(config)	×
162	return copier.Run(ctx)	×
163	}
164
165	// TODO get the tmp_ and raw_ from the job Target?
166	const tmpTable = "`{{.Project}}.tmp_{{.Job.Experiment}}.{{.Job.Datatype}}`"
167	const rawTable = "`{{.Project}}.raw_{{.Job.Experiment}}.{{.Job.Datatype}}`"
168	const joinedTable = "`{{.Project}}.{{.Job.Experiment}}.{{.Job.Datatype}}`"
169
170	var dedupTemplate = template.Must(template.New("").Parse(`
171	#standardSQL
172	# Delete all duplicate rows based on key and prefered priority ordering.
173	# This is resource intensive for tcpinfo - 20 slot hours for 12M rows with 250M snapshots,
174	# roughly proportional to the memory footprint of the table partition.
175	# The query is very cheap if there are no duplicates.
176	DELETE
177	FROM ` + tmpTable + ` AS target
178	WHERE {{.Date}} = "{{.Job.Date.Format "2006-01-02"}}"
179	# This identifies all rows that don't match rows to preserve.
180	AND NOT EXISTS (
181	# This creates list of rows to preserve, based on key and priority.
182	WITH keep AS (
183	SELECT * EXCEPT(row_number) FROM (
184	SELECT
185	{{range $k, $v := .PartitionKeys}}{{$v}}, {{end}}
186	parser.Time,
187	ROW_NUMBER() OVER (
188	PARTITION BY {{range $k, $v := .PartitionKeys}}{{$v}}, {{end}}date
189	ORDER BY {{.OrderKeys}} parser.Time DESC
190	) row_number
191	FROM (
192	SELECT * FROM ` + tmpTable + `
193	WHERE {{.Date}} = "{{.Job.Date.Format "2006-01-02"}}"
194	)
195	)
196	WHERE row_number = 1
197	)
198	SELECT * FROM keep
199	# This matches against the keep table based on keys. Sufficient select keys must be
200	# used to distinguish the preferred row from the others.
201	WHERE
202	{{range $k, $v := .PartitionKeys}}target.{{$v}} = keep.{{$k}} AND {{end}}
203	target.parser.Time = keep.Time
204	)`))
205
206	// DeleteTmp deletes the tmp table partition.
207	func (to TableOps) DeleteTmp(ctx context.Context) error {	×
208	if to.client == nil {	×
209	return dataset.ErrNilBqClient	×
210	}	×
211	// TODO - name should be field in queryer.
212	tmp := to.client.Dataset("tmp_" + to.Job.Experiment).Table(	×
213	fmt.Sprintf("%s$%s", to.Job.Datatype, to.Job.Date.Format("20060102")))	×
214	log.Println("Deleting", tmp.FullyQualifiedName())	×
215	return tmp.Delete(ctx)	×
216	}
217
218	// joinTemplate is used to create join queries, based on Job details.
219	// TODO - explore whether using query parameters would improve readability
220	// instead of executing this template. Query params cannot be used for
221	// table names, but would work for most other variables.
222	var joinTemplate = template.Must(template.New("").Parse(`
223	#standardSQL
224	# Join the ndt7 data with server and client annotation.
225	WITH {{.Job.Datatype}} AS (
226	SELECT *
227	FROM ` + rawTable + `
228	WHERE {{.Date}} = "{{.Job.Date.Format "2006-01-02"}}"
229	),
230
231	# Need to remove dups?
232	ann AS (
233	SELECT *
234	FROM ` + "`{{.Project}}.raw_{{.Job.Experiment}}.annotation`" + `
235	WHERE {{.Date}} BETWEEN DATE_SUB("{{.Job.Date.Format "2006-01-02"}}", INTERVAL 1 DAY) AND "{{.Job.Date.Format "2006-01-02"}}"
236	)
237
238	SELECT {{.Job.Datatype}}.id, {{.Job.Datatype}}.date, {{.Job.Datatype}}.parser,
239	ann.* EXCEPT(id, date, parser), {{.Job.Datatype}}.* EXCEPT(id, date, parser)
240	FROM {{.Job.Datatype}} LEFT JOIN ann USING (id)
241	`))
242
243	// Join joins the raw tables into annotated tables.
244	func (to TableOps) Join(ctx context.Context, dryRun bool) (bqiface.Job, error) {	1✔
245	qs := to.makeQuery(joinTemplate)	1✔
246		1✔
247	if len(qs) == 0 {	1✔
248	return nil, dataset.ErrNilQuery	×
249	}	×
250	if to.client == nil {	1✔
251	return nil, dataset.ErrNilBqClient	×
252	}	×
253	q := to.client.Query(qs)	1✔
254	if q == nil {	1✔
255	return nil, dataset.ErrNilQuery	×
256	}	×
257	// The destintation is a partition in a table based on the job
258	// type and date. Initially, this will only be ndt7.
259	dest := to.client.Dataset(to.Job.Experiment).Table(	1✔
260	to.Job.Datatype + "$" + to.Job.Date.Format("20060102"))	1✔
261	qc := bqiface.QueryConfig{	1✔
262	QueryConfig: bigquery.QueryConfig{	1✔
263	DryRun: dryRun,	1✔
264	Q: qs,	1✔
265	// We want to replace the whole partition	1✔
266	WriteDisposition: bigquery.WriteTruncate,	1✔
267	// Create the table if it doesn't exist	1✔
268	CreateDisposition: bigquery.CreateIfNeeded,	1✔
269	// Allow additional fields introduced by the raw tables to be automatically	1✔
270	// added to the joined, materialized output table.	1✔
271	SchemaUpdateOptions: []string{"ALLOW_FIELD_ADDITION", "ALLOW_FIELD_RELAXATION"},	1✔
272	// Partitioning spec, in event we have to create the table.	1✔
273	TimePartitioning: &bigquery.TimePartitioning{	1✔
274	Field: "date",	1✔
275	RequirePartitionFilter: true,	1✔
276	},	1✔
277	},	1✔
278	Dst: dest,	1✔
279	}	1✔
280	q.SetQueryConfig(qc)	1✔
281	return q.Run(ctx)	1✔
282	}

m-lab / etl-gardener / 3286

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous