diff --git a/README.md b/README.md
index 51b1233..deb59e6 100644
--- a/README.md
+++ b/README.md
@@ -1,370 +1,420 @@
-# GeoFlink and TStream: Distributed Frameworks for the Real-Time Processing of Spatial Data Streams and Trajectory Streams, Respectively
-
-## Table of Contents
-
-- [Introduction](#introduction)
-- [GeoFlink: Spatial Stream Processing](#geoFlinkStreamProcessing)
- * [Creating a Grid Index](#creating-a-grid-index)
- * [Defining a Spatial Data Stream](#defining-a-spatial-data-stream)
- * [Continuous Spatial Range Query](#continuous-spatial-range-query)
- + [1-Point-Point Spatial Range Query](#1-point-point-spatial-range-query)
- + [2-Point-Polygon Spatial Range Query](#2-point-polygon-spatial-range-query)
- + [3-Polygon-Polygon Spatial Range Query](#3-polygon-polygon-spatial-range-query)
- * [Continuous Spatial kNN Query](#continuous-spatial-knn-query)
- * [Continuous Spatial Join Query](#continuous-spatial-join-query)
- * [Sample GeoFlink Code for a Spatial Range Query](#sample-geoflink-code-for-a-spatial-range-query)
-- [TStream: Trajectory Stream Processing](#tStreamProcessing)
- * [TStream Queries](#tstream-queries)
- + [Continuous Range Query](#continuous-range-query)
- + [Continuous kNN Query](#continuous-knn-query)
- + [Continuous Join Query](#continuous-join-query)
-- [Getting Started](#getting-started)
- * [Requirements](#requirements)
- * [Running Your First GeoFlink/TStream Job](#firstJob)
-- [Publications](#publications)
-- [Contact Us!](#contact)
-
-
-
-## Introduction
-
-**GeoFlink** is an extension of Apache Flink — a scalable opensource distributed streaming engine — for the real-time processing of unbounded spatial streams. GeoFlink leverages a grid-based index for preserving spatial data proximity and pruning of objects which cannot be part of a spatial query result. Thus, providing effective data distribution that guarantees reduced query processing time.
-
-GeoFlink supports spatial range, spatial *k*NN and spatial join queries. Please refer to the [ Publications ](#publications) section for details of the architecture and experimental study demonstrating GeoFlink achieving higher query performance than other ordinary distributed approaches.
-
-**TStream** is a distributed and scalable open source framework for the real-time processing of trajectory data streams. TStream supports range, *k*NN and join queries on trajectory streams.
-
-
-## GeoFlink: Spatial Stream Processing
-
-GeoFlink currently supports GeoJSON and CSV input formats from Apache Kafka and Point spatial object. Future releases will extend support to other input formats and spatial object types including line and polygon.
-
-GeoJSON is a format for encoding a variety of geographic data structures. Its basic element consists of the *type*, *geometry* and *properties* members. The geometry member contains its type (`Point`, `LineString`, `Polygon`, `MultiPoint`, `MultiLineString`, and `MultiPolygon`) and coordinates [longitude, latitude]. For details, please see [https://geojson.org/](https://geojson.org/).
-
-
-```json
-{
- "type": "Feature",
- "geometry": {
- "type": "Point",
- "coordinates": [139.8107, 35.7101]
- },
- "properties": {
- "name": "Tokyo Skytree"
- }
-}
-```
-
-As for the stream in CSV format, the first and second attribute must be longitude and latitude, respectively.
-
-All queries illustrated in this section make use of aggregation windows and are continuous in nature, i.e., they generate window-based continuous results on the continuous data stream. Namely, one output is generated per window aggregation.
-
-In the following code snippets, longitude is referred as X and latitude as Y.
-
-
-### Creating a Grid Index
-Before running queries on GeoFlink, a grid index needs to be defined. These are the spatial bounds where all of the spatial objects and query points are expected to lie. This step generates a grid index, which forms the backbone of GeoFlink's optimized query processing.
-
-The Grid index is constructed by partitioning the 2D space given by its boundary *(MinX, MinY), (MaxX, MaxY)* *(MaxX-MinX = MaxY-MinY)* into square shaped cells of length *l*. Smaller *l* results in the finer data distribution and pruning. However, very small *l* increases number of cells exponentially incurring higher processing costs and lowering throughput.
-
-In the following example, the grid spans over Beijing, China.
-```
-/Defining dataStream boundaries & creating index
-double minX = 115.50, maxX = 117.60, minY = 39.60, maxY = 41.10;
-int gridSize = 100;
-UniformGrid uGrid = new UniformGrid(gridSize, minX, maxX, minY, maxY);
-```
-where *GridSize* of 100 generates a grid index of 100x100 cells, with the bottom-left (minX, minY) and top-right (maxX, maxY) coordinates, respectively.
-
-
-### Defining a Spatial Data Stream
-GeoFlink users need to make an appropriate Apache Kafka connection by specifying the topic name and bootstrap server(s). Once the connection is established, the user can construct spatial stream using the GeoFlink Java/Scala API.
-
-Currently, GeoFlink supports only 'point' type objects. A Point class is a GeoFlink class that initializes incoming spatial data by creating it's Point object using x,y coordinates.
-```
-//Kafka GeoJSON stream to Spatial points stream
-DataStream spatialStream = SpatialStream.PointStream(kafkaStream, "GeoJSON", uGrid);
-```
-where *uGrid* is the grid index.
-
-
-### Continuous Spatial Range Query
-Given a data stream *S*, query point *q*, radius *r* and window parameters, range query returns the *r*-neighbours of *q* in *S* for each aggregation window.
-
-To execute a spatial range query via GeoFlink, Java/Scala API *SpatialRangeQuery* method of the *RangeQuery* class is used.
-
-GeoFlink supports three kinds of spatial range queries.
-1- Point-Point
-2- Point-Polygon
-3- Polygon-Polygon
-
-
-#### 1-Point-Point Spatial Range Query
-Both the query and the spatial stream consist of point objects. An example of point-point spatial range query is as follows:
-```
-// Query point creation using coordinates (longitude, latitude)
-Point queryPoint = new Point(116.414899, 39.920374, uGrid);
-
-// Continous range query
-int windowSize = 10; // window size in seconds
-int windowSlideStep = 5; // window slide step in seconds
-int queryRadius = 0.5;
-
-DataStream rNeighborsStream = RangeQuery.SpatialRangeQuery(spatialStream, queryPoint, queryRadius, windowSize, windowSlideStep, uGrid);
-```
-where *spatialStream* is a spatial data stream, *queryPoint* denotes a query point, *radius* denotes the range query radius, *windowSize* and *windowSlideStep* denote the sliding window size and slide step respectively and *uGrid* denotes the grid index. The query output is generated continuously for each window slide based on size and slide step.
-
-
-#### 2-Point-Polygon Spatial Range Query
-The query is a point object and the spatial stream consists of polygon objects. An example of point-polygon spatial range query is as follows:
-```
-// Query point creation using coordinates (longitude, latitude)
-Point queryPoint = new Point(116.414899, 39.920374, uGrid);
-
-// Continous range query
-int windowSize = 10; // window size in seconds
-int windowSlideStep = 5; // window slide step in seconds
-int queryRadius = 0.5;
-
-DataStream pointPolygonRangeQueryOutput = RangeQuery.SpatialRangeQuery(spatialPolygonStream, queryPoint, radius, uGrid, windowSize, windowSlideStep);
-```
-where *spatialStream* is a spatial data stream, *queryPoint* denotes a query point, *radius* denotes the range query radius, *windowSize* and *windowSlideStep* denote the sliding window size and slide step respectively and *uGrid* denotes the grid index. The query output is generated continuously for each window slide based on size and slide step.
-
-
-#### 3-Polygon-Polygon Spatial Range Query
-The query is a polygon object and the spatial stream consists of polygon objects. An example of polygon-polygon spatial range query is as follows:
-```
-// Query polygon creation using coordinates (longitude, latitude)
-ArrayList queryPolygonCoordinates = new ArrayList();
-queryPolygonCoordinates.add(new Coordinate(-73.984416, 40.675882));
-queryPolygonCoordinates.add(new Coordinate(-73.984511, 40.675767));
-queryPolygonCoordinates.add(new Coordinate(-73.984719, 40.675867));
-queryPolygonCoordinates.add(new Coordinate(-73.984726, 40.67587));
-queryPolygonCoordinates.add(new Coordinate(-73.984718, 40.675881));
-queryPolygonCoordinates.add(new Coordinate(-73.984631, 40.675986));
-queryPolygonCoordinates.add(new Coordinate(-73.984416, 40.675882));
-Polygon queryPolygon = new Polygon(queryPolygonCoordinates, uGrid);
-
-// Continous range query
-int windowSize = 10; // window size in seconds
-int windowSlideStep = 5; // window slide step in seconds
-int queryRadius = 0.5;
-
-DataStream polygonPolygonRangeQueryOutput = RangeQuery.SpatialRangeQuery(spatialPolygonStream, queryPolygon, radius, uGrid, windowSize, windowSlideStep);
-```
-where *spatialStream* is a spatial data stream, *queryPoint* denotes a query point, *radius* denotes the range query radius, *windowSize* and *windowSlideStep* denote the sliding window size and slide step respectively and *uGrid* denotes the grid index. The query output is generated continuously for each window slide based on size and slide step.
-
-
-
-
-### Continuous Spatial kNN Query
-Given a data stream *S*, a query point *q*, a query radius *r*, a positive integer *k* and window parameters, *k*NN query returns the nearest *k* *r*-neighbours of *q* in *S* for each slide of the aggregation window. If less than *k* nearest neighbours of *q* lie within the *r* distance of *q* in *S*, then all the *r*-neighbours are returned.
-
-To execute a spatial *k*NN query in GeoFlink, *SpatialKNNQuery* method of the *KNNQuery* class is used.
-```
-// Query point creation
-Point queryPoint = new Point(116.414899, 39.920374, uGrid);
-
-// Define input parameters
-k = 50
-int windowSize = 10; // window size in seconds
-int windowSlideStep = 5; // window slide step in seconds
-int queryRadius = 0.5;
-
-// Register a spatial kNN query
-DataStream >> outputStream = KNNQuery.SpatialKNNQuery(spatialStream, queryPoint, queryRadius, k, windowSize, windowSlideStep, uGrid);
-```
-where *k* denotes the number of points required in the *k*NN output, *spatialStream* denotes a spatial data stream, *queryPoint* denotes a query point, *queryRadius* denotes the max query radius to search for *k*NN, *windowSize* and *windowSlideStep* denote the sliding window size and slide step respectively and *uGrid* denotes the grid index.
-
-
- Please note that the output stream is a stream of priority queue which is a sorted list of *k*NNs with respect to the distance from the query point *q*. The query output is generated continuously for each window slide based on the *windowSize* and *windowSlideStep*.
-
-
-
-
-
-### Continuous Spatial Join Query
-Given two streams *S1* (Ordinary stream) and *S2* (Query stream), a radius *r* and window parameters, spatial join query returns all the points in *S1* that lie within the radius *r* of *S2* points for each aggregation window.
-
-To execute a spatial join query via the GeoFlink Java/Scala API *SpatialJoinQuery* method of the *JoinQuery* class is used. The query output is generated continuously for each window slide based on the *windowSize* and *windowSlideStep*.
-
-```
-// Create a query stream
-DataStream queryStream = SpatialStream.
-PointStream(geoJSONQryStream,"GeoJSON",uGrid);
-
-// Define input parameters
-int queryRadius = 0.5;
-int windowSize = 10; // window size in seconds
-int windowSlideStep = 5; // window slide step in seconds
-
-// Register a spatial join query
-DataStream>outputStream = JoinQuery.SpatialJoinQuery(spatialStream, queryStream, queryRadius, windowSize, windowSlideStep, uGrid);
-```
-where *spatialStream* and *queryStream* denote the ordinary stream and query stream, respectively and *uGrid* denotes the grid index.
- The query output is generated continuously for each window slide based on the *windowSize* and *windowSlideStep*.
-
-
-### Sample GeoFlink Code for a Spatial Range Query
-
-A sample GeoFlink code written in Java to execute Range Query on a spatial data stream
-
- //Defining dataStream boundaries & creating index
- double minX = 115.50, maxX = 117.60, minY = 39.60, maxY = 41.10;
- int gridSize = 100;
- UniformGrid uGrid = new UniformGrid(gridSize, minX, maxX, minY, maxY);
-
- //Kafka GeoJSON stream to Spatial points stream
- DataStream spatialStream = SpatialStream.PointStream(kafkaStream, "GeoJSON", uGrid);
-
- //Query point creation
- Point queryPoint = new Point(116.414899, 39.920374, uGrid);
-
- //Continous range query
- int windowSize = 10; // window size in seconds
- int windowSlideStep = 5; // window slide step in seconds
- int queryRadius = 0.5;
- DataStream rNeighborsStream = RangeQuery.SpatialRangeQuery(spatialStream, queryPoint, radius, windowSize, windowSlideStep, uGrid);
-
-
-
-
-## TStream: Trajectory Stream Processing
-TStream takes trajectory stream as input and generates a stream of sub-trajectories corresponding to the window size.
-
-
-### TStream Queries
-- Range
-- *k*NN
-- Join
-
-
-#### Continuous Range Query
-
- //Creating uniform grid index by defining its boundaries
- double minX = 115.50, maxX = 117.60, minY = 39.60, maxY = 41.10;
- int gridSize = 100;
- UniformGrid uGrid = new UniformGrid(gridSize, minX, maxX, minY, maxY);
-
- //Generating trajecotry stream from Kafka GeoJSON stream
- DataStream spatialTrajectoryStream = Deserialization.TrajectoryStream(kafkaStream, "GeoJSON", ..., uGrid);
-
- //Query polygon set creation
- Set queryPolygonSet = HelperClass.generateQueryPolygons(n, minX, maxX, minY, maxY, uGrid);
-
- //Configuring query window
- QueryConfiguration windowConf = new QueryConfiguration(QueryType.WindowBased);
- windowConf.setWindowSize(windowSize);
- windowConf.setSlideStep(windowSlideStep);
-
- // Trajectory range query execution
- new PointPolygonTRangeQuery(windowConf, uGrid).run(spatialTrajectoryStream, queryPolygonSet);
-
-
-#### Continuous kNN Query
-
- //Creating uniform grid index by defining its boundaries
- double minX = 115.50, maxX = 117.60, minY = 39.60, maxY = 41.10;
- int gridSize = 100;
- UniformGrid uGrid = new UniformGrid(gridSize, minX, maxX, minY, maxY);
-
- //Declaring query variables
- double radius = 0.05;
- int k = 30;
-
- //Generating trajecotry stream from Kafka GeoJSON stream
- DataStream spatialTrajectoryStream = Deserialization.TrajectoryStream(kafkaStream, "GeoJSON", ..., uGrid);
-
- //Query point set creation
- Set queryPointSet = HelperClass.generateQueryPoints(n, minX, maxX, minY, maxY, uGrid);
-
- //Configuring query window
- QueryConfiguration windowConf = new QueryConfiguration(QueryType.WindowBased);
- windowConf.setWindowSize(windowSize);
- windowConf.setSlideStep(windowSlideStep);
-
- // Trajectory knn query execution
- new PointPointTKNNQuery(windowConf, uGrid).run(spatialTrajectoryStream, queryPointSet, radius, k);
-
-
-#### Continuous Join Query
-
- //Creating uniform grid index by defining its boundaries
- double minX = 115.50, maxX = 117.60, minY = 39.60, maxY = 41.10;
- int gridSize = 100;
- UniformGrid uGrid = new UniformGrid(gridSize, minX, maxX, minY, maxY);
-
- //Declaring query variables
- double radius = 0.05;
- int k = 30;
-
- //Generating two trajecotry streams from Kafka GeoJSON stream
- DataStream spatialTrajectoryStream = Deserialization.TrajectoryStream(kafkaStream, "GeoJSON", ..., uGrid);
- DataStream spatialTrajectoryStream2 = Deserialization.TrajectoryStream(kafkaStream2, "GeoJSON", ..., uGrid);
-
- //Configuring query window
- QueryConfiguration windowConf = new QueryConfiguration(QueryType.WindowBased);
- windowConf.setWindowSize(windowSize);
- windowConf.setSlideStep(windowSlideStep);
-
- // Trajectory join query execution
- new PointPointTJoinQuery(windowConf, uGrid).run(spatialTrajectoryStream, spatialTrajectoryStream2, radius);
-
-
-
-## Getting Started
-
- ### Requirements
- - Java 8
- - Maven 3.0.4 (or higher)
-- Scala 2.11 or 2.12 (optional for running Scala API)
-- Apache Flink cluster v.1.9.x or higher
-- Apache Kafka cluster v.2.x.x
-
-Please ensure that your Apache Flink and Apache Kafka clusters are configured correctly before running GeoFlink.
-
-
-### Running Your First GeoFlink/TStream Job
-- Set up your Kafka cluster and load it with a spatial data stream.
-- Download or clone the GeoFlink from [https://github.com/salmanahmedshaikh/GeoFlink](https://github.com/salmanahmedshaikh/GeoFlink).
-- Use your favourite IDE to open the downloaded GeoFlink project. We recommend using intelliJ Idea IDE.
-- Use ``` StreamingJob ``` class to write your custom code utilizing the GeoFlink's methods discussed above. In the following we provide a sample GeoFlink code for a spatial range query.
-- One can use IntelliJ IDE to execute the GeoFlink's project on a single node.
-- For a cluster execution, a project's jar file need to be created. To generate the ```.jar``` file, go to the project directory through command line and run ```mvn clean package```.
-- The ```.jar``` file can be uploaded and executed through the flink WebUI usually available at [http://localhost:8081](http://localhost:8081/).
-
-
-## Publications
-
- - GeoFlink @ CIKM2020 [GeoFlink: A Distributed and Scalable Framework for the Real-time Processing of Spatial Streams](https://dl.acm.org/doi/10.1145/3340531.3412761)
- - GeoFlink @ IEEE Access2022 [GeoFlink: An Efficient and Scalable Spatial Data Stream Management System](https://doi.org/10.1109/ACCESS.2022.3154063)
-
-
-
-## Contact Us!
-For queries and suggestions, please contact us @ shaikh.salman@aist.go.jp
+# GeoFlink and TStream: Distributed Frameworks for the Real-Time Processing of Spatial Data Streams and Trajectory Streams, Respectively
+
+## Table of Contents
+
+- [Introduction](#introduction)
+- [GeoFlink: Spatial Stream Processing](#geoFlinkStreamProcessing)
+ * [Creating a Grid Index](#creating-a-grid-index)
+ * [Defining a Spatial Data Stream](#defining-a-spatial-data-stream)
+ * [Continuous Spatial Range Query](#continuous-spatial-range-query)
+ + [1-Point-Point Spatial Range Query](#1-point-point-spatial-range-query)
+ + [2-Point-Polygon Spatial Range Query](#2-point-polygon-spatial-range-query)
+ + [3-Polygon-Polygon Spatial Range Query](#3-polygon-polygon-spatial-range-query)
+ * [Continuous Spatial kNN Query](#continuous-spatial-knn-query)
+ * [Continuous Spatial Join Query](#continuous-spatial-join-query)
+ * [Sample GeoFlink Code for a Spatial Range Query](#sample-geoflink-code-for-a-spatial-range-query)
+- [TStream: Trajectory Stream Processing](#tStreamProcessing)
+ * [TStream Queries](#tstream-queries)
+ + [Continuous Range Query](#continuous-range-query)
+ + [Continuous kNN Query](#continuous-knn-query)
+ + [Continuous Join Query](#continuous-join-query)
+- [Getting Started](#getting-started)
+ * [Requirements](#requirements)
+ * [Running Your First GeoFlink/TStream Job](#firstJob)
+- [Building the JNI RVV Distance Library](#jniRVV)
+- [Publications](#publications)
+- [Contact Us!](#contact)
+
+
+
+## Introduction
+
+**GeoFlink** is an extension of Apache Flink — a scalable opensource distributed streaming engine — for the real-time processing of unbounded spatial streams. GeoFlink leverages a grid-based index for preserving spatial data proximity and pruning of objects which cannot be part of a spatial query result. Thus, providing effective data distribution that guarantees reduced query processing time.
+
+GeoFlink supports spatial range, spatial *k*NN and spatial join queries. Please refer to the [ Publications ](#publications) section for details of the architecture and experimental study demonstrating GeoFlink achieving higher query performance than other ordinary distributed approaches.
+
+**TStream** is a distributed and scalable open source framework for the real-time processing of trajectory data streams. TStream supports range, *k*NN and join queries on trajectory streams.
+
+
+## GeoFlink: Spatial Stream Processing
+
+GeoFlink currently supports GeoJSON and CSV input formats from Apache Kafka and Point spatial object. Future releases will extend support to other input formats and spatial object types including line and polygon.
+
+GeoJSON is a format for encoding a variety of geographic data structures. Its basic element consists of the *type*, *geometry* and *properties* members. The geometry member contains its type (`Point`, `LineString`, `Polygon`, `MultiPoint`, `MultiLineString`, and `MultiPolygon`) and coordinates [longitude, latitude]. For details, please see [https://geojson.org/](https://geojson.org/).
+
+
+```json
+{
+ "type": "Feature",
+ "geometry": {
+ "type": "Point",
+ "coordinates": [139.8107, 35.7101]
+ },
+ "properties": {
+ "name": "Tokyo Skytree"
+ }
+}
+```
+
+As for the stream in CSV format, the first and second attribute must be longitude and latitude, respectively.
+
+All queries illustrated in this section make use of aggregation windows and are continuous in nature, i.e., they generate window-based continuous results on the continuous data stream. Namely, one output is generated per window aggregation.
+
+In the following code snippets, longitude is referred as X and latitude as Y.
+
+
+### Creating a Grid Index
+Before running queries on GeoFlink, a grid index needs to be defined. These are the spatial bounds where all of the spatial objects and query points are expected to lie. This step generates a grid index, which forms the backbone of GeoFlink's optimized query processing.
+
+The Grid index is constructed by partitioning the 2D space given by its boundary *(MinX, MinY), (MaxX, MaxY)* *(MaxX-MinX = MaxY-MinY)* into square shaped cells of length *l*. Smaller *l* results in the finer data distribution and pruning. However, very small *l* increases number of cells exponentially incurring higher processing costs and lowering throughput.
+
+In the following example, the grid spans over Beijing, China.
+```
+/Defining dataStream boundaries & creating index
+double minX = 115.50, maxX = 117.60, minY = 39.60, maxY = 41.10;
+int gridSize = 100;
+UniformGrid uGrid = new UniformGrid(gridSize, minX, maxX, minY, maxY);
+```
+where *GridSize* of 100 generates a grid index of 100x100 cells, with the bottom-left (minX, minY) and top-right (maxX, maxY) coordinates, respectively.
+
+
+### Defining a Spatial Data Stream
+GeoFlink users need to make an appropriate Apache Kafka connection by specifying the topic name and bootstrap server(s). Once the connection is established, the user can construct spatial stream using the GeoFlink Java/Scala API.
+
+Currently, GeoFlink supports only 'point' type objects. A Point class is a GeoFlink class that initializes incoming spatial data by creating it's Point object using x,y coordinates.
+```
+//Kafka GeoJSON stream to Spatial points stream
+DataStream spatialStream = SpatialStream.PointStream(kafkaStream, "GeoJSON", uGrid);
+```
+where *uGrid* is the grid index.
+
+
+### Continuous Spatial Range Query
+Given a data stream *S*, query point *q*, radius *r* and window parameters, range query returns the *r*-neighbours of *q* in *S* for each aggregation window.
+
+To execute a spatial range query via GeoFlink, Java/Scala API *SpatialRangeQuery* method of the *RangeQuery* class is used.
+
+GeoFlink supports three kinds of spatial range queries.
+1- Point-Point
+2- Point-Polygon
+3- Polygon-Polygon
+
+
+#### 1-Point-Point Spatial Range Query
+Both the query and the spatial stream consist of point objects. An example of point-point spatial range query is as follows:
+```
+// Query point creation using coordinates (longitude, latitude)
+Point queryPoint = new Point(116.414899, 39.920374, uGrid);
+
+// Continous range query
+int windowSize = 10; // window size in seconds
+int windowSlideStep = 5; // window slide step in seconds
+int queryRadius = 0.5;
+
+DataStream rNeighborsStream = RangeQuery.SpatialRangeQuery(spatialStream, queryPoint, queryRadius, windowSize, windowSlideStep, uGrid);
+```
+where *spatialStream* is a spatial data stream, *queryPoint* denotes a query point, *radius* denotes the range query radius, *windowSize* and *windowSlideStep* denote the sliding window size and slide step respectively and *uGrid* denotes the grid index. The query output is generated continuously for each window slide based on size and slide step.
+
+
+#### 2-Point-Polygon Spatial Range Query
+The query is a point object and the spatial stream consists of polygon objects. An example of point-polygon spatial range query is as follows:
+```
+// Query point creation using coordinates (longitude, latitude)
+Point queryPoint = new Point(116.414899, 39.920374, uGrid);
+
+// Continous range query
+int windowSize = 10; // window size in seconds
+int windowSlideStep = 5; // window slide step in seconds
+int queryRadius = 0.5;
+
+DataStream pointPolygonRangeQueryOutput = RangeQuery.SpatialRangeQuery(spatialPolygonStream, queryPoint, radius, uGrid, windowSize, windowSlideStep);
+```
+where *spatialStream* is a spatial data stream, *queryPoint* denotes a query point, *radius* denotes the range query radius, *windowSize* and *windowSlideStep* denote the sliding window size and slide step respectively and *uGrid* denotes the grid index. The query output is generated continuously for each window slide based on size and slide step.
+
+
+#### 3-Polygon-Polygon Spatial Range Query
+The query is a polygon object and the spatial stream consists of polygon objects. An example of polygon-polygon spatial range query is as follows:
+```
+// Query polygon creation using coordinates (longitude, latitude)
+ArrayList queryPolygonCoordinates = new ArrayList();
+queryPolygonCoordinates.add(new Coordinate(-73.984416, 40.675882));
+queryPolygonCoordinates.add(new Coordinate(-73.984511, 40.675767));
+queryPolygonCoordinates.add(new Coordinate(-73.984719, 40.675867));
+queryPolygonCoordinates.add(new Coordinate(-73.984726, 40.67587));
+queryPolygonCoordinates.add(new Coordinate(-73.984718, 40.675881));
+queryPolygonCoordinates.add(new Coordinate(-73.984631, 40.675986));
+queryPolygonCoordinates.add(new Coordinate(-73.984416, 40.675882));
+Polygon queryPolygon = new Polygon(queryPolygonCoordinates, uGrid);
+
+// Continous range query
+int windowSize = 10; // window size in seconds
+int windowSlideStep = 5; // window slide step in seconds
+int queryRadius = 0.5;
+
+DataStream polygonPolygonRangeQueryOutput = RangeQuery.SpatialRangeQuery(spatialPolygonStream, queryPolygon, radius, uGrid, windowSize, windowSlideStep);
+```
+where *spatialStream* is a spatial data stream, *queryPoint* denotes a query point, *radius* denotes the range query radius, *windowSize* and *windowSlideStep* denote the sliding window size and slide step respectively and *uGrid* denotes the grid index. The query output is generated continuously for each window slide based on size and slide step.
+
+
+
+
+### Continuous Spatial kNN Query
+Given a data stream *S*, a query point *q*, a query radius *r*, a positive integer *k* and window parameters, *k*NN query returns the nearest *k* *r*-neighbours of *q* in *S* for each slide of the aggregation window. If less than *k* nearest neighbours of *q* lie within the *r* distance of *q* in *S*, then all the *r*-neighbours are returned.
+
+To execute a spatial *k*NN query in GeoFlink, *SpatialKNNQuery* method of the *KNNQuery* class is used.
+```
+// Query point creation
+Point queryPoint = new Point(116.414899, 39.920374, uGrid);
+
+// Define input parameters
+k = 50
+int windowSize = 10; // window size in seconds
+int windowSlideStep = 5; // window slide step in seconds
+int queryRadius = 0.5;
+
+// Register a spatial kNN query
+DataStream >> outputStream = KNNQuery.SpatialKNNQuery(spatialStream, queryPoint, queryRadius, k, windowSize, windowSlideStep, uGrid);
+```
+where *k* denotes the number of points required in the *k*NN output, *spatialStream* denotes a spatial data stream, *queryPoint* denotes a query point, *queryRadius* denotes the max query radius to search for *k*NN, *windowSize* and *windowSlideStep* denote the sliding window size and slide step respectively and *uGrid* denotes the grid index.
+
+
+ Please note that the output stream is a stream of priority queue which is a sorted list of *k*NNs with respect to the distance from the query point *q*. The query output is generated continuously for each window slide based on the *windowSize* and *windowSlideStep*.
+
+
+
+
+
+### Continuous Spatial Join Query
+Given two streams *S1* (Ordinary stream) and *S2* (Query stream), a radius *r* and window parameters, spatial join query returns all the points in *S1* that lie within the radius *r* of *S2* points for each aggregation window.
+
+To execute a spatial join query via the GeoFlink Java/Scala API *SpatialJoinQuery* method of the *JoinQuery* class is used. The query output is generated continuously for each window slide based on the *windowSize* and *windowSlideStep*.
+
+```
+// Create a query stream
+DataStream queryStream = SpatialStream.
+PointStream(geoJSONQryStream,"GeoJSON",uGrid);
+
+// Define input parameters
+int queryRadius = 0.5;
+int windowSize = 10; // window size in seconds
+int windowSlideStep = 5; // window slide step in seconds
+
+// Register a spatial join query
+DataStream>outputStream = JoinQuery.SpatialJoinQuery(spatialStream, queryStream, queryRadius, windowSize, windowSlideStep, uGrid);
+```
+where *spatialStream* and *queryStream* denote the ordinary stream and query stream, respectively and *uGrid* denotes the grid index.
+ The query output is generated continuously for each window slide based on the *windowSize* and *windowSlideStep*.
+
+
+### Sample GeoFlink Code for a Spatial Range Query
+
+A sample GeoFlink code written in Java to execute Range Query on a spatial data stream
+
+ //Defining dataStream boundaries & creating index
+ double minX = 115.50, maxX = 117.60, minY = 39.60, maxY = 41.10;
+ int gridSize = 100;
+ UniformGrid uGrid = new UniformGrid(gridSize, minX, maxX, minY, maxY);
+
+ //Kafka GeoJSON stream to Spatial points stream
+ DataStream spatialStream = SpatialStream.PointStream(kafkaStream, "GeoJSON", uGrid);
+
+ //Query point creation
+ Point queryPoint = new Point(116.414899, 39.920374, uGrid);
+
+ //Continous range query
+ int windowSize = 10; // window size in seconds
+ int windowSlideStep = 5; // window slide step in seconds
+ int queryRadius = 0.5;
+ DataStream rNeighborsStream = RangeQuery.SpatialRangeQuery(spatialStream, queryPoint, radius, windowSize, windowSlideStep, uGrid);
+
+
+
+
+## TStream: Trajectory Stream Processing
+TStream takes trajectory stream as input and generates a stream of sub-trajectories corresponding to the window size.
+
+
+### TStream Queries
+- Range
+- *k*NN
+- Join
+
+
+#### Continuous Range Query
+
+ //Creating uniform grid index by defining its boundaries
+ double minX = 115.50, maxX = 117.60, minY = 39.60, maxY = 41.10;
+ int gridSize = 100;
+ UniformGrid uGrid = new UniformGrid(gridSize, minX, maxX, minY, maxY);
+
+ //Generating trajecotry stream from Kafka GeoJSON stream
+ DataStream spatialTrajectoryStream = Deserialization.TrajectoryStream(kafkaStream, "GeoJSON", ..., uGrid);
+
+ //Query polygon set creation
+ Set queryPolygonSet = HelperClass.generateQueryPolygons(n, minX, maxX, minY, maxY, uGrid);
+
+ //Configuring query window
+ QueryConfiguration windowConf = new QueryConfiguration(QueryType.WindowBased);
+ windowConf.setWindowSize(windowSize);
+ windowConf.setSlideStep(windowSlideStep);
+
+ // Trajectory range query execution
+ new PointPolygonTRangeQuery(windowConf, uGrid).run(spatialTrajectoryStream, queryPolygonSet);
+
+
+#### Continuous kNN Query
+
+ //Creating uniform grid index by defining its boundaries
+ double minX = 115.50, maxX = 117.60, minY = 39.60, maxY = 41.10;
+ int gridSize = 100;
+ UniformGrid uGrid = new UniformGrid(gridSize, minX, maxX, minY, maxY);
+
+ //Declaring query variables
+ double radius = 0.05;
+ int k = 30;
+
+ //Generating trajecotry stream from Kafka GeoJSON stream
+ DataStream spatialTrajectoryStream = Deserialization.TrajectoryStream(kafkaStream, "GeoJSON", ..., uGrid);
+
+ //Query point set creation
+ Set queryPointSet = HelperClass.generateQueryPoints(n, minX, maxX, minY, maxY, uGrid);
+
+ //Configuring query window
+ QueryConfiguration windowConf = new QueryConfiguration(QueryType.WindowBased);
+ windowConf.setWindowSize(windowSize);
+ windowConf.setSlideStep(windowSlideStep);
+
+ // Trajectory knn query execution
+ new PointPointTKNNQuery(windowConf, uGrid).run(spatialTrajectoryStream, queryPointSet, radius, k);
+
+
+#### Continuous Join Query
+
+ //Creating uniform grid index by defining its boundaries
+ double minX = 115.50, maxX = 117.60, minY = 39.60, maxY = 41.10;
+ int gridSize = 100;
+ UniformGrid uGrid = new UniformGrid(gridSize, minX, maxX, minY, maxY);
+
+ //Declaring query variables
+ double radius = 0.05;
+ int k = 30;
+
+ //Generating two trajecotry streams from Kafka GeoJSON stream
+ DataStream spatialTrajectoryStream = Deserialization.TrajectoryStream(kafkaStream, "GeoJSON", ..., uGrid);
+ DataStream spatialTrajectoryStream2 = Deserialization.TrajectoryStream(kafkaStream2, "GeoJSON", ..., uGrid);
+
+ //Configuring query window
+ QueryConfiguration windowConf = new QueryConfiguration(QueryType.WindowBased);
+ windowConf.setWindowSize(windowSize);
+ windowConf.setSlideStep(windowSlideStep);
+
+ // Trajectory join query execution
+ new PointPointTJoinQuery(windowConf, uGrid).run(spatialTrajectoryStream, spatialTrajectoryStream2, radius);
+
+
+
+## Getting Started
+
+ ### Requirements
+ - Java 8
+ - Maven 3.0.4 (or higher)
+- Scala 2.11 or 2.12 (optional for running Scala API)
+- Apache Flink cluster v.1.9.x or higher
+- Apache Kafka cluster v.2.x.x
+
+Please ensure that your Apache Flink and Apache Kafka clusters are configured correctly before running GeoFlink.
+
+
+### Running Your First GeoFlink/TStream Job
+- Set up your Kafka cluster and load it with a spatial data stream.
+- Download or clone the GeoFlink from [https://github.com/salmanahmedshaikh/GeoFlink](https://github.com/salmanahmedshaikh/GeoFlink).
+- Use your favourite IDE to open the downloaded GeoFlink project. We recommend using intelliJ Idea IDE.
+- Use ``` StreamingJob ``` class to write your custom code utilizing the GeoFlink's methods discussed above. In the following we provide a sample GeoFlink code for a spatial range query.
+- One can use IntelliJ IDE to execute the GeoFlink's project on a single node.
+- For a cluster execution, a project's jar file need to be created. To generate the ```.jar``` file, go to the project directory through command line and run ```mvn clean package```.
+- The ```.jar``` file can be uploaded and executed through the flink WebUI usually available at [http://localhost:8081](http://localhost:8081/).
+
+
+## Publications
+
+ - GeoFlink @ CIKM2020 [GeoFlink: A Distributed and Scalable Framework for the Real-time Processing of Spatial Streams](https://dl.acm.org/doi/10.1145/3340531.3412761)
+ - GeoFlink @ IEEE Access2022 [GeoFlink: An Efficient and Scalable Spatial Data Stream Management System](https://doi.org/10.1109/ACCESS.2022.3154063)
+
+
+
+## Building the JNI RVV Distance Library
+
+Note on Java-only build
+- The project can be built and run with Java-only code without compiling the C/RVV JNI library. A pure-Java build (e.g., mvn clean package) succeeds without the native .so.
+- The JNI RVV library is optional and only needed if you want RISC-V Vector (RVV) accelerated distance computations.
+
+Prerequisites
+- JDK installed and JAVA_HOME set
+- RISC-V cross compiler: riscv64-unknown-linux-gnu-gcc available in PATH
+- A Linux target runtime for the produced librvvdist.so (the Makefile targets Linux headers include/linux and generates a riscv64 .so)
+
+1) Generate JNI headers
+- From the project's Java source directory, generate the header for the RvvDistanceCalculator native class:
+ cd src/main/java
+ javac -h ../c GeoFlink/utils/RvvDistanceCalculator.java
+- This creates GeoFlink_utils_RvvDistanceCalculator.h under src/main/c.
+
+2) Configure Makefile
+- Open src/main/c/Makefile and set JAVA_HOME to your local JDK installation path (it must contain include/jni.h and include/linux/jni_md.h).
+- Ensure riscv64-unknown-linux-gnu-gcc is in PATH.
+
+3) Build the shared library
+- From src/main/c, build the library:
+ make
+- Output: librvvdist.so in src/main/c.
+
+4) Make the library discoverable at runtime
+- Java will load librvvdist.so via System.loadLibrary("rvvdist"). Ensure the library path includes src/main/c or place the .so in a standard library directory.
+ Options:
+ - Run with JVM option:
+ -Djava.library.path=src/main/c
+ - Or set LD_LIBRARY_PATH before running:
+ export LD_LIBRARY_PATH=src/main/c:$LD_LIBRARY_PATH
+- Deploying on different architectures:
+ - The provided Makefile builds a riscv64 Linux .so. Load it only on a matching RISC-V Linux runtime. Do not attempt to load it on x86_64 hosts.
+
+5) Clean
+- From src/main/c:
+ make clean
+
+Troubleshooting
+- jni.h not found: Check JAVA_HOME and header paths (JAVA_HOME/include and JAVA_HOME/include/linux must exist).
+- riscv64-unknown-linux-gnu-gcc not found: Install the RISC-V cross toolchain and ensure it's in PATH.
+- UnsatisfiedLinkError at runtime:
+ - Verify the library name (rvvdist) matches System.loadLibrary.
+ - Ensure -Djava.library.path or LD_LIBRARY_PATH contains the directory holding librvvdist.so.
+ - Confirm the target architecture (riscv64) matches the runtime system.
+
+
+## Contact Us!
+For queries and suggestions, please contact us @ shaikh.salman@aist.go.jp
diff --git a/src/main/c/RvvDistanceCalculator.c b/src/main/c/RvvDistanceCalculator.c
new file mode 100644
index 0000000..d6ba653
--- /dev/null
+++ b/src/main/c/RvvDistanceCalculator.c
@@ -0,0 +1,75 @@
+#include
+#include
+#include
+#include
+
+/*
+ * Class: GeoFlink_utils_RvvDistanceCalculator
+ * Method: calculateDistances
+ * Signature: (DD[D[D)[D
+ */
+JNIEXPORT jdoubleArray JNICALL Java_GeoFlink_utils_RvvDistanceCalculator_calculateDistances(
+ JNIEnv *env,
+ jclass cls,
+ jdouble qx,
+ jdouble qy,
+ jdoubleArray streamX_j,
+ jdoubleArray streamY_j) {
+
+ // 1. Get pointers to the Java arrays
+ jdouble *streamX = (*env)->GetDoubleArrayElements(env, streamX_j, NULL);
+ jdouble *streamY = (*env)->GetDoubleArrayElements(env, streamY_j, NULL);
+ if (streamX == NULL || streamY == NULL) {
+ // Error handling: failed to get array elements
+ return NULL;
+ }
+
+ jsize n = (*env)->GetArrayLength(env, streamX_j);
+
+ // 2. Create the result array to be returned to Java
+ jdoubleArray distances_j = (*env)->NewDoubleArray(env, n);
+ jdouble *distances = (*env)->GetDoubleArrayElements(env, distances_j, NULL);
+ if (distances == NULL) {
+ // Clean up and return
+ (*env)->ReleaseDoubleArrayElements(env, streamX_j, streamX, JNI_ABORT);
+ (*env)->ReleaseDoubleArrayElements(env, streamY_j, streamY, JNI_ABORT);
+ return NULL;
+ }
+
+ // 3. Main loop for RVV-based computation
+ size_t gvl; // Group vector length
+ for (size_t i = 0; i < n; i += gvl) {
+
+ gvl = __riscv_vsetvl_e64m8(n - i); // Set vector length for this iteration
+
+ // Load stream coordinates into vector registers
+ vfloat64m8_t vec_x = __riscv_vle64_v_f64m8(&streamX[i], gvl);
+ vfloat64m8_t vec_y = __riscv_vle64_v_f64m8(&streamY[i], gvl);
+
+ // Calculate dx = streamX - qx
+ vfloat64m8_t vec_dx = __riscv_vfsub_vf_f64m8(vec_x, qx, gvl);
+ // Calculate dy = streamY - qy
+ vfloat64m8_t vec_dy = __riscv_vfsub_vf_f64m8(vec_y, qy, gvl);
+
+ // Calculate dx*dx
+ vfloat64m8_t vec_dx_sq = __riscv_vfmul_vv_f64m8(vec_dx, vec_dx, gvl);
+ // Calculate dy*dy
+ vfloat64m8_t vec_dy_sq = __riscv_vfmul_vv_f64m8(vec_dy, vec_dy, gvl);
+
+ // Calculate dist_sq = dx*dx + dy*dy
+ vfloat64m8_t vec_dist_sq = __riscv_vfadd_vv_f64m8(vec_dx_sq, vec_dy_sq, gvl);
+
+ // Calculate dist = sqrt(dist_sq)
+ vfloat64m8_t vec_dist = __riscv_vfsqrt_v_f64m8(vec_dist_sq, gvl);
+
+ // Store results back to the distances array
+ __riscv_vse64_v_f64m8(&distances[i], vec_dist, gvl);
+ }
+
+ // 4. Release arrays and return the result
+ (*env)->ReleaseDoubleArrayElements(env, streamX_j, streamX, JNI_ABORT); // JNI_ABORT: we don't need to copy back
+ (*env)->ReleaseDoubleArrayElements(env, streamY_j, streamY, JNI_ABORT);
+ (*env)->ReleaseDoubleArrayElements(env, distances_j, distances, 0); // 0: copy back and free buffer
+
+ return distances_j;
+}
diff --git a/src/main/java/GeoFlink/spatialOperators/knn/PointPointKNNQuery.java b/src/main/java/GeoFlink/spatialOperators/knn/PointPointKNNQuery.java
index b315926..c478a16 100644
--- a/src/main/java/GeoFlink/spatialOperators/knn/PointPointKNNQuery.java
+++ b/src/main/java/GeoFlink/spatialOperators/knn/PointPointKNNQuery.java
@@ -164,23 +164,49 @@ public String getKey(Point p) throws Exception {
public void apply(String gridID, TimeWindow timeWindow, Iterable inputTuples, Collector>> outputStream) throws Exception {
kNNPQ.clear();
- for (Point p : inputTuples) {
+ // Step 1: Collect all points from the window to enable batch processing.
+ java.util.List pointList = new java.util.ArrayList<>();
+ inputTuples.forEach(pointList::add);
+
+ int numPoints = pointList.size();
+ if (numPoints == 0) {
+ outputStream.collect(kNNPQ);
+ return;
+ }
- if (kNNPQ.size() < k) {
- double distance = DistanceFunctions.getDistance(queryPoint, p);
- if(distance <= queryRadius) {
- kNNPQ.offer(new Tuple2(p, distance));
- }
+ // Step 2: Prepare coordinate arrays for the JNI call.
+ double[] streamX = new double[numPoints];
+ double[] streamY = new double[numPoints];
+ int i = 0;
+ for (Point p : pointList) {
+ streamX[i] = p.point.getX();
+ streamY[i] = p.point.getY();
+ i++;
+ }
- } else {
- double distance = DistanceFunctions.getDistance(queryPoint, p);
- if(distance <= queryRadius) {
- // PQ is maintained in descending order with the object with the largest distance from query point at the top/peek
+ // Step 3: Perform batch distance calculation using RVV-optimized native code or a Java fallback.
+ double[] distances;
+ if (GeoFlink.utils.RvvDistanceCalculator.isLibraryLoaded()) {
+ System.out.println("Using Native RVV distance calculation.");
+ distances = GeoFlink.utils.RvvDistanceCalculator.calculateDistances(queryPoint.point.getX(), queryPoint.point.getY(), streamX, streamY);
+ } else {
+ System.out.println("Falling back to Java distance calculation.");
+ distances = GeoFlink.utils.RvvDistanceCalculator.calculateDistancesJava(queryPoint.point.getX(), queryPoint.point.getY(), streamX, streamY);
+ }
+
+ // Step 4: Process the results to find the k-nearest neighbors.
+ for (i = 0; i < numPoints; i++) {
+ double distance = distances[i];
+ Point p = pointList.get(i);
+
+ if (distance <= queryRadius) {
+ if (kNNPQ.size() < k) {
+ kNNPQ.offer(new Tuple2<>(p, distance));
+ } else {
assert kNNPQ.peek() != null;
- double largestDistInPQ = kNNPQ.peek().f1;
- if (largestDistInPQ > distance) { // remove element with the largest distance and add the new element
+ if (kNNPQ.peek().f1 > distance) {
kNNPQ.poll();
- kNNPQ.offer(new Tuple2(p, distance));
+ kNNPQ.offer(new Tuple2<>(p, distance));
}
}
}
@@ -200,4 +226,3 @@ public void apply(String gridID, TimeWindow timeWindow, Iterable inputTup
.apply(new kNNWinAllEvaluationPointStream(k));
}
}
-
diff --git a/src/main/java/GeoFlink/utils/RvvDistanceCalculator.java b/src/main/java/GeoFlink/utils/RvvDistanceCalculator.java
new file mode 100644
index 0000000..89fdfc1
--- /dev/null
+++ b/src/main/java/GeoFlink/utils/RvvDistanceCalculator.java
@@ -0,0 +1,65 @@
+package GeoFlink.utils;
+
+/**
+ * A JNI wrapper class to call RVV-optimized native code for distance calculations.
+ */
+public class RvvDistanceCalculator {
+
+ private static boolean libraryLoaded = false;
+
+ static {
+ try {
+ // The name of the library should match the one generated by the Makefile,
+ // e.g., lib-rvvdist.so on Linux. System.loadLibrary handles the platform-specific naming.
+ System.loadLibrary("rvvdist");
+ libraryLoaded = true;
+ System.out.println("Native RVV distance calculation library loaded successfully.");
+ } catch (UnsatisfiedLinkError e) {
+ // This allows the program to fall back to a pure Java implementation
+ // if the native library is not available for any reason (e.g., running on a non-RISC-V platform).
+ System.err.println("Native RVV library not found. Falling back to Java implementation. Error: " + e.getMessage());
+ }
+ }
+
+ /**
+ * Checks if the native library was loaded successfully.
+ * @return true if the library is available, false otherwise.
+ */
+ public static boolean isLibraryLoaded() {
+ return libraryLoaded;
+ }
+
+ /**
+ * Calculates the Euclidean distance between a query point (qx, qy) and a batch of stream points.
+ * This method delegates the computation to a native C function optimized with RISC-V Vector Extension.
+ *
+ * @param qx The x-coordinate of the query point.
+ * @param qy The y-coordinate of the query point.
+ * @param streamX An array of x-coordinates for the stream points.
+ * @param streamY An array of y-coordinates for the stream points.
+ * @return An array of calculated distances. The length is the same as the input arrays.
+ */
+ public static native double[] calculateDistances(double qx, double qy, double[] streamX, double[] streamY);
+
+ /**
+ * A pure Java fallback implementation for distance calculation.
+ * This is used if the native library cannot be loaded.
+ *
+ * @param qx The x-coordinate of the query point.
+ * @param qy The y-coordinate of the query point.
+ * @param streamX An array of x-coordinates for the stream points.
+ * @param streamY An array of y-coordinates for the stream points.
+ * @return An array of calculated distances.
+ */
+ public static double[] calculateDistancesJava(double qx, double qy, double[] streamX, double[] streamY) {
+ int n = streamX.length;
+ double[] distances = new double[n];
+
+ for (int i = 0; i < n; i++) {
+ double dx = streamX[i] - qx;
+ double dy = streamY[i] - qy;
+ distances[i] = Math.sqrt(dx * dx + dy * dy);
+ }
+ return distances;
+ }
+}