We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Endpoints "/fuzz-strategy-selection" and "/fuzzer-and-job-weights" under cron-service failed. 
Hi,
Deploy was done from the master, not from the latest tag. 


The reason in both cases is the same:
**"Table afl_stats.TestcaseRun does not have a schema.">. Retrying again.**

Cron-jobs:

Below details for an endpoint "/fuzz-strategy-selection":
![image](https://user-images.githubusercontent.com/60612203/165520715-937f4ed5-3787-4a3c-8100-7d2ef620fbbb.png)

Stacktrace:
```
<HttpError 400 when requesting https://bigquery.googleapis.com/bigquery/v2/projects/sta-np-11528163/queries?alt=json returned "Table afl_stats.TestcaseRun does not have a schema.">
Traceback (most recent call last):
  File "/srv/handlers/base_handler.py", line 277, in dispatch_request
    return super(Handler, self).dispatch_request(*args, **kwargs)
  File "third_party/flask/views.py", line 163, in dispatch_request
    return meth(*args, **kwargs)
  File "/srv/libs/handler.py", line 100, in wrapper
    result = func(self)
  File "/srv/handlers/cron/fuzz_strategy_selection.py", line 218, in get
    _query_and_upload_strategy_probabilities(engine)
  File "/srv/handlers/cron/fuzz_strategy_selection.py", line 185, in _query_and_upload_strategy_probabilities
    data = _query_multi_armed_bandit_probabilities(engine)
  File "/srv/handlers/cron/fuzz_strategy_selection.py", line 151, in _query_multi_armed_bandit_probabilities
    return client.query(query=formatted_query).rows
  File "/srv/clusterfuzz/_internal/google_cloud_utils/big_query.py", line 247, in query
    result = self.raw_query(query, max_results=0)
  File "/srv/clusterfuzz/_internal/base/retry.py", line 88, in _wrapper
    result = func(*args, **kwargs)
  File "/srv/clusterfuzz/_internal/google_cloud_utils/big_query.py", line 190, in raw_query
    projectId=self.project_id, body=body).execute()
  File "third_party/googleapiclient/_helpers.py", line 134, in positional_wrapper
    return wrapped(*args, **kwargs)
  File "third_party/googleapiclient/http.py", line 907, in execute
    raise HttpError(resp, content, uri=self.uri)
googleapiclient.errors.HttpError: <HttpError 400 when requesting https://bigquery.googleapis.com/bigquery/v2/projects/sta-np-11528163/queries?alt=json returned "Table afl_stats.TestcaseRun does not have a schema.">
```

**Is it normal behaviour?**

**Do you have any idea what can be wrong? How to fix it?**

So far we have only tested libfuzzer, afl wasn't touched yet.

Thanks,

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/appengine/handlers/cron/load_bigquery_stats.py`
Content:
```
1 # Copyright 2019 Google LLC
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #      http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 """Handler used for loading bigquery data."""
15 
16 import datetime
17 import random
18 import time
19 
20 from googleapiclient.errors import HttpError
21 import httplib2
22 
23 from clusterfuzz._internal.base import utils
24 from clusterfuzz._internal.datastore import data_types
25 from clusterfuzz._internal.google_cloud_utils import big_query
26 from clusterfuzz._internal.metrics import fuzzer_stats
27 from clusterfuzz._internal.metrics import fuzzer_stats_schema
28 from clusterfuzz._internal.metrics import logs
29 from handlers import base_handler
30 from libs import handler
31 
32 STATS_KINDS = [fuzzer_stats.JobRun, fuzzer_stats.TestcaseRun]
33 
34 NUM_RETRIES = 2
35 RETRY_SLEEP_TIME = 5
36 
37 
38 class Handler(base_handler.Handler):
39   """Cron handler for loading bigquery stats."""
40 
41   def _utc_now(self):
42     """Return datetime.datetime.utcnow()."""
43     return datetime.datetime.utcnow()
44 
45   def _execute_insert_request(self, request):
46     """Executes a table/dataset insert request, retrying on transport errors."""
47     for i in range(NUM_RETRIES + 1):
48       try:
49         request.execute()
50         return True
51       except HttpError as e:
52         if e.resp.status == 409:
53           # Already exists.
54           return True
55 
56         logs.log_error('Failed to insert table/dataset.')
57         return False
58       except httplib2.HttpLib2Error:
59         # Transport error.
60         time.sleep(random.uniform(0, (1 << i) * RETRY_SLEEP_TIME))
61         continue
62 
63     logs.log_error('Failed to insert table/dataset.')
64     return False
65 
66   def _create_dataset_if_needed(self, bigquery, dataset_id):
67     """Create a new dataset if necessary."""
68     project_id = utils.get_application_id()
69     dataset_body = {
70         'datasetReference': {
71             'datasetId': dataset_id,
72             'projectId': project_id,
73         },
74     }
75     dataset_insert = bigquery.datasets().insert(
76         projectId=project_id, body=dataset_body)
77 
78     return self._execute_insert_request(dataset_insert)
79 
80   def _create_table_if_needed(self, bigquery, dataset_id, table_id):
81     """Create a new table if needed."""
82     project_id = utils.get_application_id()
83     table_body = {
84         'tableReference': {
85             'datasetId': dataset_id,
86             'projectId': project_id,
87             'tableId': table_id,
88         },
89         'timePartitioning': {
90             'type': 'DAY',
91         },
92     }
93 
94     table_insert = bigquery.tables().insert(
95         projectId=project_id, datasetId=dataset_id, body=table_body)
96     return self._execute_insert_request(table_insert)
97 
98   def _load_data(self, bigquery, fuzzer):
99     """Load yesterday's stats into BigQuery."""
100     project_id = utils.get_application_id()
101 
102     yesterday = (self._utc_now().date() - datetime.timedelta(days=1))
103     date_string = yesterday.strftime('%Y%m%d')
104     timestamp = utils.utc_date_to_timestamp(yesterday)
105 
106     dataset_id = fuzzer_stats.dataset_name(fuzzer)
107     if not self._create_dataset_if_needed(bigquery, dataset_id):
108       return
109 
110     for kind in STATS_KINDS:
111       kind_name = kind.__name__
112       table_id = kind_name
113       if not self._create_table_if_needed(bigquery, dataset_id, table_id):
114         continue
115 
116       if kind == fuzzer_stats.TestcaseRun:
117         schema = fuzzer_stats_schema.get(fuzzer)
118       else:
119         schema = kind.SCHEMA
120 
121       gcs_path = fuzzer_stats.get_gcs_stats_path(kind_name, fuzzer, timestamp)
122       load = {
123           'destinationTable': {
124               'projectId': project_id,
125               'tableId': table_id + '$' + date_string,
126               'datasetId': dataset_id,
127           },
128           'schemaUpdateOptions': ['ALLOW_FIELD_ADDITION',],
129           'sourceFormat': 'NEWLINE_DELIMITED_JSON',
130           'sourceUris': ['gs:/' + gcs_path + '*.json'],
131           'writeDisposition': 'WRITE_TRUNCATE',
132       }
133       if schema is not None:
134         load['schema'] = schema
135 
136       job_body = {
137           'configuration': {
138               'load': load,
139           },
140       }
141 
142       logs.log("Uploading job to BigQuery.", job_body=job_body)
143       request = bigquery.jobs().insert(projectId=project_id, body=job_body)
144       response = request.execute()
145 
146       # We cannot really check the response here, as the query might be still
147       # running, but having a BigQuery jobId in the log would make our life
148       # simpler if we ever have to manually check the status of the query.
149       # See https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs/query.
150       logs.log('Response from BigQuery: %s' % response)
151 
152   @handler.cron()
153   def get(self):
154     """Load bigquery stats from GCS."""
155     if not big_query.get_bucket():
156       logs.log_error('Loading stats to BigQuery failed: missing bucket name.')
157       return
158 
159     # Retrieve list of fuzzers before iterating them, since the query can expire
160     # as we create the load jobs.
161     bigquery_client = big_query.get_api_client()
162     for fuzzer in list(data_types.Fuzzer.query()):
163       logs.log('Loading stats to BigQuery for %s.' % fuzzer.name)
164       self._load_data(bigquery_client, fuzzer.name)
165
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/appengine/handlers/cron/load_bigquery_stats.py b/src/appengine/handlers/cron/load_bigquery_stats.py
--- a/src/appengine/handlers/cron/load_bigquery_stats.py
+++ b/src/appengine/handlers/cron/load_bigquery_stats.py
@@ -77,7 +77,7 @@
 
     return self._execute_insert_request(dataset_insert)
 
-  def _create_table_if_needed(self, bigquery, dataset_id, table_id):
+  def _create_table_if_needed(self, bigquery, dataset_id, table_id, schema):
     """Create a new table if needed."""
     project_id = utils.get_application_id()
     table_body = {
@@ -91,6 +91,9 @@
         },
     }
 
+    if schema is not None:
+      table_body['schema'] = schema
+
     table_insert = bigquery.tables().insert(
         projectId=project_id, datasetId=dataset_id, body=table_body)
     return self._execute_insert_request(table_insert)
@@ -110,14 +113,16 @@
     for kind in STATS_KINDS:
       kind_name = kind.__name__
       table_id = kind_name
-      if not self._create_table_if_needed(bigquery, dataset_id, table_id):
-        continue
 
       if kind == fuzzer_stats.TestcaseRun:
         schema = fuzzer_stats_schema.get(fuzzer)
       else:
         schema = kind.SCHEMA
 
+      if not self._create_table_if_needed(bigquery, dataset_id, table_id,
+                                          schema):
+        continue
+
       gcs_path = fuzzer_stats.get_gcs_stats_path(kind_name, fuzzer, timestamp)
       load = {
           'destinationTable': {

{"golden_diff": "diff --git a/src/appengine/handlers/cron/load_bigquery_stats.py b/src/appengine/handlers/cron/load_bigquery_stats.py\n--- a/src/appengine/handlers/cron/load_bigquery_stats.py\n+++ b/src/appengine/handlers/cron/load_bigquery_stats.py\n@@ -77,7 +77,7 @@\n \n     return self._execute_insert_request(dataset_insert)\n \n-  def _create_table_if_needed(self, bigquery, dataset_id, table_id):\n+  def _create_table_if_needed(self, bigquery, dataset_id, table_id, schema):\n     \"\"\"Create a new table if needed.\"\"\"\n     project_id = utils.get_application_id()\n     table_body = {\n@@ -91,6 +91,9 @@\n         },\n     }\n \n+    if schema is not None:\n+      table_body['schema'] = schema\n+\n     table_insert = bigquery.tables().insert(\n         projectId=project_id, datasetId=dataset_id, body=table_body)\n     return self._execute_insert_request(table_insert)\n@@ -110,14 +113,16 @@\n     for kind in STATS_KINDS:\n       kind_name = kind.__name__\n       table_id = kind_name\n-      if not self._create_table_if_needed(bigquery, dataset_id, table_id):\n-        continue\n \n       if kind == fuzzer_stats.TestcaseRun:\n         schema = fuzzer_stats_schema.get(fuzzer)\n       else:\n         schema = kind.SCHEMA\n \n+      if not self._create_table_if_needed(bigquery, dataset_id, table_id,\n+                                          schema):\n+        continue\n+\n       gcs_path = fuzzer_stats.get_gcs_stats_path(kind_name, fuzzer, timestamp)\n       load = {\n           'destinationTable': {\n", "issue": "Endpoints \"/fuzz-strategy-selection\" and \"/fuzzer-and-job-weights\" under cron-service failed. \nHi,\r\nDeploy was done from the master, not from the latest tag. \r\n\r\n\r\nThe reason in both cases is the same:\r\n**\"Table afl_stats.TestcaseRun does not have a schema.\">. Retrying again.**\r\n\r\nCron-jobs:\r\n\r\nBelow details for an endpoint \"/fuzz-strategy-selection\":\r\n![image](https://user-images.githubusercontent.com/60612203/165520715-937f4ed5-3787-4a3c-8100-7d2ef620fbbb.png)\r\n\r\nStacktrace:\r\n```\r\n<HttpError 400 when requesting https://bigquery.googleapis.com/bigquery/v2/projects/sta-np-11528163/queries?alt=json returned \"Table afl_stats.TestcaseRun does not have a schema.\">\r\nTraceback (most recent call last):\r\n  File \"/srv/handlers/base_handler.py\", line 277, in dispatch_request\r\n    return super(Handler, self).dispatch_request(*args, **kwargs)\r\n  File \"third_party/flask/views.py\", line 163, in dispatch_request\r\n    return meth(*args, **kwargs)\r\n  File \"/srv/libs/handler.py\", line 100, in wrapper\r\n    result = func(self)\r\n  File \"/srv/handlers/cron/fuzz_strategy_selection.py\", line 218, in get\r\n    _query_and_upload_strategy_probabilities(engine)\r\n  File \"/srv/handlers/cron/fuzz_strategy_selection.py\", line 185, in _query_and_upload_strategy_probabilities\r\n    data = _query_multi_armed_bandit_probabilities(engine)\r\n  File \"/srv/handlers/cron/fuzz_strategy_selection.py\", line 151, in _query_multi_armed_bandit_probabilities\r\n    return client.query(query=formatted_query).rows\r\n  File \"/srv/clusterfuzz/_internal/google_cloud_utils/big_query.py\", line 247, in query\r\n    result = self.raw_query(query, max_results=0)\r\n  File \"/srv/clusterfuzz/_internal/base/retry.py\", line 88, in _wrapper\r\n    result = func(*args, **kwargs)\r\n  File \"/srv/clusterfuzz/_internal/google_cloud_utils/big_query.py\", line 190, in raw_query\r\n    projectId=self.project_id, body=body).execute()\r\n  File \"third_party/googleapiclient/_helpers.py\", line 134, in positional_wrapper\r\n    return wrapped(*args, **kwargs)\r\n  File \"third_party/googleapiclient/http.py\", line 907, in execute\r\n    raise HttpError(resp, content, uri=self.uri)\r\ngoogleapiclient.errors.HttpError: <HttpError 400 when requesting https://bigquery.googleapis.com/bigquery/v2/projects/sta-np-11528163/queries?alt=json returned \"Table afl_stats.TestcaseRun does not have a schema.\">\r\n```\r\n\r\n**Is it normal behaviour?**\r\n\r\n**Do you have any idea what can be wrong? How to fix it?**\r\n\r\nSo far we have only tested libfuzzer, afl wasn't touched yet.\r\n\r\nThanks,\n", "before_files": [{"content": "# Copyright 2019 Google LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"Handler used for loading bigquery data.\"\"\"\n\nimport datetime\nimport random\nimport time\n\nfrom googleapiclient.errors import HttpError\nimport httplib2\n\nfrom clusterfuzz._internal.base import utils\nfrom clusterfuzz._internal.datastore import data_types\nfrom clusterfuzz._internal.google_cloud_utils import big_query\nfrom clusterfuzz._internal.metrics import fuzzer_stats\nfrom clusterfuzz._internal.metrics import fuzzer_stats_schema\nfrom clusterfuzz._internal.metrics import logs\nfrom handlers import base_handler\nfrom libs import handler\n\nSTATS_KINDS = [fuzzer_stats.JobRun, fuzzer_stats.TestcaseRun]\n\nNUM_RETRIES = 2\nRETRY_SLEEP_TIME = 5\n\n\nclass Handler(base_handler.Handler):\n  \"\"\"Cron handler for loading bigquery stats.\"\"\"\n\n  def _utc_now(self):\n    \"\"\"Return datetime.datetime.utcnow().\"\"\"\n    return datetime.datetime.utcnow()\n\n  def _execute_insert_request(self, request):\n    \"\"\"Executes a table/dataset insert request, retrying on transport errors.\"\"\"\n    for i in range(NUM_RETRIES + 1):\n      try:\n        request.execute()\n        return True\n      except HttpError as e:\n        if e.resp.status == 409:\n          # Already exists.\n          return True\n\n        logs.log_error('Failed to insert table/dataset.')\n        return False\n      except httplib2.HttpLib2Error:\n        # Transport error.\n        time.sleep(random.uniform(0, (1 << i) * RETRY_SLEEP_TIME))\n        continue\n\n    logs.log_error('Failed to insert table/dataset.')\n    return False\n\n  def _create_dataset_if_needed(self, bigquery, dataset_id):\n    \"\"\"Create a new dataset if necessary.\"\"\"\n    project_id = utils.get_application_id()\n    dataset_body = {\n        'datasetReference': {\n            'datasetId': dataset_id,\n            'projectId': project_id,\n        },\n    }\n    dataset_insert = bigquery.datasets().insert(\n        projectId=project_id, body=dataset_body)\n\n    return self._execute_insert_request(dataset_insert)\n\n  def _create_table_if_needed(self, bigquery, dataset_id, table_id):\n    \"\"\"Create a new table if needed.\"\"\"\n    project_id = utils.get_application_id()\n    table_body = {\n        'tableReference': {\n            'datasetId': dataset_id,\n            'projectId': project_id,\n            'tableId': table_id,\n        },\n        'timePartitioning': {\n            'type': 'DAY',\n        },\n    }\n\n    table_insert = bigquery.tables().insert(\n        projectId=project_id, datasetId=dataset_id, body=table_body)\n    return self._execute_insert_request(table_insert)\n\n  def _load_data(self, bigquery, fuzzer):\n    \"\"\"Load yesterday's stats into BigQuery.\"\"\"\n    project_id = utils.get_application_id()\n\n    yesterday = (self._utc_now().date() - datetime.timedelta(days=1))\n    date_string = yesterday.strftime('%Y%m%d')\n    timestamp = utils.utc_date_to_timestamp(yesterday)\n\n    dataset_id = fuzzer_stats.dataset_name(fuzzer)\n    if not self._create_dataset_if_needed(bigquery, dataset_id):\n      return\n\n    for kind in STATS_KINDS:\n      kind_name = kind.__name__\n      table_id = kind_name\n      if not self._create_table_if_needed(bigquery, dataset_id, table_id):\n        continue\n\n      if kind == fuzzer_stats.TestcaseRun:\n        schema = fuzzer_stats_schema.get(fuzzer)\n      else:\n        schema = kind.SCHEMA\n\n      gcs_path = fuzzer_stats.get_gcs_stats_path(kind_name, fuzzer, timestamp)\n      load = {\n          'destinationTable': {\n              'projectId': project_id,\n              'tableId': table_id + '$' + date_string,\n              'datasetId': dataset_id,\n          },\n          'schemaUpdateOptions': ['ALLOW_FIELD_ADDITION',],\n          'sourceFormat': 'NEWLINE_DELIMITED_JSON',\n          'sourceUris': ['gs:/' + gcs_path + '*.json'],\n          'writeDisposition': 'WRITE_TRUNCATE',\n      }\n      if schema is not None:\n        load['schema'] = schema\n\n      job_body = {\n          'configuration': {\n              'load': load,\n          },\n      }\n\n      logs.log(\"Uploading job to BigQuery.\", job_body=job_body)\n      request = bigquery.jobs().insert(projectId=project_id, body=job_body)\n      response = request.execute()\n\n      # We cannot really check the response here, as the query might be still\n      # running, but having a BigQuery jobId in the log would make our life\n      # simpler if we ever have to manually check the status of the query.\n      # See https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs/query.\n      logs.log('Response from BigQuery: %s' % response)\n\n  @handler.cron()\n  def get(self):\n    \"\"\"Load bigquery stats from GCS.\"\"\"\n    if not big_query.get_bucket():\n      logs.log_error('Loading stats to BigQuery failed: missing bucket name.')\n      return\n\n    # Retrieve list of fuzzers before iterating them, since the query can expire\n    # as we create the load jobs.\n    bigquery_client = big_query.get_api_client()\n    for fuzzer in list(data_types.Fuzzer.query()):\n      logs.log('Loading stats to BigQuery for %s.' % fuzzer.name)\n      self._load_data(bigquery_client, fuzzer.name)\n", "path": "src/appengine/handlers/cron/load_bigquery_stats.py"}], "after_files": [{"content": "# Copyright 2019 Google LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"Handler used for loading bigquery data.\"\"\"\n\nimport datetime\nimport random\nimport time\n\nfrom googleapiclient.errors import HttpError\nimport httplib2\n\nfrom clusterfuzz._internal.base import utils\nfrom clusterfuzz._internal.datastore import data_types\nfrom clusterfuzz._internal.google_cloud_utils import big_query\nfrom clusterfuzz._internal.metrics import fuzzer_stats\nfrom clusterfuzz._internal.metrics import fuzzer_stats_schema\nfrom clusterfuzz._internal.metrics import logs\nfrom handlers import base_handler\nfrom libs import handler\n\nSTATS_KINDS = [fuzzer_stats.JobRun, fuzzer_stats.TestcaseRun]\n\nNUM_RETRIES = 2\nRETRY_SLEEP_TIME = 5\n\n\nclass Handler(base_handler.Handler):\n  \"\"\"Cron handler for loading bigquery stats.\"\"\"\n\n  def _utc_now(self):\n    \"\"\"Return datetime.datetime.utcnow().\"\"\"\n    return datetime.datetime.utcnow()\n\n  def _execute_insert_request(self, request):\n    \"\"\"Executes a table/dataset insert request, retrying on transport errors.\"\"\"\n    for i in range(NUM_RETRIES + 1):\n      try:\n        request.execute()\n        return True\n      except HttpError as e:\n        if e.resp.status == 409:\n          # Already exists.\n          return True\n\n        logs.log_error('Failed to insert table/dataset.')\n        return False\n      except httplib2.HttpLib2Error:\n        # Transport error.\n        time.sleep(random.uniform(0, (1 << i) * RETRY_SLEEP_TIME))\n        continue\n\n    logs.log_error('Failed to insert table/dataset.')\n    return False\n\n  def _create_dataset_if_needed(self, bigquery, dataset_id):\n    \"\"\"Create a new dataset if necessary.\"\"\"\n    project_id = utils.get_application_id()\n    dataset_body = {\n        'datasetReference': {\n            'datasetId': dataset_id,\n            'projectId': project_id,\n        },\n    }\n    dataset_insert = bigquery.datasets().insert(\n        projectId=project_id, body=dataset_body)\n\n    return self._execute_insert_request(dataset_insert)\n\n  def _create_table_if_needed(self, bigquery, dataset_id, table_id, schema):\n    \"\"\"Create a new table if needed.\"\"\"\n    project_id = utils.get_application_id()\n    table_body = {\n        'tableReference': {\n            'datasetId': dataset_id,\n            'projectId': project_id,\n            'tableId': table_id,\n        },\n        'timePartitioning': {\n            'type': 'DAY',\n        },\n    }\n\n    if schema is not None:\n      table_body['schema'] = schema\n\n    table_insert = bigquery.tables().insert(\n        projectId=project_id, datasetId=dataset_id, body=table_body)\n    return self._execute_insert_request(table_insert)\n\n  def _load_data(self, bigquery, fuzzer):\n    \"\"\"Load yesterday's stats into BigQuery.\"\"\"\n    project_id = utils.get_application_id()\n\n    yesterday = (self._utc_now().date() - datetime.timedelta(days=1))\n    date_string = yesterday.strftime('%Y%m%d')\n    timestamp = utils.utc_date_to_timestamp(yesterday)\n\n    dataset_id = fuzzer_stats.dataset_name(fuzzer)\n    if not self._create_dataset_if_needed(bigquery, dataset_id):\n      return\n\n    for kind in STATS_KINDS:\n      kind_name = kind.__name__\n      table_id = kind_name\n\n      if kind == fuzzer_stats.TestcaseRun:\n        schema = fuzzer_stats_schema.get(fuzzer)\n      else:\n        schema = kind.SCHEMA\n\n      if not self._create_table_if_needed(bigquery, dataset_id, table_id,\n                                          schema):\n        continue\n\n      gcs_path = fuzzer_stats.get_gcs_stats_path(kind_name, fuzzer, timestamp)\n      load = {\n          'destinationTable': {\n              'projectId': project_id,\n              'tableId': table_id + '$' + date_string,\n              'datasetId': dataset_id,\n          },\n          'schemaUpdateOptions': ['ALLOW_FIELD_ADDITION',],\n          'sourceFormat': 'NEWLINE_DELIMITED_JSON',\n          'sourceUris': ['gs:/' + gcs_path + '*.json'],\n          'writeDisposition': 'WRITE_TRUNCATE',\n      }\n      if schema is not None:\n        load['schema'] = schema\n\n      job_body = {\n          'configuration': {\n              'load': load,\n          },\n      }\n\n      logs.log(\"Uploading job to BigQuery.\", job_body=job_body)\n      request = bigquery.jobs().insert(projectId=project_id, body=job_body)\n      response = request.execute()\n\n      # We cannot really check the response here, as the query might be still\n      # running, but having a BigQuery jobId in the log would make our life\n      # simpler if we ever have to manually check the status of the query.\n      # See https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs/query.\n      logs.log('Response from BigQuery: %s' % response)\n\n  @handler.cron()\n  def get(self):\n    \"\"\"Load bigquery stats from GCS.\"\"\"\n    if not big_query.get_bucket():\n      logs.log_error('Loading stats to BigQuery failed: missing bucket name.')\n      return\n\n    # Retrieve list of fuzzers before iterating them, since the query can expire\n    # as we create the load jobs.\n    bigquery_client = big_query.get_api_client()\n    for fuzzer in list(data_types.Fuzzer.query()):\n      logs.log('Loading stats to BigQuery for %s.' % fuzzer.name)\n      self._load_data(bigquery_client, fuzzer.name)\n", "path": "src/appengine/handlers/cron/load_bigquery_stats.py"}]}