Fix DbApiHook.insert_rows logging incorrect number of rows commited (a…

…pache#45789)

* Fix DbApiHook.insert_rows logging incorrect number of rows commited

* add test for checking the log message matches the amount of rows committed

providers/common/sql/src/airflow/providers/common/sql/hooks/sql.py

@@ -744,8 +744,8 @@ def insert_rows(
                         self.log.debug("Generated sql: %s", sql)
                         cur.executemany(sql, values)
                         conn.commit()
-                        self.log.info("Loaded %s rows into %s so far", len(chunked_rows), table)
                         nb_rows += len(chunked_rows)
+                        self.log.info("Loaded %s rows into %s so far", nb_rows, table)
                 else:
                     for i, row in enumerate(rows, 1):
                         values = self._serialize_cells(row, conn)

providers/common/sql/tests/provider_tests/common/sql/hooks/test_dbapi.py

@@ -602,3 +602,15 @@ def test_run_fetch_one_handler_print(self):
 
         self.cur.fetchone.side_effect = Exception("Should not get called !")
         assert rows == self.db_hook.run(sql=query, handler=fetch_one_handler)
+
+    def test_insert_rows_with_executemany_correctly_logs_amount_of_commited_rows(self, caplog):
+        table = "table"
+        rows = [("test",) for _ in range(9)]
+
+        with caplog.at_level(logging.DEBUG):
+            self.db_hook.supports_executemany = True
+            self.db_hook.insert_rows(table, iter(rows), commit_every=3)
+
+        assert any(f"Loaded 3 rows into {table} so far" in message for message in caplog.messages)
+        assert any(f"Loaded 6 rows into {table} so far" in message for message in caplog.messages)
+        assert any(f"Loaded 9 rows into {table} so far" in message for message in caplog.messages)