feat(spanner): add batch_create_session calls to session pools

[larkee]

Currently, session pools are filled by creating single sessions. For large session pool sizes (>1000) this can be incredibly slow as it requires >1000 requests to be made to the server sequentially. Using batch_create_session calls to fill the pools will reduce the number of requests by a factor of 100 which will greatly reduce the start up time.

The BurstyPool does not use batch_create_session as it is designed to create sessions on an "as needed" basis rather than filling a queue.

[larkee]

Build failure is due to a known flaky test.

[tseaver]

@larkee I have a PR pending review, #9393, which fixes that flaky test.

[crwilcox]

Failure seems to be in an unrelated area.

_________ TestSessionAPI.test_transaction_batch_update_and_execute_dml _________
self = <tests.system.test_system.TestSessionAPI testMethod=test_transaction_batch_update_and_execute_dml>
def test_transaction_batch_update_and_execute_dml(self):
retry = RetryInstanceState(_has_all_ddl)
retry(self._db.reload)()
session = self._db.session()
session.create()
self.to_delete.append(session)
with session.batch() as batch:
batch.delete(self.TABLE, self.ALL)
insert_statements = list(self._generate_insert_statements())
update_statements = [
(
"UPDATE contacts SET email = @email " "WHERE contact_id = @contact_id;",
{"contact_id": 1, "email": "phreddy@example.com"},
{"contact_id": Type(code=INT64), "email": Type(code=STRING)},
)
]
delete_statement = "DELETE contacts WHERE TRUE;"
def unit_of_work(transaction, self):
rows = list(transaction.read(self.TABLE, self.COLUMNS, self.ALL))
self.assertEqual(rows, [])
status, row_counts = transaction.batch_update(
insert_statements + update_statements
)
self.assertEqual(status.code, 0) # XXX: where are values defined?
self.assertEqual(len(row_counts), len(insert_statements) + 1)
for row_count in row_counts:
self.assertEqual(row_count, 1)
row_count = transaction.execute_update(delete_statement)
self.assertEqual(row_count, len(insert_statements))
> session.run_in_transaction(unit_of_work, self)
tests/system/test_system.py:861:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
google/cloud/spanner_v1/session.py:299: in run_in_transaction
return_value = func(txn, *args, **kw)
tests/system/test_system.py:852: in unit_of_work
self.assertEqual(status.code, 0) # XXX: where are values defined?
E AssertionError: 10 != 0