Soft deletion support

[abhishekk72850]

I have a pg table with is_deleted column.

My requirement is to not sync or desync the rows where is_deleted = True

I wrote the plugin:

class ExcludeSoftDeletedPlugin(plugin.Plugin):
    """
    This plugins excludes the rows where is_deleted = True
    """

    name: str = "ExcludeSoftDeleted"

    def transform(self, doc: dict, **kwargs) -> dict:
        """
        Doesn't not sync rows with is_deleted True
        """
        doc_id = kwargs["_id"]
        doc_index = kwargs["_index"]

        # Ignores row with is deleted true
        if doc["is_deleted"]:
            return None

        return doc

And this does ignore the rows during initial sync or first run.

But the issue is it doesn't work when pgsync is running and i update the column value from False to True, it simply ignores the update and doc for that row still remains in the elastic.

It would be great help if anyone can tell how to achieve this.

[abhishekk72850]

For now i have solved this by updating the pgsync db FUNCTION, to trigger DELETE when updating is_deleted to true

CREATE OR REPLACE FUNCTION public.table_notify()
 RETURNS trigger
 LANGUAGE plpgsql
AS $function$
DECLARE
  channel TEXT;
  old_row JSON;
  new_row JSON;
  notification JSON;
  xmin BIGINT;
  _indices TEXT [];
  _primary_keys TEXT [];
  _foreign_keys TEXT [];
  temp_op TEXT;

BEGIN
    -- database is also the channel name.
    channel := CURRENT_DATABASE();
    temp_op := TG_OP;

    IF temp_op = 'DELETE' THEN

        SELECT primary_keys, indices
        INTO _primary_keys, _indices
        FROM public._view
        WHERE table_name = TG_TABLE_NAME;

        old_row = ROW_TO_JSON(OLD);
        old_row := (
            SELECT JSONB_OBJECT_AGG(key, value)
            FROM JSON_EACH(old_row)
            WHERE key = ANY(_primary_keys)
        );
        xmin := OLD.xmin;
    ELSE
        IF temp_op <> 'TRUNCATE' THEN

            SELECT primary_keys, foreign_keys, indices
            INTO _primary_keys, _foreign_keys, _indices
            FROM public._view
            WHERE table_name = TG_TABLE_NAME;

            new_row = ROW_TO_JSON(NEW);
            new_row := (
                SELECT JSONB_OBJECT_AGG(key, value)
                FROM JSON_EACH(new_row)
                WHERE key = ANY(_primary_keys || _foreign_keys)
            );
            IF temp_op = 'UPDATE' THEN
                IF TG_TABLE_NAME = 'profile_service' THEN

                    IF NEW.is_deleted THEN

                        -- Update TG_OP to "DELETE" and proceed with notification logic
                        temp_op := 'DELETE';
                    END IF;
                END IF;

                old_row = ROW_TO_JSON(OLD);
                old_row := (
                    SELECT JSONB_OBJECT_AGG(key, value)
                    FROM JSON_EACH(old_row)
                    WHERE key = ANY(_primary_keys || _foreign_keys)
                );
            END IF;
            xmin := NEW.xmin;
        END IF;
    END IF;
    
--    temp_op := 'DELETE';

    -- construct the notification as a JSON object.
    notification = JSON_BUILD_OBJECT(
        'xmin', xmin,
        'new', new_row,
        'old', old_row,
        'indices', _indices,
        'tg_op', temp_op,
        'table', TG_TABLE_NAME,
        'schema', TG_TABLE_SCHEMA
    );

    -- Notify/Listen updates occur asynchronously,
    -- so this doesn't block the Postgres trigger procedure.
    PERFORM PG_NOTIFY(channel, notification::TEXT);

  RETURN NEW;
END;
$function$
;

[toluaina]

• Were you expecting the doc to be deleted in ES?
• This means it's (conceptually) soft deleted in the db but hard deleted in ES or OS.
• My suggestion here would be to delegate more logic to the plugin and actually perform the intended delete operation in Elasticsearch or Opensearch.