Kvalobs importer

db/drop_indices.sql

@@ -1,7 +1,11 @@
--- Remove indices before bulk insertion
-DROP INDEX IF EXISTS data_timestamp_index,
-                     data_timeseries_index,
-                     nonscalar_data_timestamp_index,
-                     nonscalar_data_timeseries_index,
-                     old_flags_obtime_index,
-                     old_flags_timeseries_index;
+DO $$
+DECLARE
+ i RECORD;
+BEGIN
+    FOR i IN (SELECT schemaname, indexname fROM pg_indexes
+                WHERE schemaname IN ('public', 'flags')
+                AND NOT indexdef LIKE '%UNIQUE%')
+    LOOP
+        EXECUTE format('DROP INDEX IF EXISTS %s.%s', i.schemaname, i.indexname);
+    END LOOP;
+END $$;

db/flags.sql

@@ -7,20 +7,8 @@ CREATE TABLE IF NOT EXISTS flags.kvdata (
     corrected REAL NULL,
     controlinfo TEXT NULL,
     useinfo TEXT NULL,
-    cfailed INT4 NULL,
+    cfailed TEXT NULL,
     CONSTRAINT unique_kvdata_timeseries_obstime UNIQUE (timeseries, obstime)
 );
-CREATE INDEX IF NOT EXISTS kvdata_obtime_index ON flags.kvdata (obstime);
-CREATE INDEX IF NOT EXISTS kvdata_timeseries_index ON flags.kvdata USING HASH (timeseries); 
-
-CREATE TABLE IF NOT EXISTS flags.old_databases (
-    timeseries INT4 REFERENCES public.timeseries,
-    obstime TIMESTAMPTZ NOT NULL,
-    corrected REAL NULL,
-    controlinfo TEXT NULL,
-    useinfo TEXT NULL,
-    cfailed INT4 NULL ,
-    CONSTRAINT unique_old_flags_timeseries_obstime UNIQUE (timeseries, obstime)
-);
-CREATE INDEX IF NOT EXISTS old_flags_obtime_index ON flags.old_databases (obstime);
-CREATE INDEX IF NOT EXISTS old_flags_timeseries_index ON flags.old_databases USING HASH (timeseries); 
+CREATE INDEX IF NOT EXISTS kvdata_obstime_index ON flags.kvdata (obstime);
+CREATE INDEX IF NOT EXISTS kvdata_timeseries_index ON flags.kvdata USING HASH (timeseries);

db/partitions_generated.sql

@@ -1,4 +1,6 @@
 -- Generated by simple script for testing
+CREATE TABLE IF NOT EXISTS data_y1850_to_y1950 PARTITION OF public.data
+FOR VALUES FROM ('1850-01-01 00:00:00+00') TO ('1950-01-01 00:00:00+00');
 CREATE TABLE IF NOT EXISTS data_y1950_to_y2000 PARTITION OF public.data
 FOR VALUES FROM ('1950-01-01 00:00:00+00') TO ('2000-01-01 00:00:00+00');
 CREATE TABLE IF NOT EXISTS data_y2000_to_y2010 PARTITION OF public.data
@@ -35,6 +37,8 @@ CREATE TABLE IF NOT EXISTS data_y2028_to_y2029 PARTITION OF public.data
 FOR VALUES FROM ('2028-01-01 00:00:00+00') TO ('2029-01-01 00:00:00+00');
 CREATE TABLE IF NOT EXISTS data_y2029_to_y2030 PARTITION OF public.data
 FOR VALUES FROM ('2029-01-01 00:00:00+00') TO ('2030-01-01 00:00:00+00');
+CREATE TABLE IF NOT EXISTS nonscalar_data_y1850_to_y1950 PARTITION OF public.nonscalar_data
+FOR VALUES FROM ('1850-01-01 00:00:00+00') TO ('1950-01-01 00:00:00+00');
 CREATE TABLE IF NOT EXISTS nonscalar_data_y1950_to_y2000 PARTITION OF public.nonscalar_data
 FOR VALUES FROM ('1950-01-01 00:00:00+00') TO ('2000-01-01 00:00:00+00');
 CREATE TABLE IF NOT EXISTS nonscalar_data_y2000_to_y2010 PARTITION OF public.nonscalar_data

ingestion/src/kvkafka.rs

@@ -109,7 +109,7 @@ pub struct Kvdata {
     #[serde(default, deserialize_with = "optional")]
     useinfo: Option<String>,
     #[serde(default, deserialize_with = "optional")]
-    cfailed: Option<i32>,
+    cfailed: Option<String>,
 }
 
 // If the field is either empty or missing it should deserialize to None.

integration_tests/src/main.rs

@@ -9,22 +9,6 @@ async fn insert_schema(client: &tokio_postgres::Client, filename: &str) -> Resul
     client.batch_execute(schema.as_str()).await
 }
 
-fn format_partition(start: &str, end: &str, table: &str) -> String {
-    // TODO: add multiple partitions?
-    format!(
-        "CREATE TABLE {table}_y{start}_to_y{end} PARTITION OF {table} \
-        FOR VALUES FROM ('{start}-01-01 00:00:00+00') TO ('{end}-01-01 00:00:00+00')",
-    )
-}
-
-async fn create_data_partitions(client: &tokio_postgres::Client) -> Result<(), Error> {
-    let scalar_string = format_partition("1950", "2100", "public.data");
-    let nonscalar_string = format_partition("1950", "2100", "public.nonscalar_data");
-
-    client.batch_execute(scalar_string.as_str()).await?;
-    client.batch_execute(nonscalar_string.as_str()).await
-}
-
 #[tokio::main]
 async fn main() {
     let (client, connection) = tokio_postgres::connect(CONNECT_STRING, NoTls)
@@ -38,10 +22,13 @@ async fn main() {
     });
 
     // NOTE: order matters
-    let schemas = ["db/public.sql", "db/labels.sql", "db/flags.sql"];
+    let schemas = [
+        "db/public.sql",
+        "db/partitions_generated.sql",
+        "db/labels.sql",
+        "db/flags.sql",
+    ];
     for schema in schemas {
         insert_schema(&client, schema).await.unwrap();
     }
-
-    create_data_partitions(&client).await.unwrap();
 }

justfile

@@ -6,20 +6,25 @@ test_all: setup && clean
 	cargo test --workspace --no-fail-fast -- --nocapture --test-threads=1
 
 test_end_to_end: setup && clean
-	cargo test --test end_to_end --no-fail-fast -- --nocapture --test-threads=1
+	-cargo test --test end_to_end --no-fail-fast -- --nocapture --test-threads=1
 
 test_migrations: debug_migrations && clean
 
 # Debug commands don't perfom the clean up action after running.
 # This allows to manually check the state of the database.
+
 debug_kafka: setup
-	cargo test --test end_to_end test_kafka --features debug --no-fail-fast -- --nocapture --test-threads=1
+	-cargo test --test end_to_end test_kafka --features debug --no-fail-fast -- --nocapture --test-threads=1
 
 debug_test TEST: setup
-	cargo test {{TEST}} --features debug --no-fail-fast -- --nocapture --test-threads=1
+	-cargo test {{TEST}} --features debug --no-fail-fast -- --nocapture --test-threads=1
 
 debug_migrations: setup
-    @ cd migrations && go test -v ./...
+    -@ cd migrations && go test -v ./...
+
+# psql into the container database
+psql:
+    @docker exec -it lard_tests psql -U postgres
 
 setup:
 	@ echo "Starting Postgres docker container..."

migrations/README.md

@@ -1,6 +1,6 @@
 # Migrations
 
-Go package used to dump tables from old databases (KDVH, Kvalobs) and import them into LARD.
+Go package that dumps tables from old databases (KDVH, Kvalobs) and imports them into LARD.
 
 ## Usage
 
@@ -10,16 +10,18 @@ Go package used to dump tables from old databases (KDVH, Kvalobs) and import the
    go build
    ```
 
-1. Dump tables from KDVH
+1. Dump tables
 
    ```terminal
    ./migrate kdvh dump
+   ./migrate kvalobs dump
    ```
 
 1. Import dumps into LARD
 
    ```terminal
    ./migrate kdvh import
+   ./migrate kvalobs import
    ```
 
 For each command, you can use the `--help` flag to see all available options.

migrations/go.mod

@@ -3,15 +3,16 @@ module migrate
 go 1.22.3
 
 require (
+	github.com/alexflint/go-arg v1.5.1
 	github.com/gocarina/gocsv v0.0.0-20240520201108-78e41c74b4b1
 	github.com/jackc/pgx/v5 v5.6.0
-	github.com/jessevdk/go-flags v1.6.1
 	github.com/joho/godotenv v1.5.1
 	github.com/rickb777/period v1.0.5
 	github.com/schollz/progressbar/v3 v3.16.1
 )
 
 require (
+	github.com/alexflint/go-scalar v1.2.0 // indirect
 	github.com/govalues/decimal v0.1.29 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
@@ -25,5 +26,3 @@ require (
 	golang.org/x/term v0.25.0 // indirect
 	golang.org/x/text v0.16.0 // indirect
 )
-
-replace github.com/jessevdk/go-flags => github.com/Lun4m/go-flags v0.0.0-20241118100134-6375192b7985

migrations/go.sum

@@ -1,7 +1,7 @@
-github.com/Lun4m/go-flags v0.0.0-20241113125827-68757125e949 h1:7xyEGIr1X5alOjBjlNTDF+aRBcRIo60YX5sdlziLE5w=
-github.com/Lun4m/go-flags v0.0.0-20241113125827-68757125e949/go.mod h1:42/L0FDbP0qe91I+81tBqjU3uoz1tn1GDMZAhcCE2PE=
-github.com/Lun4m/go-flags v0.0.0-20241118100134-6375192b7985 h1:eUA/sFZ1CtY9+9y/fPpUivYW8fJBlXqB4/8CjC+yXqk=
-github.com/Lun4m/go-flags v0.0.0-20241118100134-6375192b7985/go.mod h1:42/L0FDbP0qe91I+81tBqjU3uoz1tn1GDMZAhcCE2PE=
+github.com/alexflint/go-arg v1.5.1 h1:nBuWUCpuRy0snAG+uIJ6N0UvYxpxA0/ghA/AaHxlT8Y=
+github.com/alexflint/go-arg v1.5.1/go.mod h1:A7vTJzvjoaSTypg4biM5uYNTkJ27SkNTArtYXnlqVO8=
+github.com/alexflint/go-scalar v1.2.0 h1:WR7JPKkeNpnYIOfHRa7ivM21aWAdHD0gEWHCx+WQBRw=
+github.com/alexflint/go-scalar v1.2.0/go.mod h1:LoFvNMqS1CPrMVltza4LvnGKhaSpc3oyLEBUZVhhS2o=
 github.com/chengxilo/virtualterm v1.0.4 h1:Z6IpERbRVlfB8WkOmtbHiDbBANU7cimRIof7mk9/PwM=
 github.com/chengxilo/virtualterm v1.0.4/go.mod h1:DyxxBZz/x1iqJjFxTFcr6/x+jSpqN0iwWCOK1q10rlY=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -21,8 +21,6 @@ github.com/jackc/pgx/v5 v5.6.0 h1:SWJzexBzPL5jb0GEsrPMLIsi/3jOo7RHlzTjcAeDrPY=
 github.com/jackc/pgx/v5 v5.6.0/go.mod h1:DNZ/vlrUnhWCoFGxHAG8U2ljioxukquj7utPDgtQdTw=
 github.com/jackc/puddle/v2 v2.2.1 h1:RhxXJtFG022u4ibrCSMSiu5aOq1i77R3OHKNJj77OAk=
 github.com/jackc/puddle/v2 v2.2.1/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
-github.com/jessevdk/go-flags v1.6.1 h1:Cvu5U8UGrLay1rZfv/zP7iLpSHGUZ/Ou68T0iX1bBK4=
-github.com/jessevdk/go-flags v1.6.1/go.mod h1:Mk8T1hIAWpOiJiHa9rJASDK2UGWji0EuPGBnNLMooyc=
 github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=
 github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4=
 github.com/mattn/go-runewidth v0.0.16 h1:E5ScNMtiwvlvB5paMFdw9p4kSQzbXFikJ5SQO6TULQc=
@@ -42,6 +40,7 @@ github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUc
 github.com/schollz/progressbar/v3 v3.16.1 h1:RnF1neWZFzLCoGx8yp1yF7SDl4AzNDI5y4I0aUJRrZQ=
 github.com/schollz/progressbar/v3 v3.16.1/go.mod h1:I2ILR76gz5VXqYMIY/LdLecvMHDPVcQm3W/MSKi1TME=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=