[CORE] Fix non-deterministic filter executed twice when push down to …

…scan (#6296)

Co-authored-by: wangguangxin.cn <[email protected]>

backends-velox/src/test/scala/org/apache/gluten/execution/TestOperator.scala

@@ -1892,4 +1892,23 @@ class TestOperator extends VeloxWholeStageTransformerSuite with AdaptiveSparkPla
         }
     }
   }
+
+  test("fix non-deterministic filter executed twice when push down to scan") {
+    val df = sql("select * from lineitem where rand() <= 0.5")
+    // plan check
+    val plan = df.queryExecution.executedPlan
+    val scans = plan.collect { case scan: FileSourceScanExecTransformer => scan }
+    val filters = plan.collect { case filter: FilterExecTransformer => filter }
+    assert(scans.size == 1)
+    assert(filters.size == 1)
+    assert(scans(0).dataFilters.size == 1)
+    val remainingFilters = FilterHandler.getRemainingFilters(
+      scans(0).dataFilters,
+      splitConjunctivePredicates(filters(0).condition))
+    assert(remainingFilters.size == 0)
+
+    // result length check, table lineitem has 60,000 rows
+    val resultLength = df.collect().length
+    assert(resultLength > 25000 && resultLength < 35000)
+  }
 }

gluten-core/src/main/scala/org/apache/gluten/execution/BasicPhysicalOperatorTransformer.scala

@@ -365,7 +365,7 @@ object FilterHandler extends PredicateHelper {
    *   the filter conditions not pushed down into Scan.
    */
   def getRemainingFilters(scanFilters: Seq[Expression], filters: Seq[Expression]): Seq[Expression] =
-    (ExpressionSet(filters) -- ExpressionSet(scanFilters)).toSeq
+    (filters.toSet -- scanFilters.toSet).toSeq
 
   // Separate and compare the filter conditions in Scan and Filter.
   // Try to push down the remaining conditions in Filter into Scan.