To learn how to index and query geospatial points in Apache Pinot, watch the following video, or complete the tutorial below. To learn about geospatial objects, see the Geospatial objects developer guide.
Pinot Version1.0.0
Codestartreedata/pinot-recipes/geospatial-indexing

​
Prerequisites

To follow the code examples in this guide, you must install Docker locally and download recipes.
  1. If you haven’t already, download recipes.
  2. In terminal, go to the recipe by running the following command:
cd pinot-recipes/recipes/geospatial-indexing

​
Launch Pinot Cluster

You can spin up a Pinot Cluster by running the following command: 
docker-compose up
This command will run a single instance of the Pinot Controller, Pinot Server, Pinot Broker, Kafka, and Zookeeper. You can find the docker-compose.yml file on GitHub.

​
Generating Geospatial data

This recipe contains a data generator that produces JSON documents that contain various geospatial objects in Well-known text (WKT) format. You’ll need to first install the following dependencies: 
pip install geofactory faker geojson shapely
Once that’s done you can run the data generator and grab just the first generated document, by running the following command: 
python datagen.py 2>/dev/null | head -n1 | jq
Output is shown below: 
{
  "tsString": 1677845139617,
  "uuid": "362ec33c-6bd3-4928-8123-11ebe6a50c07",
  "count": 552,
  "pointString": "POINT (108.249223 5.520676)"
}
You can see from this output that we have a geospatial point. Pinot also supports polygons, mutli polygons, line strings, multi points, multi line strings, and geometry collections. You can read more about this in the Geospatial documention.

​
Kafka ingestion

We’re going to ingest this data into an Apache Kafka topic using the kcat command line tool. We’ll also use jq to structure the data in the key:payload structure that Kafka expects: 
python datagen.py --sleep 0.0001 2>/dev/null |
jq -cr --arg sep ø '[.uuid, tostring] | join($sep)' |
kcat -P -b localhost:9092 -t events -Kø
We can check that Kafka has some data by running the following command: 
docker exec -it kafka-geospatial kafka-run-class.sh \
  kafka.tools.GetOffsetShell \
  --broker-list localhost:9092 \
  --topic events
We’ll see something like the following: 
events:0:138780

​
Pinot Schema and Table

Now let’s create a Pinot Schema and Table. First, the schema: 
{
    "schemaName": "events_geo",
    "dimensionFieldSpecs": [
      {"name": "uuid", "dataType": "STRING"},
      {"name": "point", "dataType": "BYTES"}
    ],
    "metricFieldSpecs": [{"name": "count", "dataType": "INT"}],
    "dateTimeFieldSpecs": [
      {
        "name": "ts",
        "dataType": "TIMESTAMP",
        "format": "1:MILLISECONDS:EPOCH",
        "granularity": "1:MILLISECONDS"
      }
    ]
}
Note that the column for point has a data type of BYTES. Geospatial columns must use the BYTES type because Pinot will serialize the Geospatial objects into bytes for storage purposes. Now for the table config: 
{
    "tableName": "events_geo",
    "tableType": "REALTIME",
    "segmentsConfig": {
      "timeColumnName": "ts",
      "schemaName": "events_geo",
      "replication": "1",
      "replicasPerPartition": "1"
    },
    "fieldConfigList": [
      {
        "name": "point",
        "encodingType": "RAW",
        "indexType": "H3",
        "properties": {
          "resolutions": "5"
        }
      }
    ],
    "tableIndexConfig": {
      "loadMode": "MMAP",
      "noDictionaryColumns": ["point"],
      "streamConfigs": {
        "streamType": "kafka",
        "stream.kafka.topic.name": "events",
        "stream.kafka.broker.list": "kafka-geospatial:9093",
        "stream.kafka.consumer.type": "lowlevel",
        "stream.kafka.consumer.prop.auto.offset.reset": "smallest",
        "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
        "stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
        "realtime.segment.flush.threshold.rows":"100000",
        "realtime.segment.flush.threshold.time":"1h"
      }
    },
    "ingestionConfig": {
      "transformConfigs": [
        {
          "columnName": "point",
          "transformFunction": "toSphericalGeography(ST_GeomFromText(pointString))"
        }
      ]
    },
    "tenants": {},
    "metadata": {}
}
The highlighted lines define a Geospatial index on the point column. If you scroll down a bit, under ingestionConfig.transformConfigs you can see transformation functions that converts the WKT string from our Kafka streams into a Geospatial object. We’ll create the table by running the following: 
docker run \
   --network geospatial \
   -v $PWD/config:/config \
   apachepinot/pinot:1.0.0 AddTable \
     -schemaFile /config/schema.json \
     -tableConfigFile /config/table.json \
     -controllerHost "pinot-controller-geospatial" \
    -exec

​
When is the Geospatial index used?

Now that we’ve created a table that contains a column with a Geospatial index, it’d be useful to know when that index will be used. This index is used when a variety of predicates appear in the WHERE clause of a query.

​
ST_Distance

The ST_Distance function computes the distance in meters between two sets of coordinates. When used with this function, one of the arguments must be an identifer (i.e. column name) and the other a literal value. The result of calling ST_Distance must then be used in a range query i.e. it should be less than or greater than a specified distance. Examples are shown below: 
WHERE ST_Distance(point, '<Literal representing Point, Polygon, LineString, or GeometryCollection>') > 5000
WHERE ST_Distance('<Literal representing Point, Polygon, LineString, or GeometryCollection>', point) < 2000`

​
ST_Within

ST_Within checks the containment of two geospatial objects. It returns true if and only if the first geometry is completely inside the second geometry. When used with this function, the first argument must be an identifier (i.e. column name) and the other a literal value. An example is shown below: 
WHERE ST_Within(point, '<Literal representing a geometry object>')

​
ST_Contains

ST_Contains also checks the containment of two geospatial objects. More specifically, it returns true if and only if no points of the second geometry lie in the exterior of the first geometry, and at least one point of the interior of the first geometry lies in the interior of the second geometry. When used with this function, the first argument must be a literal and the other an identifier (i.e. column name). An example is shown below: 
WHERE ST_Contains('<Literal representing a geometry object>', point)

​
Geospatial Querying

Now let’s head on over to the Pinot UI and write some queries that use the Geospatial index. The following query counts how many points fit inside a polygon that covers an area from Delaware to West Virginia. A polygon covering an area from Delaware to West Virginia A polygon covering an area from Delaware to West Virginia 
select count(*)
from events_geo 
WHERE ST_Within(
         point,
         toSphericalGeography(ST_GeomFromText('POLYGON((
           -79.68778610229492 39.475226764883985,
           -76.83970928192137 40.48289486417028,
           -75.6193685531616 38.75281151479021,
           -77.61510372161864 37.51568305958246,
           -81.04884624481201 38.86621021801801,
           -79.68778610229492 39.475226764883985))'))
	   ) = 1
limit 1
count(*)
747
Query Results We can flip that query around to use ST_Contains like this: 
select count(*)
from events_geo 
WHERE ST_Contains(
         toSphericalGeography(ST_GeomFromText('POLYGON((
           -79.68778610229492 39.475226764883985,
           -76.83970928192137 40.48289486417028,
           -75.6193685531616 38.75281151479021,
           -77.61510372161864 37.51568305958246,
           -81.04884624481201 38.86621021801801,
           -79.68778610229492 39.475226764883985))')),
           point
	   ) = 1
limit 1
This query will return the same results as the previous one. Finally, we can find some of the points within 50km of Washington: 
select uuid, ST_AsText(point), ts,
       ST_Distance(point, toSphericalGeography(ST_GeomFromText(
		 'POINT(-77.39327430725099 38.93217314143698)'))) AS distance
from events_geo 
WHERE  distance < 50000
limit 10
uuidstastext(point)tsdistance
1a9a7314-7550-4327-9fb5-911f636b869cPOINT (-77.729246 39.255027)2023-03-06 14:27:59.21846146.138375698916
f6d3260b-93d7-478c-b7b1-2dfc214a09dcPOINT (-77.796573 39.060015)2023-03-06 14:40:17.59137640.39151993725
51693e80-cdd3-48d6-af53-78156357a5dcPOINT (-77.751046 39.115125)2023-03-06 14:40:17.59137000.762022801944
fafe2a9c-ca39-4dd2-bf55-37013a1e2719POINT (-77.569928 39.292096)2023-03-06 14:04:46.50942825.54468861071
d5329e92-edf7-4733-816e-9f2e2c580c2dPOINT (-77.188892 39.187191)2023-03-06 14:04:46.50933399.262939756285
9b7cba1f-a2e4-4690-bba9-48b71f50b915POINT (-77.053764 38.896884)2023-03-06 14:04:46.50929635.04964656463
1b5d5a0e-1a6b-46fb-85b4-42e5c0e208e5POINT (-77.618947 38.881897)2023-03-06 13:38:18.33120311.553153107612
1639f1eb-fb60-4f94-9c59-d7e595c8f416POINT (-76.960903 38.772332)2023-03-06 14:12:30.66741445.58284663461
Query Results

​
How do I check that the Geospatial index is being used?

We can check that the Geospatial index is being used by prefixing a query with EXPLAIN PLAN FOR, which will return the query plan. To get the query plan for the ST_Distance function, we’d write the following: 
EXPLAIN PLAN FOR
select uuid, ST_AsText(point), ts,
       ST_Distance(point, toSphericalGeography(ST_GeomFromText(
		 'POINT(-77.39327430725099 38.93217314143698)'))) AS distance
from events_geo 
WHERE  distance < 50000
limit 10
OperatorOperator_IdParent_Id
BROKER_REDUCE(limit:10)10
COMBINE_SELECT21
PLAN_START(numSegmentsForThisPlan:71)-1-1
SELECT(selectList:uuid, stastext(point), ts, st_distance(point,‘80c053592b680000014043775173125738’))32
TRANSFORM(st_distance(point,‘80c053592b680000014043775173125738’), stastext(point), ts, uuid)43
PROJECT(uuid, point, ts)54
DOC_ID_SET65
FILTER_H3_INDEX(indexLookUp:h3_index,operator:RANGE, predicate:st_distance( point,‘80c053592b680000014043775173125738’) < ‘50000’)76
Query Results We must see FILTER_H3_INDEX as one of the operators, otherwise the index isn’t being used. We can do the same thing for the ST_Within and ST_Contains queries. We should see the INCLUSION_FILTER_H3_INDEX operator for both these queries: For ST_Within, we’ll see the following plan:
OperatorOperator_IdParent_Id
BROKER_REDUCE(limit:1)10
COMBINE_AGGREGATE21
PLAN_START(numSegmentsForThisPlan:75)-1-1
AGGREGATE(aggregations:count(*))32
TRANSFORM_PASSTHROUGH()43
PROJECT()54
DOC_ID_SET65
INCLUSION_FILTER_H3_INDEX(inclusionIndex:h3_index,operator:EQ, predicate:stwithin(point,‘84000000010000000600000000c053ec04b00000004043bcd43b0aae 28c05335bdcbffffff40443dcf7fb88244c052e7a3bbffffff4043605c20b209c0c053675ddb ffffff4042c201e70a0102c05443204c00000040436edff9f6ec98c053e c04b00000004043bcd43b0aae28’) = ‘1’)76
Query Results Notice that the INCLUSION_FILTER_H3_INDEX operator contains the stwithin predicate. And for ST_Contains, we’ll see this plan:
OperatorOperator_IdParent_Id
BROKER_REDUCE(limit:1)10
COMBINE_AGGREGATE21
PLAN_START(numSegmentsForThisPlan:75)-1-1
AGGREGATE(aggregations:count(*))32
TRANSFORM_PASSTHROUGH()43
PROJECT()54
DOC_ID_SET65
INCLUSION_FILTER_H3_INDEX(inclusionIndex:h3_index,operator:EQ, predicate:stcontains(‘84000000010000000600000000c053ec04b00000004043bcd43b0aae 28c05335bdcbffffff40443dcf7fb88244c052e7a3bbffffff4043605c20b209c0c053675ddb ffffff4042c201e70a0102c05443204c00000040436edff9f6ec98c053e c04b00000004043bcd43b0aae28’,point) = ‘1’)76
Query Results And notice on this one that the INCLUSION_FILTER_H3_INDEX operator contains the stcontains predicate.