Records are missing after creating the table from spark temp table in Spark2

I have created a dataframe from below sequence.

val df = sc.parallelize(Seq((100,23,9.50),
(100,23,9.51),
(100,24,9.52),
(100,25,9.54),
(100,23,9.55),
(101,21,8.51),
(101,23,8.52),
(101,24,8.55),
(101,20,8.56))).toDF("id", "temp","time")

I wanted to update the DF by addin few more rows where data is missing for the time. So I have iterated the DF from mapPartitions to add new rows.

import org.apache.spark.sql.functions._
import org.apache.spark.sql.{DataFrame, Row, Column}

@transient val w = org.apache.spark.sql.expressions.Window.partitionBy("id").orderBy("time")

val leadDf = df.withColumn("time_diff", ((lead("time", 1).over(w) - df("time")).cast("Float")*100).cast("int"))

Dataframe iteration goes here:

val result =   leadDf.rdd.mapPartitions(itr =>
  new Iterator[Row] {
    var prevRow = null: Row
    var prevDone = true
    var firstRow = true
    var outputRow: Row = null: Row
    var counter  = 0
    var currRecord = null :Row
    var currRow: Row = if (itr.hasNext) {currRecord = itr.next;  currRecord } else null
    prevRow = currRow
    override def hasNext: Boolean = {
      if (!prevDone) {
        prevRow = incrementValue(prevRow,2)
        outputRow = prevRow
        counter = counter -1
        if(counter == 0) {
          prevDone = true
        }
        true
      } else if (itr.hasNext) {
        prevRow = currRow
        if(counter == 0 && prevRow.getAs[Int](3) != 1 && !isNullValue(prevRow,3 )){
          outputRow = prevRow
          counter = prevRow.getAs[Int](3) - 1
          prevDone = false
        }else if(counter > 0) {
          counter = counter -1
          prevDone = false
        }
        else {
          outputRow = currRow
        }
        //if(counter == 0){
        currRow = itr.next
        true
      } else if (currRow != null) {
        outputRow = currRow
        currRow =null
        true
      } else {
        false
      }
    }
    override def next(): Row = outputRow
  })
  val newDf = spark.createDataFrame(result,leadDf.schema)

After this, I can see 12 records in dataframe. But got 10 records from the physical table created by the temp table created from "newDf" dataframe.

newDf.registerTempTable("test")
spark.sql("create table newtest as select * from test")

scala> newDf.count
res14: Long = 12

scala> spark.sql("select * from newtest").count
res15: Long = 10

The same code works fine in Spark 1.6 and final table count matches with dataframe record count.

Can someone explain why this is happening ? and any solution or workaround to solve the problem