meta := ex.state.mu.txn.GetTxnCoordMeta(ctx)
if meta.CommandCount > 0 {
  err := pgerror.Newf(pgcode.Syntax,
    "SAVEPOINT %s needs to be the first statement in a transaction", RestartSavepointName)
}

if ex.state.mu.txn.Active() {
  err := pgerror.Newf(pgcode.Syntax,
    "SAVEPOINT %s needs to be the first statement in a transaction", RestartSavepointName)
}

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from cockroachdb.sqlalchemy import run_transaction

import _thread
import time
import logging

logging.basicConfig()
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

#engine = create_engine("cockroachdb://root@localhost:26257/defaultdb?sslmode=disable", pool_size=2, max_overflow=0)
engine = create_engine("cockroachdb://root@localhost:26257/defaultdb?sslmode=disable")
session_maker = sessionmaker(bind=engine)

def db_func(conn, i):
    mod = i % 10
    if i < 10:
        conn.execute(f"insert into a values({i}, {i}) on conflict do nothing")
    rs = conn.execute(f"select a, b from a where a = {mod}")
    conn.execute("select crdb_internal.force_retry('5s')")
    for row in rs:
        print(f"thread {i}: results {row}")

def f(thread_id):
    print(f"thread id {thread_id}")
    run_transaction(session_maker, lambda conn: db_func(conn, thread_id))

run_transaction(session_maker, lambda c: c.execute("create table if not exists a (a int primary key, b int)"))
try:
    for x in range(1):
        _thread.start_new_thread(f, (x,))
except:
    print("Error: unable to start thread")
    raise

while 1:
    pass

INFO:sqlalchemy.engine.base.Engine:select current_schema()
INFO:sqlalchemy.engine.base.Engine:{}
INFO:sqlalchemy.engine.base.Engine:SELECT CAST('test plain returns' AS VARCHAR(60)) AS anon_1
INFO:sqlalchemy.engine.base.Engine:{}
INFO:sqlalchemy.engine.base.Engine:SELECT CAST('test unicode returns' AS VARCHAR(60)) AS anon_1
INFO:sqlalchemy.engine.base.Engine:{}
INFO:sqlalchemy.engine.base.Engine:select version()
INFO:sqlalchemy.engine.base.Engine:{}
INFO:sqlalchemy.engine.base.Engine:BEGIN (implicit)
INFO:sqlalchemy.engine.base.Engine:SAVEPOINT cockroach_restart
INFO:sqlalchemy.engine.base.Engine:{}
INFO:sqlalchemy.engine.base.Engine:create table if not exists a (a int primary key, b int)
INFO:sqlalchemy.engine.base.Engine:{}
INFO:sqlalchemy.engine.base.Engine:RELEASE SAVEPOINT cockroach_restart
INFO:sqlalchemy.engine.base.Engine:{}
INFO:sqlalchemy.engine.base.Engine:COMMIT
thread id 0
INFO:sqlalchemy.engine.base.Engine:BEGIN (implicit)
INFO:sqlalchemy.engine.base.Engine:SAVEPOINT cockroach_restart
INFO:sqlalchemy.engine.base.Engine:{}
INFO:sqlalchemy.engine.base.Engine:insert into a values(0, 0) on conflict do nothing
INFO:sqlalchemy.engine.base.Engine:{}
INFO:sqlalchemy.engine.base.Engine:select a, b from a where a = 0
INFO:sqlalchemy.engine.base.Engine:{}
INFO:sqlalchemy.engine.base.Engine:select crdb_internal.force_retry('5s')
INFO:sqlalchemy.engine.base.Engine:{}
INFO:sqlalchemy.engine.base.Engine:ROLLBACK TO SAVEPOINT cockroach_restart
INFO:sqlalchemy.engine.base.Engine:{}
INFO:sqlalchemy.engine.base.Engine:SAVEPOINT cockroach_restart
INFO:sqlalchemy.engine.base.Engine:{}
INFO:sqlalchemy.engine.base.Engine:ROLLBACK
Unhandled exception in thread started by <function f at 0x10481b8c0>
Traceback (most recent call last):
  File "/Users/rafiss/.virtualenvs/sqlalchemy-full/lib/python3.7/site-packages/SQLAlchemy-1.3.13.dev0-py3.7-macosx-10.14-x86_64.egg/sqlalchemy/engine/base.py", line 1246, in _execute_context
    cursor, statement, parameters, context
  File "/Users/rafiss/.virtualenvs/sqlalchemy-full/lib/python3.7/site-packages/SQLAlchemy-1.3.13.dev0-py3.7-macosx-10.14-x86_64.egg/sqlalchemy/engine/default.py", line 588, in do_execute
    cursor.execute(statement, parameters)
psycopg2.errors.SyntaxError: SAVEPOINT cockroach_restart needs to be the first statement in a transaction