From 013a0680ab2cac5d384d7901b9235dd9d5a6ab0a Mon Sep 17 00:00:00 2001
From: Joris Van den Bossche <jorisvandenbossche@gmail.com>
Date: Tue, 3 Jun 2014 23:58:37 +0200
Subject: [PATCH] BUG: fix read_sql delegation for queries without select
 statement (GH7324)

The select check was introduced for mysql legacy compatibility, but introduced problem if the query did not contain a select statement (eg a stored procedure)
---
 doc/source/v0.14.1.txt      |  3 +++
 pandas/io/sql.py            | 21 +++++++--------------
 pandas/io/tests/test_sql.py | 32 +++++++++++++++++++++++++++++++-
 3 files changed, 41 insertions(+), 15 deletions(-)

diff --git a/doc/source/v0.14.1.txt b/doc/source/v0.14.1.txt
index 3a53b5629c6e3..0a89806c899a4 100644
--- a/doc/source/v0.14.1.txt
+++ b/doc/source/v0.14.1.txt
@@ -117,3 +117,6 @@ Bug Fixes
 - Bug in ``MultiIndex.append``, ``concat`` and ``pivot_table`` don't preserve timezone (:issue:`6606`)
 - Bug in ``.loc`` with a list of indexers on a single-multi index level (that is not nested) (:issue:`7349`)
 - Bug in ``Series.map`` when mapping a dict with tuple keys of different lengths (:issue:`7333`)
+- Bug all ``StringMethods`` now work on empty Series (:issue:`7242`)
+- Fix delegation of `read_sql` to `read_sql_query` when query does not contain
+ 'select' (:issue:`7324`).
diff --git a/pandas/io/sql.py b/pandas/io/sql.py
index aa08c95c4f1c3..0ac4b5f3fcc2b 100644
--- a/pandas/io/sql.py
+++ b/pandas/io/sql.py
@@ -374,26 +374,19 @@ def read_sql(sql, con, index_col=None, coerce_float=True, params=None,
     """
     pandas_sql = pandasSQL_builder(con)
 
-    if 'select' in sql.lower():
-        try:
-            if pandas_sql.has_table(sql):
-                return pandas_sql.read_table(
-                    sql, index_col=index_col, coerce_float=coerce_float,
-                    parse_dates=parse_dates, columns=columns)
-        except:
-            pass
-
+    if isinstance(pandas_sql, PandasSQLLegacy):
         return pandas_sql.read_sql(
             sql, index_col=index_col, params=params,
             coerce_float=coerce_float, parse_dates=parse_dates)
-    else:
-        if isinstance(pandas_sql, PandasSQLLegacy):
-            raise ValueError("Reading a table with read_sql is not supported "
-                             "for a DBAPI2 connection. Use an SQLAlchemy "
-                             "engine or specify an sql query")
+
+    if pandas_sql.has_table(sql):
         return pandas_sql.read_table(
             sql, index_col=index_col, coerce_float=coerce_float,
             parse_dates=parse_dates, columns=columns)
+    else:
+        return pandas_sql.read_sql(
+            sql, index_col=index_col, params=params,
+            coerce_float=coerce_float, parse_dates=parse_dates)
 
 
 def to_sql(frame, name, con, flavor='sqlite', if_exists='fail', index=True,
diff --git a/pandas/io/tests/test_sql.py b/pandas/io/tests/test_sql.py
index 2796ab48ec894..a34f278fc5a96 100644
--- a/pandas/io/tests/test_sql.py
+++ b/pandas/io/tests/test_sql.py
@@ -671,7 +671,7 @@ def test_read_sql_delegate(self):
                               "read_sql and read_sql_query have not the same"
                               " result with a query")
 
-        self.assertRaises(ValueError, sql.read_sql, 'iris', self.conn)
+        self.assertRaises(sql.DatabaseError, sql.read_sql, 'iris', self.conn)
 
     def test_safe_names_warning(self):
         # GH 6798
@@ -1078,6 +1078,36 @@ def test_default_type_conversion(self):
         self.assertTrue(issubclass(df.BoolColWithNull.dtype.type, np.floating),
                         "BoolColWithNull loaded with incorrect type")
 
+    def test_read_procedure(self):
+        # see GH7324. Although it is more an api test, it is added to the
+        # mysql tests as sqlite does not have stored procedures
+        df = DataFrame({'a': [1, 2, 3], 'b':[0.1, 0.2, 0.3]})
+        df.to_sql('test_procedure', self.conn, index=False)
+
+        proc = """DROP PROCEDURE IF EXISTS get_testdb;
+
+        CREATE PROCEDURE get_testdb ()
+
+        BEGIN
+            SELECT * FROM test_procedure;
+        END"""
+
+        connection = self.conn.connect()
+        trans = connection.begin()
+        try:
+            r1 = connection.execute(proc)
+            trans.commit()
+        except:
+            trans.rollback()
+            raise
+
+        res1 = sql.read_sql_query("CALL get_testdb();", self.conn)
+        tm.assert_frame_equal(df, res1)
+
+        # test delegation to read_sql_query
+        res2 = sql.read_sql("CALL get_testdb();", self.conn)
+        tm.assert_frame_equal(df, res2)
+
 
 class TestPostgreSQLAlchemy(_TestSQLAlchemy):
     """