PolarDB:Electronic fence computing

Create the table and populate it with sample data

The following SQL creates a table of 10,000 random polygon geofences. Each fence is generated by creating a random point and expanding it into a circle buffer with a radius of 0.01 degrees (approximately 1 km).

-- Helper: generate a random float in [low, high]
CREATE OR REPLACE FUNCTION test_random_float(low float, high float)
    RETURNS float AS $$
BEGIN
    RETURN random() * (high-low) + low;
END;
$$ LANGUAGE 'plpgsql' STRICT;

-- Helper: generate a random geometry point within a bounding box
CREATE OR REPLACE FUNCTION test_random_geogpoint(lowx float, highx float,
                                            lowy float, highy float)
    RETURNS geometry AS $$
BEGIN
    RETURN ST_SetSRID(st_point(test_random_float(lowx, highx), test_random_float(lowy, highy)),4326);
END;
$$ LANGUAGE 'plpgsql' STRICT;

-- Geofence table
CREATE TABLE IF NOT EXISTS geofencings (
    name VARCHAR(100) PRIMARY KEY,
    geofencing Geometry
);

-- Insert 10,000 random geofences
INSERT INTO geofencings SELECT 'g_' || s, test_random_geogpoint(129, 130, 49, 50) FROM generate_series(1, 10000) s;

-- Expand each point into a polygon buffer (~1 km radius)
UPDATE geofencings SET geofencing = ST_Buffer(geofencing, 0.01);

Register the geofence table as a Flink dimension table

This step uses the lookup join pattern: Flink queries the fence table from PolarDB at the processing time of each incoming event (FOR SYSTEM_TIME AS OF points.proctime).

Register the PolarDB geofencings table as a Flink dimension table using the polardb connector. The join.predicate = 'CONTAINS' option tells GanosBase to use the CONTAINS spatial predicate for the join: for each incoming point, Flink checks which fence polygon contains that coordinate.

CREATE TEMPORARY TABLE geofencings (
    name STRING NOT NULL,
    geofencing STRING NOT NULL,
    PRIMARY KEY (name) NOT ENFORCED
) WITH (
    'connector' = 'polardb',
    'jdbcUrl' = 'jdbc:postgresql://<yourHostname>:<yourPort>/<dbname>',
    'username' = '<yourUserName>',
    'password' = '<yourPassWord>',
    'query' = 'SELECT name, ST_AsText(geofencing) FROM geofencings;',
    'join.columnName' = 'geofencing',
    'join.predicate' = 'CONTAINS'
);

Replace the following placeholders:

Create the result table in PolarDB

CREATE TABLE IF NOT EXISTS point_in_geofencing (
  id serial PRIMARY KEY,          -- Primary key
  point_id VARCHAR(100),          -- Point ID
  geofencing_name VARCHAR(100),   -- Fence name
  point Geometry                  -- Point geometry
);

Register the result table as a Flink table

Records are inserted in batches of 100 for efficiency.

CREATE TEMPORARY TABLE point_in_geofencing (
  point_id STRING NOT NULL,
  geofencing_name STRING NOT NULL,
  point STRING NOT NULL
) WITH (
    'connector' = 'polardb',
    'jdbcUrl' = 'jdbc:postgresql://<yourHostname>:<yourPort>/<dbname>',
    'username' = '<yourUserName>',
    'password' = '<yourPassWord>',
    'query' = 'INSERT INTO point_in_geofencing(point_id,geofencing_name,point) values(?,?,ST_GeomFromText(?));',
    'batchSize' = '100'
);

Simulate a data source

The following Flink SQL uses the datagen connector to simulate a location event stream at 100,000 records per second, with random coordinates inside the bounding box used to generate the fences (longitude 129–130, latitude 49–50).

This datagen source is for testing only. In production, replace it with a Kafka connector using the same column schema (id, lng, lat, proctime).

CREATE TEMPORARY TABLE points (
  id STRING NOT NULL,
  lng DOUBLE NOT NULL,          -- Longitude
  lat DOUBLE NOT NULL,          -- Latitude
  proctime AS PROCTIME(),       -- Processing-time attribute (required for lookup join)
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '100000',
  'fields.lng.kind' = 'random',
  'fields.lng.min' = '129',
  'fields.lng.max' = '130',
  'fields.lat.kind' = 'random',
  'fields.lat.min' = '49',
  'fields.lat.max' = '50',
  'fields.id.kind' = 'random',
  'fields.id.length' = '5'
);

Run the join query

The query joins the incoming point stream against the geofence dimension table. FOR SYSTEM_TIME AS OF points.proctime is the standard Flink lookup join syntax: it queries PolarDB at the processing time of each event. GanosBase evaluates the CONTAINS predicate — equivalent to geofencings.geofencing CONTAINS ST_MakePoint(lng, lat) — using the in-memory spatial index, then writes matching records to the result table.

INSERT INTO point_in_geofencing
SELECT points.id, geofencings.name, ST_MakePoint(lng, lat)
FROM points
JOIN geofencings FOR SYSTEM_TIME AS OF points.proctime
ON geofencings.geofencing = ST_MakePoint(lng, lat);