Skip to content
This repository was archived by the owner on Jun 8, 2026. It is now read-only.

perf: use batch_update instead of sending a request for each statement #572

Closed
larkee wants to merge 6 commits into from
Closed

perf: use batch_update instead of sending a request for each statement #572

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
ef18bbb
perf: use batch_update instead of sending a separate request for each…
larkee Sep 15, 2021
3b62eef
fix: check status for errors
larkee Sep 16, 2021
858fb4a
test: add and update tests
larkee Sep 16, 2021
35adea9
🦉 Updates from OwlBot
gcf-owl-bot[bot] Oct 13, 2021
69bb7f7
Merge branch 'main' into dbapi-use-batch
larkee Oct 13, 2021
9c6f6bd
Merge branch 'main' into dbapi-use-batch
larkee Oct 21, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 7 additions & 3 deletions google/cloud/spanner_dbapi/_helpers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,12 +11,12 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from google.cloud.spanner_dbapi.exceptions import OperationalError
from google.cloud.spanner_dbapi.parse_utils import get_param_types
from google.cloud.spanner_dbapi.parse_utils import parse_insert
from google.cloud.spanner_dbapi.parse_utils import sql_pyformat_args_to_spanner
from google.cloud.spanner_v1 import param_types

from google.rpc.code_pb2 import OK

SQL_LIST_TABLES = """
SELECT
Expand Down Expand Up @@ -56,10 +56,14 @@


def _execute_insert_heterogenous(transaction, sql_params_list):
statements = []
for sql, params in sql_params_list:
sql, params = sql_pyformat_args_to_spanner(sql, params)
param_types = get_param_types(params)
transaction.execute_update(sql, params=params, param_types=param_types)
statements.append((sql, params, param_types))
status, _ = transaction.batch_update(statements)
if status.code != OK:
raise OperationalError(status.message)


def _execute_insert_homogenous(transaction, parts):
Expand Down
37 changes: 35 additions & 2 deletions tests/unit/spanner_dbapi/test__helpers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@
class TestHelpers(unittest.TestCase):
def test__execute_insert_heterogenous(self):
from google.cloud.spanner_dbapi import _helpers
from google.rpc.status_pb2 import Status
from google.rpc.code_pb2 import OK

sql = "sql"
params = (sql, None)
Expand All @@ -32,12 +34,43 @@ def test__execute_insert_heterogenous(self):
"google.cloud.spanner_dbapi._helpers.get_param_types", return_value=None
) as mock_param_types:
transaction = mock.MagicMock()
transaction.execute_update = mock_execute = mock.MagicMock()
status = Status(code=OK)
transaction.batch_update = mock_batch = mock.MagicMock(
return_value=(status, 1)
)
_helpers._execute_insert_heterogenous(transaction, [params])

mock_pyformat.assert_called_once_with(params[0], params[1])
mock_param_types.assert_called_once_with(None)
mock_execute.assert_called_once_with(sql, params=None, param_types=None)
mock_batch.assert_called_once_with([(sql, None, None)])

def test__execute_insert_heterogenous_error(self):
from google.cloud.spanner_dbapi import _helpers
from google.cloud.spanner_dbapi import OperationalError
from google.rpc.status_pb2 import Status
from google.rpc.code_pb2 import UNKNOWN

sql = "sql"
params = (sql, None)
with mock.patch(
"google.cloud.spanner_dbapi._helpers.sql_pyformat_args_to_spanner",
return_value=params,
) as mock_pyformat:
with mock.patch(
"google.cloud.spanner_dbapi._helpers.get_param_types", return_value=None
) as mock_param_types:
transaction = mock.MagicMock()
status = Status(code=UNKNOWN)
transaction.batch_update = mock_batch = mock.MagicMock(
return_value=(status, 0)
)

with self.assertRaises(OperationalError):
_helpers._execute_insert_heterogenous(transaction, [params])

mock_pyformat.assert_called_once_with(params[0], params[1])
mock_param_types.assert_called_once_with(None)
mock_batch.assert_called_once_with([(sql, None, None)])

def test__execute_insert_homogenous(self):
from google.cloud.spanner_dbapi import _helpers
Expand Down
12 changes: 10 additions & 2 deletions tests/unit/spanner_dbapi/test_connection.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -387,13 +387,17 @@ def test_run_statement_w_heterogenous_insert_statements(self):
"""Check that Connection executed heterogenous insert statements."""
from google.cloud.spanner_dbapi.checksum import ResultsChecksum
from google.cloud.spanner_dbapi.cursor import Statement
from google.rpc.status_pb2 import Status
from google.rpc.code_pb2 import OK

sql = "INSERT INTO T (f1, f2) VALUES (1, 2)"
params = None
param_types = None

connection = self._make_connection()
connection.transaction_checkout = mock.Mock()
transaction = mock.MagicMock()
connection.transaction_checkout = mock.Mock(return_value=transaction)
transaction.batch_update = mock.Mock(return_value=(Status(code=OK), 1))
statement = Statement(sql, params, param_types, ResultsChecksum(), True)

connection.run_statement(statement, retried=True)
Expand All @@ -404,13 +408,17 @@ def test_run_statement_w_homogeneous_insert_statements(self):
"""Check that Connection executed homogeneous insert statements."""
from google.cloud.spanner_dbapi.checksum import ResultsChecksum
from google.cloud.spanner_dbapi.cursor import Statement
from google.rpc.status_pb2 import Status
from google.rpc.code_pb2 import OK

sql = "INSERT INTO T (f1, f2) VALUES (%s, %s), (%s, %s)"
params = ["a", "b", "c", "d"]
param_types = {"f1": str, "f2": str}

connection = self._make_connection()
connection.transaction_checkout = mock.Mock()
transaction = mock.MagicMock()
connection.transaction_checkout = mock.Mock(return_value=transaction)
transaction.batch_update = mock.Mock(return_value=(Status(code=OK), 1))
statement = Statement(sql, params, param_types, ResultsChecksum(), True)

connection.run_statement(statement, retried=True)
Expand Down