Restore but only missing data on the table

Question

I have 2 database already restored on 2 different database on the same instance, call it geo and restore1 the schema is identical, the only difference is that geo database contain only records before <= 2023-11-23 and after >= 2023-12-01, while restore1 contains <= 2023-11-30, how can I copy those missing values (2023-11-23 until 2023-11-30 from restore1 to geo database) for each table easily?

one way that I could think of is, create a program, for each table, selecting latest id before 2023-11-24 from geo (call it X), then select all record with id greater than X, and then insert it manually to geo database.

Is there any simpler way?

there's hundreds of tables, but most of them something like this:

restore1=# \d station_logs_599
                                              Table "public.station_logs_599"
         Column         |              Type              | Collation | Nullable |                 Default
------------------------+--------------------------------+-----------+----------+------------------------------------------
 id                     | integer                        |           | not null | nextval('station_logs_id_seq'::regclass)
 created_at             | timestamp(6) without time zone |           |          |
 updated_at             | timestamp(6) without time zone |           |          |
 deleted_at             | timestamp(6) without time zone |           |          |
 submitted_at           | timestamp(6) without time zone |           | not null |
 sequence               | integer                        |           |          |
 level_sensor           | double precision               |           |          |
 accel_x                | double precision               |           |          |
 accel_y                | double precision               |           |          |
 accel_z                | double precision               |           |          |
 power_current          | double precision               |           |          |
 ip_address             | character varying(50)          |           |          |
 log_type               | integer                        |           |          |
 station_id             | integer                        |           | not null |
 power_voltage          | double precision               |           |          |
 data                   | jsonb                          |           |          |
 is_deleted             | boolean                        |           | not null | false
 temperature            | double precision               |           |          |
 wind_speed             | double precision               |           |          |
 soil_moisture          | double precision               |           |          |
 wind_direction         | double precision               |           |          |
 raindrop               | double precision               |           |          |
 humidity               | integer                        |           |          |
 barometric_pressure    | double precision               |           |          |
 wind_speed_average     | double precision               |           |          |
 wind_gust              | double precision               |           |          |
 wind_direction_average | double precision               |           |          |
 rain_rate              | double precision               |           |          |
Indexes:
    "station_logs_599_epoch" btree (date_part('epoch'::text, submitted_at))
    "station_logs_599_submitted_at" btree (submitted_at)
    "uniq_sid_sat_599" UNIQUE CONSTRAINT, btree (submitted_at)
Foreign-key constraints:
    "station_logs_599_station_fk" FOREIGN KEY (station_id) REFERENCES stations(id)

Answer 1

I use golang to solve this:

tl;dw:

1. read all stations table
2. if both exists, select recap per date for source and destination (DATE(submitted_at), COUNT(1))
3. if destination has different number than source, insert to destination

    var src, dest *Pg.RDBMS
    {
        opt := `host=127.0.0.1 user=x password=x dbname=x sslmode=disable` 
        conn := sqlx.MustConnect(`postgres`, opt)
        conn.DB.SetMaxOpenConns(61) 
        dest = &Pg.RDBMS{
            Name:    `destination`,
            Adapter: conn,
        }
    }

    {
        opt := `host=127.0.0.1 user=y password=y dbname=y sslmode=disable` 
        conn := sqlx.MustConnect(`postgres`, opt)
        src = &Pg.RDBMS{
            Name:    `source`,
            Adapter: conn,
        }
    }


    srcStations := src.QStrMapMap(`imei`, `SELECT * FROM stations`)
    destStations := dest.QStrMapMap(`imei`, `SELECT * FROM stations`)

    // compare stations
    removeUnecessary := func(m M.SX) {
        delete(m, `last_level_sensor`)
        delete(m, `last_submitted_at`)
        delete(m, `updated_at`)
        delete(m, `updated_by`)
        delete(m, `name`)
        delete(m, `lat`)
        delete(m, `long`)
    }

    for imei := range srcStations {
        ss := srcStations.GetMSX(imei)
        removeUnecessary(ss)
        ds := destStations.GetMSX(imei)
        removeUnecessary(ds)

        // check the diff of stations table
        if len(ds) == 0 {
            fmt.Println(`missing from destination: `+imei, ss)
            // TODO: insert on destination
            continue
        } else if !reflect.DeepEqual(ss, ds) {
            fmt.Println(`different: ` + imei)
            diffs := pretty.Diff(ss, ds)
            for _, diff := range diffs {
                fmt.Println(diff)
            }
        }

        // if source different from destination, there might be one more than station created not with same order after moving server
        srcStationId := fmt.Sprint(ss[`id`])
        dstStationId := fmt.Sprint(ds[`id`])
        if srcStationId != dstStationId {
            fmt.Println(`source station id not the same as destination`, srcStationId, dstStationId)
        }

        // check whether tables exists
        if !src.TableExists(`station_logs_` + srcStationId) {
            fmt.Println(`missing source table station_logs_` + srcStationId)
            continue
        }
        if !dest.TableExists(`station_logs_` + dstStationId) {
            fmt.Println(`missing destination table station_logs_` + dstStationId)
            continue
        }

        srcSelect1 := fmt.Sprintf(`SELECT DATE(submitted_at)::TEXT, COUNT(1) FROM station_logs_%s GROUP BY 1`, srcStationId)
        srcDtCount := src.QStrIntMap(srcSelect1)

        dstSelect1 := fmt.Sprintf(`SELECT DATE(submitted_at)::TEXT, COUNT(1) FROM station_logs_%s GROUP BY 1`, dstStationId)
        dstDtCount := dest.QStrIntMap(dstSelect1)

        fmt.Println(`comparing station_logs_`, srcStationId, dstStationId)

        for submittedAt, srcCount := range srcDtCount {
            dstCount := dstDtCount[submittedAt]
            if srcCount != dstCount {
                fmt.Println(`different count: `, srcStationId, submittedAt, srcCount, dstCount)

                query2 := fmt.Sprintf(`SELECT * FROM station_logs_%s WHERE DATE(submitted_at) = '%s'`, srcStationId, submittedAt)

                srcLogs := src.QStrMapMap(`id`, query2)

                dstInsertLogs := fmt.Sprintf(`
INSERT INTO station_logs_%s (
    created_at, updated_at, submitted_at, sequence, level_sensor, accel_x, accel_y, accel_z, power_current, ip_address, log_type, station_id, power_voltage, data, is_deleted, temperature, wind_speed, soil_moisture, wind_direction, raindrop, humidity, barometric_pressure, wind_speed_average, wind_gust, wind_direction_average, rain_rate
) VALUES (
            $1,         $2,        $3,           $4,           $5,         $6,      $7,      $8,      $9,            $10,       $11,      $12,        $13,           $14,   $15,        $16,         $17,        $18,           $19,         $20,      $21,                 $22,                 $23,       $24,                  $25,        $26 
)`, dstStationId)
                query3 := fmt.Sprintf(`
SELECT COUNT(1) FROM station_logs_%s WHERE submitted_at = $1`, dstStationId)

                skipped := 0
                dest.DoTransaction(func(tx *Pg.Tx) string {
                    for id := range srcLogs {
                        log := srcLogs.GetMSX(id)
                        if tx.QInt(query3, log[`submitted_at`]) > 0 {
                            skipped++
                            continue
                        }

                        tx.DoExec(dstInsertLogs,
                            log[`created_at`],
                            log[`updated_at`],
                            log[`submitted_at`],
                            log[`sequence`],
                            log[`level_sensor`],
                            log[`accel_x`],
                            log[`accel_y`],
                            log[`accel_z`],
                            log[`power_current`],
                            log[`ip_address`],
                            log[`log_type`],
                            log[`station_id`],
                            log[`power_voltage`],
                            log[`data`],
                            log[`is_deleted`],
                            log[`temperature`],
                            log[`wind_speed`],
                            log[`soil_moisture`],
                            log[`wind_direction`],
                            log[`raindrop`],
                            log[`humidity`],
                            log[`barometric_pressure`],
                            log[`wind_speed_average`],
                            log[`wind_gust`],
                            log[`wind_direction_average`],
                            log[`rain_rate`],
                        )
                    }
                    return ``
                })

                fmt.Println(`inserted: `, srcStationId, submittedAt, len(srcLogs), `skipped`, skipped)

            }

        }
   }