[GitHub] [hudi] psendyk commented on a diff in pull request #8627: keep a single random record instance

["psendyk (via GitHub)" <gi...@apache.org>]

psendyk commented on code in PR #8627:
URL: https://github.com/apache/hudi/pull/8627#discussion_r1184277562


##########
hudi-cli/src/main/scala/org/apache/hudi/cli/DedupeSparkJob.scala:
##########
@@ -100,81 +106,31 @@ class DedupeSparkJob(basePath: String,
     getDedupePlan(dupeMap)
   }
 
-  private def getDedupePlan(dupeMap: Map[String, Buffer[Row]]): HashMap[String, HashSet[String]] = {
-    val fileToDeleteKeyMap = new HashMap[String, HashSet[String]]()
+  private def getDedupePlan(
+    dupeMap: Map[String, Buffer[Row]]
+  ): HashMap[String, HashMap[String, Int]] = {
+    val fileToDeleteKeyMap = new HashMap[String, HashMap[String, Int]]()
     dupeMap.foreach(rt => {
       val (key, rows) = rt
 
       dedupeType match {
-        case DeDupeType.UPDATE_TYPE =>
-          /*
-          This corresponds to the case where all duplicates have been updated at least once.
-          Once updated, duplicates are bound to have same commit time unless forcefully modified.
-          */
-          rows.init.foreach(r => {
-            val f = r(2).asInstanceOf[String].split("_")(0)
-            if (!fileToDeleteKeyMap.contains(f)) {
-              fileToDeleteKeyMap(f) = HashSet[String]()
-            }
-            fileToDeleteKeyMap(f).add(key)
-          })
-
         case DeDupeType.INSERT_TYPE =>
-          /*
-          This corresponds to the case where duplicates got created due to INSERT and have never been updated.
-          */
-          var maxCommit = -1L
-
-          rows.foreach(r => {
-            val c = r(3).asInstanceOf[String].toLong
-            if (c > maxCommit)
-              maxCommit = c
-          })
-
-          rows.foreach(r => {
-            val c = r(3).asInstanceOf[String].toLong
-            if (c != maxCommit) {
-              val f = r(2).asInstanceOf[String].split("_")(0)
-              if (!fileToDeleteKeyMap.contains(f)) {
-                fileToDeleteKeyMap(f) = HashSet[String]()
-              }
-              fileToDeleteKeyMap(f).add(key)
-            }
-          })
-
-        case DeDupeType.UPSERT_TYPE =>
-          /*
-          This corresponds to the case where duplicates got created as a result of inserts as well as updates,
-          i.e few duplicate records have been updated, while others were never updated.
-           */
-          var maxCommit = -1L
-
-          rows.foreach(r => {
-            val c = r(3).asInstanceOf[String].toLong
-            if (c > maxCommit)
-              maxCommit = c
-          })
-          val rowsWithMaxCommit = new ListBuffer[Row]()
-          rows.foreach(r => {
-            val c = r(3).asInstanceOf[String].toLong
-            if (c != maxCommit) {
-              val f = r(2).asInstanceOf[String].split("_")(0)
-              if (!fileToDeleteKeyMap.contains(f)) {
-                fileToDeleteKeyMap(f) = HashSet[String]()
+          val recordsMarkedForDeletion = new HashSet[String]()
+
+          rows
+            .slice(1, rows.length)

Review Comment:
   keep the first record instance -- doesn't need to be ordered by commit time



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org