Repository: googleapis/python-bigquery

Issue:
None or empty array parameter values aren't handled in the dbapi
```
>>> import google.cloud.bigquery.dbapi
>>> conn = google.cloud.bigquery.dbapi.connect()
/home/jim/p/g/python-bigquery/google/cloud/bigquery/client.py:444: UserWarning: Cannot create BigQuery Storage client, the dependency google-cloud-bigquery-storage is not installed.
  warnings.warn(
>>> cursor = conn.cursor()
>>> cursor.execute("select %(x)s", dict(x=None))
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/jim/p/g/python-bigquery/google/cloud/bigquery/dbapi/_helpers.py", line 270, in with_closed_check
    return method(self, *args, **kwargs)
  File "/home/jim/p/g/python-bigquery/google/cloud/bigquery/dbapi/cursor.py", line 173, in execute
    query_parameters = _helpers.to_query_parameters(parameters)
  File "/home/jim/p/g/python-bigquery/google/cloud/bigquery/dbapi/_helpers.py", line 167, in to_query_parameters
    return to_query_parameters_dict(parameters)
  File "/home/jim/p/g/python-bigquery/google/cloud/bigquery/dbapi/_helpers.py", line 146, in to_query_parameters_dict
    param = scalar_to_query_parameter(value, name=name)
  File "/home/jim/p/g/python-bigquery/google/cloud/bigquery/dbapi/_helpers.py", line 53, in scalar_to_query_parameter
    raise exceptions.ProgrammingError(
google.cloud.bigquery.dbapi.exceptions.ProgrammingError: encountered parameter x with value None of unexpected type
```

The same issue applies to empty array (`[]`) parameters.


Discussion:

Patch:
diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS
index 10f4ee7c..ae570eb0 100644
--- a/.github/CODEOWNERS
+++ b/.github/CODEOWNERS
@@ -5,8 +5,7 @@
 # https://help.github.com/en/github/creating-cloning-and-archiving-repositories/about-code-owners#codeowners-syntax
 
 # The @googleapis/api-bigquery is the default owner for changes in this repo
-*               @googleapis/api-bigquery
+*               @googleapis/api-bigquery @googleapis/yoshi-python
 
 # The python-samples-reviewers team is the default owner for samples changes
 /samples/   @googleapis/python-samples-owners
-
diff --git a/docs/dbapi.rst b/docs/dbapi.rst
index ca0256d3..41ec8583 100644
--- a/docs/dbapi.rst
+++ b/docs/dbapi.rst
@@ -4,3 +4,40 @@ DB-API Reference
 .. automodule:: google.cloud.bigquery.dbapi
   :members:
   :show-inheritance:
+
+
+DB-API Query-Parameter Syntax
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+The BigQuery DB-API uses the `qmark` `parameter style
+<https://www.python.org/dev/peps/pep-0249/#paramstyle>`_ for
+unnamed/positional parameters and the `pyformat` parameter style for
+named parameters.
+
+An example of a query using unnamed parameters::
+
+  insert into people (name, income) values (?, ?)
+
+and using named parameters::
+
+  insert into people (name, income) values (%(name)s, %(income)s)
+
+Providing explicit type information
+-----------------------------------
+
+BigQuery requires type information for parameters.  The The BigQuery
+DB-API can usually determine parameter types for parameters based on
+provided values.  Sometimes, however, types can't be determined (for
+example when `None` is passed) or are determined incorrectly (for
+example when passing a floating-point value to a numeric column).
+
+The BigQuery DB-API provides an extended parameter syntax.  For named
+parameters, a BigQuery type is provided after the name separated by a
+colon, as in::
+
+  insert into people (name, income) values (%(name:string)s, %(income:numeric)s)
+
+For unnamed parameters, use the named syntax with a type, but now
+name, as in::
+
+  insert into people (name, income) values (%(:string)s, %(:numeric)s)
diff --git a/google/cloud/bigquery/_helpers.py b/google/cloud/bigquery/_helpers.py
index daa14b92..ad8e3f00 100644
--- a/google/cloud/bigquery/_helpers.py
+++ b/google/cloud/bigquery/_helpers.py
@@ -275,7 +275,7 @@ def _int_to_json(value):
 
 def _float_to_json(value):
     """Coerce 'value' to an JSON-compatible representation."""
-    return value
+    return value if value is None else float(value)
 
 
 def _decimal_to_json(value):
diff --git a/google/cloud/bigquery/dbapi/_helpers.py b/google/cloud/bigquery/dbapi/_helpers.py
index beb3c5e7..3b0d8134 100644
--- a/google/cloud/bigquery/dbapi/_helpers.py
+++ b/google/cloud/bigquery/dbapi/_helpers.py
@@ -20,7 +20,7 @@ import functools
 import numbers
 
 from google.cloud import bigquery
-from google.cloud.bigquery import table
+from google.cloud.bigquery import table, enums
 from google.cloud.bigquery.dbapi import exceptions
 
 
@@ -28,7 +28,28 @@ _NUMERIC_SERVER_MIN = decimal.Decimal("-9.9999999999999999999999999999999999999E
 _NUMERIC_SERVER_MAX = decimal.Decimal("9.9999999999999999999999999999999999999E+28")
 
 
-def scalar_to_query_parameter(value, name=None):
+def _parameter_type(name, value, query_parameter_type=None, value_doc=""):
+    if query_parameter_type:
+        try:
+            parameter_type = getattr(
+                enums.SqlParameterScalarTypes, query_parameter_type.upper()
+            )._type
+        except AttributeError:
+            raise exceptions.ProgrammingError(
+                f"The given parameter type, {query_parameter_type},"
+                f" for {name} is not a valid BigQuery scalar type."
+            )
+    else:
+        parameter_type = bigquery_scalar_type(value)
+        if parameter_type is None:
+            raise exceptions.ProgrammingError(
+                f"Encountered parameter {name} with "
+                f"{value_doc} value {value} of unexpected type."
+            )
+    return parameter_type
+
+
+def scalar_to_query_parameter(value, name=None, query_parameter_type=None):
     """Convert a scalar value into a query parameter.
 
     Args:
@@ -37,6 +58,7 @@ def scalar_to_query_parameter(value, name=None):
 
         name (str):
             (Optional) Name of the query parameter.
+        query_parameter_type (Optional[str]): Given type for the parameter.
 
     Returns:
         google.cloud.bigquery.ScalarQueryParameter:
@@ -47,24 +69,19 @@ def scalar_to_query_parameter(value, name=None):
         google.cloud.bigquery.dbapi.exceptions.ProgrammingError:
             if the type cannot be determined.
     """
-    parameter_type = bigquery_scalar_type(value)
-
-    if parameter_type is None:
-        raise exceptions.ProgrammingError(
-            "encountered parameter {} with value {} of unexpected type".format(
-                name, value
-            )
-        )
-    return bigquery.ScalarQueryParameter(name, parameter_type, value)
+    return bigquery.ScalarQueryParameter(
+        name, _parameter_type(name, value, query_parameter_type), value
+    )
 
 
-def array_to_query_parameter(value, name=None):
+def array_to_query_parameter(value, name=None, query_parameter_type=None):
     """Convert an array-like value into a query parameter.
 
     Args:
         value (Sequence[Any]): The elements of the array (should not be a
             string-like Sequence).
         name (Optional[str]): Name of the query parameter.
+        query_parameter_type (Optional[str]): Given type for the parameter.
 
     Returns:
         A query parameter corresponding with the type and value of the plain
@@ -80,29 +97,30 @@ def array_to_query_parameter(value, name=None):
             "not string-like.".format(name)
         )
 
-    if not value:
+    if query_parameter_type or value:
+        array_type = _parameter_type(
+            name,
+            value[0] if value else None,
+            query_parameter_type,
+            value_doc="array element ",
+        )
+    else:
         raise exceptions.ProgrammingError(
             "Encountered an empty array-like value of parameter {}, cannot "
             "determine array elements type.".format(name)
         )
 
-    # Assume that all elements are of the same type, and let the backend handle
-    # any type incompatibilities among the array elements
-    array_type = bigquery_scalar_type(value[0])
-    if array_type is None:
-        raise exceptions.ProgrammingError(
-            "Encountered unexpected first array element of parameter {}, "
-            "cannot determine array elements type.".format(name)
-        )
-
     return bigquery.ArrayQueryParameter(name, array_type, value)
 
 
-def to_query_parameters_list(parameters):
+def to_query_parameters_list(parameters, parameter_types):
     """Converts a sequence of parameter values into query parameters.
 
     Args:
         parameters (Sequence[Any]): Sequence of query parameter values.
+        parameter_types:
+            A list of parameter types, one for each parameter.
+            Unknown types are provided as None.
 
     Returns:
         List[google.cloud.bigquery.query._AbstractQueryParameter]:
@@ -110,23 +128,27 @@ def to_query_parameters_list(parameters):
     """
     result = []
 
-    for value in parameters:
+    for value, type_ in zip(parameters, parameter_types):
         if isinstance(value, collections_abc.Mapping):
             raise NotImplementedError("STRUCT-like parameter values are not supported.")
         elif array_like(value):
-            param = array_to_query_parameter(value)
+            param = array_to_query_parameter(value, None, type_)
         else:
-            param = scalar_to_query_parameter(value)
+            param = scalar_to_query_parameter(value, None, type_)
+
         result.append(param)
 
     return result
 
 
-def to_query_parameters_dict(parameters):
+def to_query_parameters_dict(parameters, query_parameter_types):
     """Converts a dictionary of parameter values into query parameters.
 
     Args:
         parameters (Mapping[str, Any]): Dictionary of query parameter values.
+        parameter_types:
+            A dictionary of parameter types. It needn't have a key for each
+            parameter.
 
     Returns:
         List[google.cloud.bigquery.query._AbstractQueryParameter]:
@@ -140,21 +162,38 @@ def to_query_parameters_dict(parameters):
                 "STRUCT-like parameter values are not supported "
                 "(parameter {}).".format(name)
             )
-        elif array_like(value):
-            param = array_to_query_parameter(value, name=name)
         else:
-            param = scalar_to_query_parameter(value, name=name)
+            query_parameter_type = query_parameter_types.get(name)
+            if array_like(value):
+                param = array_to_query_parameter(
+                    value, name=name, query_parameter_type=query_parameter_type
+                )
+            else:
+                param = scalar_to_query_parameter(
+                    value, name=name, query_parameter_type=query_parameter_type,
+                )
+
         result.append(param)
 
     return result
 
 
-def to_query_parameters(parameters):
+def to_query_parameters(parameters, parameter_types):
     """Converts DB-API parameter values into query parameters.
 
     Args:
         parameters (Union[Mapping[str, Any], Sequence[Any]]):
             A dictionary or sequence of query parameter values.
+        parameter_types (Union[Mapping[str, str], Sequence[str]]):
+            A dictionary or list of parameter types.
+
+            If parameters is a mapping, then this must be a dictionary
+            of parameter types.  It needn't have a key for each
+            parameter.
+
+            If parameters is a sequence, then this must be a list of
+            parameter types, one for each paramater.  Unknown types
+            are provided as None.
 
     Returns:
         List[google.cloud.bigquery.query._AbstractQueryParameter]:
@@ -164,9 +203,9 @@ def to_query_parameters(parameters):
         return []
 
     if isinstance(parameters, collections_abc.Mapping):
-        return to_query_parameters_dict(parameters)
-
-    return to_query_parameters_list(parameters)
+        return to_query_parameters_dict(parameters, parameter_types)
+    else:
+        return to_query_parameters_list(parameters, parameter_types)
 
 
 def bigquery_scalar_type(value):
diff --git a/google/cloud/bigquery/dbapi/cursor.py b/google/cloud/bigquery/dbapi/cursor.py
index ca78d390..f74781df 100644
--- a/google/cloud/bigquery/dbapi/cursor.py
+++ b/google/cloud/bigquery/dbapi/cursor.py
@@ -18,6 +18,7 @@ import collections
 from collections import abc as collections_abc
 import copy
 import logging
+import re
 
 try:
     from google.cloud.bigquery_storage import ArrowSerializationOptions
@@ -161,6 +162,14 @@ class Cursor(object):
             job_config (google.cloud.bigquery.job.QueryJobConfig):
                 (Optional) Extra configuration options for the query job.
         """
+        formatted_operation, parameter_types = _format_operation(operation, parameters)
+        self._execute(
+            formatted_operation, parameters, job_id, job_config, parameter_types
+        )
+
+    def _execute(
+        self, formatted_operation, parameters, job_id, job_config, parameter_types
+    ):
         self._query_data = None
         self._query_job = None
         client = self.connection._client
@@ -169,8 +178,7 @@ class Cursor(object):
         # query parameters was not one of the standard options. Convert both
         # the query and the parameters to the format expected by the client
         # libraries.
-        formatted_operation = _format_operation(operation, parameters=parameters)
-        query_parameters = _helpers.to_query_parameters(parameters)
+        query_parameters = _helpers.to_query_parameters(parameters, parameter_types)
 
         if client._default_query_job_config:
             if job_config:
@@ -209,8 +217,19 @@ class Cursor(object):
             seq_of_parameters (Union[Sequence[Mapping[str, Any], Sequence[Any]]]):
                 Sequence of many sets of parameter values.
         """
-        for parameters in seq_of_parameters:
-            self.execute(operation, parameters)
+        if seq_of_parameters:
+            # There's no reason to format the line more than once, as
+            # the operation only barely depends on the parameters.  So
+            # we just use the first set of parameters. If there are
+            # different numbers or types of parameters, we'll error
+            # anyway.
+            formatted_operation, parameter_types = _format_operation(
+                operation, seq_of_parameters[0]
+            )
+            for parameters in seq_of_parameters:
+                self._execute(
+                    formatted_operation, parameters, None, None, parameter_types
+                )
 
     def _try_fetch(self, size=None):
         """Try to start fetching data, if not yet started.
@@ -427,7 +446,7 @@ def _format_operation_dict(operation, parameters):
         raise exceptions.ProgrammingError(exc)
 
 
-def _format_operation(operation, parameters=None):
+def _format_operation(operation, parameters):
     """Formats parameters in operation in way BigQuery expects.
 
     Args:
@@ -445,9 +464,67 @@ def _format_operation(operation, parameters=None):
             ``parameters`` argument.
     """
     if parameters is None or len(parameters) == 0:
-        return operation.replace("%%", "%")  # Still do percent de-escaping.
+        return operation.replace("%%", "%"), None  # Still do percent de-escaping.
+
+    operation, parameter_types = _extract_types(operation)
+    if parameter_types is None:
+        raise exceptions.ProgrammingError(
+            f"Parameters were provided, but {repr(operation)} has no placeholders."
+        )
 
     if isinstance(parameters, collections_abc.Mapping):
-        return _format_operation_dict(operation, parameters)
+        return _format_operation_dict(operation, parameters), parameter_types
+
+    return _format_operation_list(operation, parameters), parameter_types
+
+
+def _extract_types(
+    operation, extra_type_sub=re.compile(r"(%*)%(?:\(([^:)]*)(?::(\w+))?\))?s").sub
+):
+    """Remove type information from parameter placeholders.
+
+    For every parameter of the form %(name:type)s, replace with %(name)s and add the
+    item name->type to dict that's returned.
+
+    Returns operation without type information and a dictionary of names and types.
+    """
+    parameter_types = None
+
+    def repl(m):
+        nonlocal parameter_types
+        prefix, name, type_ = m.groups()
+        if len(prefix) % 2:
+            # The prefix has an odd number of %s, the last of which
+            # escapes the % we're looking for, so we don't want to
+            # change anything.
+            return m.group(0)
+
+        try:
+            if name:
+                if not parameter_types:
+                    parameter_types = {}
+                if type_:
+                    if name in parameter_types:
+                        if type_ != parameter_types[name]:
+                            raise exceptions.ProgrammingError(
+                                f"Conflicting types for {name}: "
+                                f"{parameter_types[name]} and {type_}."
+                            )
+                    else:
+                        parameter_types[name] = type_
+                else:
+                    if not isinstance(parameter_types, dict):
+                        raise TypeError()
+
+                return f"{prefix}%({name})s"
+            else:
+                if parameter_types is None:
+                    parameter_types = []
+                parameter_types.append(type_)
+                return f"{prefix}%s"
+        except (AttributeError, TypeError):
+            raise exceptions.ProgrammingError(
+                f"{repr(operation)} mixes named and unamed parameters."
+            )
 
-    return _format_operation_list(operation, parameters)
+    return extra_type_sub(repl, operation), parameter_types
diff --git a/google/cloud/bigquery/magics/magics.py b/google/cloud/bigquery/magics/magics.py
index 6ae7cae1..474d9a74 100644
--- a/google/cloud/bigquery/magics/magics.py
+++ b/google/cloud/bigquery/magics/magics.py
@@ -615,7 +615,7 @@ def _cell_magic(line, query):
             )
             raise NameError(msg)
 
-        params = _helpers.to_query_parameters(ast.literal_eval(params_option_value))
+        params = _helpers.to_query_parameters(ast.literal_eval(params_option_value), {})
 
     project = args.project or context.project
 
diff --git a/samples/geography/requirements.txt b/samples/geography/requirements.txt
index 7e017e28..f46b141f 100644
--- a/samples/geography/requirements.txt
+++ b/samples/geography/requirements.txt
@@ -1,4 +1,4 @@
 geojson==2.5.0
 google-cloud-bigquery==2.14.0
-google-cloud-bigquery-storage==2.1.0
+google-cloud-bigquery-storage==2.4.0
 Shapely==1.7.1
diff --git a/samples/snippets/requirements.txt b/samples/snippets/requirements.txt
index d7e60f77..f7b5cebe 100644
--- a/samples/snippets/requirements.txt
+++ b/samples/snippets/requirements.txt
@@ -1,5 +1,5 @@
 google-cloud-bigquery==2.14.0
-google-cloud-bigquery-storage==2.3.0
+google-cloud-bigquery-storage==2.4.0
 google-auth-oauthlib==0.4.4
 grpcio==1.37.0
 ipython==7.16.1; python_version < '3.7'


Test patch:
diff --git a/tests/unit/test__helpers.py b/tests/unit/test__helpers.py
index 0fdf1142..2437f356 100644
--- a/tests/unit/test__helpers.py
+++ b/tests/unit/test__helpers.py
@@ -1159,3 +1159,18 @@ def _field_isinstance_patcher():
         "google.cloud.bigquery.schema.isinstance", side_effect=fake_isinstance
     )
     return patcher
+
+
+def test_decimal_as_float_api_repr():
+    """Make sure decimals get converted to float."""
+    import google.cloud.bigquery.query
+    from decimal import Decimal
+
+    param = google.cloud.bigquery.query.ScalarQueryParameter(
+        "x", "FLOAT64", Decimal(42)
+    )
+    assert param.to_api_repr() == {
+        "parameterType": {"type": "FLOAT64"},
+        "parameterValue": {"value": 42.0},
+        "name": "x",
+    }
diff --git a/tests/unit/test_dbapi__helpers.py b/tests/unit/test_dbapi__helpers.py
index 4b2724de..250ba46d 100644
--- a/tests/unit/test_dbapi__helpers.py
+++ b/tests/unit/test_dbapi__helpers.py
@@ -18,13 +18,15 @@ import math
 import operator as op
 import unittest
 
+import pytest
+
 try:
     import pyarrow
 except ImportError:  # pragma: NO COVER
     pyarrow = None
 
 import google.cloud._helpers
-from google.cloud.bigquery import table
+from google.cloud.bigquery import table, enums
 from google.cloud.bigquery.dbapi import _helpers
 from google.cloud.bigquery.dbapi import exceptions
 from tests.unit.helpers import _to_pyarrow
@@ -142,7 +144,7 @@ class TestQueryParameters(unittest.TestCase):
 
     def test_to_query_parameters_w_dict(self):
         parameters = {"somebool": True, "somestring": "a-string-value"}
-        query_parameters = _helpers.to_query_parameters(parameters)
+        query_parameters = _helpers.to_query_parameters(parameters, {})
         query_parameter_tuples = []
         for param in query_parameters:
             query_parameter_tuples.append((param.name, param.type_, param.value))
@@ -158,7 +160,7 @@ class TestQueryParameters(unittest.TestCase):
 
     def test_to_query_parameters_w_dict_array_param(self):
         parameters = {"somelist": [10, 20]}
-        query_parameters = _helpers.to_query_parameters(parameters)
+        query_parameters = _helpers.to_query_parameters(parameters, {})
 
         self.assertEqual(len(query_parameters), 1)
         param = query_parameters[0]
@@ -171,11 +173,11 @@ class TestQueryParameters(unittest.TestCase):
         parameters = {"my_param": {"foo": "bar"}}
 
         with self.assertRaises(NotImplementedError):
-            _helpers.to_query_parameters(parameters)
+            _helpers.to_query_parameters(parameters, {})
 
     def test_to_query_parameters_w_list(self):
         parameters = [True, "a-string-value"]
-        query_parameters = _helpers.to_query_parameters(parameters)
+        query_parameters = _helpers.to_query_parameters(parameters, [None, None])
         query_parameter_tuples = []
         for param in query_parameters:
             query_parameter_tuples.append((param.name, param.type_, param.value))
@@ -186,7 +188,7 @@ class TestQueryParameters(unittest.TestCase):
 
     def test_to_query_parameters_w_list_array_param(self):
         parameters = [[10, 20]]
-        query_parameters = _helpers.to_query_parameters(parameters)
+        query_parameters = _helpers.to_query_parameters(parameters, [None])
 
         self.assertEqual(len(query_parameters), 1)
         param = query_parameters[0]
@@ -199,10 +201,10 @@ class TestQueryParameters(unittest.TestCase):
         parameters = [{"foo": "bar"}]
 
         with self.assertRaises(NotImplementedError):
-            _helpers.to_query_parameters(parameters)
+            _helpers.to_query_parameters(parameters, [None])
 
     def test_to_query_parameters_none_argument(self):
-        query_parameters = _helpers.to_query_parameters(None)
+        query_parameters = _helpers.to_query_parameters(None, None)
         self.assertEqual(query_parameters, [])
 
 
@@ -338,3 +340,79 @@ class TestRaiseOnClosedDecorator(unittest.TestCase):
 
         with self.assertRaisesRegex(RuntimeError, "I'm closed!"):
             instance.instance_method()
+
+
+VALID_BQ_TYPES = [
+    (name, getattr(enums.SqlParameterScalarTypes, name)._type)
+    for name in dir(enums.SqlParameterScalarTypes)
+    if not name.startswith("_")
+]
+
+
+@pytest.mark.parametrize("alias, type_", VALID_BQ_TYPES)
+def test_scalar_to_query_parameter_honors_given_type(alias, type_):
+    from google.cloud import bigquery
+
+    assert _helpers.scalar_to_query_parameter(1.23, None, alias) == (
+        bigquery.ScalarQueryParameter(None, type_, 1.23)
+    )
+    assert _helpers.scalar_to_query_parameter(None, "foo", alias) == (
+        bigquery.ScalarQueryParameter("foo", type_, None)
+    )
+
+
+def test_scalar_to_query_parameter_honors_given_type_errors_on_invalid():
+    with pytest.raises(
+        google.cloud.bigquery.dbapi.exceptions.ProgrammingError,
+        match="The given parameter type, INT, for foo is not a valid BigQuery scalar type.",
+    ):
+        _helpers.scalar_to_query_parameter(None, "foo", "INT")
+
+
+@pytest.mark.parametrize("alias, type_", VALID_BQ_TYPES)
+def test_array_to_query_parameter_honors_given_type(alias, type_):
+    from google.cloud import bigquery
+
+    assert _helpers.array_to_query_parameter([1.23], None, alias) == (
+        bigquery.ArrayQueryParameter(None, type_, [1.23])
+    )
+    assert _helpers.array_to_query_parameter((), "foo", alias) == (
+        bigquery.ArrayQueryParameter("foo", type_, ())
+    )
+
+
+def test_array_to_query_parameter_honors_given_type_errors_on_invalid():
+    with pytest.raises(
+        google.cloud.bigquery.dbapi.exceptions.ProgrammingError,
+        match="The given parameter type, INT, for foo is not a valid BigQuery scalar type.",
+    ):
+        _helpers.array_to_query_parameter((), "foo", "INT")
+
+
+def test_to_query_parameters_dict_w_types():
+    from google.cloud import bigquery
+
+    assert sorted(
+        _helpers.to_query_parameters(
+            dict(i=1, x=1.2, y=None, z=[]), dict(x="numeric", y="string", z="float64")
+        ),
+        key=lambda p: p.name,
+    ) == [
+        bigquery.ScalarQueryParameter("i", "INT64", 1),
+        bigquery.ScalarQueryParameter("x", "NUMERIC", 1.2),
+        bigquery.ScalarQueryParameter("y", "STRING", None),
+        bigquery.ArrayQueryParameter("z", "FLOAT64", []),
+    ]
+
+
+def test_to_query_parameters_list_w_types():
+    from google.cloud import bigquery
+
+    assert _helpers.to_query_parameters(
+        [1, 1.2, None, []], [None, "numeric", "string", "float64"]
+    ) == [
+        bigquery.ScalarQueryParameter(None, "INT64", 1),
+        bigquery.ScalarQueryParameter(None, "NUMERIC", 1.2),
+        bigquery.ScalarQueryParameter(None, "STRING", None),
+        bigquery.ArrayQueryParameter(None, "FLOAT64", []),
+    ]
diff --git a/tests/unit/test_dbapi_cursor.py b/tests/unit/test_dbapi_cursor.py
index 039ef3b4..5afe269e 100644
--- a/tests/unit/test_dbapi_cursor.py
+++ b/tests/unit/test_dbapi_cursor.py
@@ -12,10 +12,12 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
+import mock
 import operator as op
 import unittest
 
-import mock
+import pytest
+
 
 try:
     import pyarrow
@@ -612,6 +614,15 @@ class TestCursor(unittest.TestCase):
         self.assertIsNone(cursor.description)
         self.assertEqual(cursor.rowcount, 12)
 
+    def test_executemany_empty(self):
+        from google.cloud.bigquery.dbapi import connect
+
+        connection = connect(self._mock_client(rows=[], num_dml_affected_rows=12))
+        cursor = connection.cursor()
+        cursor.executemany((), ())
+        self.assertIsNone(cursor.description)
+        self.assertEqual(cursor.rowcount, -1)
+
     def test_is_iterable(self):
         from google.cloud.bigquery import dbapi
 
@@ -638,13 +649,15 @@ class TestCursor(unittest.TestCase):
     def test__format_operation_w_dict(self):
         from google.cloud.bigquery.dbapi import cursor
 
-        formatted_operation = cursor._format_operation(
-            "SELECT %(somevalue)s, %(a `weird` one)s;",
+        parameter_types = {}
+        formatted_operation, parameter_types = cursor._format_operation(
+            "SELECT %(somevalue)s, %(a `weird` one:STRING)s;",
             {"somevalue": "hi", "a `weird` one": "world"},
         )
         self.assertEqual(
             formatted_operation, "SELECT @`somevalue`, @`a \\`weird\\` one`;"
         )
+        self.assertEqual(parameter_types, {"a `weird` one": "STRING"})
 
     def test__format_operation_w_wrong_dict(self):
         from google.cloud.bigquery import dbapi
@@ -660,7 +673,7 @@ class TestCursor(unittest.TestCase):
     def test__format_operation_w_redundant_dict_key(self):
         from google.cloud.bigquery.dbapi import cursor
 
-        formatted_operation = cursor._format_operation(
+        formatted_operation, _ = cursor._format_operation(
             "SELECT %(somevalue)s;", {"somevalue": "foo", "value-not-used": "bar"}
         )
         self.assertEqual(formatted_operation, "SELECT @`somevalue`;")
@@ -668,7 +681,7 @@ class TestCursor(unittest.TestCase):
     def test__format_operation_w_sequence(self):
         from google.cloud.bigquery.dbapi import cursor
 
-        formatted_operation = cursor._format_operation(
+        formatted_operation, _ = cursor._format_operation(
             "SELECT %s, %s;", ("hello", "world")
         )
         self.assertEqual(formatted_operation, "SELECT ?, ?;")
@@ -698,19 +711,19 @@ class TestCursor(unittest.TestCase):
     def test__format_operation_w_empty_dict(self):
         from google.cloud.bigquery.dbapi import cursor
 
-        formatted_operation = cursor._format_operation("SELECT '%f'", {})
+        formatted_operation, _ = cursor._format_operation("SELECT '%f'", {})
         self.assertEqual(formatted_operation, "SELECT '%f'")
 
     def test__format_operation_wo_params_single_percent(self):
         from google.cloud.bigquery.dbapi import cursor
 
-        formatted_operation = cursor._format_operation("SELECT '%'", {})
+        formatted_operation, _ = cursor._format_operation("SELECT '%'", {})
         self.assertEqual(formatted_operation, "SELECT '%'")
 
     def test__format_operation_wo_params_double_percents(self):
         from google.cloud.bigquery.dbapi import cursor
 
-        formatted_operation = cursor._format_operation("SELECT '%%'", {})
+        formatted_operation, _ = cursor._format_operation("SELECT '%%'", {})
         self.assertEqual(formatted_operation, "SELECT '%'")
 
     def test__format_operation_unescaped_percent_w_dict_param(self):
@@ -734,3 +747,80 @@ class TestCursor(unittest.TestCase):
             "SELECT %s, %s, '100 %';",
             ["foo", "bar"],
         )
+
+    def test__format_operation_no_placeholders(self):
+        from google.cloud.bigquery import dbapi
+        from google.cloud.bigquery.dbapi import cursor
+
+        self.assertRaises(
+            dbapi.ProgrammingError,
+            cursor._format_operation,
+            "SELECT 42",
+            ["foo", "bar"],
+        )
+
+
+@pytest.mark.parametrize(
+    "inp,expect",
+    [
+        ("", ("", None)),
+        ("values(%(foo)s, %(bar)s)", ("values(%(foo)s, %(bar)s)", {})),
+        (
+            "values('%%(oof:INT64)s', %(foo)s, %(bar)s)",
+            ("values('%%(oof:INT64)s', %(foo)s, %(bar)s)", {}),
+        ),
+        (
+            "values(%(foo:INT64)s, %(bar)s)",
+            ("values(%(foo)s, %(bar)s)", dict(foo="INT64")),
+        ),
+        (
+            "values('%%(oof:INT64)s, %(foo:INT64)s, %(foo)s)",
+            ("values('%%(oof:INT64)s, %(foo)s, %(foo)s)", dict(foo="INT64")),
+        ),
+        (
+            "values(%(foo:INT64)s, %(foo:INT64)s)",
+            ("values(%(foo)s, %(foo)s)", dict(foo="INT64")),
+        ),
+        (
+            "values(%(foo:INT64)s, %(bar:NUMERIC)s) 100 %",
+            ("values(%(foo)s, %(bar)s) 100 %", dict(foo="INT64", bar="NUMERIC")),
+        ),
+        (" %s %()s %(:int64)s ", (" %s %s %s ", [None, None, "int64"])),
+        (" %%s %s %()s %(:int64)s ", (" %%s %s %s %s ", [None, None, "int64"])),
+        (
+            "values(%%%(foo:INT64)s, %(bar)s)",
+            ("values(%%%(foo)s, %(bar)s)", dict(foo="INT64")),
+        ),
+        (
+            "values(%%%%(foo:INT64)s, %(bar)s)",
+            ("values(%%%%(foo:INT64)s, %(bar)s)", dict()),
+        ),
+        (
+            "values(%%%%%(foo:INT64)s, %(bar)s)",
+            ("values(%%%%%(foo)s, %(bar)s)", dict(foo="INT64")),
+        ),
+    ],
+)
+def test__extract_types(inp, expect):
+    from google.cloud.bigquery.dbapi.cursor import _extract_types as et
+
+    assert et(inp) == expect
+
+
+@pytest.mark.parametrize(
+    "match,inp",
+    [
+        (
+            "Conflicting types for foo: numeric and int64.",
+            " %(foo:numeric)s %(foo:int64)s ",
+        ),
+        (r"' %s %\(foo\)s ' mixes named and unamed parameters.", " %s %(foo)s "),
+        (r"' %\(foo\)s %s ' mixes named and unamed parameters.", " %(foo)s %s "),
+    ],
+)
+def test__extract_types_fail(match, inp):
+    from google.cloud.bigquery.dbapi.cursor import _extract_types as et
+    from google.cloud.bigquery.dbapi import exceptions
+
+    with pytest.raises(exceptions.ProgrammingError, match=match):
+        et(inp)