Visitar URL original
dbapi: use a single transaction for the lifetime of a Cursor · googleapis/google-cloud-python@5bd6a1b · GitHub
Skip to content

Commit 5bd6a1b

Browse files
odeke-emodeke-em
committed
dbapi: use a single transaction for the lifetime of a Cursor
* Changed INSERT statements to ALWAYS invoke transaction.execute_update instead of transaction.insert for homogenous insertions i.e. 'INSERT INTO T (a, b) VALUES (%s, %s), (%s, %s)', [1, 2, 3, 4] which would get transformed into transaction.insert('T', ['a', 'b'], [(1, 2), (3, 4)]) but unfortunately the results would be batched until .commit() was invoked. * Also while here, added a method to list tables because in the same held transaction, Cloud Spanner complains about issues with concurrency when trying to list tables i.e. 400 Unsupported concurrency mode in query using INFORMATION_SCHEMA. * Caveat: The long standing caveat is that after grabbing a transaction, if a schema update is performed e.g. a CREATE statement, the transaction is aborted with: ```shell Traceback (most recent call last): File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/google/api_core/grpc_helpers.py", line 57, in error_remapped_callable return callable_(*args, **kwargs) File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/grpc/_channel.py", line 565, in __call__ return _end_unary_response_blocking(state, call, False, None) File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/grpc/_channel.py", line 467, in _end_unary_response_blocking raise _Rendezvous(state, None, None, deadline) grpc._channel._Rendezvous: <_Rendezvous of RPC that terminated with: status = StatusCode.ABORTED details = "Transaction aborted. Database schema probably changed during transaction, retry may succeed." debug_error_string = "{"created":"@1579980002.811673000","description":"Error received from peer ipv4:216.58.223.106:443","file":"src/core/lib/surface/call.cc","file_line":1046,"grpc_message":"Transaction aborted. Database schema probably changed during transaction, retry may succeed.","grpc_status":10}" > ``` Requires review from the Cloud Spanner team to figure out how to deal with changing schema statements. Updates #274
1 parent 4b57443 commit 5bd6a1b

6 files changed

Lines changed: 152 additions & 100 deletions

File tree

‎packages/django-google-spanner/spanner/dbapi/connection.py‎

Lines changed: 58 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -5,11 +5,15 @@
55
# https://developers.google.com/open-source/licenses/bsd
66

77
from .cursor import Cursor
8-
from .exceptions import Error
8+
from .exceptions import Error, ProgrammingError
99

1010

1111
class Connection(object):
1212
def __init__(self, db_handle):
13+
sess = db_handle.session()
14+
if not sess.exists():
15+
sess.create()
16+
self.__sess = sess
1317
self.__dbhandle = db_handle
1418
self.__closed = False
1519
self.__ddl_statements = []
@@ -23,12 +27,16 @@ def __raise_if_already_closed(self):
2327

2428
def close(self):
2529
self.rollback()
26-
self.__dbhandle = None
30+
self.__clear()
2731
self.__closed = True
2832

2933
def __enter__(self):
3034
return self
3135

36+
def __clear(self):
37+
self.__dbhandle = None
38+
self.__sess.delete()
39+
3240
def __exit__(self, etype, value, traceback):
3341
self.commit()
3442
self.close()
@@ -38,15 +46,24 @@ def commit(self):
3846

3947
self.run_prior_DDL_statements()
4048

49+
if not self.__txn:
50+
raise ProgrammingError('attempting to invoke commit on a non-existent transaction')
51+
52+
if not self.__txn.committed:
53+
self.__txn.commit()
54+
self.__txn = None
55+
4156
def rollback(self):
4257
self.__raise_if_already_closed()
4358

44-
# TODO: to be added.
45-
59+
return self.__txn.rollback()
60+
4661
def cursor(self):
4762
self.__raise_if_already_closed()
4863

49-
return Cursor(self)
64+
self.__txn = self.__sess.transaction()
65+
self.__txn.begin()
66+
return Cursor(self.__txn, self)
5067

5168
def __handle_update_ddl(self, ddl_statements):
5269
"""
@@ -64,16 +81,6 @@ def __handle_update_ddl(self, ddl_statements):
6481
# Synchronously wait on the operation's completion.
6582
return self.__dbhandle.update_ddl(ddl_statements).result()
6683

67-
def read_snapshot(self):
68-
self.__raise_if_already_closed()
69-
70-
return self.__dbhandle.snapshot()
71-
72-
def in_transaction(self, fn, *args, **kwargs):
73-
self.__raise_if_already_closed()
74-
75-
return self.__dbhandle.run_in_transaction(fn, *args, **kwargs)
76-
7784
def append_ddl_statement(self, ddl_statement):
7885
self.__raise_if_already_closed()
7986

@@ -88,3 +95,39 @@ def run_prior_DDL_statements(self):
8895
ddl_statements = self.__ddl_statements
8996
self.__ddl_statements = []
9097
return self.__handle_update_ddl(ddl_statements)
98+
99+
def __update_ddl(self, ddl_statements):
100+
"""
101+
Runs the list of Data Definition Language (DDL) statements on the specified
102+
database. Note that each DDL statement MUST NOT contain a semicolon.
103+
Args:
104+
ddl_statements: a list of DDL statements, each without a semicolon.
105+
Returns:
106+
google.api_core.operation.Operation
107+
"""
108+
# Synchronously wait on the operation's completion.
109+
return self.__dbhandle.update_ddl(ddl_statements).result()
110+
111+
def list_tables(self):
112+
# We CANNOT list tables with
113+
# SELECT
114+
# t.table_name
115+
# FROM
116+
# information_schema.tables AS t
117+
# WHERE
118+
# t.table_catalog = '' and t.table_schema = ''
119+
# with a transaction otherwise we get back:
120+
# 400 Unsupported concurrency mode in query using INFORMATION_SCHEMA.
121+
# hence this specialized method.
122+
self.run_prior_DDL_statements()
123+
124+
with self.__dbhandle.snapshot() as snapshot:
125+
res = snapshot.execute_sql("""
126+
SELECT
127+
t.table_name
128+
FROM
129+
information_schema.tables AS t
130+
WHERE
131+
t.table_catalog = '' and t.table_schema = ''
132+
""")
133+
return list(res)

‎packages/django-google-spanner/spanner/dbapi/cursor.py‎

Lines changed: 45 additions & 52 deletions
Original file line numberDiff line numberDiff line change
@@ -19,10 +19,11 @@
1919

2020

2121
class Cursor(object):
22-
def __init__(self, db_handle=None):
22+
def __init__(self, txn, db_handle=None):
2323
self.__itr = None
2424
self.__res = None
2525
self.__row_count = _UNSET_COUNT
26+
self.__txn = txn
2627
self.__db_handle = db_handle
2728
self.__last_op = None
2829
self.__closed = False
@@ -89,7 +90,7 @@ def execute(self, sql, args=None):
8990
self.__run_prior_DDL_statements()
9091

9192
if classification == STMT_NON_UPDATING:
92-
self.__handle_DQL(sql, args or None)
93+
self.__handle_DQL(self.__get_txn(), sql, args or None)
9394
elif classification == STMT_INSERT:
9495
self.__handle_insert(sql, args or None)
9596
else:
@@ -102,16 +103,10 @@ def execute(self, sql, args=None):
102103
raise OperationalError(e.details if hasattr(e, 'details') else e)
103104

104105
def __handle_update(self, sql, params):
105-
self.__db_handle.in_transaction(
106-
self.__do_execute_update,
107-
sql, params,
108-
)
109-
110-
def __do_execute_update(self, transaction, sql, params, param_types=None):
111106
sql = ensure_where_clause(sql)
112107
sql, params = sql_pyformat_args_to_spanner(sql, params)
113108

114-
res = transaction.execute_update(sql, params=params, param_types=get_param_types(params))
109+
res = self.__txn.execute_update(sql, params=params, param_types=get_param_types(params))
115110
self.__itr = None
116111
if type(res) == int:
117112
self.__row_count = res
@@ -134,27 +129,11 @@ def __handle_insert(self, sql, params):
134129
# transaction.execute_sql(sql, params, param_types)
135130
# which invokes more RPCs and is more costly.
136131

137-
if parts.get('homogenous'):
138-
# The common case of multiple values being passed in
139-
# non-complex pyformat args and need to be uploaded in one RPC.
140-
return self.__db_handle.in_transaction(
141-
self.__do_execute_insert_homogenous,
142-
parts,
143-
)
144-
else:
145-
# All the other cases that are esoteric and need
146-
# transaction.execute_sql
147-
sql_params_list = parts.get('sql_params_list')
148-
return self.__db_handle.in_transaction(
149-
self.__do_execute_insert_heterogenous,
150-
sql_params_list,
151-
)
152-
153-
def __do_execute_insert_heterogenous(self, transaction, sql_params_list):
132+
sql_params_list = parts.get('sql_params_list')
154133
for sql, params in sql_params_list:
155134
sql, params = sql_pyformat_args_to_spanner(sql, params)
156135
param_types = get_param_types(params)
157-
res = transaction.execute_sql(sql, params=params, param_types=param_types)
136+
res = txn.execute_update(sql, params=params, param_types=param_types)
158137
# TODO: File a bug with Cloud Spanner and the Python client maintainers
159138
# about a lost commit when res isn't read from.
160139
_ = list(res)
@@ -166,40 +145,41 @@ def __do_execute_insert_homogenous(self, transaction, parts):
166145
values = parts.get('values')
167146
return transaction.insert(table, columns, values)
168147

169-
def __handle_DQL(self, sql, params):
170-
with self.__db_handle.read_snapshot() as snapshot:
171-
# Reference
172-
# https://googleapis.dev/python/spanner/latest/session-api.html#google.cloud.spanner_v1.session.Session.execute_sql
173-
sql, params = sql_pyformat_args_to_spanner(sql, params)
174-
res = snapshot.execute_sql(sql, params=params, param_types=get_param_types(params))
175-
if type(res) == int:
176-
self.__row_count = res
177-
self.__itr = None
178-
else:
179-
# Immediately using:
180-
# iter(response)
181-
# here, because this Spanner API doesn't provide
182-
# easy mechanisms to detect when only a single item
183-
# is returned or many, yet mixing results that
184-
# are for .fetchone() with those that would result in
185-
# many items returns a RuntimeError if .fetchone() is
186-
# invoked and vice versa.
187-
self.__res = res
188-
# Read the first element so that StreamedResult can
189-
# return the metadata after a DQL statement. See issue #155.
190-
self.__itr = PeekIterator(self.__res)
191-
# Unfortunately, Spanner doesn't seem to send back
192-
# information about the number of rows available.
193-
self.__row_count = _UNSET_COUNT
148+
def __handle_DQL(self, txn, sql, params):
149+
# Reference
150+
# https://googleapis.dev/python/spanner/latest/session-api.html#google.cloud.spanner_v1.session.Session.execute_sql
151+
sql, params = sql_pyformat_args_to_spanner(sql, params)
152+
res = txn.execute_sql(sql, params=params, param_types=get_param_types(params))
153+
if type(res) == int:
154+
self.__row_count = res
155+
self.__itr = None
156+
else:
157+
# Immediately using:
158+
# iter(response)
159+
# here, because this Spanner API doesn't provide
160+
# easy mechanisms to detect when only a single item
161+
# is returned or many, yet mixing results that
162+
# are for .fetchone() with those that would result in
163+
# many items returns a RuntimeError if .fetchone() is
164+
# invoked and vice versa.
165+
self.__res = res
166+
# Read the first element so that StreamedResult can
167+
# return the metadata after a DQL statement. See issue #155.
168+
self.__itr = PeekIterator(self.__res)
169+
# Unfortunately, Spanner doesn't seem to send back
170+
# information about the number of rows available.
171+
self.__row_count = _UNSET_COUNT
194172

195173
def __enter__(self):
196174
return self
197175

198176
def __exit__(self, etype, value, traceback):
177+
self.__txn.commit()
199178
self.__clear()
200179

201180
def __clear(self):
202181
self.__db_handle = None
182+
self.__txn = None
203183

204184
def executemany(self, operation, seq_of_params):
205185
self.__raise_if_already_closed()
@@ -272,6 +252,19 @@ def setoutputsize(size, column=None):
272252
def __run_prior_DDL_statements(self):
273253
return self.__db_handle.run_prior_DDL_statements()
274254

255+
def list_tables(self):
256+
# We CANNOT list tables with
257+
# SELECT
258+
# t.table_name
259+
# FROM
260+
# information_schema.tables AS t
261+
# WHERE
262+
# t.table_catalog = '' and t.table_schema = ''
263+
# with a transaction otherwise we get back:
264+
# 400 Unsupported concurrency mode in query using INFORMATION_SCHEMA.
265+
# hence this specialized method.
266+
return self.__db_handle.list_tables()
267+
275268

276269
class Column:
277270
def __init__(self, name, type_code, display_size=None, internal_size=None,

‎packages/django-google-spanner/spanner/dbapi/parse_utils.py‎

Lines changed: 9 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -160,22 +160,21 @@ def parse_insert(insert_sql, params):
160160
if values.homogenous():
161161
# Case c)
162162

163-
# We MUST strip backticks for the table_name and columns,
164-
# only in this code path for homogenous statements.
165-
table = strip_backticks(match.group('table_name'))
166-
167163
columns = [
168-
strip_backticks(mi.strip())
169-
for mi in match.group('columns').split(',')
164+
strip_backticks(mi.strip())
165+
for mi in match.group('columns').split(',')
170166
]
167+
sql_params_list = []
168+
insert_sql_preamble = 'INSERT INTO %s (%s) VALUES %s' % (
169+
match.group('table_name'), match.group('columns'), values.argv[0],
170+
)
171171
values_pyformat = [str(arg) for arg in values.argv]
172172
rows_list = rows_for_insert_or_update(columns, params, values_pyformat)
173+
for row in rows_list:
174+
sql_params_list.append((insert_sql_preamble, row,))
173175

174176
return {
175-
'homogenous': True,
176-
'table': table,
177-
'values': rows_list,
178-
'columns': columns,
177+
'sql_params_list': sql_params_list,
179178
}
180179

181180
# Case d)

‎packages/django-google-spanner/spanner/django/features.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@ class DatabaseFeatures(BaseDatabaseFeatures):
1515
supports_partial_indexes = False
1616
supports_regex_backreferencing = False
1717
supports_timezones = False
18-
supports_transactions = False
18+
supports_transactions = True
1919
supports_column_check_constraints = False
2020
supports_table_check_constraints = False
2121
uses_savepoints = False

‎packages/django-google-spanner/spanner/django/introspection.py‎

Lines changed: 1 addition & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -12,13 +12,5 @@
1212
class DatabaseIntrospection(BaseDatabaseIntrospection):
1313
def get_table_list(self, cursor):
1414
"""Return a list of table and view names in the current database."""
15-
cursor.execute("""
16-
SELECT
17-
t.table_name
18-
FROM
19-
information_schema.tables AS t
20-
WHERE
21-
t.table_catalog = '' and t.table_schema = ''
22-
""")
2315
# The second TableInfo field is 't' for table or 'v' for view.
24-
return [TableInfo(row[0], 't') for row in cursor.fetchall()]
16+
return [TableInfo(row[0], 't') for row in cursor.list_tables()]

0 commit comments

Comments
 (0)