Repository: OpenEnergyPlatform/open-MaStR
Issue category: Performance Issue

Issue:
Increase parsing speed
This task contains several steps:

1. Search different ways that might increase parsing speed. Parsing is done right now by the `pandas.read_xml` method [here](https://github.com/OpenEnergyPlatform/open-MaStR/blob/d17a92c17912f7f96814d309859dc495f942c07a/open_mastr/xml_download/utils_write_to_database.py#L158). Several alternatives are:
  * `polars`, `duckdb`, `pyspark` might have xml parsers and might be faster
  * use plain xml parsing from python (without pandas)
  * ...

2. Writing to sqlite database right now is done by `pandas.to_sql` [here](https://github.com/OpenEnergyPlatform/open-MaStR/blob/d17a92c17912f7f96814d309859dc495f942c07a/open_mastr/xml_download/utils_write_to_database.py#L182C5-L182C26). There might be other faster methods depending on step 1.
3. Construct a benchmark in an own repository. Use a benchmark xml file from the Marktstammdatenregister and test different implementations for parsing them.
4. Decide for a best method and implement it in `open-mastr`


Hints:
Hi! I started working on this task and decided to change the steps slightly:

1. Construct the benchmark
    - Use the Marktstammdatenregister to construct a few datasets of various size - ✅  ([link](https://github.com/AlexandraImbrisca/open-MaStR/tree/develop/benchmark))
    - Create a script to automate the calculation and comparison of the parsing speed between various optimisations - ✅  ([link](https://github.com/AlexandraImbrisca/open-MaStR/pull/6))
2. Explore faster methods of parsing the XML
     - Research the options and implement the changes
     - Run the benchmark and analyse the results
3. Explore faster methods of writing to the sqlite database
     - Research the options and implement the changes
     - Run the benchmark and analyse the results
4. Decide on the best method and add it to this repository
I was at DACH Energy Informatics Conference and took two points from there:
- Many researchers use open-mastr
- The feature request I heard most often was the question, if we can decrease the time it needs to download and parse the data
I think people will be really happy if this issue is successful 😃 
> I was at DACH Energy Informatics Conference and took two points from there:
> 
> * Many researchers use open-mastr
> * The feature request I heard most often was the question, if we can decrease the time it needs to download and parse the data
>   I think people will be really happy if this issue is successful 😃

Sounds great. I think we cannot do much about the dl speed but I'm really looking forward to the parsing enhancement @AlexandraImbrisca :smile: 

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 42432fd8..404fb9ac 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -13,6 +13,8 @@ and the versioning aims to respect [Semantic Versioning](http://semver.org/spec/
 ### Changed
 - Repair Header image in Readme
   [#587](https://github.com/OpenEnergyPlatform/open-MaStR/pull/587)
+- Increase XML parsing speed
+  [#598](https://github.com/OpenEnergyPlatform/open-MaStR/pull/598)
 - Update bug issue templates
   [#593](https://github.com/OpenEnergyPlatform/open-MaStR/pull/593)
 ### Removed
diff --git a/CITATION.cff b/CITATION.cff
index 591a84ea..25c61bab 100644
--- a/CITATION.cff
+++ b/CITATION.cff
@@ -30,6 +30,10 @@ authors:
     alias: "@nesnoj"
     affiliation: "Reiner Lemoine Institut"
     orcid: " https://orcid.org/0000-0002-8563-5261"
+  - family-names: "Imbrisca"
+    given-names: "Alexandra-Andreea"
+    alias: "@AlexandraImbrisca"
+    affiliation: "Technical University of Munich"
 title: "open-MaStR"
 type: software
 license: AGPL-3.0
diff --git a/open_mastr/xml_download/utils_write_to_database.py b/open_mastr/xml_download/utils_write_to_database.py
index 4917e9d9..d0e9b524 100644
--- a/open_mastr/xml_download/utils_write_to_database.py
+++ b/open_mastr/xml_download/utils_write_to_database.py
@@ -1,3 +1,4 @@
+from io import StringIO
 from shutil import Error
 from zipfile import ZipFile
 
@@ -5,14 +6,14 @@
 import numpy as np
 import pandas as pd
 import sqlalchemy
-from sqlalchemy import select
+from sqlalchemy import inspect, select
 from sqlalchemy.sql import text
+from sqlalchemy.sql.sqltypes import Date, DateTime
 
 from open_mastr.utils.config import setup_logger
 from open_mastr.utils.helpers import data_to_include_tables
 from open_mastr.utils.orm import tablename_mapping
 from open_mastr.xml_download.utils_cleansing_bulk import cleanse_bulk_data
-from io import StringIO
 
 
 def write_mastr_xml_to_database(
@@ -26,74 +27,79 @@ def write_mastr_xml_to_database(
     include_tables = data_to_include_tables(data, mapping="write_xml")
 
     with ZipFile(zipped_xml_file_path, "r") as f:
-        files_list = f.namelist()
-        files_list = correct_ordering_of_filelist(files_list)
+        files_list = correct_ordering_of_filelist(f.namelist())
+
         for file_name in files_list:
-            # xml_tablename is the beginning of the filename without the number in lowercase
-            xml_tablename = file_name.split("_")[0].split(".")[0].lower()
-
-            if is_table_relevant(
-                xml_tablename=xml_tablename, include_tables=include_tables
-            ):
-                sql_tablename = tablename_mapping[xml_tablename]["__name__"]
-
-                if is_first_file(file_name):
-                    create_database_table(engine=engine, xml_tablename=xml_tablename)
-                    print(
-                        f"Table '{sql_tablename}' is filled with data '{xml_tablename}' "
-                        "from the bulk download."
-                    )
-                print(f"File '{file_name}' is parsed.")
+            xml_table_name = extract_xml_table_name(file_name)
 
-                df = preprocess_table_for_writing_to_database(
-                    f=f,
-                    file_name=file_name,
-                    xml_tablename=xml_tablename,
-                    bulk_download_date=bulk_download_date,
-                )
+            if not is_table_relevant(xml_table_name, include_tables):
+                continue
 
-                # Convert date and datetime columns into the datatype datetime
-                df = cast_date_columns_to_datetime(xml_tablename, df)
+            sql_table_name = extract_sql_table_name(xml_table_name)
 
-                if bulk_cleansing:
-                    df = cleanse_bulk_data(df, zipped_xml_file_path)
+            if is_first_file(file_name):
+                create_database_table(engine, xml_table_name)
+                print(
+                    f"Table '{sql_table_name}' is filled with data '{xml_table_name}' "
+                    "from the bulk download."
+                )
+            print(f"File '{file_name}' is parsed.")
+
+            df = read_xml_file(f, file_name)
+            df = process_table_before_insertion(
+                df,
+                xml_table_name,
+                zipped_xml_file_path,
+                bulk_download_date,
+                bulk_cleansing,
+            )
 
-                add_table_to_database(
-                    df=df,
-                    xml_tablename=xml_tablename,
-                    sql_tablename=sql_tablename,
-                    if_exists="append",
-                    engine=engine,
+            if engine.dialect.name == "sqlite":
+                add_table_to_sqlite_database(df, xml_table_name, sql_table_name, engine)
+            else:
+                add_table_to_non_sqlite_database(
+                    df, xml_table_name, sql_table_name, engine
                 )
+
     print("Bulk download and data cleansing were successful.")
 
 
-def is_table_relevant(xml_tablename: str, include_tables: list) -> bool:
+def extract_xml_table_name(file_name: str) -> str:
+    """Extract the XML table name from the file name."""
+    return file_name.split("_")[0].split(".")[0].lower()
+
+
+def extract_sql_table_name(xml_table_name: str) -> str:
+    """Extract the SQL table name from the xml table name."""
+    return tablename_mapping[xml_table_name]["__name__"]
+
+
+def is_table_relevant(xml_table_name: str, include_tables: list) -> bool:
     """Checks if the table contains relevant data and if the user wants to
     have it in the database."""
     # few tables are only needed for data cleansing of the xml files and contain no
     # information of relevance
     try:
         boolean_write_table_to_sql_database = (
-            tablename_mapping[xml_tablename]["__class__"] is not None
+            tablename_mapping[xml_table_name]["__class__"] is not None
         )
     except KeyError:
         print(
-            f"Table '{xml_tablename}' is not supported by your open-mastr version and "
+            f"Table '{xml_table_name}' is not supported by your open-mastr version and "
             f"will be skipped."
         )
         return False
     # check if the table should be written to sql database (depends on user input)
-    include_count = include_tables.count(xml_tablename)
+    include_count = include_tables.count(xml_table_name)
 
     return include_count == 1 and boolean_write_table_to_sql_database
 
 
-def create_database_table(engine: sqlalchemy.engine.Engine, xml_tablename: str) -> None:
-    orm_class = tablename_mapping[xml_tablename]["__class__"]
-    # drop the content from table
+def create_database_table(
+    engine: sqlalchemy.engine.Engine, xml_table_name: str
+) -> None:
+    orm_class = tablename_mapping[xml_table_name]["__class__"]
     orm_class.__table__.drop(engine, checkfirst=True)
-    # create table schema
     orm_class.__table__.create(engine)
 
 
@@ -105,8 +111,10 @@ def is_first_file(file_name: str) -> bool:
     )
 
 
-def cast_date_columns_to_datetime(xml_tablename: str, df: pd.DataFrame) -> pd.DataFrame:
-    sqlalchemy_columnlist = tablename_mapping[xml_tablename][
+def cast_date_columns_to_datetime(
+    xml_table_name: str, df: pd.DataFrame
+) -> pd.DataFrame:
+    sqlalchemy_columnlist = tablename_mapping[xml_table_name][
         "__class__"
     ].__table__.columns.items()
     for column in sqlalchemy_columnlist:
@@ -117,15 +125,31 @@ def cast_date_columns_to_datetime(xml_tablename: str, df: pd.DataFrame) -> pd.Da
     return df
 
 
+def cast_date_columns_to_string(xml_table_name: str, df: pd.DataFrame) -> pd.DataFrame:
+    column_list = tablename_mapping[xml_table_name][
+        "__class__"
+    ].__table__.columns.items()
+    for column in column_list:
+        column_name = column[0]
+
+        if not (column[0] in df.columns and is_date_column(column, df)):
+            continue
+
+        df[column_name] = pd.to_datetime(df[column_name], errors="coerce")
+
+        if type(column[1].type) is Date:
+            df[column_name] = (
+                df[column_name].dt.strftime("%Y-%m-%d").replace("NaT", None)
+            )
+        elif type(column[1].type) is DateTime:
+            df[column_name] = (
+                df[column_name].dt.strftime("%Y-%m-%d %H:%M:%S.%f").replace("NaT", None)
+            )
+    return df
+
+
 def is_date_column(column, df: pd.DataFrame) -> bool:
-    return (
-        type(column[1].type)
-        in [
-            sqlalchemy.sql.sqltypes.Date,
-            sqlalchemy.sql.sqltypes.DateTime,
-        ]
-        and column[0] in df.columns
-    )
+    return type(column[1].type) in [Date, DateTime] and column[0] in df.columns
 
 
 def correct_ordering_of_filelist(files_list: list) -> list:
@@ -154,49 +178,35 @@ def correct_ordering_of_filelist(files_list: list) -> list:
     return files_list
 
 
-def preprocess_table_for_writing_to_database(
-    f: ZipFile,
-    file_name: str,
-    xml_tablename: str,
-    bulk_download_date: str,
-) -> pd.DataFrame:
-    data = f.read(file_name)
-    try:
-        df = pd.read_xml(data, encoding="UTF-16", compression="zip")
-    except lxml.etree.XMLSyntaxError as err:
-        df = handle_xml_syntax_error(data.decode("utf-16"), err)
-
-    df = add_zero_as_first_character_for_too_short_string(df)
-    df = change_column_names_to_orm_format(df, xml_tablename)
-
-    # Add Column that refers to the source of the data
-    df["DatenQuelle"] = "bulk"
-    df["DatumDownload"] = bulk_download_date
-    return df
+def read_xml_file(f: ZipFile, file_name: str) -> pd.DataFrame:
+    """Read the xml file from the zip file and return it as a DataFrame."""
+    with f.open(file_name) as xml_file:
+        try:
+            return pd.read_xml(xml_file, encoding="UTF-16", parser="etree")
+        except lxml.etree.XMLSyntaxError as error:
+            return handle_xml_syntax_error(xml_file.read().decode("utf-16"), error)
 
 
 def change_column_names_to_orm_format(
-    df: pd.DataFrame, xml_tablename: str
+    df: pd.DataFrame, xml_table_name: str
 ) -> pd.DataFrame:
-    if tablename_mapping[xml_tablename]["replace_column_names"]:
+    if tablename_mapping[xml_table_name]["replace_column_names"]:
         df.rename(
-            columns=tablename_mapping[xml_tablename]["replace_column_names"],
+            columns=tablename_mapping[xml_table_name]["replace_column_names"],
             inplace=True,
         )
     return df
 
 
-def add_table_to_database(
+def add_table_to_non_sqlite_database(
     df: pd.DataFrame,
-    xml_tablename: str,
-    sql_tablename: str,
-    if_exists: str,
+    xml_table_name: str,
+    sql_table_name: str,
     engine: sqlalchemy.engine.Engine,
 ) -> None:
     # get a dictionary for the data types
-
     table_columns_list = list(
-        tablename_mapping[xml_tablename]["__class__"].__table__.columns
+        tablename_mapping[xml_table_name]["__class__"].__table__.columns
     )
     dtypes_for_writing_sql = {
         column.name: column.type
@@ -204,16 +214,22 @@ def add_table_to_database(
         if column.name in df.columns
     }
 
-    add_missing_columns_to_table(engine, xml_tablename, column_list=df.columns.tolist())
+    # Convert date and datetime columns into the datatype datetime.
+    df = cast_date_columns_to_datetime(xml_table_name, df)
+
+    add_missing_columns_to_table(
+        engine, xml_table_name, column_list=df.columns.tolist()
+    )
+
     for _ in range(10000):
         try:
             with engine.connect() as con:
                 with con.begin():
                     df.to_sql(
-                        sql_tablename,
+                        sql_table_name,
                         con=con,
                         index=False,
-                        if_exists=if_exists,
+                        if_exists="append",
                         dtype=dtypes_for_writing_sql,
                     )
                     break
@@ -224,7 +240,7 @@ def add_table_to_database(
         except sqlalchemy.exc.IntegrityError:
             # error resulting from Unique constraint failed
             df = write_single_entries_until_not_unique_comes_up(
-                df=df, xml_tablename=xml_tablename, engine=engine
+                df, xml_table_name, engine
             )
 
 
@@ -261,14 +277,14 @@ def add_zero_as_first_character_for_too_short_string(df: pd.DataFrame) -> pd.Dat
 
 
 def write_single_entries_until_not_unique_comes_up(
-    df: pd.DataFrame, xml_tablename: str, engine: sqlalchemy.engine.Engine
+    df: pd.DataFrame, xml_table_name: str, engine: sqlalchemy.engine.Engine
 ) -> pd.DataFrame:
     """
     Remove from dataframe these rows, which are already existing in the database table
     Parameters
     ----------
     df
-    xml_tablename
+    xml_table_name
     engine
 
     Returns
@@ -276,7 +292,7 @@ def write_single_entries_until_not_unique_comes_up(
     Filtered dataframe
     """
 
-    table = tablename_mapping[xml_tablename]["__class__"].__table__
+    table = tablename_mapping[xml_table_name]["__class__"].__table__
     primary_key = next(c for c in table.columns if c.primary_key)
 
     with engine.connect() as con:
@@ -295,14 +311,14 @@ def write_single_entries_until_not_unique_comes_up(
         labels=key_list, errors="ignore"
     )  # drop primary keys that already exist in the table
     df = df.reset_index()
-    print(f"{len_df_before-len(df)} entries already existed in the database.")
+    print(f"{len_df_before - len(df)} entries already existed in the database.")
 
     return df
 
 
 def add_missing_columns_to_table(
     engine: sqlalchemy.engine.Engine,
-    xml_tablename: str,
+    xml_table_name: str,
     column_list: list,
 ) -> None:
     """
@@ -312,8 +328,8 @@ def add_missing_columns_to_table(
     Parameters
     ----------
     engine
-    xml_tablename
-    df
+    xml_table_name
+    column_list
 
     Returns
     -------
@@ -323,7 +339,7 @@ def add_missing_columns_to_table(
 
     # get the columns name from the existing database
     inspector = sqlalchemy.inspect(engine)
-    table_name = tablename_mapping[xml_tablename]["__class__"].__table__.name
+    table_name = tablename_mapping[xml_table_name]["__class__"].__table__.name
     columns = inspector.get_columns(table_name)
     column_names_from_database = [column["name"] for column in columns]
 
@@ -392,3 +408,59 @@ def find_nearest_brackets(xml_string: str, position: int) -> tuple[int, int]:
             continue
 
     raise Error("An error occured when parsing the xml file. Maybe it is corrupted?")
+
+
+def process_table_before_insertion(
+    df: pd.DataFrame,
+    xml_table_name: str,
+    zipped_xml_file_path: str,
+    bulk_download_date: str,
+    bulk_cleansing: bool,
+) -> pd.DataFrame:
+    df = add_zero_as_first_character_for_too_short_string(df)
+    df = change_column_names_to_orm_format(df, xml_table_name)
+
+    # Add Column that refers to the source of the data
+    df["DatenQuelle"] = "bulk"
+    df["DatumDownload"] = bulk_download_date
+
+    if bulk_cleansing:
+        df = cleanse_bulk_data(df, zipped_xml_file_path)
+    return df
+
+
+def add_table_to_sqlite_database(
+    df: pd.DataFrame,
+    xml_table_name: str,
+    sql_table_name: str,
+    engine: sqlalchemy.engine.Engine,
+) -> None:
+    column_list = df.columns.tolist()
+    add_missing_columns_to_table(engine, xml_table_name, column_list)
+
+    # Convert NaNs to None.
+    df = df.where(pd.notnull(df), None)
+
+    # Convert date columns to strings. Dates are not supported directly by SQLite.
+    df = cast_date_columns_to_string(xml_table_name, df)
+
+    # Create SQL statement for bulk insert. ON CONFLICT DO NOTHING prevents duplicates.
+    insert_stmt = f"INSERT INTO {sql_table_name} ({','.join(column_list)}) VALUES ({','.join(['?' for _ in column_list])}) ON CONFLICT DO NOTHING"
+
+    for _ in range(10000):
+        try:
+            with engine.connect() as con:
+                with con.begin():
+                    con.connection.executemany(insert_stmt, df.to_numpy())
+                    break
+        except sqlalchemy.exc.DataError as err:
+            delete_wrong_xml_entry(err, df)
+        except sqlalchemy.exc.IntegrityError:
+            # error resulting from Unique constraint failed
+            df = write_single_entries_until_not_unique_comes_up(
+                df, xml_table_name, engine
+            )
+        except:
+            # If any unexpected error occurs, we'll switch back to the non-SQLite method.
+            add_table_to_non_sqlite_database(df, xml_table_name, sql_table_name, engine)
+            break


Test patch:
diff --git a/tests/xml_download/test_utils_write_to_database.py b/tests/xml_download/test_utils_write_to_database.py
index dc5aaacd..e72e5501 100644
--- a/tests/xml_download/test_utils_write_to_database.py
+++ b/tests/xml_download/test_utils_write_to_database.py
@@ -1,25 +1,35 @@
+import os
+import sqlite3
 import sys
+from datetime import datetime
+from os.path import expanduser
 from zipfile import ZipFile
 
+import numpy as np
+import pandas as pd
+import pytest
+from sqlalchemy import create_engine, inspect
+from sqlalchemy.sql import text
+
 from open_mastr.utils import orm
-from open_mastr.xml_download.utils_cleansing_bulk import (
-    replace_mastr_katalogeintraege,
-)
+from open_mastr.utils.orm import RetrofitUnits, NuclearExtended, tablename_mapping
 from open_mastr.xml_download.utils_write_to_database import (
-    cast_date_columns_to_datetime,
-    preprocess_table_for_writing_to_database,
-    add_table_to_database,
+    add_missing_columns_to_table,
     add_zero_as_first_character_for_too_short_string,
+    cast_date_columns_to_string,
+    change_column_names_to_orm_format,
     correct_ordering_of_filelist,
+    create_database_table,
+    extract_sql_table_name,
+    extract_xml_table_name,
+    is_date_column,
+    is_first_file,
+    is_table_relevant,
+    process_table_before_insertion,
+    read_xml_file,
+    add_table_to_non_sqlite_database,
+    add_table_to_sqlite_database,
 )
-import os
-from os.path import expanduser
-import sqlite3
-from sqlalchemy import create_engine
-import pandas as pd
-import pytest
-import numpy as np
-from datetime import datetime
 
 # Check if xml file exists
 _xml_file_exists = False
@@ -70,103 +80,93 @@ def engine_testdb():
     yield create_engine(testdb_url)
 
 
-@pytest.mark.skipif(
-    not _xml_file_exists, reason="The zipped xml file could not be found."
-)
-def test_preprocess_table_for_writing_to_database(zipped_xml_file_path):
-    # Prepare
-    file_name = "EinheitenKernkraft.xml"
-    xml_tablename = "einheitenkernkraft"
-    bulk_download_date = zipped_xml_file_path.split("_")[-1].replace(".zip", "")
-    # Check if bulk_download_date is derived correctly like 20220323
-    assert len(bulk_download_date) == 8
+def test_extract_xml_table_name():
+    file_name = "Netzanschlusspunkte_31.xml"
+    assert extract_xml_table_name(file_name) == "netzanschlusspunkte"
 
-    with ZipFile(zipped_xml_file_path, "r") as f:
-        # Act
-        df = preprocess_table_for_writing_to_database(
-            f=f,
-            file_name=file_name,
-            xml_tablename=xml_tablename,
-            bulk_download_date=bulk_download_date,
-        )
 
-    # Assert
-    assert df["DatenQuelle"].unique().tolist() == ["bulk"]
-    assert df["DatumDownload"].unique().tolist() == [bulk_download_date]
-    assert df["Gemeindeschluessel"].apply(len).unique().tolist() == [8]
-    assert df["Postleitzahl"].apply(len).unique().tolist() == [5]
+def text_extract_sql_table_name():
+    xml_table_name = "netzanschlusspunkte"
+    assert extract_sql_table_name(xml_table_name) == "network_connection_points"
 
 
-@pytest.mark.skipif(
-    not _xml_file_exists, reason="The zipped xml file could not be found."
-)
-def test_add_table_to_database(zipped_xml_file_path, engine_testdb):
-    # Prepare
+def test_is_table_relevant():
+    include_tables = ["anlagengasspeicher", "marktakteure"]
+    assert is_table_relevant("anlagengasspeicher", include_tables) is True
+    assert is_table_relevant("netzanschlusspunkte", include_tables) is False
+
+
+def test_create_database_table(engine_testdb):
     orm.Base.metadata.create_all(engine_testdb)
-    file_name = "EinheitenKernkraft.xml"
-    xml_tablename = "einheitenkernkraft"
-    sql_tablename = "nuclear_extended"
-    bulk_download_date = zipped_xml_file_path.split("_")[-1].replace(".zip", "")
-    # Check if bulk_download_date is derived correctly like 20220323
-    assert len(bulk_download_date) == 8
-    with ZipFile(zipped_xml_file_path, "r") as f:
-        df_write = preprocess_table_for_writing_to_database(
-            f=f,
-            file_name=file_name,
-            xml_tablename=xml_tablename,
-            bulk_download_date=bulk_download_date,
-        )
+    xml_table_name = "einheitenkernkraft"
+    sql_table_name = "nuclear_extended"
 
-    # Convert date and datetime columns into the datatype datetime
-    df_write = cast_date_columns_to_datetime(xml_tablename, df_write)
+    create_database_table(engine_testdb, xml_table_name)
 
-    # Katalogeintraege: int -> string value
-    df_write = replace_mastr_katalogeintraege(
-        zipped_xml_file_path=zipped_xml_file_path, df=df_write
-    )
+    assert inspect(engine_testdb).has_table(sql_table_name) is True
 
-    # Act
-    add_table_to_database(
-        df=df_write,
-        xml_tablename=xml_tablename,
-        sql_tablename=sql_tablename,
-        if_exists="replace",
-        engine=engine_testdb,
+
+def test_is_first_file():
+    assert is_first_file("EinheitenKernkraft.xml") is True
+    assert is_first_file("EinheitenKernkraft_1.xml") is True
+    assert is_first_file("EinheitenKernkraft_2.xml") is False
+
+
+def test_cast_date_columns_to_string():
+    initial_df = pd.DataFrame(
+        {
+            "EegMastrNummer": [1, 2, 3],
+            "Registrierungsdatum": [
+                datetime(2024, 3, 11).date(),
+                datetime(1999, 2, 1).date(),
+                np.datetime64("nat"),
+            ],
+            "DatumLetzteAktualisierung": [
+                datetime(2022, 3, 22),
+                datetime(2020, 1, 2, 10, 12, 46),
+                np.datetime64("nat"),
+            ],
+        }
     )
-    with engine_testdb.connect() as con:
-        with con.begin():
-            df_read = pd.read_sql_table(table_name=sql_tablename, con=con)
-    # Drop the empty columns which come from orm and are not filled by bulk download
-    df_read.dropna(how="all", axis=1, inplace=True)
-    # Rename LokationMaStRNummer -> LokationMastrNummer unresolved error
-    df_write.rename(
-        columns={"LokationMaStRNummer": "LokationMastrNummer"}, inplace=True
-    )
-    # Reorder columns of the df_write since the order of columns doesn't play a role
-    df_read = df_read[df_write.columns.tolist()]
-    # Cast the dtypes of df_write like df_read dtypes
-    # Datatypes are changed during inserting via sqlalchemy and orm
-    df_read = df_read.astype(dict(zip(df_write.columns.tolist(), df_write.dtypes)))
-    # Assert
+    expected_df = pd.DataFrame(
+        {
+            "EegMastrNummer": [1, 2, 3],
+            "Registrierungsdatum": ["2024-03-11", "1999-02-01", np.nan],
+            "DatumLetzteAktualisierung": [
+                "2022-03-22 00:00:00.000000",
+                "2020-01-02 10:12:46.000000",
+                np.nan,
+            ],
+        }
+    )
+
     pd.testing.assert_frame_equal(
-        df_write.select_dtypes(exclude="datetime"),
-        df_read.select_dtypes(exclude="datetime"),
+        expected_df, cast_date_columns_to_string("anlageneegwasser", initial_df)
     )
 
 
-def test_add_zero_as_first_character_for_too_short_string():
-    # Prepare
-    df_raw = pd.DataFrame(
-        {"ID": [0, 1, 2], "Gemeindeschluessel": [9162000, np.nan, 19123456]}
-    )
-    df_correct = pd.DataFrame(
-        {"ID": [0, 1, 2], "Gemeindeschluessel": ["09162000", np.nan, "19123456"]}
+def test_is_date_column():
+    columns = RetrofitUnits.__table__.columns.items()
+    df = pd.DataFrame(
+        {
+            "Id": [1],
+            "DatumLetzteAktualisierung": [datetime(2022, 3, 22)],
+            "WiederinbetriebnahmeDatum": [datetime(2024, 3, 11).date()],
+        }
     )
 
-    # Act
-    df_edited = add_zero_as_first_character_for_too_short_string(df_raw)
-    # Assert
-    pd.testing.assert_frame_equal(df_edited, df_correct)
+    date_column = list(filter(lambda col: col[0] == "Id", columns))[0]
+    assert is_date_column(date_column, df) is False
+
+    datetime_column = list(
+        filter(lambda col: col[0] == "DatumLetzteAktualisierung", columns)
+    )[0]
+    assert is_date_column(datetime_column, df) is True
+
+    date_column = list(
+        filter(lambda col: col[0] == "WiederinbetriebnahmeDatum", columns)
+    )[0]
+    assert is_date_column(date_column, df) is True
 
 
 def test_correct_ordering_of_filelist():
@@ -215,24 +215,181 @@ def test_correct_ordering_of_filelist():
     ]
 
 
-def test_cast_date_columns_to_datetime():
+@pytest.mark.skipif(
+    not _xml_file_exists, reason="The zipped xml file could not be found."
+)
+def test_read_xml_file(zipped_xml_file_path):
+    with ZipFile(zipped_xml_file_path, "r") as f:
+        df = read_xml_file(f, "EinheitenKernkraft.xml")
+
+    assert df.shape[0] > 0
+
+    # Since the file is from the latest download, its content can vary over time. To make sure that the table is
+    # correctly created, we check that all of its columns are associated are included in our mapping.
+    for column in df.columns:
+        if column in tablename_mapping["einheitenkernkraft"]["replace_column_names"]:
+            column = tablename_mapping["einheitenkernkraft"]["replace_column_names"][
+                column
+            ]
+        assert column in NuclearExtended.__table__.columns.keys()
+
+
+def test_add_zero_as_first_character_for_too_short_string():
+    # Prepare
     df_raw = pd.DataFrame(
+        {"ID": [0, 1, 2], "Gemeindeschluessel": [9162000, np.nan, 19123456]}
+    )
+    df_correct = pd.DataFrame(
+        {"ID": [0, 1, 2], "Gemeindeschluessel": ["09162000", np.nan, "19123456"]}
+    )
+
+    # Act
+    df_edited = add_zero_as_first_character_for_too_short_string(df_raw)
+    # Assert
+    pd.testing.assert_frame_equal(df_edited, df_correct)
+
+
+def test_change_column_names_to_orm_format():
+    initial_df = pd.DataFrame(
         {
-            "ID": [0, 1, 2],
-            "Registrierungsdatum": ["2022-03-22", "2020-01-02", "2022-03-35"],
+            "VerknuepfteEinheitenMaStRNummern": ["test1", "test2"],
+            "NetzanschlusspunkteMaStRNummern": [1, 2],
         }
     )
-    df_replaced = pd.DataFrame(
+    expected_df = pd.DataFrame(
         {
-            "ID": [0, 1, 2],
-            "Registrierungsdatum": [
-                datetime(2022, 3, 22),
-                datetime(2020, 1, 2),
-                np.datetime64("nat"),
-            ],
+            "VerknuepfteEinheiten": ["test1", "test2"],
+            "Netzanschlusspunkte": [1, 2],
+        }
+    )
+
+    pd.testing.assert_frame_equal(
+        expected_df, change_column_names_to_orm_format(initial_df, "lokationen")
+    )
+
+
+def test_process_table_before_insertion(zipped_xml_file_path):
+    bulk_download_date = datetime.now().date().strftime("%Y%m%d")
+    initial_df = pd.DataFrame(
+        {
+            "Gemeindeschluessel": [9162000, 19123456],
+            "Postleitzahl": [1234, 54321],
+            "NameKraftwerk": ["test1", "test2"],
+            "LokationMaStRNummer": ["test3", "test4"],
+        }
+    )
+    expected_df = pd.DataFrame(
+        {
+            "Gemeindeschluessel": ["09162000", "19123456"],
+            "Postleitzahl": ["01234", "54321"],
+            "NameKraftwerk": ["test1", "test2"],
+            "LokationMastrNummer": ["test3", "test4"],
+            "DatenQuelle": ["bulk", "bulk"],
+            "DatumDownload": [bulk_download_date, bulk_download_date],
         }
     )
 
     pd.testing.assert_frame_equal(
-        df_replaced, cast_date_columns_to_datetime("anlageneegwasser", df_raw)
+        expected_df,
+        process_table_before_insertion(
+            initial_df,
+            "einheitenkernkraft",
+            zipped_xml_file_path,
+            bulk_download_date,
+            bulk_cleansing=False,
+        ),
+    )
+
+
+def test_add_missing_columns_to_table(engine_testdb):
+    with engine_testdb.connect() as con:
+        with con.begin():
+            # We must recreate the table to be sure that the new colum is not present.
+            con.execute(text("DROP TABLE IF EXISTS gas_consumer"))
+            create_database_table(engine_testdb, "einheitengasverbraucher")
+
+            initial_data_in_db = pd.DataFrame(
+                {
+                    "EinheitMastrNummer": ["id1"],
+                    "DatumLetzteAktualisierung": [datetime(2022, 2, 2)],
+                }
+            )
+            initial_data_in_db.to_sql(
+                "gas_consumer", con=con, if_exists="append", index=False
+            )
+
+    add_missing_columns_to_table(
+        engine_testdb, "einheitengasverbraucher", ["NewColumn"]
     )
+
+    expected_df = pd.DataFrame(
+        {
+            "EinheitMastrNummer": ["id1"],
+            "DatumLetzteAktualisierung": [datetime(2022, 2, 2)],
+            "NewColumn": [None],
+        }
+    )
+    with engine_testdb.connect() as con:
+        with con.begin():
+            actual_df = pd.read_sql_table("gas_consumer", con=con)
+            # The actual_df will contain more columns than the expected_df, so we can't use assert_frame_equal.
+            assert expected_df.index.isin(actual_df.index).all()
+
+
+@pytest.mark.parametrize(
+    "add_table_to_database_function",
+    [add_table_to_sqlite_database, add_table_to_non_sqlite_database],
+)
+def test_add_table_to_sqlite_database(engine_testdb, add_table_to_database_function):
+    with engine_testdb.connect() as con:
+        with con.begin():
+            # We must recreate the table to be sure that no other data is present.
+            con.execute(text("DROP TABLE IF EXISTS gsgk_eeg"))
+            create_database_table(
+                engine_testdb, "anlageneeggeothermiegrubengasdruckentspannung"
+            )
+
+    df = pd.DataFrame(
+        {
+            "Registrierungsdatum": ["2022-02-02", "2024-03-20"],
+            "EegMastrNummer": ["id1", "id2"],
+            "DatumLetzteAktualisierung": [
+                "2022-12-02 10:10:10.000300",
+                "2024-10-10 00:00:00.000000",
+            ],
+            "AusschreibungZuschlag": [True, False],
+            "Netzbetreiberzuordnungen": ["test1", "test2"],
+            "InstallierteLeistung": [1.0, 100.4],
+        }
+    )
+    expected_df = pd.DataFrame(
+        {
+            "InstallierteLeistung": [1.0, 100.4],
+            "AnlageBetriebsstatus": [None, None],
+            "Registrierungsdatum": [datetime(2022, 2, 2), datetime(2024, 3, 20)],
+            "EegMastrNummer": ["id1", "id2"],
+            "Meldedatum": [np.datetime64("NaT"), np.datetime64("NaT")],
+            "DatumLetzteAktualisierung": [
+                datetime(2022, 12, 2, 10, 10, 10, 300),
+                datetime(2024, 10, 10),
+            ],
+            "EegInbetriebnahmedatum": [np.datetime64("NaT"), np.datetime64("NaT")],
+            "VerknuepfteEinheit": [None, None],
+            "AnlagenschluesselEeg": [None, None],
+            "AusschreibungZuschlag": [True, False],
+            "AnlagenkennzifferAnlagenregister": [None, None],
+            "AnlagenkennzifferAnlagenregister_nv": [None, None],
+            "Netzbetreiberzuordnungen": ["test1", "test2"],
+            "DatenQuelle": [None, None],
+            "DatumDownload": [np.datetime64("NaT"), np.datetime64("NaT")],
+        }
+    )
+
+    add_table_to_database_function(
+        df, "anlageneeggeothermiegrubengasdruckentspannung", "gsgk_eeg", engine_testdb
+    )
+    with engine_testdb.connect() as con:
+        with con.begin():
+            pd.testing.assert_frame_equal(
+                expected_df, pd.read_sql_table("gsgk_eeg", con=con)
+            )