f625f9e060e258c903e44f2d28a6689a123842af,lore/io/connection.py,Connection,insert,#Connection#Any#Any#Any#,162

Before Change

    def insert(self, table, dataframe, batch_size=10 ** 5):
        if self._engine.dialect.name in ["postgresql", "redshift"]:
            for batch in range(int(math.ceil(float(len(dataframe)) / batch_size))):
                if sys.version_info[0] == 2:
                    rows = io.BytesIO()
                else:
                    rows = io.StringIO()
                slice = dataframe.iloc[batch * batch_size:(batch + 1) * batch_size]
                slice.to_csv(rows, index=False, header=False, sep="|", na_rep="\\N", quoting=csv.QUOTE_NONE)
                rows.seek(0)
                self._connection.connection.cursor().copy_from(rows, table, null="\\N", sep="|", columns=dataframe.columns)
                self._connection.connection.commit()
                del rows
                gc.collect()
        else:
            dataframe.to_sql(
                table,
                self._connection,

After Change

                    tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".csv.gz")
                    tmp.close()
                    batch.to_csv(tmp.name, index=False, header=False, sep="|", na_rep="\\N", quoting=csv.QUOTE_NONE, compression="gzip")
                    self._connection.connection.cursor().execute("PUT file://%(path)s @~/staged;" % {"path": tmp.name})
                    self._connection.connection.cursor().execute(
                        "COPY INTO %(table)s "
                        "FROM @~/staged "
                        "FILE_FORMAT = (TYPE = CSV FIELD_DELIMITER = \"|\" SKIP_HEADER = 0 COMPRESSION = GZIP) "
                        "PURGE = TRUE" % {
                            "table": table
                        }
                    )
                finally:
                    os.remove(tmp.name)

Instances