[SEDONA-731] add osm nodes parser

docs/tutorial/sql.md

@@ -480,7 +480,7 @@ Since v1.7.1, Sedona supports loading OSM PBF file format as a DataFrame.
 	```
 
 OSM PBF files can contain nodes, ways, and relations. Currently Sedona support
-DenseNodes, Ways and Relations. When you load the data you get a DataFrame with the following schema.
+Nodes, DenseNodes, Ways and Relations. When you load the data you get a DataFrame with the following schema.
 
 ```
 root

spark/common/src/main/java/org/apache/sedona/sql/datasources/osmpbf/extractors/DenseNodeExtractor.java

@@ -81,6 +81,8 @@ HashMap<String, String> parseTags(Osmformat.StringTable stringTable) {
       keyIndex = keyIndex + 2;
     }
 
+    keyIndex = keyIndex + 1;
+
     return tags;
   }
 }

spark/common/src/main/java/org/apache/sedona/sql/datasources/osmpbf/extractors/NodeExtractor.java

@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.sedona.sql.datasources.osmpbf.extractors;
+
+import java.util.HashMap;
+import org.apache.sedona.sql.datasources.osmpbf.build.Osmformat;
+import org.apache.sedona.sql.datasources.osmpbf.features.TagsResolver;
+import org.apache.sedona.sql.datasources.osmpbf.model.OsmNode;
+
+public class NodeExtractor {
+
+  Osmformat.PrimitiveGroup primitiveGroup;
+  Osmformat.PrimitiveBlock primitiveBlock;
+
+  public NodeExtractor(
+      Osmformat.PrimitiveGroup primitiveGroup, Osmformat.PrimitiveBlock primitiveBlock) {
+    this.primitiveGroup = primitiveGroup;
+    this.primitiveBlock = primitiveBlock;
+  }
+
+  public OsmNode extract(int idx, Osmformat.StringTable stringTable) {
+    return parse(idx, stringTable);
+  }
+
+  private OsmNode parse(int idx, Osmformat.StringTable stringTable) {
+    Osmformat.Node node = primitiveGroup.getNodes(idx);
+
+    long id = node.getId();
+    long latitude = node.getLat();
+    long longitude = node.getLon();
+
+    long latOffset = primitiveBlock.getLatOffset();
+    long lonOffset = primitiveBlock.getLonOffset();
+    long granularity = primitiveBlock.getGranularity();
+
+    // https://wiki.openstreetmap.org/wiki/PBF_Format
+    // latitude = .000000001 * (lat_offset + (granularity * lat))
+    // longitude = .000000001 * (lon_offset + (granularity * lon))
+    float lat = (float) (.000000001 * (latOffset + (latitude * granularity)));
+    float lon = (float) (.000000001 * (lonOffset + (longitude * granularity)));
+
+    HashMap<String, String> tags =
+        TagsResolver.resolveTags(node.getKeysCount(), node::getKeys, node::getVals, stringTable);
+
+    return new OsmNode(id, lat, lon, tags);
+  }
+}

spark/common/src/main/java/org/apache/sedona/sql/datasources/osmpbf/iterators/BlobIterator.java

@@ -27,6 +27,7 @@
 import org.apache.sedona.sql.datasources.osmpbf.build.Fileformat.Blob;
 import org.apache.sedona.sql.datasources.osmpbf.build.Osmformat;
 import org.apache.sedona.sql.datasources.osmpbf.extractors.DenseNodeExtractor;
+import org.apache.sedona.sql.datasources.osmpbf.extractors.NodeExtractor;
 import org.apache.sedona.sql.datasources.osmpbf.extractors.RelationExtractor;
 import org.apache.sedona.sql.datasources.osmpbf.extractors.WaysExtractor;
 import org.apache.sedona.sql.datasources.osmpbf.model.OSMEntity;
@@ -68,7 +69,7 @@ public OSMEntity next() {
     }
 
     if (!currentPrimitiveGroup.getNodesList().isEmpty()) {
-      return null;
+      return extractNodePrimitiveGroup();
     }
 
     if (!currentPrimitiveGroup.getWaysList().isEmpty()) {
@@ -86,6 +87,18 @@ public OSMEntity next() {
     return null;
   }
 
+  private OSMEntity extractNodePrimitiveGroup() {
+    osmEntityIdx += 1;
+    if (currentPrimitiveGroup.getNodesList().size() == osmEntityIdx) {
+      nextEntity();
+    }
+
+    Osmformat.StringTable stringTable = primitiveBlock.getStringtable();
+
+    return new NodeExtractor(currentPrimitiveGroup, primitiveBlock)
+        .extract(osmEntityIdx, stringTable);
+  }
+
   public OSMEntity extractDenseNodePrimitiveGroup() {
     if (denseNodesIterator == null) {
       denseNodesIterator =

spark/common/src/test/scala/org/apache/sedona/sql/OsmReaderTest.scala

@@ -25,8 +25,12 @@ import org.testcontainers.containers.MinIOContainer
 
 import java.io.FileInputStream
 
+case class Node(id: Long, latitude: Double, longitude: Double, tags: Map[String, String])
+
 class OsmReaderTest extends TestBaseScala with Matchers {
   val monacoPath: String = resourceFolder + "osmpbf/monaco-latest.osm.pbf"
+  val densePath: String = resourceFolder + "osmpbf/dense.pbf"
+  val nodesPath: String = resourceFolder + "osmpbf/nodes.pbf"
 
   import sparkSession.implicits._
 
@@ -44,6 +48,40 @@ class OsmReaderTest extends TestBaseScala with Matchers {
       assert(cnt > 0)
     }
 
+    it("should parse normal nodes") {
+      sparkSession.read
+        .format("osmpbf")
+        .load(nodesPath)
+        .select("id", "location.*", "tags")
+        .selectExpr(
+          "id",
+          "ROUND(latitude, 2) AS latitude",
+          "ROUND(longitude, 2) AS longitude",
+          "tags")
+        .as[Node]
+        .collect() should contain theSameElementsAs Array(
+        Node(1002, 48.86, 2.35, Map("amenity" -> "cafe", "name" -> "Cafe de Paris")),
+        Node(1003, 30.12, 22.23, Map("amenity" -> "bakery", "name" -> "Delicious Pastries")),
+        Node(1001, 52.52, 13.40, Map("amenity" -> "restaurant", "name" -> "Curry 36")))
+    }
+
+    it("should parse dense nodes") {
+      sparkSession.read
+        .format("osmpbf")
+        .load(densePath)
+        .select("id", "location.*", "tags")
+        .selectExpr(
+          "id",
+          "ROUND(latitude, 2) AS latitude",
+          "ROUND(longitude, 2) AS longitude",
+          "tags")
+        .as[Node]
+        .collect() should contain theSameElementsAs Array(
+        Node(1002, 48.86, 2.35, Map("amenity" -> "cafe", "name" -> "Cafe de Paris")),
+        Node(1003, 30.12, 22.23, Map("amenity" -> "bakery", "name" -> "Delicious Pastries")),
+        Node(1001, 52.52, 13.40, Map("amenity" -> "restaurant", "name" -> "Curry 36")))
+    }
+
     it("should be able to read from osm file on s3") {
       val container = new MinIOContainer("minio/minio:latest")