[SEDONA-712] add toSpatialRdd method for RDD[InternalRow] (#1816)

spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/adapters/StructuredAdapter.scala

@@ -84,17 +84,30 @@ object StructuredAdapter {
   }
 
   /**
-   * Convert DataFrame to SpatialRDD. It puts InternalRow as user data of Geometry. It allows only
-   * one geometry column.
+   * Convert DataFrame to SpatialRDD. It puts InternalRow as user data of Geometry.
    *
    * @param dataFrame
    * @param geometryFieldName
    */
-  def toSpatialRdd(dataFrame: DataFrame, geometryFieldName: String): SpatialRDD[Geometry] = {
+  def toSpatialRdd(dataFrame: DataFrame, geometryFieldName: String): SpatialRDD[Geometry] =
+    toSpatialRdd(dataFrame.queryExecution.toRdd, dataFrame.schema, geometryFieldName)
+
+  /**
+   * Convert RDD[InternalRow] to SpatialRDD. It puts InternalRow as user data of Geometry.
+   *
+   * @param rdd
+   * @param schema
+   * @param geometryFieldName
+   * @return
+   */
+  def toSpatialRdd(
+      rdd: RDD[InternalRow],
+      schema: StructType,
+      geometryFieldName: String): SpatialRDD[Geometry] = {
     val spatialRDD = new SpatialRDD[Geometry]
-    spatialRDD.schema = dataFrame.schema
+    spatialRDD.schema = schema
     val ordinal = spatialRDD.schema.fieldIndex(geometryFieldName)
-    spatialRDD.rawSpatialRDD = dataFrame.queryExecution.toRdd
+    spatialRDD.rawSpatialRDD = rdd
       .map(row => {
         val geom = GeometrySerializer.deserialize(row.getBinary(ordinal))
         geom.setUserData(row.copy())