graphql-engine: hasura + timescaledb on pg11 > Upserts not working anymore

Hello, @0x777 @marionschleifer @tirumaraiselvan, did you guys plan to release a fix to this bug? It’s soon going to be 2 years this issue is open. Reproducing should be quite straightforward:

1. docker pull timescale/timescaledb:latest-pg12
2. docker pull hasura/graphql-engine
3. Connect to timescaledb (which is a postgres instance + an extension) and run CREATE EXTENSION IF NOT EXISTS timescaledb CASCADE; as well as CREATE SCHEMA "schema1"; & CREATE TABLE "schema1"."table1" ("id" int AS IDENTITY PRIMARY KEY,"value" text NOT NULL, "timestamp" timestamp with time zone NOT NULL); and finally run SELECT create_hypertable('schema1.table1', 'timestamp'); to convert schema.table1 to a timescale “hypertable”.
4. Send this mutation through Hasura:

mutation upsert_schema1_table1(
    $objects: [schema1_table1_insert_input!]!
    $update_columns: [schema1_table1_update_column!]!
  ) {
    insert_schema1_table1(
      objects: $objects
      on_conflict: { constraint: table1_pkey, update_columns: $update_columns }
    ) {
      returning {
        id
        value
       timestamp
      }
    }
  }

with these query variables:

{
  "objects": {
    "id": 1,
    "value": "value",
    "timestamp": "2021-02-26T12:00:00+00:00",
  },
  "update_columns": ["id", "value", "timestamp"]
}

This will throw an “ON CONFLICT ON CONSTRAINT” error, as with Hypertables one can’t reference a primary key by its name, but only by its column names:

Your help would be much appreciated ❤️

Hello, any resolution to this issue?

Hi @marionschleifer , any update on this issue? Thanks!

@ondrejspilkaABB sure. So, I’ve defined a SQL function named ingest_packets and registered it in Hasura to be able to use it later via the GraphQL API.

CREATE OR REPLACE FUNCTION unit.ingest_packets(_data jsonb)
 RETURNS SETOF unit.packet
 LANGUAGE plpgsql
AS $function$
DECLARE
    unit_id         bigint;
    r               jsonb;
    p               jsonb;
    _packet_content jsonb;
BEGIN
    for p in (select jsonb_array_elements((_data ->> 'packets')::jsonb) element)
        loop
            _packet_content := ((p ->> 'packet_content')::jsonb);
            unit_id := unit.insert_or_get_unit((p ->> 'unit_imei')::text);
            
            RETURN NEXT unit.insert_or_get_packet(unit_id, (p ->> 'tcp_session_id')::uuid,
                                                  (_packet_content ->> 'packet')::jsonb);
                                                  
            for r in (select jsonb_array_elements((_packet_content ->> 'records')::jsonb) element)
                loop
                    PERFORM unit.insert_or_update_record(
                            unit_id,
                            (((_packet_content ->> 'packet')::jsonb) ->> 'mongo_id')::text,
                            (((_packet_content ->> 'packet')::jsonb) ->> 'timestamp')::timestamp with time zone,
                            r
                        );
                end loop;
        end loop;
END
$function$

and then create unit.insert_or_get_packet:

CREATE OR REPLACE FUNCTION unit.insert_or_get_packet(_unit_id bigint, _session_id uuid, packet jsonb)
 RETURNS SETOF unit.packet
 LANGUAGE plpgsql
AS $function$
BEGIN
    RETURN QUERY INSERT INTO unit.packet (
                                          mongo_id,
                                          session_id,
                                          "timestamp",
                                          data_hex
        )
        VALUES ((packet ->> 'mongo_id')::text,
                _session_id,
                (packet ->> 'timestamp')::timestamp with time zone,
                (packet ->> 'data_hex')::text,) ON CONFLICT (mongo_id, timestamp) DO
            UPDATE
            SET
                mongo_id = EXCLUDED.mongo_id,
                timestamp = EXCLUDED.timestamp,
                session_id = EXCLUDED.session_id,
                data_hex = EXCLUDED.data_hex,
        RETURNING *;
END
$function$

and insert_or_update_record:

CREATE OR REPLACE FUNCTION unit.insert_or_update_record(_unit_id bigint, _packet_mongo_id text, _packet_timestamp timestamp with time zone, _record jsonb)
 RETURNS void
 LANGUAGE plpgsql
AS $function$
BEGIN
    INSERT INTO unit.record(unit_id,
                            packet_mongo_id,
                            packet_timestamp,
                            "timestamp",
                            priority,
                            gps_longitude,
                            gps_latitude,
                            gps_altitude,
                            gps_angle,
                            gps_satellites,
                            gps_speed
                            )
    VALUES (_unit_id,
            _packet_mongo_id,
            _packet_timestamp,
            (_record ->> 'timestamp')::timestamp with time zone,
            (_record ->> 'priority')::integer,
            (_record ->> 'gps_longitude')::double precision,
            (_record ->> 'gps_latitude')::double precision,
            (_record ->> 'gps_altitude')::double precision,
            (_record ->> 'gps_angle')::double precision,
            (_record ->> 'gps_satellites')::integer,
            (_record ->> 'gps_speed')::double precision,
            )
    ON CONFLICT (packet_mongo_id, packet_timestamp, timestamp) DO UPDATE
        SET
            unit_id = excluded.unit_id,
            priority = excluded.priority,
            gps_longitude = excluded.gps_longitude,
            gps_latitude = excluded.gps_latitude,
            gps_altitude = excluded.gps_altitude,
            gps_angle = excluded.gps_angle,
            gps_satellites = excluded.gps_satellites,
            gps_speed = excluded.gps_speed;
END
$function$

Using the below improvements, I was able to increase the Bringes ingestion rate from <50 packets/second to ~5000 packets/second.

Hope this is useful for you. If you have any questions please let me know.

Until the upstream TimescaleDB issue is fixed, would it be possible to allow ON CONFLICT DO NOTHING in graphQL without explicitly specifying any constraint? As the Postgres docs state:

For ON CONFLICT DO NOTHING, it is optional to specify a conflict_target; when omitted, conflicts with all usable constraints (and unique indexes) are handled.

This does not cover the ON CONFLICT DO UPDATE case, but might be a helpful first step.

Thanks @iosifnicolae2 that is super useful.

Would love to see this officially supported by Hasura…

@0x777 any progress on the issue?

@waterdrop01 We’ll try to add a workaround.

Hello @0x777, any chance you share with us the progress and problems you may encounter related to this issue? This would be useful for people following this issue to get a sense of the amount of work remaining to tackle it! Thanks!

Hello @tirumaraiselvan, maybe you can provide estimation, when this is going to be fixed? Use-case is for example batch upserts, so here your product loses some competitiveness…

Hi @tirumaraiselvan, any visibility regarding this issue? Would be greatly appreciated