Add KDVH migration package

.github/workflows/ci.yml

@@ -64,3 +64,21 @@ jobs:
 
       - name: Run unit and integration tests
         run: cargo test --no-fail-fast -- --nocapture --test-threads=1
+
+      - name: Setup Go 1.22.x
+        uses: actions/setup-go@v5
+        with:
+          go-version: '1.22.x'
+          cache-dependency-path: migrations/go.sum
+
+      - name: Install Go dependencies
+        working-directory: migrations
+        run: go get
+
+      - name: Build migration package
+        working-directory: migrations
+        run: go build -v ./...
+
+      - name: Run Go tests
+        working-directory: migrations
+        run: go test -v ./...

db/drop_indices.sql

@@ -0,0 +1,7 @@
+-- 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;

db/flags.sql

@@ -10,6 +10,17 @@ CREATE TABLE IF NOT EXISTS flags.kvdata (
     cfailed INT4 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); 

integration_tests/README.md

@@ -20,22 +20,21 @@ End-to-end tests are implemented inside `integration_tests\tests\end_to_end.rs`.
 > defined in the `mock_permit_tables` function, otherwise the ingestor will not be able to
 > insert the data into the database.
 
-If you have Docker installed, you can run the tests locally using the provided
-`Makefile`:
+If you have Docker installed, you can run the tests locally using the provided `justfile`:
 
 ```terminal
 # Run all tests
-make test_all
+just test_all
 
 # Run unit tests only
-make unit_tests
+just test_unit
 
 # Run integration tests only
-make end_to_end
+just test_end_to_end
 
-# Debug a specific test (does not clean up the DB if `my_test_name` is an integration test)
-TEST=my_test_name make debug_test
+# Debug a specific test (does not clean up the DB if `test_name` is an integration test)
+just debug_test test_name
 
 # If any error occurs while running integration tests, you might need to reset the DB container manually
-make clean
+just clean
 ```

integration_tests/src/main.rs

@@ -37,6 +37,7 @@ async fn main() {
         }
     });
 
+    // NOTE: order matters
     let schemas = ["db/public.sql", "db/labels.sql", "db/flags.sql"];
     for schema in schemas {
         insert_schema(&client, schema).await.unwrap();

justfile

@@ -0,0 +1,36 @@
+test_unit:
+	cargo build --workspace --tests
+	cargo test --no-fail-fast --workspace --exclude lard_tests -- --nocapture
+
+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
+
+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
+
+debug_test TEST: setup
+	cargo test {{TEST}} --features debug --no-fail-fast -- --nocapture --test-threads=1
+
+debug_migrations: setup
+    @ cd migrations && go test -v ./...
+
+setup:
+	@ echo "Starting Postgres docker container..."
+	docker run --name lard_tests -e POSTGRES_PASSWORD=postgres -p 5432:5432 -d postgres
+	@ echo; sleep 5
+	cargo build --workspace --tests
+	@ echo; echo "Loading DB schema..."; echo
+	@target/debug/prepare_postgres
+
+clean:
+	@ echo "Stopping Postgres container..."
+	docker stop lard_tests
+	@ echo "Removing Postgres container..."
+	docker rm lard_tests

migrations/.gitignore

@@ -0,0 +1,7 @@
+*.txt
+*.sh
+migrate
+tables*/
+test_*/
+.env
+dumps/

migrations/README.md

@@ -0,0 +1,29 @@
+# Migrations
+
+Go package used to dump tables from old databases (KDVH, Kvalobs) and import them into LARD.
+
+## Usage
+
+1. Compile it with
+
+   ```terminal
+   go build
+   ```
+
+1. Dump tables from KDVH
+
+   ```terminal
+   ./migrate kdvh dump
+   ```
+
+1. Import dumps into LARD
+
+   ```terminal
+   ./migrate kdvh import
+   ```
+
+For each command, you can use the `--help` flag to see all available options.
+
+## Other notes
+
+Insightful talk on migrations: [here](https://www.youtube.com/watch?v=wqXqJfQMrqI&t=280s)

migrations/go.mod

@@ -0,0 +1,29 @@
+module migrate
+
+go 1.22.3
+
+require (
+	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/govalues/decimal v0.1.29 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
+	github.com/jackc/puddle/v2 v2.2.1 // indirect
+	github.com/mitchellh/colorstring v0.0.0-20190213212951-d06e56a500db // indirect
+	github.com/rickb777/plural v1.4.2 // indirect
+	github.com/rivo/uniseg v0.4.7 // indirect
+	golang.org/x/crypto v0.25.0 // indirect
+	golang.org/x/sync v0.7.0 // indirect
+	golang.org/x/sys v0.26.0 // indirect
+	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

@@ -0,0 +1,64 @@
+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/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=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/gocarina/gocsv v0.0.0-20240520201108-78e41c74b4b1 h1:FWNFq4fM1wPfcK40yHE5UO3RUdSNPaBC+j3PokzA6OQ=
+github.com/gocarina/gocsv v0.0.0-20240520201108-78e41c74b4b1/go.mod h1:5YoVOkjYAQumqlV356Hj3xeYh4BdZuLE0/nRkf2NKkI=
+github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
+github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/govalues/decimal v0.1.29 h1:GKC5g9y9oWxKIy51czdHTShOABwHm/shVuOVPwG415M=
+github.com/govalues/decimal v0.1.29/go.mod h1:LUlHHucpCmA4rJfNrDvMgrWibDpYnDNWqJuNU1/gxW8=
+github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
+github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk=
+github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+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=
+github.com/mattn/go-runewidth v0.0.16/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
+github.com/mitchellh/colorstring v0.0.0-20190213212951-d06e56a500db h1:62I3jR2EmQ4l5rM/4FEfDWcRD+abF5XlKShorW5LRoQ=
+github.com/mitchellh/colorstring v0.0.0-20190213212951-d06e56a500db/go.mod h1:l0dey0ia/Uv7NcFFVbCLtqEBQbrT4OCwCSKTEv6enCw=
+github.com/onsi/gomega v1.33.1 h1:dsYjIxxSR755MDmKVsaFQTE22ChNBcuuTWgkUDSubOk=
+github.com/onsi/gomega v1.33.1/go.mod h1:U4R44UsT+9eLIaYRB2a5qajjtQYn0hauxvRm16AVYg0=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/rickb777/period v1.0.5 h1:jAzlI2knYam5VMy0X8eYgqJBl0ew57N+J1djJSBOulM=
+github.com/rickb777/period v1.0.5/go.mod h1:AmEwpgIShi3EEw34qbafoPJxVeRbv9VVtjLyOeRwK6c=
+github.com/rickb777/plural v1.4.2 h1:Kl/syFGLFZ5EbuV8c9SVud8s5HI2HpCCtOMw2U1kS+A=
+github.com/rickb777/plural v1.4.2/go.mod h1:kdmXUpmKBJTS0FtG/TFumd//VBWsNTD7zOw7x4umxNw=
+github.com/rivo/uniseg v0.4.7 h1:WUdvkW8uEhrYfLC4ZzdpI2ztxP1I582+49Oc5Mq64VQ=
+github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88=
+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.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=
+github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+golang.org/x/crypto v0.25.0 h1:ypSNr+bnYL2YhwoMt2zPxHFmbAN1KZs/njMG3hxUp30=
+golang.org/x/crypto v0.25.0/go.mod h1:T+wALwcMOSE0kXgUAnPAHqTLW+XHgcELELW8VaDgm/M=
+golang.org/x/net v0.27.0 h1:5K3Njcw06/l2y9vpGCSdcxWOYHOUk3dVNGDXN+FvAys=
+golang.org/x/net v0.27.0/go.mod h1:dDi0PyhWNoiUOrAS8uXv/vnScO4wnHQO4mj9fn/RytE=
+golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M=
+golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/sys v0.26.0 h1:KHjCJyddX0LoSTb3J+vWpupP9p0oznkqVk/IfjymZbo=
+golang.org/x/sys v0.26.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/term v0.25.0 h1:WtHI/ltw4NvSUig5KARz9h521QvRC8RmF/cuYqifU24=
+golang.org/x/term v0.25.0/go.mod h1:RPyXicDX+6vLxogjjRxjgD2TKtmAO6NZBsBRfrOLu7M=
+golang.org/x/text v0.16.0 h1:a94ExnEXNtEwYLGJSIUxnWoxoRz/ZcCsV63ROupILh4=
+golang.org/x/text v0.16.0/go.mod h1:GhwF1Be+LQoKShO3cGOHzqOgRrGaYc9AvblQOmPVHnI=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

migrations/kdvh/db/main.go

@@ -0,0 +1,47 @@
+package db
+
+// Map of all tables found in KDVH, with set max import year
+type KDVH struct {
+	Tables map[string]*Table
+}
+
+func Init() *KDVH {
+	return &KDVH{map[string]*Table{
+		// Section 1: tables that need to be migrated entirely
+		// TODO: figure out if we need to use the elem_code_paramid_level_sensor_t_edata table?
+		"T_EDATA":     NewTable("T_EDATA", "T_EFLAG", "T_ELEM_EDATA").SetImportYear(3000),
+		"T_METARDATA": NewTable("T_METARDATA", "", "T_ELEM_METARDATA").SetImportYear(3000),
+
+		// Section 2: tables with some data in kvalobs, import only up to 2005-12-31
+		"T_ADATA":      NewTable("T_ADATA", "T_AFLAG", "T_ELEM_OBS").SetImportYear(2006),
+		"T_MDATA":      NewTable("T_MDATA", "T_MFLAG", "T_ELEM_OBS").SetImportYear(2006),
+		"T_TJ_DATA":    NewTable("T_TJ_DATA", "T_TJ_FLAG", "T_ELEM_OBS").SetImportYear(2006),
+		"T_PDATA":      NewTable("T_PDATA", "T_PFLAG", "T_ELEM_OBS").SetImportYear(2006),
+		"T_NDATA":      NewTable("T_NDATA", "T_NFLAG", "T_ELEM_OBS").SetImportYear(2006),
+		"T_VDATA":      NewTable("T_VDATA", "T_VFLAG", "T_ELEM_OBS").SetImportYear(2006),
+		"T_UTLANDDATA": NewTable("T_UTLANDDATA", "T_UTLANDFLAG", "T_ELEM_OBS").SetImportYear(2006),
+
+		// Section 3: tables that should only be dumped
+		"T_10MINUTE_DATA": NewTable("T_10MINUTE_DATA", "T_10MINUTE_FLAG", "T_ELEM_OBS"),
+		"T_ADATA_LEVEL":   NewTable("T_ADATA_LEVEL", "T_AFLAG_LEVEL", "T_ELEM_OBS"),
+		"T_MINUTE_DATA":   NewTable("T_MINUTE_DATA", "T_MINUTE_FLAG", "T_ELEM_OBS"),
+		"T_SECOND_DATA":   NewTable("T_SECOND_DATA", "T_SECOND_FLAG", "T_ELEM_OBS"),
+		"T_CDCV_DATA":     NewTable("T_CDCV_DATA", "T_CDCV_FLAG", "T_ELEM_EDATA"),
+		"T_MERMAID":       NewTable("T_MERMAID", "T_MERMAID_FLAG", "T_ELEM_EDATA"),
+		"T_SVVDATA":       NewTable("T_SVVDATA", "T_SVVFLAG", "T_ELEM_OBS"),
+
+		// Section 4: special cases, namely digitized historical data
+		"T_MONTH":           NewTable("T_MONTH", "T_MONTH_FLAG", "T_ELEM_MONTH").SetImportYear(1957),
+		"T_DIURNAL":         NewTable("T_DIURNAL", "T_DIURNAL_FLAG", "T_ELEM_DIURNAL").SetImportYear(2006),
+		"T_HOMOGEN_DIURNAL": NewTable("T_HOMOGEN_DIURNAL", "", "T_ELEM_HOMOGEN_MONTH"),
+		"T_HOMOGEN_MONTH":   NewTable("T_HOMOGEN_MONTH", "T_ELEM_HOMOGEN_MONTH", ""),
+
+		// Section 5: tables missing in the KDVH proxy:
+		// 1. these exist in a separate database
+		"T_AVINOR":   NewTable("T_AVINOR", "T_AVINOR_FLAG", "T_ELEM_OBS"),
+		"T_PROJDATA": NewTable("T_PROJDATA", "T_PROJFLAG", "T_ELEM_PROJ"),
+		// 2. these are not in active use and don't need to be imported in LARD
+		"T_DIURNAL_INTERPOLATED": NewTable("T_DIURNAL_INTERPOLATED", "", ""),
+		"T_MONTH_INTERPOLATED":   NewTable("T_MONTH_INTERPOLATED", "", ""),
+	}}
+}

migrations/kdvh/db/table.go

@@ -0,0 +1,54 @@
+package db
+
+// In KDVH for each table name we usually have three separate tables:
+// 1. A DATA table containing observation values;
+// 2. A FLAG table containing quality control (QC) flags;
+// 3. A ELEM table containing metadata about the validity of the timeseries.
+//
+// DATA and FLAG tables have the same schema:
+// | dato | stnr | ... |
+// where 'dato' is the timestamp of the observation, 'stnr' is the station
+// where the observation was measured, and '...' is a varying number of columns
+// each with different observations, where the column name is the 'elem_code'
+// (e.g. for air temperature, 'ta').
+//
+// The ELEM tables have the following schema:
+// | stnr | elem_code | fdato | tdato | table_name | flag_table_name | audit_dato
+
+// This struct contains basic metadata for a KDVH table
+type Table struct {
+	TableName     string // Name of the DATA table
+	FlagTableName string // Name of the FLAG table
+	ElemTableName string // Name of the ELEM table
+	Path          string // Directory name of where the dumped table is stored
+	importUntil   int    // Import data only until the year specified by this field. Table import will be skipped, if `SetImportYear` is not called.
+}
+
+// Creates default Table
+func NewTable(data, flag, elem string) *Table {
+	return &Table{
+		TableName:     data,
+		FlagTableName: flag,
+		ElemTableName: elem,
+		// NOTE: '_combined' kept for backward compatibility with original scripts
+		Path: data + "_combined",
+	}
+}
+
+// Specify the year until data should be imported
+func (t *Table) SetImportYear(year int) *Table {
+	if year > 0 {
+		t.importUntil = year
+	}
+	return t
+}
+
+// Checks if the table is set for import
+func (t *Table) ShouldImport() bool {
+	return t.importUntil > 0
+}
+
+// Checks if the table max import year was reached
+func (t *Table) MaxImportYearReached(year int) bool {
+	return t.importUntil < 0 || year >= t.importUntil
+}