Synchronize data from relational tables to a graph database - PolarDB

This topic describes how to synchronize data from relational tables to a graph database.

Prerequisites

Use a high-privilege account to create theganos_graph extension.
Note
- This extension relies on functionalities provided by Apache AGE. Before you install this extension, install and enable the Apache AGE extension.
- If you encounter an error similar to ERROR: invalid extension name: "ganos_graph" when you install the extension, contact us.
```
CREATE EXTENSION IF NOT EXISTS ganos_graph;
```

Create a graph and labels.

SELECT create_graph('<graph_name>');
SELECT create_vlabel('<graph_name>', '<label_name>');
SELECT create_elabel('<graph_name>', '<label_name>');

Synchronize data to a graph database

Triggers are used to synchronize operations from a relational database to a graph database. When INSERT, UPDATE, or DELETE operations are performed on tables in the relational database, the triggers automatically capture these changes and synchronize the operations to the graph database. This ensures data consistency.

Vertex triggers

age_create_vertex_insert_trigger

Create an INSERT trigger for a vertex table.

Syntax

age_create_vertex_insert_trigger('<graph_name>', 
                                '<label_name>', 
                                '<table_name>',
                                '<vertex_primary_key_field_name>' default 'id',
                                '<custom_property_processing_function>')

Note

<vertex_primary_key_field_name> must be an integer. It can be of the int4 (32-bit integer) or int8 (64-bit integer) type.

<custom_property_processing_function> is a function that you specify to convert property values in relational tables into vertex property values in the graph. Function syntax:

CREATE OR REPLACE FUNCTION custom_insert_function(NEW RECORD, columns TEXT) 
RETURNS JSON AS $$
BEGIN
    RETURN json_build_object('name', NEW.name);
END;
$$ LANGUAGE plpgsql;

age_create_vertex_update_trigger

Create an UPDATE trigger for a vertex table.

Syntax

age_create_vertex_update_trigger('<graph_name>', 
                                '<label_name>', 
                                '<table_name>',
                                '<vertex_primary_key_field_name>' default 'id',
                                '<custom_property_processing_function>')

Note

<vertex_primary_key_field_name> must be an integer. It can be of the int4 (32-bit integer) or int8 (64-bit integer) type.
<custom_property_processing_function> is a function that you specify to convert property values in relational tables into vertex property values in the graph. This function supports retaining property values from the old graph. Function syntax:
```
CREATE OR REPLACE FUNCTION custom_update_function(OLD RECORD, NEW RECORD, columns TEXT) 
RETURNS JSON AS $$
BEGIN
    RETURN json_build_object('name_old', OLD.name, 'name_new', NEW.name);
END;
$$ LANGUAGE plpgsql;
```

age_create_vertex_delete_trigger

Create a DELETE trigger for a vertex table.

Syntax

age_create_vertex_delete_trigger('<graph_name>', 
                                '<label_name>', 
                                '<table_name>',
                                '<vertex_primary_key_field_name>' default 'id',
                                '<custom_property_processing_function>')

Note

<vertex_primary_key_field_name> must be an integer. It can be of the int4 (32-bit integer) or int8 (64-bit integer) type.
<custom_property_processing_function> is a function that you specify to execute additional operations when deleting a vertex, such as recording deleted data or performing cleanups. Function syntax:
```
CREATE OR REPLACE FUNCTION custom_delete_function(OLD RECORD,  columns TEXT) 
RETURNS void AS $$
BEGIN
    RAISE NOTICE 'delete: %', OLD;
END;
$$ LANGUAGE plpgsql;
```

Edge triggers

age_create_edge_insert_trigger

Create an INSERT trigger for an edge table.

Syntax

age_create_edge_insert_trigger('<graph_name>', 
                                '<edge_label_name>', 
                                '<edge_table_name>',
                                '<start_vertex_label_name>',
                                '<end_vertex_label_name>',
                                '<edge_table_primary_key_field_name>' default 'id',
                                '<start_vertex_table_primary_key_field_name>' default 'from_id',
                                '<end_vertex_table_primary_key_field_name>' default 'to_id',
                                '<custom_property_processing_function>')

Note

The value of a <primary_key_field_name> parameter must be an integer. It can be of the int4 (32-bit integer) or int8 (64-bit integer) type.

<custom_property_processing_function> is a function that you specify to convert property values in relational tables into edge property values in the graph. Function syntax:

CREATE OR REPLACE FUNCTION custom_insert_function(NEW RECORD, columns TEXT) 
RETURNS JSON AS $$
BEGIN
    RETURN json_build_object('name', NEW.name);
END;
$$ LANGUAGE plpgsql;

age_create_edge_update_trigger

Create an UPDATE trigger for an edge table.

Syntax

age_create_edge_update_trigger('<graph_name>', 
                                '<edge_label_name>', 
                                '<edge_table_name>',
                                '<start_vertex_label_name>',
                                '<end_vertex_label_name>',
                                '<edge_table_primary_key_field_name>' default 'id',
                                '<start_vertex_table_primary_key_field_name>' default 'from_id',
                                '<end_vertex_table_primary_key_field_name>' default 'to_id',
                                '<custom_property_processing_function>')

Note

The value of a <primary_key_field_name> parameter must be an integer. It can be of the int4 (32-bit integer) or int8 (64-bit integer) type.

<custom_property_processing_function> is a function that you specify to convert property values in relational tables into edge property values in the graph. Function syntax:

CREATE OR REPLACE FUNCTION custom_update_function(OLD RECORD, NEW RECORD, columns TEXT) 
RETURNS JSON AS $$
BEGIN
    RETURN json_build_object('name_old', OLD.name, 'name_new', NEW.name);
END;
$$ LANGUAGE plpgsql;

age_create_edge_delete_trigger

Create a DELETE trigger for an edge table.

Syntax

age_create_edge_delete_trigger ('<graph_name>', 
                                '<edge_label_name>', 
                                '<edge_table_name>',
                                '<edge_table_primary_key_field_name>' default 'id',                                
                                '<custom_property_processing_function>')

Note

<edge_table_primary_key_field_name> must be an integer. It can be of the int4 (32-bit integer) or int8 (64-bit integer) type.

<custom_property_processing_function> is a function that you specify to execute additional operations when deleting an edge. Function syntax:

CREATE OR REPLACE FUNCTION custom_delete_function(OLD RECORD,  columns TEXT) 
RETURNS void AS $$
BEGIN
    RAISE NOTICE 'delete: %', OLD;
END;
$$ LANGUAGE plpgsql;

Example

The following example provides SQL statements that associates relational tables with graph vertexes and creates triggers. When data in relational tables changes (INSERT/UPDATE/DELETE), the triggers automatically update the graph vertexes.

Relational table	Graph vertex
person_table	Person
city_table	City
lives_in_table	LivesIn

Create graph structure

SELECT ag_catalog.create_graph('graph_age_create_trigger');

-- Create vertex labels
SELECT ag_catalog.create_vlabel('graph_age_create_trigger', 'Person');
SELECT ag_catalog.create_vlabel('graph_age_create_trigger', 'City');

-- Create edge label
SELECT ag_catalog.create_elabel('graph_age_create_trigger', 'LivesIn');

Create relational tables

-- Create tables
CREATE TABLE person_table (
    id SERIAL PRIMARY KEY,
    name TEXT,
    age INT
);

CREATE TABLE city_table (
    id SERIAL PRIMARY KEY,
    name TEXT,
    population INT
);

CREATE TABLE lives_in_table (
    id SERIAL PRIMARY KEY,
    person_id INT,
    city_id INT,
    since DATE
);

Create triggers

SELECT age_create_vertex_insert_trigger('graph_age_create_trigger', 'Person', 'person_table');
SELECT age_create_vertex_update_trigger('graph_age_create_trigger', 'Person', 'person_table');
SELECT age_create_vertex_delete_trigger('graph_age_create_trigger', 'Person', 'person_table');

SELECT age_create_vertex_insert_trigger('graph_age_create_trigger', 'City', 'city_table');
SELECT age_create_vertex_update_trigger('graph_age_create_trigger', 'City', 'city_table');
SELECT age_create_vertex_delete_trigger('graph_age_create_trigger', 'City', 'city_table');

SELECT age_create_edge_insert_trigger('graph_age_create_trigger', 'LivesIn', 'lives_in_table', 'Person', 'City', 'id', 'person_id', 'city_id');
SELECT age_create_edge_update_trigger('graph_age_create_trigger', 'LivesIn', 'lives_in_table', 'Person', 'City', 'id', 'person_id', 'city_id');
SELECT age_create_edge_delete_trigger('graph_age_create_trigger', 'LivesIn', 'lives_in_table', 'id');

Insert data

-- Insert data into tables
INSERT INTO person_table (name, age) VALUES ('Alice', 30);
INSERT INTO person_table (name, age) VALUES ('Bob', 25);

INSERT INTO city_table (name, population) VALUES ('New York', 8419000);
INSERT INTO city_table (name, population) VALUES ('Los Angeles', 3980000);

INSERT INTO lives_in_table (person_id, city_id, since) VALUES (1, 1, '2010-01-01');
INSERT INTO lives_in_table (person_id, city_id, since) VALUES (2, 2, '2015-06-15');

Verify that the data has been inserted into the graph:

SELECT * from cypher('graph_age_create_trigger', $$    
MATCH (p1)-[r]->(p2)    
RETURN properties(p1), properties(r), properties(p2)$$) as (p agtype, r agtype, p2 agtype);

{"id": 1, "age": 30, "name": "Alice"} | {"id": 1, "since": "2010-01-01", "city_id": 1, "person_id": 1} | {"id": 1, "name": "New York", "population": 8419000} 
{"id": 2, "age": 25, "name": "Bob"}   | {"id": 2, "since": "2015-06-15", "city_id": 2, "person_id": 2} | {"id": 2, "name": "Los Angeles", "population": 3980000}

Modify data

UPDATE person_table SET age = 31 WHERE id = 1;
UPDATE city_table SET population = 100000 WHERE id = 1;
UPDATE lives_in_table SET since = '2025-01-01' WHERE id = 1;

Verify that the data has been synchronized and updated:

SELECT * FROM cypher('graph_age_create_trigger', $$    
MATCH (p1)-[r]->(p2)    
RETURN properties(p1), properties(r), properties(p2)$$) as (p agtype, r agtype, p2 agtype);

{"id": 1, "age": 31, "name": "Alice"} | {"id": 1, "since": "2025-01-01", "city_id": 1, "person_id": 1} | {"id": 1, "name": "New York", "population": 100000} 
{"id": 2, "age": 25, "name": "Bob"}   | {"id": 2, "since": "2015-06-15", "city_id": 2, "person_id": 2} | {"id": 2, "name": "Los Angeles", "population": 3980000}

Delete data

DELETE FROM lives_in_table WHERE person_id = (SELECT id FROM person_table WHERE name = 'Bob' limit 1);
DELETE FROM person_table WHERE name = 'Bob';

SELECT * FROM cypher('graph_age_create_trigger', $$    
MATCH (p1)-[r]->(p2)    
RETURN properties(p1), properties(r), properties(p2)$$) as (p agtype, r agtype, p2 agtype); 

{"id": 1, "age": 30, "name": "Alice"} | {"id": 1, "since": "2010-01-01", "city_id": 1, "person_id": 1} | {"id": 1, "name": "New York", "population": 8419000}