Let 'commitTime' be independent with modificationTime

Author: HeartSaVioR

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSinkLog.scala

@@ -57,8 +57,8 @@ case class SinkFileStatus(
       blockReplication: Int,
       blockSize: Long,
       action: String) {
-    // use modification time if we don't know about exact commit time
-    this(path, size, isDir, modificationTime, blockReplication, blockSize, action, modificationTime)
+    // use Long.MaxValue if we don't know about exact commit time, which means they will not evicted
+    this(path, size, isDir, modificationTime, blockReplication, blockSize, action, Long.MaxValue)
   }
 
   def toFileStatus: FileStatus = {
@@ -69,15 +69,14 @@ case class SinkFileStatus(
 
 object SinkFileStatus {
   def apply(f: FileStatus): SinkFileStatus = {
-    SinkFileStatus(
-      path = f.getPath.toUri.toString,
-      size = f.getLen,
-      isDir = f.isDirectory,
-      modificationTime = f.getModificationTime,
-      blockReplication = f.getReplication,
-      blockSize = f.getBlockSize,
-      action = FileStreamSinkLog.ADD_ACTION,
-      commitTime = f.getModificationTime)
+    new SinkFileStatus(
+      f.getPath.toUri.toString,
+      f.getLen,
+      f.isDirectory,
+      f.getModificationTime,
+      f.getReplication,
+      f.getBlockSize,
+      FileStreamSinkLog.ADD_ACTION)
   }
 }
 

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/FileStreamSinkLogSuite.scala

@@ -239,17 +239,18 @@ class FileStreamSinkLogSuite extends SparkFunSuite with SharedSQLContext {
   }
 
   test("read Spark 2.1.0 log format") {
+    val maxLong = Long.MaxValue
     assert(readFromResource("file-sink-log-version-2.1.0") === Seq(
       // SinkFileStatus("/a/b/0", 100, false, 100, 1, 100, FileStreamSinkLog.ADD_ACTION), -> deleted
-      SinkFileStatus("/a/b/1", 100, false, 100, 1, 100, FileStreamSinkLog.ADD_ACTION, 100),
-      SinkFileStatus("/a/b/2", 200, false, 200, 1, 100, FileStreamSinkLog.ADD_ACTION, 200),
-      SinkFileStatus("/a/b/3", 300, false, 300, 1, 100, FileStreamSinkLog.ADD_ACTION, 300),
-      SinkFileStatus("/a/b/4", 400, false, 400, 1, 100, FileStreamSinkLog.ADD_ACTION, 400),
-      SinkFileStatus("/a/b/5", 500, false, 500, 1, 100, FileStreamSinkLog.ADD_ACTION, 500),
-      SinkFileStatus("/a/b/6", 600, false, 600, 1, 100, FileStreamSinkLog.ADD_ACTION, 600),
-      SinkFileStatus("/a/b/7", 700, false, 700, 1, 100, FileStreamSinkLog.ADD_ACTION, 700),
-      SinkFileStatus("/a/b/8", 800, false, 800, 1, 100, FileStreamSinkLog.ADD_ACTION, 800),
-      SinkFileStatus("/a/b/9", 900, false, 900, 3, 200, FileStreamSinkLog.ADD_ACTION, 900)
+      SinkFileStatus("/a/b/1", 100, false, 100, 1, 100, FileStreamSinkLog.ADD_ACTION, maxLong),
+      SinkFileStatus("/a/b/2", 200, false, 200, 1, 100, FileStreamSinkLog.ADD_ACTION, maxLong),
+      SinkFileStatus("/a/b/3", 300, false, 300, 1, 100, FileStreamSinkLog.ADD_ACTION, maxLong),
+      SinkFileStatus("/a/b/4", 400, false, 400, 1, 100, FileStreamSinkLog.ADD_ACTION, maxLong),
+      SinkFileStatus("/a/b/5", 500, false, 500, 1, 100, FileStreamSinkLog.ADD_ACTION, maxLong),
+      SinkFileStatus("/a/b/6", 600, false, 600, 1, 100, FileStreamSinkLog.ADD_ACTION, maxLong),
+      SinkFileStatus("/a/b/7", 700, false, 700, 1, 100, FileStreamSinkLog.ADD_ACTION, maxLong),
+      SinkFileStatus("/a/b/8", 800, false, 800, 1, 100, FileStreamSinkLog.ADD_ACTION, maxLong),
+      SinkFileStatus("/a/b/9", 900, false, 900, 3, 200, FileStreamSinkLog.ADD_ACTION, maxLong)
     ))
   }
 
@@ -258,7 +259,7 @@ class FileStreamSinkLogSuite extends SparkFunSuite with SharedSQLContext {
    * in SinkFileStatus.
    */
   private def newFakeSinkFileStatus(path: String, action: String): SinkFileStatus =
-    newFakeSinkFileStatus(path, action, 100L)
+    newFakeSinkFileStatus(path, action, Long.MaxValue)
 
   /**
    * Create a fake SinkFileStatus using path and action, and commit time.