diff --git a/google/cloud/spanner_dbapi/cursor.py b/google/cloud/spanner_dbapi/cursor.py
index 91bccedd4c..330aeb2c72 100644
--- a/google/cloud/spanner_dbapi/cursor.py
+++ b/google/cloud/spanner_dbapi/cursor.py
@@ -315,6 +315,10 @@ def executemany(self, operation, seq_of_params):
                 "Executing DDL statements with executemany() method is not allowed."
             )
 
+        # For every operation, we've got to ensure that any prior DDL
+        # statements were run.
+        self.connection.run_prior_DDL_statements()
+
         many_result_set = StreamedManyResultSets()
 
         if class_ in (parse_utils.STMT_INSERT, parse_utils.STMT_UPDATING):
diff --git a/tests/system/test_dbapi.py b/tests/system/test_dbapi.py
index 29617ad614..f3c5da1f46 100644
--- a/tests/system/test_dbapi.py
+++ b/tests/system/test_dbapi.py
@@ -27,7 +27,6 @@
 from google.cloud.spanner_v1 import gapic_version as package_version
 from . import _helpers
 
-
 DATABASE_NAME = "dbapi-txn"
 
 DDL_STATEMENTS = (
@@ -344,6 +343,156 @@ def test_DDL_autocommit(shared_instance, dbapi_database):
             op.result()
 
 
+def test_ddl_execute_autocommit_true(shared_instance, dbapi_database):
+    """Check that DDL statement in autocommit mode results in successful
+    DDL statement execution for execute method."""
+
+    conn = Connection(shared_instance, dbapi_database)
+    conn.autocommit = True
+    cur = conn.cursor()
+    cur.execute(
+        """
+        CREATE TABLE DdlExecuteAutocommit (
+            SingerId     INT64 NOT NULL,
+            Name    STRING(1024),
+        ) PRIMARY KEY (SingerId)
+        """
+    )
+    table = dbapi_database.table("DdlExecuteAutocommit")
+    assert table.exists() is True
+
+    cur.close()
+    conn.close()
+
+
+def test_ddl_executemany_autocommit_true(shared_instance, dbapi_database):
+    """Check that DDL statement in autocommit mode results in exception for
+    executemany method ."""
+
+    conn = Connection(shared_instance, dbapi_database)
+    conn.autocommit = True
+    cur = conn.cursor()
+    with pytest.raises(ProgrammingError):
+        cur.executemany(
+            """
+            CREATE TABLE DdlExecuteManyAutocommit (
+                SingerId     INT64 NOT NULL,
+                Name    STRING(1024),
+            ) PRIMARY KEY (SingerId)
+            """,
+            [],
+        )
+    table = dbapi_database.table("DdlExecuteManyAutocommit")
+    assert table.exists() is False
+
+    cur.close()
+    conn.close()
+
+
+def test_ddl_executemany_autocommit_false(shared_instance, dbapi_database):
+    """Check that DDL statement in non-autocommit mode results in exception for
+    executemany method ."""
+
+    conn = Connection(shared_instance, dbapi_database)
+    cur = conn.cursor()
+    with pytest.raises(ProgrammingError):
+        cur.executemany(
+            """
+            CREATE TABLE DdlExecuteManyAutocommit (
+                SingerId     INT64 NOT NULL,
+                Name    STRING(1024),
+            ) PRIMARY KEY (SingerId)
+            """,
+            [],
+        )
+    table = dbapi_database.table("DdlExecuteManyAutocommit")
+    assert table.exists() is False
+
+    cur.close()
+    conn.close()
+
+
+def test_ddl_execute(shared_instance, dbapi_database):
+    """Check that DDL statement followed by non-DDL execute statement in
+    non autocommit mode results in successful DDL statement execution."""
+
+    conn = Connection(shared_instance, dbapi_database)
+    want_row = (
+        1,
+        "first-name",
+    )
+    cur = conn.cursor()
+    cur.execute(
+        """
+        CREATE TABLE DdlExecute (
+            SingerId     INT64 NOT NULL,
+            Name    STRING(1024),
+        ) PRIMARY KEY (SingerId)
+        """
+    )
+    table = dbapi_database.table("DdlExecute")
+    assert table.exists() is False
+
+    cur.execute(
+        """
+        INSERT INTO DdlExecute (SingerId, Name)
+        VALUES (1, "first-name")
+        """
+    )
+    assert table.exists() is True
+    conn.commit()
+
+    # read the resulting data from the database
+    cur.execute("SELECT * FROM DdlExecute")
+    got_rows = cur.fetchall()
+
+    assert got_rows == [want_row]
+
+    cur.close()
+    conn.close()
+
+
+def test_ddl_executemany(shared_instance, dbapi_database):
+    """Check that DDL statement followed by non-DDL executemany statement in
+    non autocommit mode results in successful DDL statement execution."""
+
+    conn = Connection(shared_instance, dbapi_database)
+    want_row = (
+        1,
+        "first-name",
+    )
+    cur = conn.cursor()
+    cur.execute(
+        """
+        CREATE TABLE DdlExecuteMany (
+            SingerId     INT64 NOT NULL,
+            Name    STRING(1024),
+        ) PRIMARY KEY (SingerId)
+        """
+    )
+    table = dbapi_database.table("DdlExecuteMany")
+    assert table.exists() is False
+
+    cur.executemany(
+        """
+        INSERT INTO DdlExecuteMany (SingerId, Name)
+        VALUES (%s, %s)
+        """,
+        [want_row],
+    )
+    assert table.exists() is True
+    conn.commit()
+
+    # read the resulting data from the database
+    cur.execute("SELECT * FROM DdlExecuteMany")
+    got_rows = cur.fetchall()
+
+    assert got_rows == [want_row]
+
+    cur.close()
+    conn.close()
+
+
 @pytest.mark.skipif(_helpers.USE_EMULATOR, reason="Emulator does not support json.")
 def test_autocommit_with_json_data(shared_instance, dbapi_database):
     """