Notification Configuration for PostgreSQL

To handle notifications from a PostgreSQL database, the GRAVITY system uses the database server’s pg_notify mechanism.

This is a built-in function that accepts two arguments: channel and payload (see PostgreSQL documentation: https://www.postgresql.org/docs/current/sql-notify.html).

The channel is the channel that GRAVITY will listen on, and the payload is the data that will be received. The payload must be sent in JSON format as an object consisting of key–value pairs. Each property is passed to the project invoked by the trigger as a parameter. If the project defines a parameter with a name matching the key, it will be populated with the corresponding value.

It is also possible to send values to the DATA INPUT operator. In this case, the key is the operator’s name, and the value is a JSON object that matches the template defined on that operator.

Database Server Configuration

Because pg_notify is a function and we want to react to changes to specific records in a table, the simplest approach is to use a database trigger on that table.

Assume that a table with the following structure exists:

CREATE TABLE etl.articles
(
  id integer,
  name character(100),
)
WITH (
  OIDS=FALSE
);

The project is triggered automatically whenever a new item is inserted into the table or an existing record is updated.

Let’s define a function responsible for sending the notification.

CREATE OR REPLACE FUNCTION etl.notify_articles_add()
  RETURNS trigger AS
$BODY$
BEGIN
  PERFORM pg_notify(
	'notify_articles_add',   
	json_build_object(
	'id', NEW.id,
	'name', NEW.name)::text
  );

  RETURN NEW;
END;
$BODY$
  LANGUAGE plpgsql VOLATILE;

The function will send a notification on the notify_articles_add channel with the following payload:

{
	"id": [new_record_id],
	"name": [item_name]
}

Let’s define a trigger for the articles table.

CREATE TRIGGER trigger_notify_articles_add
  AFTER INSERT OR UPDATE
  ON etl.articles
  FOR EACH ROW
  EXECUTE PROCEDURE etl.notify_articles_add();

At this point, when a record is added to the articles table, the defined trigger will call the notify_articles_add function, which will send a notification on the notify_articles_add channel. If a trigger is defined in the GRAVITY system that listens on the specified channel, it will start a project with parameters that correspond to the fields of the JSON object sent in the notification payload.

If we want to use the INPUT DATA operator in the project to receive information from the notification, the pg_notify function should be defined as follows:

CREATE OR REPLACE FUNCTION etl.notify_articles_add()
  RETURNS trigger AS
$BODY$
BEGIN
  PERFORM pg_notify(
	'notify_articles_add',   
	json_build_object(
	'InputData1',
  	json_build_object(   		 
    	'id', NEW.id,
    	'name', NEW.name,
	)::text)::text
  );

  RETURN NEW;
END;
$BODY$
  LANGUAGE plpgsql VOLATILE;

That is, it should pass a JSON object with the following structure:

{
	InputData1: {
		"id": [new_record_id],
		"name": "[item_name]"
	}
}

where InputData1 is the name of the DATA INPUT operator, and that DATA INPUT operator is configured with a data template that corresponds to the InputData1 object.

GRAVITY System Configuration

• When configuring the trigger for the project, after selecting a PostgreSQL-based database connection, enter the channel that will be used for listening.
• Use parameters whose names correspond to the structure of the JSON object in the notification payload, or use the INPUT DATA operator.