Sedona LEFT JOIN using SPATIAL INDEX

[michelelaporta]

Good morning Folks,

I'm trying to use sedona spatial index but I can't figure out how to obtain the same results as if I use a LEFT JOIN.

Given a polygon I need to determinate if a given set of points are WITHIN the polygon. The test is very simple:

1 point is inside
1 point is outside

are these points WITHIN the polygon ? Using the LEFT JOIN I obtain the correct results ( I also want points not in polygon):

+-------------------+----------------------------------------------------------------------------------------------------------------------+
|point              |geometry                                                                                                                                 |
+-------------------+----------------------------------------------------------------------------------------------------------------------+
|POINT (38.7 13.17) |POLYGON ((38.72 13.14, 38.72 13.21, 38.68 13.21, 38.68 13.14, 38.72 13.14))|
|POINT (38.74 13.05)|NULL                                                                                                                            |
+-------------------+----------------------------------------------------------------------------------------------------------------------+

If I use the spatial index the results are different ( the outside point is missing):

+------------------+----------------------------------------------------------------------------------------------------------------------+
|point             |geometry                                                                                                                                |
+------------------+----------------------------------------------------------------------------------------------------------------------+
|POINT (38.7 13.17)|POLYGON ((38.72 13.14, 38.72 13.21, 38.68 13.21, 38.68 13.14, 38.72 13.14))|
+------------------+----------------------------------------------------------------------------------------------------------------------+

How to use spatial index to have the LEFT JOIN results?

Any suggestions would be appreciated.

Attaching a java maven project to show the case:

sedona-spatial-indexing.zip

Thanks in advance

Michele

P.S. without using index the "LEFT JOIN" is quite slow using a big dataset.

[michelelaporta]

as mentioned by petern48 seems this functionality has been already merge here #247 but I cannot find any usage examples. Any help would be appreciated. Thanks

[jiayuasu]

This support is only available via Spatial SQL / DataFrame API. Not the old RDD SpatialJoinQuery API.

You can trigger this via SQL left join operator.

[michelelaporta]

Thanks jiayuasu. In the attached example https://github.com/user-attachments/files/23711205/sedona-spatial-indexing.zip I already try to used the left join operator via SQL but the indexing isn't trigger. With large dataset this approach it's too slow. Could you please clarify "This support is only available via Spatial SQL / DataFrame API. Not the old RDD SpatialJoinQuery API." ? Should this Spatial SQL / DataFrame API support available for java or other languages? Any example/testcase showing up the left join usage with index?
Thanks Michele

[michelelaporta]

Solved follow a working examples:

package test;

import static org.apache.spark.sql.functions.broadcast;
import static org.apache.spark.sql.functions.expr;

import java.util.ArrayList;
import java.util.List;

import org.apache.sedona.spark.SedonaContext;
import org.apache.sedona.sql.SedonaSqlExtensions;
import org.apache.sedona.viz.core.Serde.SedonaVizKryoRegistrator;
import org.apache.sedona.viz.sql.utils.SedonaVizRegistrator;
import org.apache.spark.SparkConf;
import org.apache.spark.serializer.KryoSerializer;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
/**
 * @see https://github.com/apache/sedona/issues/247

--add-opens=java.base/java.nio=ALL-UNNAMED
--add-opens=java.base/java.lang.invoke=ALL-UNNAMED
--add-opens=java.base/java.util=ALL-UNNAMED 
--add-exports java.base/sun.nio.ch=ALL-UNNAMED
--add-exports java.base/sun.util.calendar=ALL-UNNAMED
 */
public class SedonBroadcastTest {

    public static void main(String[] args) {
    	try {
    		SparkConf sparkConf = new SparkConf()
                    .setMaster("local[" + Runtime.getRuntime().availableProcessors() + "]")
                    .set("spark.sedona.global.index", "true")
                    .set("spark.kryo.registrator", SedonaVizKryoRegistrator.class.getName())
                    .set("spark.sql.extensions", SedonaSqlExtensions.class.getName())
                    .set("spark.serializer", KryoSerializer.class.getName())
                    ;

            SparkSession sedona = SedonaContext.builder().config(sparkConf).getOrCreate();
            SedonaVizRegistrator.registerAll(sedona);
			
            String polygonData = "POLYGON ((38.72 13.14,38.72 13.21,38.68 13.21,38.68 13.14,38.72 13.14))";
            String polygonSql = "select ST_GeomFromWKT('" + polygonData + "') as geometry";
            
            Dataset<Row> polygon =sedona.sql(polygonSql);
            
            // point inside
            double lat_in1 = 13.17d;
            double lon_in1 = 38.70d;
    		Row point_in1 = RowFactory.create(lat_in1,lon_in1);
			
			// point outside
			double lat_out1 = 13.502197d;
            double lon_out1 = 38.464880d;
    		Row point_out1 = RowFactory.create(lat_out1,lon_out1);

    		List<Row> rowData = new ArrayList<>();
			rowData.add(point_in1);
			rowData.add(point_out1);
    		
    		StructType schema = DataTypes.createStructType(
				new StructField[] {
					DataTypes.createStructField("lat", DataTypes.DoubleType, true),
					DataTypes.createStructField("lon", DataTypes.DoubleType, true),
			});
			Dataset<Row> data = sedona.createDataFrame(rowData, schema);
			data.createOrReplaceTempView("inputSpatialDf");

			String sqlQuery = "SELECT ST_GeomFromText(CONCAT('POINT('," + "lon" + ",' '," + "lat" + ",')')) as point from inputSpatialDf";
			Dataset<Row> pointDs = sedona.sql(sqlQuery);
			
			// USING BROADCAST
			Dataset<Row> broadcastJoinDf = pointDs.alias("pointDf").join(
			        broadcast(polygon).alias("polygonDf"), expr("ST_Within(pointDf.point, polygonDf.geometry)"), "left");
			broadcastJoinDf.show();
			
    		
		} catch (Exception e) {
			e.printStackTrace();
		}
    }
}

The results:

+--------------------+--------------------+
|               point|            geometry|
+--------------------+--------------------+
|  POINT (38.7 13.17)|POLYGON ((38.72 1...|
|POINT (38.46488 1...|                NULL|
+--------------------+--------------------+

In logs I can see SpatialRDD has been triggered:

25/12/01 12:54:14 DEBUG SedonaSqlParser:64 Parsing command: inputSpatialDf
25/12/01 12:55:28 INFO SpatialRDD:271 Collected 1361 samples

I am correct now?

[jiayuasu]

Please do df.explain() of the spatial join result DataFrame. If you see broadcastindexjoin in the plan, then you are correct

And I believe your code is correct actually

[michelelaporta]

Thanks jiayuasu I see BroadcastIndexJoin in the plan:

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- BroadcastIndexJoin point#96: geometry, RightSide, LeftSide, LeftOuter, WITHIN ST_WITHIN(point#96, geometry#15)
   :- Project [ **org.apache.spark.sql.sedona_sql.expressions.ST_GeomFromText**   AS point#96]
   :  +- FileScan parquet [latitude#42,longitude#43] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/data/adv/ITA/mobilewalla/20250426], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<latitude:double,longitude:double>
   +- SpatialIndex geometry#15: geometry, RTREE, false, false
      +- Filter isnotnull(geometry#15)
         +- InMemoryTableScan [alpha2#0, alpha3#1, geometry#15], [isnotnull(geometry#15)]
               +- InMemoryRelation [alpha2#0, alpha3#1, geometry#15], StorageLevel(disk, memory, deserialized, 1 replicas)
                     +- AdaptiveSparkPlan isFinalPlan=true
                        +- == Final Plan ==
                           ShuffleQueryStage 0
                           +- Exchange SinglePartition, REPARTITION_BY_NUM, [plan_id=95]
                              +- Project [alpha2#0, alpha3#1, st_geomfromwkb(geometry#3) AS geometry#15]
                                 +- *(1) ColumnarToRow
                                    +- FileScan parquet [alpha2#0,alpha3#1,geometry#3] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/tmp/sedona-spatial-indexing/countries.parquet], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<alpha2:string,alpha3:string,geometry:string>
                        +- == Initial Plan ==
                           Exchange SinglePartition, REPARTITION_BY_NUM, [plan_id=8]
                           +- Project [alpha2#0, alpha3#1, st_geomfromwkb(geometry#3) AS geometry#15]
                              +- FileScan parquet [alpha2#0,alpha3#1,geometry#3] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/tmp/sedona-spatial-indexing/countries.parquet], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<alpha2:string,alpha3:string,geometry:string>

[alexeyegorov]

Hi @michelelaporta and @jiayuasu,

after reading this discussion here, I started again trying to somehow enable LEFT JOIN strategy while using Spatial Join in Sedona.

LEFT JOIN problem

Let me explain shortly our problem. First, we are running Sedona on Databricks with Delta Lake (in case, there are any Databricks specific spark issues). Then, we have country-wide datasets comparable to Overture Buildings dataset. Now, to compute some building parameters we want to LEFT JOIN a data set to it which is also country-wide (e.g. parcels). Using broadcast index join is not really working out in the setup:

• it is unfeasable to broadcast one of these datasets
• building an spatial index on one of the datasets is also kind of unfeasable

How we solved this issue so far?

Databricks and Sedona suggest to use grid systems like H3 or S2. We applied this approach:

• create H3 cells
• explode the resulting array
• perform left join using cell id and then applying WHERE filter using ST_INTERSECTS
• deduplicate the result (based on multiple cells per polygon possible)

This approach works, however it requires to create a way bigger data file in-between, the code complexity increases, etc.

Range Join as a solution

I spent yesterday several hours and tested different approaches starting with the INNER JOIN that is using range join by Sedona. This as a starting point is already performing way better then using an intermediate grid creation step. However, we need a LEFT JOIN.

Thanks to several suggestions by Copilot, I found a solution to achieve exactly what we need. Here is a sketch:

WITH inner_join AS (
    SELECT
        buildings.building_id
        , parcels.parcel_id
    FROM buildings
    INNER JOIN parcels
        ON ST_INTERSECTS(buildings.building_geometry, parcels.parcels_geometry)
)

, left_join AS (
    SELECT
        buildings.*
        , COALESCE(inner_join.parcel_id, NULL) AS parcel_id
        , COALESCE(inner_join.parcels_geometry, NULL) AS parcels_geometry
    FROM buildings
    FULL OUTER JOIN inner_join ON buildings.building_id = inner_join.building_id
)

SELECT * FROM left_join

1. First CTE performs an inner-join to retrieve all the matches. As this uses RangeJoin strategy, the performance is great.
2. Second CTE extends the original buildings table by adding the matched parcels entries where applicable.

I ran this setup a few times with a warmed-up cluster (8 nodes with 4 cores). The previous grid solution took me (only the join of the exploded H3 tables and its deduplication) around 15 min. This above solution doesn't require the deduplication and H3 grid steps. Without even considering the H3 cell explosion (writing files to disk etc.) the JOIN took me around 7 min.

Summary

I wanted to share this solution to get some feedback and allow others to see that. I am sure I am not the first one to run into it.
As a side note: the intermediate H3 dataset required around 173GB of storage, while the original is somewhere around 15GB. Meaning, this saves storage costs and also read/write IO.

As we are deploying Sedona almost completely using DBT, we can put this solution as a Jinja macro and reuse it with one line of code. Otherwise this could be exported as function.

@jiayuasu would it be worth to extend the documentation with this solution? I am not sure if Sedona would otherwise be able to use any spatial join to perform the LEFT JOIN in the near future. If this solution is good, I can try to submit a PR for documentation update. In this case, I would be happy if you could hint me, where to start with that. ;)

[jiayuasu]

this is exciting. Please feel free to submit a PR to add the LEFT Join + Range Join to the doc! Please add it here:https://sedona.apache.org/latest/api/sql/Optimizer/#range-join

[alexeyegorov]

I added a PR for that. I assume it matches both range and distance joins as it should not make a difference. I can give it also another try.

#2561

I would also be happy if anyone could verify that it also works as intended on a non-databricks setup.