ENH: to_sql() add parameter "method" to control insertions method (#8…

doc/source/whatsnew/v0.24.0.txt

@@ -16,6 +16,7 @@ Other Enhancements
 - :func:`Series.mode` and :func:`DataFrame.mode` now support the ``dropna`` parameter which can be used to specify whether NaN/NaT values should be considered (:issue:`17534`)
 - :func:`to_csv` now supports ``compression`` keyword when a file handle is passed. (:issue:`21227`)
 - :meth:`Index.droplevel` is now implemented also for flat indexes, for compatibility with MultiIndex (:issue:`21115`)
+- :func:`~pandas.DataFrame.to_sql` add parameter ``method`` to control SQL insertion clause (:8953:)
 
 
 .. _whatsnew_0240.api_breaking:

pandas/core/generic.py

@@ -2014,7 +2014,7 @@ def to_msgpack(self, path_or_buf=None, encoding='utf-8', **kwargs):
  **kwargs)
 
  def to_sql(self, name, con, schema=None, if_exists='fail', index=True,
- index_label=None, chunksize=None, dtype=None):
+ index_label=None, chunksize=None, dtype=None, method='default'):
  """
  Write records stored in a DataFrame to a SQL database.
 
@@ -2052,6 +2052,8 @@ def to_sql(self, name, con, schema=None, if_exists='fail', index=True,
  Specifying the datatype for columns. The keys should be the column
  names and the values should be the SQLAlchemy types or strings for
  the sqlite3 legacy mode.
+ method : {'default', 'multi', callable}, default 'default'
+ Controls the SQL insertion clause used.
 
  Raises
  ------
@@ -2120,11 +2122,59 @@ def to_sql(self, name, con, schema=None, if_exists='fail', index=True,
 
  >>> engine.execute("SELECT * FROM integers").fetchall()
  [(1,), (None,), (2,)]
+
+ Insertion method:
+
+ .. versionadded:: 0.24.0
+
+ The parameter ``method`` controls the SQL insertion clause used.
+ Possible values are:
+
+ - `'default'`: Uses standard SQL `INSERT` clause
+ - `'multi'`: Pass multiple values in a single `INSERT` clause.
+ It uses a **special** SQL syntax not supported by all backends.
+ This usually provides a big performance for Analytic databases
+ like *Presto* and *Redshit*, but has worse performance for
+ traditional SQL backend if the table contains many columns.
+ For more information check SQLAlchemy `documention <http://docs.sqlalchemy.org/en/latest/core/dml.html?highlight=multivalues#sqlalchemy.sql.expression.Insert.values.params.*args>`__.
+ - callable: with signature `(pd_table, conn, keys, data_iter)`.
+ This can be used to implement more performant insertion based on
+ specific backend dialect features.
+ I.e. using *Postgresql* `COPY clause
+ <https://www.postgresql.org/docs/current/static/sql-copy.html>`__.
+ Check API for details and a sample implementation
+ :func:`~pandas.DataFrame.to_sql`.
+
+
+ Example of callable for Postgresql *COPY*::
+
+ # Alternative to_sql() *method* for DBs that support COPY FROM
+ import csv
+ from io import StringIO
+
+ def psql_insert_copy(table, conn, keys, data_iter):
+ # gets a DBAPI connection that can provide a cursor
+ dbapi_conn = conn.connection
+ with dbapi_conn.cursor() as cur:
+ s_buf = StringIO()
+ writer = csv.writer(s_buf)
+ writer.writerows(data_iter)
+ s_buf.seek(0)
+
+ columns = ', '.join('"{}"'.format(k) for k in keys)
+ if table.schema:
+ table_name = '{}.{}'.format(table.schema, table.name)
+ else:
+ table_name = table.name
+
+ sql = 'COPY {} ({}) FROM STDIN WITH CSV'.format(
+ table_name, columns)
+ cur.copy_expert(sql=sql, file=s_buf)
  """
  from pandas.io import sql
  sql.to_sql(self, name, con, schema=schema, if_exists=if_exists,
  index=index, index_label=index_label, chunksize=chunksize,
- dtype=dtype)
+ dtype=dtype, method=method)
 
  def to_pickle(self, path, compression='infer',
  protocol=pkl.HIGHEST_PROTOCOL):

pandas/io/sql.py

@@ -6,6 +6,7 @@
 
 from __future__ import print_function, division
 from datetime import datetime, date, time
+from functools import partial
 
 import warnings
 import re
@@ -398,7 +399,7 @@ def read_sql(sql, con, index_col=None, coerce_float=True, params=None,
 
 
 def to_sql(frame, name, con, schema=None, if_exists='fail', index=True,
- index_label=None, chunksize=None, dtype=None):
+ index_label=None, chunksize=None, dtype=None, method='default'):
  """
  Write records stored in a DataFrame to a SQL database.
 
@@ -432,6 +433,8 @@ def to_sql(frame, name, con, schema=None, if_exists='fail', index=True,
  Optional specifying the datatype for columns. The SQL type should
  be a SQLAlchemy type, or a string for sqlite3 fallback connection.
  If all columns are of the same type, one single value can be used.
+ method : {'default', 'multi', callable}, default 'default'
+ Controls the SQL insertion clause used.
 
  """
  if if_exists not in ('fail', 'replace', 'append'):
@@ -447,7 +450,7 @@ def to_sql(frame, name, con, schema=None, if_exists='fail', index=True,
 
  pandas_sql.to_sql(frame, name, if_exists=if_exists, index=index,
  index_label=index_label, schema=schema,
- chunksize=chunksize, dtype=dtype)
+ chunksize=chunksize, dtype=dtype, method=method)
 
 
 def has_table(table_name, con, schema=None):
@@ -572,8 +575,29 @@ def create(self):
  else:
  self._execute_create()
 
- def insert_statement(self):
- return self.table.insert()
+ def _execute_insert(self, conn, keys, data_iter):
+ """Execute SQL statement inserting data
+
+ Parameters
+ ----------
+ conn : sqlalchemy.engine.Engine or sqlalchemy.engine.Connection
+ keys : list of str
+ Column names
+ data_iter : generator of list
+ Each item contains a list of values to be inserted
+ """
+ data = [{k: v for k, v in zip(keys, row)} for row in data_iter]
+ conn.execute(self.table.insert(), data)
+
+ def _execute_insert_multi(self, conn, keys, data_iter):
+ """Alternative to _execute_insert for DBs support multivalue INSERT.
+
+ Note: multi-value insert is usually faster for analytics DBs
+ and tables containing a few columns
+ but performance degrades quickly with increase of columns.
+ """
+ data = [{k: v for k, v in zip(keys, row)} for row in data_iter]
+ conn.execute(self.table.insert(data))
 
  def insert_data(self):
  if self.index is not None:
@@ -611,11 +635,18 @@ def insert_data(self):
 
  return column_names, data_list
 
- def _execute_insert(self, conn, keys, data_iter):
- data = [{k: v for k, v in zip(keys, row)} for row in data_iter]
- conn.execute(self.insert_statement(), data)
+ def insert(self, chunksize=None, method=None):
+
+ # set insert method
+ if method in (None, 'default'):
+ exec_insert = self._execute_insert
+ elif method == 'multi':
+ exec_insert = self._execute_insert_multi
+ elif callable(method):
+ exec_insert = partial(method, self)
+ else:
+ raise ValueError('Invalid parameter `method`: {}'.format(method))
 
- def insert(self, chunksize=None):
  keys, data_list = self.insert_data()
 
  nrows = len(self.frame)
@@ -638,7 +669,7 @@ def insert(self, chunksize=None):
  break
 
  chunk_iter = zip(*[arr[start_i:end_i] for arr in data_list])
- self._execute_insert(conn, keys, chunk_iter)
+ exec_insert(conn, keys, chunk_iter)
 
  def _query_iterator(self, result, chunksize, columns, coerce_float=True,
  parse_dates=None):
@@ -1078,7 +1109,8 @@ def read_query(self, sql, index_col=None, coerce_float=True,
  read_sql = read_query
 
  def to_sql(self, frame, name, if_exists='fail', index=True,
- index_label=None, schema=None, chunksize=None, dtype=None):
+ index_label=None, schema=None, chunksize=None, dtype=None,
+ method='default'):
  """
  Write records stored in a DataFrame to a SQL database.
 
@@ -1108,7 +1140,8 @@ def to_sql(self, frame, name, if_exists='fail', index=True,
  Optional specifying the datatype for columns. The SQL type should
  be a SQLAlchemy type. If all columns are of the same type, one
  single value can be used.
-
+ method : {'default', 'multi', callable}, default 'default'
+ Controls the SQL insertion clause used.
  """
  if dtype and not is_dict_like(dtype):
  dtype = {col_name: dtype for col_name in frame}
@@ -1124,7 +1157,7 @@ def to_sql(self, frame, name, if_exists='fail', index=True,
  if_exists=if_exists, index_label=index_label,
  schema=schema, dtype=dtype)
  table.create()
- table.insert(chunksize)
+ table.insert(chunksize, method=method)
  if (not name.isdigit() and not name.islower()):
  # check for potentially case sensitivity issues (GH7815)
  # Only check when name is not a number and name is not lower case
@@ -1434,7 +1467,8 @@ def _fetchall_as_list(self, cur):
  return result
 
  def to_sql(self, frame, name, if_exists='fail', index=True,
- index_label=None, schema=None, chunksize=None, dtype=None):
+ index_label=None, schema=None, chunksize=None, dtype=None,
+ method='default'):
  """
  Write records stored in a DataFrame to a SQL database.
 
@@ -1463,7 +1497,8 @@ def to_sql(self, frame, name, if_exists='fail', index=True,
  Optional specifying the datatype for columns. The SQL type should
  be a string. If all columns are of the same type, one single value
  can be used.
-
+ method : {'default', 'multi', callable}, default 'default'
+ Controls the SQL insertion clause used.
  """
  if dtype and not is_dict_like(dtype):
  dtype = {col_name: dtype for col_name in frame}
@@ -1478,7 +1513,7 @@ def to_sql(self, frame, name, if_exists='fail', index=True,
  if_exists=if_exists, index_label=index_label,
  dtype=dtype)
  table.create()
- table.insert(chunksize)
+ table.insert(chunksize, method)
 
  def has_table(self, name, schema=None):
  # TODO(wesm): unused?

pandas/tests/io/test_sql.py

@@ -372,12 +372,16 @@ def _read_sql_iris_named_parameter(self):
  iris_frame = self.pandasSQL.read_query(query, params=params)
  self._check_iris_loaded_frame(iris_frame)
 
- def _to_sql(self):
+ def _to_sql(self, method=None):
  self.drop_table('test_frame1')
 
- self.pandasSQL.to_sql(self.test_frame1, 'test_frame1')
+ self.pandasSQL.to_sql(self.test_frame1, 'test_frame1', method=method)
  assert self.pandasSQL.has_table('test_frame1')
 
+ num_entries = len(self.test_frame1)
+ num_rows = self._count_rows('test_frame1')
+ assert num_rows == num_entries
+
  # Nuke table
  self.drop_table('test_frame1')
 
@@ -431,6 +435,25 @@ def _to_sql_append(self):
  assert num_rows == num_entries
  self.drop_table('test_frame1')
 
+ def _to_sql_method_callable(self):
+ check = [] # used to double check function below is really being used
+
+ def sample(pd_table, conn, keys, data_iter):
+ check.append(1)
+ data = [{k: v for k, v in zip(keys, row)} for row in data_iter]
+ conn.execute(pd_table.table.insert(), data)
+ self.drop_table('test_frame1')
+
+ self.pandasSQL.to_sql(self.test_frame1, 'test_frame1', method=sample)
+ assert self.pandasSQL.has_table('test_frame1')
+
+ assert check == [1]
+ num_entries = len(self.test_frame1)
+ num_rows = self._count_rows('test_frame1')
+ assert num_rows == num_entries
+ # Nuke table
+ self.drop_table('test_frame1')
+
  def _roundtrip(self):
  self.drop_table('test_frame_roundtrip')
  self.pandasSQL.to_sql(self.test_frame1, 'test_frame_roundtrip')
@@ -1180,7 +1203,7 @@ def setup_connect(self):
  pytest.skip(
  "Can't connect to {0} server".format(self.flavor))
 
- def test_aread_sql(self):
+ def test_read_sql(self):
  self._read_sql_iris()
 
  def test_read_sql_parameter(self):
@@ -1204,6 +1227,12 @@ def test_to_sql_replace(self):
  def test_to_sql_append(self):
  self._to_sql_append()
 
+ def test_to_sql_method_multi(self):
+ self._to_sql(method='multi')
+
+ def test_to_sql_method_callable(self):
+ self._to_sql_method_callable()
+
  def test_create_table(self):
  temp_conn = self.connect()
  temp_frame = DataFrame(