We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Extend the OTEL traces to track bytes billed and bytes processed
Extend OTEL traces to track bytes billed and bytes processed. Allowing developers to see how much data and cost was consumed as part of BigQueryJob trace 

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `google/cloud/bigquery/opentelemetry_tracing.py`
Content:
```
1 # Copyright 2020 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 
15 import logging
16 from contextlib import contextmanager
17 from google.api_core.exceptions import GoogleAPICallError  # type: ignore
18 
19 logger = logging.getLogger(__name__)
20 try:
21     from opentelemetry import trace  # type: ignore
22     from opentelemetry.instrumentation.utils import http_status_to_status_code  # type: ignore
23     from opentelemetry.trace.status import Status  # type: ignore
24 
25     HAS_OPENTELEMETRY = True
26     _warned_telemetry = True
27 
28 except ImportError:
29     HAS_OPENTELEMETRY = False
30     _warned_telemetry = False
31 
32 _default_attributes = {
33     "db.system": "BigQuery"
34 }  # static, default values assigned to all spans
35 
36 
37 @contextmanager
38 def create_span(name, attributes=None, client=None, job_ref=None):
39     """Creates a ContextManager for a Span to be exported to the configured exporter.
40     If no configuration exists yields None.
41 
42         Args:
43             name (str): Name that will be set for the span being created
44             attributes (Optional[dict]):
45                 Additional attributes that pertain to
46                 the specific API call (i.e. not a default attribute)
47             client (Optional[google.cloud.bigquery.client.Client]):
48                 Pass in a Client object to extract any attributes that may be
49                 relevant to it and add them to the created spans.
50             job_ref (Optional[google.cloud.bigquery.job._AsyncJob])
51                 Pass in a _AsyncJob object to extract any attributes that may be
52                 relevant to it and add them to the created spans.
53 
54         Yields:
55             opentelemetry.trace.Span: Yields the newly created Span.
56 
57         Raises:
58             google.api_core.exceptions.GoogleAPICallError:
59                 Raised if a span could not be yielded or issue with call to
60                 OpenTelemetry.
61     """
62     global _warned_telemetry
63     final_attributes = _get_final_span_attributes(attributes, client, job_ref)
64     if not HAS_OPENTELEMETRY:
65         if not _warned_telemetry:
66             logger.debug(
67                 "This service is instrumented using OpenTelemetry. "
68                 "OpenTelemetry or one of its components could not be imported; "
69                 "please add compatible versions of opentelemetry-api and "
70                 "opentelemetry-instrumentation packages in order to get BigQuery "
71                 "Tracing data."
72             )
73             _warned_telemetry = True
74 
75         yield None
76         return
77     tracer = trace.get_tracer(__name__)
78 
79     # yield new span value
80     with tracer.start_as_current_span(name=name, attributes=final_attributes) as span:
81         try:
82             yield span
83         except GoogleAPICallError as error:
84             if error.code is not None:
85                 span.set_status(Status(http_status_to_status_code(error.code)))
86             raise
87 
88 
89 def _get_final_span_attributes(attributes=None, client=None, job_ref=None):
90     """Compiles attributes from: client, job_ref, user-provided attributes.
91 
92     Attributes from all of these sources are merged together. Note the
93     attributes are added sequentially based on perceived order of precedence:
94     i.e. attributes added last may overwrite attributes added earlier.
95 
96     Args:
97         attributes (Optional[dict]):
98             Additional attributes that pertain to
99             the specific API call (i.e. not a default attribute)
100 
101         client (Optional[google.cloud.bigquery.client.Client]):
102             Pass in a Client object to extract any attributes that may be
103             relevant to it and add them to the final_attributes
104 
105         job_ref (Optional[google.cloud.bigquery.job._AsyncJob])
106             Pass in a _AsyncJob object to extract any attributes that may be
107             relevant to it and add them to the final_attributes.
108 
109     Returns: dict
110     """
111 
112     collected_attributes = _default_attributes.copy()
113 
114     if client:
115         collected_attributes.update(_set_client_attributes(client))
116     if job_ref:
117         collected_attributes.update(_set_job_attributes(job_ref))
118     if attributes:
119         collected_attributes.update(attributes)
120 
121     final_attributes = {k: v for k, v in collected_attributes.items() if v is not None}
122     return final_attributes
123 
124 
125 def _set_client_attributes(client):
126     return {"db.name": client.project, "location": client.location}
127 
128 
129 def _set_job_attributes(job_ref):
130     job_attributes = {
131         "db.name": job_ref.project,
132         "job_id": job_ref.job_id,
133         "state": job_ref.state,
134     }
135 
136     job_attributes["hasErrors"] = job_ref.error_result is not None
137 
138     if job_ref.created is not None:
139         job_attributes["timeCreated"] = job_ref.created.isoformat()
140 
141     if job_ref.started is not None:
142         job_attributes["timeStarted"] = job_ref.started.isoformat()
143 
144     if job_ref.ended is not None:
145         job_attributes["timeEnded"] = job_ref.ended.isoformat()
146 
147     if job_ref.location is not None:
148         job_attributes["location"] = job_ref.location
149 
150     if job_ref.parent_job_id is not None:
151         job_attributes["parent_job_id"] = job_ref.parent_job_id
152 
153     if job_ref.num_child_jobs is not None:
154         job_attributes["num_child_jobs"] = job_ref.num_child_jobs
155 
156     return job_attributes
157
```
--- 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/google/cloud/bigquery/opentelemetry_tracing.py b/google/cloud/bigquery/opentelemetry_tracing.py
--- a/google/cloud/bigquery/opentelemetry_tracing.py
+++ b/google/cloud/bigquery/opentelemetry_tracing.py
@@ -153,4 +153,12 @@
     if job_ref.num_child_jobs is not None:
         job_attributes["num_child_jobs"] = job_ref.num_child_jobs
 
+    total_bytes_billed = getattr(job_ref, "total_bytes_billed", None)
+    if total_bytes_billed is not None:
+        job_attributes["total_bytes_billed"] = total_bytes_billed
+
+    total_bytes_processed = getattr(job_ref, "total_bytes_processed", None)
+    if total_bytes_processed is not None:
+        job_attributes["total_bytes_processed"] = total_bytes_processed
+
     return job_attributes

{"golden_diff": "diff --git a/google/cloud/bigquery/opentelemetry_tracing.py b/google/cloud/bigquery/opentelemetry_tracing.py\n--- a/google/cloud/bigquery/opentelemetry_tracing.py\n+++ b/google/cloud/bigquery/opentelemetry_tracing.py\n@@ -153,4 +153,12 @@\n     if job_ref.num_child_jobs is not None:\n         job_attributes[\"num_child_jobs\"] = job_ref.num_child_jobs\n \n+    total_bytes_billed = getattr(job_ref, \"total_bytes_billed\", None)\n+    if total_bytes_billed is not None:\n+        job_attributes[\"total_bytes_billed\"] = total_bytes_billed\n+\n+    total_bytes_processed = getattr(job_ref, \"total_bytes_processed\", None)\n+    if total_bytes_processed is not None:\n+        job_attributes[\"total_bytes_processed\"] = total_bytes_processed\n+\n     return job_attributes\n", "issue": "Extend the OTEL traces to track bytes billed and bytes processed\nExtend OTEL traces to track bytes billed and bytes processed. Allowing developers to see how much data and cost was consumed as part of BigQueryJob trace \n", "before_files": [{"content": "# Copyright 2020 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\nimport logging\nfrom contextlib import contextmanager\nfrom google.api_core.exceptions import GoogleAPICallError  # type: ignore\n\nlogger = logging.getLogger(__name__)\ntry:\n    from opentelemetry import trace  # type: ignore\n    from opentelemetry.instrumentation.utils import http_status_to_status_code  # type: ignore\n    from opentelemetry.trace.status import Status  # type: ignore\n\n    HAS_OPENTELEMETRY = True\n    _warned_telemetry = True\n\nexcept ImportError:\n    HAS_OPENTELEMETRY = False\n    _warned_telemetry = False\n\n_default_attributes = {\n    \"db.system\": \"BigQuery\"\n}  # static, default values assigned to all spans\n\n\n@contextmanager\ndef create_span(name, attributes=None, client=None, job_ref=None):\n    \"\"\"Creates a ContextManager for a Span to be exported to the configured exporter.\n    If no configuration exists yields None.\n\n        Args:\n            name (str): Name that will be set for the span being created\n            attributes (Optional[dict]):\n                Additional attributes that pertain to\n                the specific API call (i.e. not a default attribute)\n            client (Optional[google.cloud.bigquery.client.Client]):\n                Pass in a Client object to extract any attributes that may be\n                relevant to it and add them to the created spans.\n            job_ref (Optional[google.cloud.bigquery.job._AsyncJob])\n                Pass in a _AsyncJob object to extract any attributes that may be\n                relevant to it and add them to the created spans.\n\n        Yields:\n            opentelemetry.trace.Span: Yields the newly created Span.\n\n        Raises:\n            google.api_core.exceptions.GoogleAPICallError:\n                Raised if a span could not be yielded or issue with call to\n                OpenTelemetry.\n    \"\"\"\n    global _warned_telemetry\n    final_attributes = _get_final_span_attributes(attributes, client, job_ref)\n    if not HAS_OPENTELEMETRY:\n        if not _warned_telemetry:\n            logger.debug(\n                \"This service is instrumented using OpenTelemetry. \"\n                \"OpenTelemetry or one of its components could not be imported; \"\n                \"please add compatible versions of opentelemetry-api and \"\n                \"opentelemetry-instrumentation packages in order to get BigQuery \"\n                \"Tracing data.\"\n            )\n            _warned_telemetry = True\n\n        yield None\n        return\n    tracer = trace.get_tracer(__name__)\n\n    # yield new span value\n    with tracer.start_as_current_span(name=name, attributes=final_attributes) as span:\n        try:\n            yield span\n        except GoogleAPICallError as error:\n            if error.code is not None:\n                span.set_status(Status(http_status_to_status_code(error.code)))\n            raise\n\n\ndef _get_final_span_attributes(attributes=None, client=None, job_ref=None):\n    \"\"\"Compiles attributes from: client, job_ref, user-provided attributes.\n\n    Attributes from all of these sources are merged together. Note the\n    attributes are added sequentially based on perceived order of precedence:\n    i.e. attributes added last may overwrite attributes added earlier.\n\n    Args:\n        attributes (Optional[dict]):\n            Additional attributes that pertain to\n            the specific API call (i.e. not a default attribute)\n\n        client (Optional[google.cloud.bigquery.client.Client]):\n            Pass in a Client object to extract any attributes that may be\n            relevant to it and add them to the final_attributes\n\n        job_ref (Optional[google.cloud.bigquery.job._AsyncJob])\n            Pass in a _AsyncJob object to extract any attributes that may be\n            relevant to it and add them to the final_attributes.\n\n    Returns: dict\n    \"\"\"\n\n    collected_attributes = _default_attributes.copy()\n\n    if client:\n        collected_attributes.update(_set_client_attributes(client))\n    if job_ref:\n        collected_attributes.update(_set_job_attributes(job_ref))\n    if attributes:\n        collected_attributes.update(attributes)\n\n    final_attributes = {k: v for k, v in collected_attributes.items() if v is not None}\n    return final_attributes\n\n\ndef _set_client_attributes(client):\n    return {\"db.name\": client.project, \"location\": client.location}\n\n\ndef _set_job_attributes(job_ref):\n    job_attributes = {\n        \"db.name\": job_ref.project,\n        \"job_id\": job_ref.job_id,\n        \"state\": job_ref.state,\n    }\n\n    job_attributes[\"hasErrors\"] = job_ref.error_result is not None\n\n    if job_ref.created is not None:\n        job_attributes[\"timeCreated\"] = job_ref.created.isoformat()\n\n    if job_ref.started is not None:\n        job_attributes[\"timeStarted\"] = job_ref.started.isoformat()\n\n    if job_ref.ended is not None:\n        job_attributes[\"timeEnded\"] = job_ref.ended.isoformat()\n\n    if job_ref.location is not None:\n        job_attributes[\"location\"] = job_ref.location\n\n    if job_ref.parent_job_id is not None:\n        job_attributes[\"parent_job_id\"] = job_ref.parent_job_id\n\n    if job_ref.num_child_jobs is not None:\n        job_attributes[\"num_child_jobs\"] = job_ref.num_child_jobs\n\n    return job_attributes\n", "path": "google/cloud/bigquery/opentelemetry_tracing.py"}], "after_files": [{"content": "# Copyright 2020 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\nimport logging\nfrom contextlib import contextmanager\nfrom google.api_core.exceptions import GoogleAPICallError  # type: ignore\n\nlogger = logging.getLogger(__name__)\ntry:\n    from opentelemetry import trace  # type: ignore\n    from opentelemetry.instrumentation.utils import http_status_to_status_code  # type: ignore\n    from opentelemetry.trace.status import Status  # type: ignore\n\n    HAS_OPENTELEMETRY = True\n    _warned_telemetry = True\n\nexcept ImportError:\n    HAS_OPENTELEMETRY = False\n    _warned_telemetry = False\n\n_default_attributes = {\n    \"db.system\": \"BigQuery\"\n}  # static, default values assigned to all spans\n\n\n@contextmanager\ndef create_span(name, attributes=None, client=None, job_ref=None):\n    \"\"\"Creates a ContextManager for a Span to be exported to the configured exporter.\n    If no configuration exists yields None.\n\n        Args:\n            name (str): Name that will be set for the span being created\n            attributes (Optional[dict]):\n                Additional attributes that pertain to\n                the specific API call (i.e. not a default attribute)\n            client (Optional[google.cloud.bigquery.client.Client]):\n                Pass in a Client object to extract any attributes that may be\n                relevant to it and add them to the created spans.\n            job_ref (Optional[google.cloud.bigquery.job._AsyncJob])\n                Pass in a _AsyncJob object to extract any attributes that may be\n                relevant to it and add them to the created spans.\n\n        Yields:\n            opentelemetry.trace.Span: Yields the newly created Span.\n\n        Raises:\n            google.api_core.exceptions.GoogleAPICallError:\n                Raised if a span could not be yielded or issue with call to\n                OpenTelemetry.\n    \"\"\"\n    global _warned_telemetry\n    final_attributes = _get_final_span_attributes(attributes, client, job_ref)\n    if not HAS_OPENTELEMETRY:\n        if not _warned_telemetry:\n            logger.debug(\n                \"This service is instrumented using OpenTelemetry. \"\n                \"OpenTelemetry or one of its components could not be imported; \"\n                \"please add compatible versions of opentelemetry-api and \"\n                \"opentelemetry-instrumentation packages in order to get BigQuery \"\n                \"Tracing data.\"\n            )\n            _warned_telemetry = True\n\n        yield None\n        return\n    tracer = trace.get_tracer(__name__)\n\n    # yield new span value\n    with tracer.start_as_current_span(name=name, attributes=final_attributes) as span:\n        try:\n            yield span\n        except GoogleAPICallError as error:\n            if error.code is not None:\n                span.set_status(Status(http_status_to_status_code(error.code)))\n            raise\n\n\ndef _get_final_span_attributes(attributes=None, client=None, job_ref=None):\n    \"\"\"Compiles attributes from: client, job_ref, user-provided attributes.\n\n    Attributes from all of these sources are merged together. Note the\n    attributes are added sequentially based on perceived order of precedence:\n    i.e. attributes added last may overwrite attributes added earlier.\n\n    Args:\n        attributes (Optional[dict]):\n            Additional attributes that pertain to\n            the specific API call (i.e. not a default attribute)\n\n        client (Optional[google.cloud.bigquery.client.Client]):\n            Pass in a Client object to extract any attributes that may be\n            relevant to it and add them to the final_attributes\n\n        job_ref (Optional[google.cloud.bigquery.job._AsyncJob])\n            Pass in a _AsyncJob object to extract any attributes that may be\n            relevant to it and add them to the final_attributes.\n\n    Returns: dict\n    \"\"\"\n\n    collected_attributes = _default_attributes.copy()\n\n    if client:\n        collected_attributes.update(_set_client_attributes(client))\n    if job_ref:\n        collected_attributes.update(_set_job_attributes(job_ref))\n    if attributes:\n        collected_attributes.update(attributes)\n\n    final_attributes = {k: v for k, v in collected_attributes.items() if v is not None}\n    return final_attributes\n\n\ndef _set_client_attributes(client):\n    return {\"db.name\": client.project, \"location\": client.location}\n\n\ndef _set_job_attributes(job_ref):\n    job_attributes = {\n        \"db.name\": job_ref.project,\n        \"job_id\": job_ref.job_id,\n        \"state\": job_ref.state,\n    }\n\n    job_attributes[\"hasErrors\"] = job_ref.error_result is not None\n\n    if job_ref.created is not None:\n        job_attributes[\"timeCreated\"] = job_ref.created.isoformat()\n\n    if job_ref.started is not None:\n        job_attributes[\"timeStarted\"] = job_ref.started.isoformat()\n\n    if job_ref.ended is not None:\n        job_attributes[\"timeEnded\"] = job_ref.ended.isoformat()\n\n    if job_ref.location is not None:\n        job_attributes[\"location\"] = job_ref.location\n\n    if job_ref.parent_job_id is not None:\n        job_attributes[\"parent_job_id\"] = job_ref.parent_job_id\n\n    if job_ref.num_child_jobs is not None:\n        job_attributes[\"num_child_jobs\"] = job_ref.num_child_jobs\n\n    total_bytes_billed = getattr(job_ref, \"total_bytes_billed\", None)\n    if total_bytes_billed is not None:\n        job_attributes[\"total_bytes_billed\"] = total_bytes_billed\n\n    total_bytes_processed = getattr(job_ref, \"total_bytes_processed\", None)\n    if total_bytes_processed is not None:\n        job_attributes[\"total_bytes_processed\"] = total_bytes_processed\n\n    return job_attributes\n", "path": "google/cloud/bigquery/opentelemetry_tracing.py"}]}