Merge pull request #6679 from Flowminder/delarative-partitioning

Delarative partitioning

Author: mergify[bot]

CHANGELOG.md

@@ -13,6 +13,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 - FlowKit test and synthetic data now uses parquet foreign tables.
 > [!WARNING]
 > The location of the parquet files in the container is `/parquet_data`, if you are testing with larger amounts of data you may wish to add an additional bind mount for this location.
+- FlowDB now uses [declarative partitioning](https://www.postgresql.org/docs/current/ddl-partitioning.html#DDL-PARTITIONING-DECLARATIVE)
+- FlowETL now attached new data as partitions, rather than subtables
+> [!WARNING]
+> This change is not backwards compatible with earlier releases of FlowDB, and you will need to repopulate your deployment. We recommend combining this change with the new parquet support.
 
 ### Fixed
 

flowdb/bin/build/0020_schema_events.sql

@@ -64,7 +64,7 @@ CREATE SCHEMA IF NOT EXISTS events;
         operator_code NUMERIC,
         country_code NUMERIC
 
-        );
+        ) PARTITION BY RANGE (datetime);
 
     CREATE TABLE IF NOT EXISTS events.forwards(
 
@@ -87,7 +87,7 @@ CREATE SCHEMA IF NOT EXISTS events;
         operator_code NUMERIC,
         country_code NUMERIC
 
-        );
+        ) PARTITION BY RANGE (datetime);
 
     CREATE TABLE IF NOT EXISTS events.sms(
 
@@ -108,7 +108,7 @@ CREATE SCHEMA IF NOT EXISTS events;
         operator_code NUMERIC,
         country_code NUMERIC
 
-        );
+        ) PARTITION BY RANGE (datetime);
 
     CREATE TABLE IF NOT EXISTS events.mds(
 
@@ -131,7 +131,7 @@ CREATE SCHEMA IF NOT EXISTS events;
         operator_code NUMERIC,
         country_code NUMERIC
 
-        );
+        ) PARTITION BY RANGE (datetime);
 
     CREATE TABLE IF NOT EXISTS events.topups(
 
@@ -156,7 +156,7 @@ CREATE SCHEMA IF NOT EXISTS events;
         operator_code NUMERIC,
         country_code NUMERIC
 
-        );
+        ) PARTITION BY RANGE (datetime);
 
     CREATE TABLE IF NOT EXISTS events.location_ids (
         location_id TEXT,

flowdb/testdata/bin/generate_synthetic_data.py

@@ -224,11 +224,7 @@ def write_day_csv(subscribers, cells, date, num_calls, call_seed, output_root_di
         calls_df_twoline.to_csv(fpath, index=False)
 
     ingest_sql = """
-            CREATE TABLE IF NOT EXISTS events.calls_{table} (
-                    CHECK ( datetime >= '{table}'::TIMESTAMPTZ
-                    AND datetime < '{end_date}'::TIMESTAMPTZ)
-                ) INHERITS (events.calls);
-                ALTER TABLE events.calls_{table} NO INHERIT events.calls;
+            CREATE TABLE IF NOT EXISTS events.calls_{table} PARTITION OF events.calls FOR VALUES FROM ('{table}') TO ('{end_date}');
 
                 COPY events.calls_{table}( datetime,msisdn_counterpart,id,msisdn,location_id,outgoing,duration,tac )
                     FROM '{output_root_dir}/data/records/calls/calls_{table}.csv'
@@ -241,8 +237,7 @@ def write_day_csv(subscribers, cells, date, num_calls, call_seed, output_root_di
             CREATE INDEX ON events.calls_{table} (location_id);
             CREATE INDEX ON events.calls_{table} (datetime);
             CLUSTER events.calls_{table} USING calls_{table}_msisdn_idx;
-            ANALYZE events.calls_{table};
-            ALTER TABLE events.calls_{table} INHERIT events.calls;""".format(
+            ANALYZE events.calls_{table};""".format(
         output_root_dir=output_root_dir,
         table=date.strftime("%Y%m%d"),
         end_date=(date + datetime.timedelta(days=1)).strftime("%Y%m%d"),

flowdb/testdata/bin/generate_synthetic_data_sql.py

@@ -672,7 +672,7 @@ def log_duration(job: str, **kwargs):
                     attach_sql.append(
                         (
                             f"Attaching events.{sub}_{table}",
-                            f"ALTER TABLE events.{sub}_{table} INHERIT events.{sub};",
+                            f"ALTER TABLE events.{sub} ATTACH PARTITION events.{sub}_{table} FOR VALUES FROM ('{table}') TO ('{(date + datetime.timedelta(days=1)).strftime('%Y%m%d')}');",
                         )
                     )
                     if args.cluster:

flowdb/testdata/test_data/py/zz_convert_events_to_parquet.py

@@ -39,17 +39,14 @@
 
 MOUNT_PARQUET_SQL = """
 DROP TABLE events.{table_name};
-CREATE FOREIGN TABLE events.{table_name} ()
-INHERITS (events.{event_type})
+CREATE FOREIGN TABLE events.{table_name}
+PARTITION OF events.{event_type}
+FOR VALUES FROM ('{start_date}') TO ('{end_date}')
 SERVER parquet_srv 
 OPTIONS(
 filename '{parquet_path}'
 );
-
---ALTER TABLE events.{event_type} 
-    --ATTACH PARTITION events.{table_name}
-    --FOR VALUES FROM ('{start_date}') TO ('{end_date}');
-
+ANALYZE events.{table_name};
 """
 
 PARQUET_COL_DTYPE_MAPPING = {