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

refactor(db_api): remove parsing of insert statements #630

Closed
larkee wants to merge 13 commits into from
Closed

refactor(db_api): remove parsing of insert statements #630

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
433ba90
perf: use batch_update instead of sending a separate request for each…
larkee Sep 15, 2021
85927c9
fix: check status for errors
larkee Sep 16, 2021
007f8b1
test: add and update tests
larkee Sep 16, 2021
068289a
refactor: remove redundant mutations code
larkee Sep 16, 2021
f6125c5
perf: generate fewer insert statements for execute calls
larkee Oct 15, 2021
7a7aae2
perf: generate fewer insert statements for executemany calls
larkee Oct 15, 2021
501a625
test: fix executemany insert tests
larkee Oct 15, 2021
e677526
fix: handle aborted errors for executemany inserts
larkee Nov 1, 2021
a377683
style: fix lint
larkee Nov 3, 2021
02bca0c
refactor: stop adding WHERE clause
larkee Nov 10, 2021
307f60c
refactor: remove unnecessary parsing
larkee Nov 11, 2021
8d771a1
refactor: revert executemany changes
larkee Nov 15, 2021
6d479c0
refactor: revert where clause removal
larkee Nov 16, 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
46 changes: 6 additions & 40 deletions google/cloud/spanner_dbapi/_helpers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,6 @@
# limitations under the License.

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

Expand Down Expand Up @@ -55,48 +54,15 @@
}


def _execute_insert_heterogenous(transaction, sql_params_list):
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)


def _execute_insert_homogenous(transaction, parts):
# Perform an insert in one shot.
table = parts.get("table")
columns = parts.get("columns")
values = parts.get("values")
return transaction.insert(table, columns, values)
def _execute_insert_heterogenous(transaction, sql, params):
sql, params = sql_pyformat_args_to_spanner(sql, params)
transaction.execute_update(sql, params, get_param_types(params))


def handle_insert(connection, sql, params):
parts = parse_insert(sql, params)

# The split between the two styles exists because:
# in the common case of multiple values being passed
# with simple pyformat arguments,
# SQL: INSERT INTO T (f1, f2) VALUES (%s, %s, %s)
# Params: [(1, 2, 3, 4, 5, 6, 7, 8, 9, 10,)]
# we can take advantage of a single RPC with:
# transaction.insert(table, columns, values)
# instead of invoking:
# with transaction:
# for sql, params in sql_params_list:
# transaction.execute_sql(sql, params, param_types)
# which invokes more RPCs and is more costly.

if parts.get("homogenous"):
# The common case of multiple values being passed in
# non-complex pyformat args and need to be uploaded in one RPC.
return connection.database.run_in_transaction(_execute_insert_homogenous, parts)
else:
# All the other cases that are esoteric and need
# transaction.execute_sql
sql_params_list = parts.get("sql_params_list")
return connection.database.run_in_transaction(
_execute_insert_heterogenous, sql_params_list
)
return connection.database.run_in_transaction(
_execute_insert_heterogenous, sql, params
)


class ColumnInfo:
Expand Down
25 changes: 7 additions & 18 deletions google/cloud/spanner_dbapi/connection.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,6 @@
from google.cloud.spanner_v1.snapshot import Snapshot

from google.cloud.spanner_dbapi._helpers import _execute_insert_heterogenous
from google.cloud.spanner_dbapi._helpers import _execute_insert_homogenous
from google.cloud.spanner_dbapi._helpers import parse_insert
from google.cloud.spanner_dbapi.checksum import _compare_checksums
from google.cloud.spanner_dbapi.checksum import ResultsChecksum
from google.cloud.spanner_dbapi.cursor import Cursor
Expand Down Expand Up @@ -430,22 +428,13 @@ def run_statement(self, statement, retried=False):
self._statements.append(statement)

if statement.is_insert:

@olavloite olavloite Nov 15, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we need this if statement at all any more? Or can it be merged with the default case for any other update statement?

parts = parse_insert(statement.sql, statement.params)

if parts.get("homogenous"):
_execute_insert_homogenous(transaction, parts)
return (
iter(()),
ResultsChecksum() if retried else statement.checksum,
)
else:
_execute_insert_heterogenous(
transaction, parts.get("sql_params_list"),
)
return (
iter(()),
ResultsChecksum() if retried else statement.checksum,
)
_execute_insert_heterogenous(
transaction, statement.sql, statement.params,
)
return (
iter(()),
ResultsChecksum() if retried else statement.checksum,
)

return (
transaction.execute_sql(
Expand Down
6 changes: 1 addition & 5 deletions google/cloud/spanner_dbapi/cursor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -208,11 +208,7 @@ def execute(self, sql, args=None):
self.connection.run_prior_DDL_statements()

if not self.connection.autocommit:
if classification == parse_utils.STMT_UPDATING:
sql = parse_utils.ensure_where_clause(sql)

if classification != parse_utils.STMT_INSERT:
sql, args = sql_pyformat_args_to_spanner(sql, args or None)
sql, args = sql_pyformat_args_to_spanner(sql, args or None)

statement = Statement(
sql,
Expand Down
254 changes: 2 additions & 252 deletions google/cloud/spanner_dbapi/parse_utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,14 +17,12 @@
import datetime
import decimal
import re
from functools import reduce

import sqlparse
from google.cloud import spanner_v1 as spanner
from google.cloud.spanner_v1 import JsonObject

from .exceptions import Error, ProgrammingError
from .parser import parse_values
from .types import DateStr, TimestampStr
from .utils import sanitize_literals_for_upload

Expand Down Expand Up @@ -185,6 +183,8 @@ def classify_stmt(query):
:rtype: str
:returns: The query type name.
"""
query = sqlparse.format(query, strip_comments=True).strip()

@olavloite olavloite Nov 15, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As this is an external library, it might be good to add a comment here that it has been verified that this library supports exactly the types of comments that Cloud Spanner supports, and no other comment types. That is, the library will strip the following types of comments (from what I can see in the source code of the library this holds true):

  • # single line comment starting with a dash
  • -- single line comment starting with two hyphens
  • `/* multi line comment */

And it does not support for example the PostgreSQL style dollar quoted string constants (https://www.postgresql.org/docs/current/sql-syntax-lexical.html) (or if it does, document that the parsing would fail if a SQL string contains a comment inside what would be interpreted as a dollar quoted string)


if RE_DDL.match(query):
return STMT_DDL

Expand All @@ -199,256 +199,6 @@ def classify_stmt(query):
return STMT_UPDATING


def parse_insert(insert_sql, params):
"""
Parse an INSERT statement and generate a list of tuples of the form:
[
(SQL, params_per_row1),
(SQL, params_per_row2),
(SQL, params_per_row3),
...
]

There are 4 variants of an INSERT statement:
a) INSERT INTO <table> (columns...) VALUES (<inlined values>): no params
b) INSERT INTO <table> (columns...) SELECT_STMT: no params
c) INSERT INTO <table> (columns...) VALUES (%s,...): with params
d) INSERT INTO <table> (columns...) VALUES (%s,..<EXPR>...) with params and expressions

Thus given each of the forms, it will produce a dictionary describing
how to upload the contents to Cloud Spanner:
Case a)
SQL: INSERT INTO T (f1, f2) VALUES (1, 2)
it produces:
{
'sql_params_list': [
('INSERT INTO T (f1, f2) VALUES (1, 2)', None),
],
}

Case b)
SQL: 'INSERT INTO T (s, c) SELECT st, zc FROM cus WHERE col IN (%s, %s)',
it produces:
{
'sql_params_list': [
('INSERT INTO T (s, c) SELECT st, zc FROM cus ORDER BY fn, ln', ('a', 'b')),
]
}

Case c)
SQL: INSERT INTO T (f1, f2) VALUES (%s, %s), (%s, %s)
Params: ['a', 'b', 'c', 'd']
it produces:
{
'sql_params_list': [
('INSERT INTO T (f1, f2) VALUES (%s, %s)', ('a', 'b')),
('INSERT INTO T (f1, f2) VALUES (%s, %s)', ('c', 'd'))
],
}

Case d)
SQL: INSERT INTO T (f1, f2) VALUES (%s, LOWER(%s)), (UPPER(%s), %s)
Params: ['a', 'b', 'c', 'd']
it produces:
{
'sql_params_list': [
('INSERT INTO T (f1, f2) VALUES (%s, LOWER(%s))', ('a', 'b',)),
('INSERT INTO T (f1, f2) VALUES (UPPER(%s), %s)', ('c', 'd',))
],
}

:type insert_sql: str
:param insert_sql: A SQL insert request.

:type params: list
:param params: A list of parameters.

:rtype: dict
:returns: A dictionary that maps `sql_params_list` to the list of
parameters in cases a), b), d) or the dictionary with information
about the resulting table in case c).
""" # noqa
match = RE_INSERT.search(insert_sql)

if not match:
raise ProgrammingError(
"Could not parse an INSERT statement from %s" % insert_sql
)

after_values_sql = RE_VALUES_TILL_END.findall(insert_sql)
if not after_values_sql:
# Case b)
insert_sql = sanitize_literals_for_upload(insert_sql)
return {"sql_params_list": [(insert_sql, params)]}

if not params:
# Case a) perhaps?
# Check if any %s exists.

# pyformat_str_count = after_values_sql.count("%s")
# if pyformat_str_count > 0:
# raise ProgrammingError(
# 'no params yet there are %d "%%s" tokens' % pyformat_str_count
# )
for item in after_values_sql:
if item.count("%s") > 0:
raise ProgrammingError(
'no params yet there are %d "%%s" tokens' % item.count("%s")
)

insert_sql = sanitize_literals_for_upload(insert_sql)
# Confirmed case of:
# SQL: INSERT INTO T (a1, a2) VALUES (1, 2)
# Params: None
return {"sql_params_list": [(insert_sql, None)]}

values_str = after_values_sql[0]
_, values = parse_values(values_str)

if values.homogenous():
# Case c)

columns = [mi.strip(" `") for mi in match.group("columns").split(",")]
sql_params_list = []
insert_sql_preamble = "INSERT INTO %s (%s) VALUES %s" % (
match.group("table_name"),
match.group("columns"),
values.argv[0],
)
values_pyformat = [str(arg) for arg in values.argv]
rows_list = rows_for_insert_or_update(columns, params, values_pyformat)
insert_sql_preamble = sanitize_literals_for_upload(insert_sql_preamble)
for row in rows_list:
sql_params_list.append((insert_sql_preamble, row))

return {"sql_params_list": sql_params_list}

# Case d)
# insert_sql is of the form:
# INSERT INTO T(c1, c2) VALUES (%s, %s), (%s, LOWER(%s))

# Sanity check:
# length(all_args) == len(params)
args_len = reduce(lambda a, b: a + b, [len(arg) for arg in values.argv])
if args_len != len(params):
raise ProgrammingError(
"Invalid length: VALUES(...) len: %d != len(params): %d"
% (args_len, len(params))
)

trim_index = insert_sql.find(values_str)
before_values_sql = insert_sql[:trim_index]

sql_param_tuples = []
for token_arg in values.argv:
row_sql = before_values_sql + " VALUES%s" % token_arg
row_sql = sanitize_literals_for_upload(row_sql)
row_params, params = (
tuple(params[0 : len(token_arg)]),
params[len(token_arg) :],
)
sql_param_tuples.append((row_sql, row_params))

return {"sql_params_list": sql_param_tuples}


def rows_for_insert_or_update(columns, params, pyformat_args=None):
"""
Create a tupled list of params to be used as a single value per
value that inserted from a statement such as
SQL: 'INSERT INTO t (f1, f2, f3) VALUES (%s, %s, %s), (%s, %s, %s), (%s, %s, %s)'
Params A: [(1, 2, 3), (4, 5, 6), (7, 8, 9)]
Params B: [1, 2, 3, 4, 5, 6, 7, 8, 9]

We'll have to convert both params types into:
Params: [(1, 2, 3,), (4, 5, 6,), (7, 8, 9,)]

:type columns: list
:param columns: A list of the columns of the table.

:type params: list
:param params: A list of parameters.

:rtype: list
:returns: A properly restructured list of the parameters.
""" # noqa
if not pyformat_args:
# This is the case where we have for example:
# SQL: 'INSERT INTO t (f1, f2, f3)'
# Params A: [(1, 2, 3), (4, 5, 6), (7, 8, 9)]
# Params B: [1, 2, 3, 4, 5, 6, 7, 8, 9]
#
# We'll have to convert both params types into:
# [(1, 2, 3,), (4, 5, 6,), (7, 8, 9,)]
contains_all_list_or_tuples = True
for param in params:
if not (isinstance(param, list) or isinstance(param, tuple)):
contains_all_list_or_tuples = False
break

if contains_all_list_or_tuples:
# The case with Params A: [(1, 2, 3), (4, 5, 6)]
# Ensure that each param's length == len(columns)
columns_len = len(columns)
for param in params:
if columns_len != len(param):
raise Error(
"\nlen(`%s`)=%d\n!=\ncolum_len(`%s`)=%d"
% (param, len(param), columns, columns_len)
)
return params
else:
# The case with Params B: [1, 2, 3]
# Insert statements' params are only passed as tuples or lists,
# yet for do_execute_update, we've got to pass in list of list.
# https://googleapis.dev/python/spanner/latest/transaction-api.html\
# #google.cloud.spanner_v1.transaction.Transaction.insert
n_stride = len(columns)
else:
# This is the case where we have for example:
# SQL: 'INSERT INTO t (f1, f2, f3) VALUES (%s, %s, %s),
# (%s, %s, %s), (%s, %s, %s)'
# Params: [1, 2, 3, 4, 5, 6, 7, 8, 9]
# which should become
# Columns: (f1, f2, f3)
# new_params: [(1, 2, 3,), (4, 5, 6,), (7, 8, 9,)]

# Sanity check 1: all the pyformat_values should have the exact same
# length.
first, rest = pyformat_args[0], pyformat_args[1:]
n_stride = first.count("%s")
for pyfmt_value in rest:
n = pyfmt_value.count("%s")
if n_stride != n:
raise Error(
"\nlen(`%s`)=%d\n!=\nlen(`%s`)=%d"
% (first, n_stride, pyfmt_value, n)
)

# Sanity check 2: len(params) MUST be a multiple of n_stride aka
# len(count of %s).
# so that we can properly group for example:
# Given pyformat args:
# (%s, %s, %s)
# Params:
# [1, 2, 3, 4, 5, 6, 7, 8, 9]
# into
# [(1, 2, 3), (4, 5, 6), (7, 8, 9)]
if (len(params) % n_stride) != 0:
raise ProgrammingError(
"Invalid length: len(params)=%d MUST be a multiple of "
"len(pyformat_args)=%d" % (len(params), n_stride)
)

# Now chop up the strides.
strides = []
for step in range(0, len(params), n_stride):
stride = tuple(params[step : step + n_stride :])
strides.append(stride)

return strides


def sql_pyformat_args_to_spanner(sql, params):
"""
Transform pyformat set SQL to named arguments for Cloud Spanner.
Expand Down
Loading