Repository: OpenEnergyPlatform/open-MaStR

Issue:
Verify data integrity after speed improvements
After improving parsing speed #598 and introducing multiprocessing #601 the data integrity needs to be verified.

Discussion:
@FlorianK13: After we have checked data integrity, we could also ask some users (like goal100 for example) to install our pre release version, run the download in parallel and check if their pipeline still works.@nesnoj: I installed 2 new environments from scratch and let them run in parallel:
1. v0.14.5, time for DL+SQLite DB creation: 4220.3 s (no parallelization)
2. current dev (HEAD: e277d4f361f7beab621d5713ba6aeb9a9a8492a4), time for DL+SQLite DB creation: 1803.0 s (4 processes)

The sqldiff summary reveals that data are different only in table `locations_extended`:

<details>
<summary>Click to see sqldiff summary</summary>

`sqldiff --summary open-MaStR_0.14.5.db open-MaStR_dev.db > open-mastr.db.diff.summary`

additional_data_requested: 0 changes, 0 inserts, 0 deletes, 0 unchanged
additional_locations_requested: 0 changes, 0 inserts, 0 deletes, 0 unchanged
balancing_area: 0 changes, 0 inserts, 0 deletes, 1144 unchanged
basic_units: 0 changes, 0 inserts, 0 deletes, 0 unchanged
biomass_eeg: 0 changes, 0 inserts, 0 deletes, 15290 unchanged
biomass_extended: 0 changes, 0 inserts, 0 deletes, 23432 unchanged
changed_dso_assignment: 0 changes, 0 inserts, 0 deletes, 161167 unchanged
combustion_extended: 0 changes, 0 inserts, 0 deletes, 90983 unchanged
deleted_market_actors: 0 changes, 0 inserts, 0 deletes, 171895 unchanged
deleted_units: 0 changes, 0 inserts, 0 deletes, 178316 unchanged
electricity_consumer: 0 changes, 0 inserts, 0 deletes, 903 unchanged
gas_consumer: 0 changes, 0 inserts, 0 deletes, 824 unchanged
gas_producer: 0 changes, 0 inserts, 0 deletes, 354 unchanged
gas_storage: 0 changes, 0 inserts, 0 deletes, 53 unchanged
gas_storage_extended: 0 changes, 0 inserts, 0 deletes, 54 unchanged
grid_connections: 0 changes, 0 inserts, 0 deletes, 4544090 unchanged
grids: 0 changes, 0 inserts, 0 deletes, 1728 unchanged
gsgk_eeg: 0 changes, 0 inserts, 0 deletes, 172 unchanged
gsgk_extended: 0 changes, 0 inserts, 0 deletes, 435 unchanged
hydro_eeg: 0 changes, 0 inserts, 0 deletes, 7425 unchanged
hydro_extended: 0 changes, 0 inserts, 0 deletes, 8739 unchanged
kwk: 0 changes, 0 inserts, 0 deletes, 89478 unchanged
locations_basic: 0 changes, 0 inserts, 0 deletes, 0 unchanged
**locations_extended: 471257 changes, 0 inserts, 0 deletes, 5000000 unchanged**
market_actors: 0 changes, 0 inserts, 0 deletes, 4477572 unchanged
market_roles: 0 changes, 0 inserts, 0 deletes, 13813 unchanged
missed_additional_data: 0 changes, 0 inserts, 0 deletes, 0 unchanged
missed_extended_location_data: 0 changes, 0 inserts, 0 deletes, 0 unchanged
nuclear_extended: 0 changes, 0 inserts, 0 deletes, 6 unchanged
permit: 0 changes, 0 inserts, 0 deletes, 33629 unchanged
retrofit_units: 0 changes, 0 inserts, 0 deletes, 1515 unchanged
solar_eeg: 0 changes, 0 inserts, 0 deletes, 5027656 unchanged
solar_extended: 0 changes, 0 inserts, 0 deletes, 5055620 unchanged
storage_eeg: 0 changes, 0 inserts, 0 deletes, 1768910 unchanged
storage_extended: 0 changes, 0 inserts, 0 deletes, 1816501 unchanged
storage_units: 0 changes, 0 inserts, 0 deletes, 1816417 unchanged
wind_eeg: 0 changes, 0 inserts, 0 deletes, 37046 unchanged
wind_extended: 0 changes, 0 inserts, 0 deletes, 38542 unchanged
</details>

I checked some of the rows emitted by `sqldiff open-MaStR_0.14.5.db open-MaStR_dev.db > open-mastr.db.diff` by querying and they are equal - it seems that only the order in the table differs which makes sense due to concurrent writing.

Manually comparing the ordered tables confirms that data are identical:
```
sqlite3 open-MaStR_0.14.5.db "SELECT * FROM locations_extended ORDER BY MastrNummer;" > open-MaStR_0.14.5_locations_extended.txt
sqlite3 open-MaStR_dev.db "SELECT * FROM locations_extended ORDER BY MastrNummer;" > open-MaStR_dev_locations_extended.txt
diff -s open-MaStR_0.14.5_locations_extended.txt open-MaStR_dev_locations_extended.txt

--> Dateien open-MaStR_0.14.5_locations_extended.txt und open-MaStR_dev_locations_extended.txt sind identisch.
```

I'll check integrity with the postgres DB too.
@nesnoj: I compared the ordered table `locations_extended` with psql DB and it is consistent. 

The only drawback are the different orders when mp was employed for the import. From my point of view this is not crucial, what do you think @FlorianK13 ?@nesnoj: @FlorianK13 @FlorianK13: Looks nice, thanks for your work :) What do you suggest on how we proceed here?
@nesnoj: I close this issue then, heading for a pre-release soonish

Solution patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3c21a3a..81d5c92 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -19,6 +19,8 @@ and the versioning aims to respect [Semantic Versioning](http://semver.org/spec/
   [#593](https://github.com/OpenEnergyPlatform/open-MaStR/pull/593)
 - Delete Google Fonts from Documentation Page
   [#599](https://github.com/OpenEnergyPlatform/open-MaStR/pull/599)
+- Use multiprocessing to speed up parsing 
+  [#601](https://github.com/OpenEnergyPlatform/open-MaStR/pull/601)
 ### Removed
 - Moved old code artefacts from `scripts` folder to paper specific 
   [repository](https://github.com/FlorianK13/verify-marktstammdaten) 
diff --git a/docs/advanced.md b/docs/advanced.md
index a0db651..1ad077c 100644
--- a/docs/advanced.md
+++ b/docs/advanced.md
@@ -92,10 +92,14 @@ For more information regarding the database see [Database settings](#database-se
 
 There are some environment variables to customize open-MaStR:
 
-| Variable               | Description                                                                                                                                                              | Example                                                                                                                    |
-|------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------|
-| `SQLITE_DATABASE_PATH` | Path to the SQLite file. This allows to use to use multiple instances of the MaStR database. The database instances exist in parallel and are independent of each other. | `/home/mastr-rabbit/.open-MaStR/data/sqlite/your_custom_instance_name.db`                                                  |
-| `OUTPUT_PATH`          | Path to user-defined output directory for CSV data, XML file and database. If not specified, output directory defaults to `$HOME/.open-MaStR/`                           | Linux: `/home/mastr-rabbit/open-mastr-user-defined-output-path`, Windows: `C:\\Users\\open-mastr-user-defined-output-path` |
+| Variable                              | Description                                                                                                                                                                                                                | Example                                                                                                                    |
+|---------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------|
+| `SQLITE_DATABASE_PATH`                | Path to the SQLite file. This allows to use to use multiple instances of the MaStR database. The database instances exist in parallel and are independent of each other.                                                   | `/home/mastr-rabbit/.open-MaStR/data/sqlite/your_custom_instance_name.db`                                                  |
+| `OUTPUT_PATH`                         | Path to user-defined output directory for CSV data, XML file and database. If not specified, output directory defaults to `$HOME/.open-MaStR/`                                                                             | Linux: `/home/mastr-rabbit/open-mastr-user-defined-output-path`, Windows: `C:\\Users\\open-mastr-user-defined-output-path` |
+| `USE_RECOMMENDED_NUMBER_OF_PROCESSES` | If set to `True`, the number of processes used for the bulk download is set to the recommended number of processes. If set to `False`, the number of processes is 1 if not otherwise configured via `NUMBER_OF_PROCESSES`. | `True` or `False`                                                                                                          |
+| `NUMBER_OF_PROCESSES`                 | Number of processed to be used for the bulk download processing.                                                                                                                                                           | `4`                                                                                                                        |
+
+**Note**: When using `USE_RECOMMENDED_NUMBER_OF_PROCESSES` or `NUMBER_OF_PROCESSES` on Windows or MacOS, please make sure that you are wrapping the method call under an [if \_\_name\_\_ == "\_\_main\_\_"](https://realpython.com/if-name-main-python/) as shown in the [example](https://github.com/OpenEnergyPlatform/open-MaStR/blob/11fb568879ae8015af74f3d6b386faa5027c2721/main.py#L73).
 
 ## Bulk download
 
diff --git a/main.py b/main.py
index 881ec38..88730b4 100644
--- a/main.py
+++ b/main.py
@@ -18,6 +18,12 @@ import os
 # set custom output path for: csv-export, database, xml-export.
 # os.environ['OUTPUT_PATH'] = r"/your/custom/output_path"
 
+# optimize bulk downloads and use the recommended number of processes
+# os.environ['USE_RECOMMENDED_NUMBER_OF_PROCESSES'] = "True"
+
+# set up your own number of processes for bulk download
+# os.environ['NUMBER_OF_PROCESSES'] = "your_number"
+
 # bulk download
 bulk_date = "today"
 bulk_cleansing = True
@@ -67,10 +73,9 @@ api_location_types = [
     "location_gas_consumption",
 ]
 
-# instantiate Mastr class
-db = Mastr()
-
 if __name__ == "__main__":
+    # instantiate Mastr class
+    db = Mastr()
 
     ## download Markstammdatenregister
     # bulk download
diff --git a/open_mastr/mastr.py b/open_mastr/mastr.py
index 592701c..646a50b 100644
--- a/open_mastr/mastr.py
+++ b/open_mastr/mastr.py
@@ -235,6 +235,12 @@ class Mastr:
             )
             download_xml_Mastr(zipped_xml_file_path, date, xml_folder_path)
 
+            print(
+                f"\nWould you like to speed up the bulk download?\n"
+                f"Try our new parallelized processing by setting os.environ['USE_RECOMMENDED_NUMBER_OF_PROCESSES'] = True "
+                f"or configure your own number of processes via os.environ['NUMBER_OF_PROCESSES'] = your_number\n"
+            )
+
             write_mastr_xml_to_database(
                 engine=self.engine,
                 zipped_xml_file_path=zipped_xml_file_path,
diff --git a/open_mastr/xml_download/utils_cleansing_bulk.py b/open_mastr/xml_download/utils_cleansing_bulk.py
index 2a304ef..b48a50f 100644
--- a/open_mastr/xml_download/utils_cleansing_bulk.py
+++ b/open_mastr/xml_download/utils_cleansing_bulk.py
@@ -8,7 +8,6 @@ from zipfile import ZipFile
 
 
 def cleanse_bulk_data(df: pd.DataFrame, zipped_xml_file_path: str) -> pd.DataFrame:
-    print("Data is cleansed.")
     df = replace_ids_with_names(df, system_catalog)
     # Katalogeintraege: int -> string value
     df = replace_mastr_katalogeintraege(
diff --git a/open_mastr/xml_download/utils_write_to_database.py b/open_mastr/xml_download/utils_write_to_database.py
index d0e9b52..2963e6b 100644
--- a/open_mastr/xml_download/utils_write_to_database.py
+++ b/open_mastr/xml_download/utils_write_to_database.py
@@ -1,12 +1,16 @@
+import os
+from concurrent.futures import ProcessPoolExecutor, wait
 from io import StringIO
+from multiprocessing import cpu_count
 from shutil import Error
 from zipfile import ZipFile
 
+import re
 import lxml
 import numpy as np
 import pandas as pd
 import sqlalchemy
-from sqlalchemy import inspect, select
+from sqlalchemy import select, create_engine, inspect
 from sqlalchemy.sql import text
 from sqlalchemy.sql.sqltypes import Date, DateTime
 
@@ -24,7 +28,10 @@ def write_mastr_xml_to_database(
     bulk_download_date: str,
 ) -> None:
     """Write the Mastr in xml format into a database defined by the engine parameter."""
+    print("Starting bulk download and data cleansing...")
+
     include_tables = data_to_include_tables(data, mapping="write_xml")
+    threads_data = []
 
     with ZipFile(zipped_xml_file_path, "r") as f:
         files_list = correct_ordering_of_filelist(f.namelist())
@@ -36,15 +43,85 @@ def write_mastr_xml_to_database(
                 continue
 
             sql_table_name = extract_sql_table_name(xml_table_name)
+            threads_data.append(
+                (
+                    file_name,
+                    xml_table_name,
+                    sql_table_name,
+                    str(engine.url),
+                    engine.url.password,
+                    zipped_xml_file_path,
+                    bulk_download_date,
+                    bulk_cleansing,
+                )
+            )
+
+    interleaved_files = interleave_files(threads_data)
+    number_of_processes = get_number_of_processes()
+
+    if number_of_processes > 0:
+        with ProcessPoolExecutor(max_workers=number_of_processes) as executor:
+            futures = [
+                executor.submit(process_xml_file, *item) for item in interleaved_files
+            ]
+            for future in futures:
+                future.result()
+            wait(futures)
+    else:
+        for item in interleaved_files:
+            process_xml_file(*item)
+
+    print("Bulk download and data cleansing were successful.")
+
+
+def get_number_of_processes():
+    """Get the number of processes to use for the bulk download. Returns -1 if the user has not opted for the
+    parallelized implementation. Otherwise, we recommend using the number of available CPUs - 1. If the user wants to
+    use more processes, they can set the custom environment variable."""
+    if "NUMBER_OF_PROCESSES" in os.environ:
+        try:
+            number_of_processes = int(os.environ.get("NUMBER_OF_PROCESSES"))
+        except ValueError:
+            print("Warning: Invalid value for NUMBER_OF_PROCESSES. Fallback to 1.")
+            return 1
+        if number_of_processes >= cpu_count():
+            print(
+                f"Warning: Your system supports {cpu_count()} CPUs. Using "
+                f"more processes than available CPUs may cause excessive "
+                f"context-switching overhead."
+            )
+        return number_of_processes
+    if "USE_RECOMMENDED_NUMBER_OF_PROCESSES" in os.environ:
+        return cpu_count() - 1
+    return -1
+
+
+def process_xml_file(
+    file_name: str,
+    xml_table_name: str,
+    sql_table_name: str,
+    connection_url: str,
+    password: str,
+    zipped_xml_file_path: str,
+    bulk_download_date: str,
+    bulk_cleansing: bool,
+) -> None:
+    """Process a single xml file and write it to the database."""
+    try:
+        # If set, the connection url obfuscates the password. We must replace the masked password with the actual password.
+        if password:
+            connection_url = re.sub(
+                r"://([^:]+):\*+@", r"://\1:" + password + "@", connection_url
+            )
 
+        # Each process will create its own engine to ensure isolation and efficient resource management.
+        # The connection url obfuscates the password. We must replace the masked password with the actual password.
+        engine = create_efficient_engine(connection_url)
+        with ZipFile(zipped_xml_file_path, "r") as f:
+            print(f"Processing file '{file_name}'...")
             if is_first_file(file_name):
+                print(f"Creating table '{sql_table_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,
@@ -53,15 +130,78 @@ def write_mastr_xml_to_database(
                 bulk_download_date,
                 bulk_cleansing,
             )
-
             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
                 )
+    except Exception as e:
+        print(f"Error processing file '{file_name}': '{e}'")
+
+
+def create_efficient_engine(connection_url: str) -> sqlalchemy.engine.Engine:
+    """Create an efficient engine for the SQLite database."""
+    is_sqlite = connection_url.startswith("sqlite://")
+
+    connect_args = {}
+
+    if is_sqlite:
+        # Wait for max 5 minutes before timing out.
+        connect_args["timeout"] = 300
+        # Lock the database only it is necessary to improve concurrency and performance.
+        connect_args["isolation_level"] = "DEFERRED"
+        # Allow multiple threads to access the database.
+        connect_args["check_same_thread"] = False
+    else:
+        # Wait for max 5 minutes before timing out.
+        connect_args["connect_timeout"] = 300
+
+    return create_engine(
+        connection_url,
+        connect_args=connect_args,
+        # Before returning a connection from the pool, check if the connection is still valid.
+        pool_pre_ping=True,
+        # Max number of connections in the pool.
+        pool_size=10,
+        # Create up to 20 more connections when the demand for connections is high.
+        max_overflow=20,
+        # Recycle inactive connections after 180 seconds to prevent stale connections.
+        pool_recycle=180,
+        # Wait for 30 seconds before raising an exception when the pool is full.
+        pool_timeout=30,
+    )
 
-    print("Bulk download and data cleansing were successful.")
+
+def interleave_files(threads_data: list):
+    """
+    Multiple threads will process different files at once. If the files target the same table, the risk of a
+    "database lock" error (i.e., 2 threads attempting to modify the same table at the same time) is increased.
+    To reduce this probability, we can "interleave" the files based on the table they belong to.
+    Example:
+        Initial order: AnlagenEegSolar_1, AnlagenEegSolar_2, ..., AnlagenEegSpeicher_1, AnlagenEegSpeicher_2, ...
+        Interleaved order: AnlagenEegSolar_1, AnlagenEegSpeicher_1, ..., AnlagenEegSolar_2, AnlagenEegSpeicher_2, ...
+    """
+    files_grouped_by_table = {}
+
+    for item in threads_data:
+        table_name = item[2]
+        if table_name not in files_grouped_by_table:
+            files_grouped_by_table[table_name] = []
+        files_grouped_by_table[table_name].append(item)
+
+    sorted_threads_data = []
+    max_no_files_per_table = max(
+        len(group) for group in files_grouped_by_table.values()
+    )
+
+    for idx in range(max_no_files_per_table):
+        for table_name in files_grouped_by_table.keys():
+            files = files_grouped_by_table[table_name]
+            if idx < len(files):
+                sorted_threads_data.append(files[idx])
+
+    return sorted_threads_data
 
 
 def extract_xml_table_name(file_name: str) -> str:
@@ -346,17 +486,25 @@ def add_missing_columns_to_table(
     missing_columns = set(column_list) - set(column_names_from_database)
 
     for column_name in missing_columns:
-        alter_query = 'ALTER TABLE %s ADD "%s" VARCHAR NULL;' % (
-            table_name,
-            column_name,
-        )
-        with engine.connect().execution_options(autocommit=True) as con:
-            with con.begin():
-                con.execute(text(alter_query).execution_options(autocommit=True))
-        log.info(
-            "From the downloaded xml files following new attribute was "
-            f"introduced: {table_name}.{column_name}"
-        )
+        if not column_exists(engine, table_name, column_name):
+            alter_query = 'ALTER TABLE %s ADD "%s" VARCHAR NULL;' % (
+                table_name,
+                column_name,
+            )
+            try:
+                with engine.connect().execution_options(autocommit=True) as con:
+                    with con.begin():
+                        con.execute(
+                            text(alter_query).execution_options(autocommit=True)
+                        )
+            except sqlalchemy.exc.OperationalError as err:
+                # If the column already exists, we can ignore the error.
+                if "duplicate column name" not in str(err):
+                    raise err
+            log.info(
+                "From the downloaded xml files following new attribute was "
+                f"introduced: {table_name}.{column_name}"
+            )
 
 
 def delete_wrong_xml_entry(err: Error, df: pd.DataFrame) -> pd.DataFrame:
@@ -464,3 +612,9 @@ def add_table_to_sqlite_database(
             # 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
+
+
+def column_exists(engine, table_name, column_name):
+    inspector = inspect(engine)
+    columns = [col["name"] for col in inspector.get_columns(table_name)]
+    return column_name in columns


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 e72e550..9910d7f 100644
--- a/tests/xml_download/test_utils_write_to_database.py
+++ b/tests/xml_download/test_utils_write_to_database.py
@@ -29,6 +29,7 @@ from open_mastr.xml_download.utils_write_to_database import (
     read_xml_file,
     add_table_to_non_sqlite_database,
     add_table_to_sqlite_database,
+    interleave_files,
 )
 
 # Check if xml file exists
@@ -393,3 +394,24 @@ def test_add_table_to_sqlite_database(engine_testdb, add_table_to_database_funct
             pd.testing.assert_frame_equal(
                 expected_df, pd.read_sql_table("gsgk_eeg", con=con)
             )
+
+
+def test_interleave_files():
+    input_data = [
+        ("AnlagenEegBiomasse.xml", "anlageneegbiomasse", "anlageneegbiomasse"),
+        ("AnlagenEegSolar_1.xml", "anlageneegsolar", "anlageneegsolar"),
+        ("AnlagenEegSolar_2.xml", "anlageneegsolar", "anlageneegsolar"),
+        ("AnlagenEegWind_1.xml", "anlageneegwind", "anlageneegwind"),
+        ("AnlagenEegWind_2.xml", "anlageneegwind", "anlageneegwind"),
+        ("AnlagenEegWind_3.xml", "anlageneegwind", "anlageneegwind"),
+        ("Bilanzierungsgebiete.xml", "bilanzierungsgebiete", "bilanzierungsgebiete"),
+    ]
+    assert interleave_files(input_data) == [
+        ("AnlagenEegBiomasse.xml", "anlageneegbiomasse", "anlageneegbiomasse"),
+        ("AnlagenEegSolar_1.xml", "anlageneegsolar", "anlageneegsolar"),
+        ("AnlagenEegWind_1.xml", "anlageneegwind", "anlageneegwind"),
+        ("Bilanzierungsgebiete.xml", "bilanzierungsgebiete", "bilanzierungsgebiete"),
+        ("AnlagenEegSolar_2.xml", "anlageneegsolar", "anlageneegsolar"),
+        ("AnlagenEegWind_2.xml", "anlageneegwind", "anlageneegwind"),
+        ("AnlagenEegWind_3.xml", "anlageneegwind", "anlageneegwind"),
+    ]