Fix timestamp spark calculation

simplexspatial · Nov 4, 2021 · 4a51f50 · 4a51f50
1 parent 545d924
commit 4a51f50
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 1 deletion.
diff --git a/spark/src/main/scala/com/acervera/osm4scala/spark/OsmPbfRowIterator.scala b/spark/src/main/scala/com/acervera/osm4scala/spark/OsmPbfRowIterator.scala
@@ -80,7 +80,7 @@ object OsmPbfRowIterator {
 
     private def populateInfo(info: Info): InternalRow = InternalRow.fromSeq(infoSchema.fieldNames.map{
       case FIELD_INFO_VERSION   => info.version.getOrElse(null)
-      case FIELD_INFO_TIMESTAMP => info.timestamp.map(inst => inst.toEpochMilli).getOrElse(null)
+      case FIELD_INFO_TIMESTAMP => info.timestamp.map(inst => inst.toEpochMilli * 1000).getOrElse(null)
       case FIELD_INFO_CHANGESET => info.changeset.getOrElse(null)
       case FIELD_INFO_USER_ID   => info.userId.getOrElse(null)
       case FIELD_INFO_USER_NAME => info.userName.map(UTF8String.fromString).orNull

diff --git a/spark/src/test/scala/com/acervera/osm4scala/spark/OsmPbfFormatSpec.scala b/spark/src/test/scala/com/acervera/osm4scala/spark/OsmPbfFormatSpec.scala
@@ -36,6 +36,8 @@ import org.scalatest.prop.TableDrivenPropertyChecks
 import org.scalatest.wordspec.AnyWordSpec
 
 import java.io.File
+import java.sql.Timestamp
+import java.time.Instant
 import scala.util.Random
 
 object SourcesForTesting {
@@ -145,6 +147,21 @@ class OsmPbfFormatSpec extends AnyWordSpec with Matchers with SparkSessionBefore
           node171946.getAs[Long]("id") shouldBe 171946L
         }
 
+        "read info" in {
+          // <node id="1699777711" version="2" timestamp="2018-03-26T07:24:26Z" lat="43.7402163" lon="7.4281505"/>
+          // Epoch 1522049066000 = Monday, March 26, 2018 7:24:26 AM
+          val node1699777711 = loadOsmPbf(spark, monacoPath)
+            .select(
+              col("id"),
+              col("info.version") as "version",
+              col("info.timestamp") as "timestamp"
+            ).filter("id == 1699777711").collect()(0)
+
+          node1699777711.getAs[Long]("id") shouldBe 1699777711L
+          node1699777711.getAs[Integer]("version") should be(2)
+          node1699777711.getAs[Timestamp]("timestamp").toInstant.toEpochMilli shouldBe 1522049066000L
+        }
+
         "read null info" in {
           val node171946 = loadOsmPbf(spark, madridPath).select("id", "info").filter("id == 171946").collect()(0)
           node171946.getAs[Long]("id") shouldBe 171946L