Fix localexchange propertie when spill is enabled

Author: feilong-liu

presto-main/src/main/java/com/facebook/presto/sql/planner/optimizations/ActualProperties.java

@@ -47,7 +47,6 @@
 import static com.facebook.presto.sql.planner.optimizations.PartitioningUtils.isRepartitionEffective;
 import static com.facebook.presto.sql.planner.optimizations.PartitioningUtils.translatePartitioningRowExpression;
 import static com.facebook.presto.sql.planner.optimizations.PartitioningUtils.translateToCoalesce;
-import static com.facebook.presto.util.MoreLists.filteredCopy;
 import static com.google.common.base.MoreObjects.toStringHelper;
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.collect.Iterables.transform;
@@ -323,7 +322,16 @@ public ActualProperties build()
         {
             List<LocalProperty<VariableReferenceExpression>> localProperties = this.localProperties;
             if (unordered) {
-                localProperties = filteredCopy(this.localProperties, property -> !property.isOrderSensitive());
+                ImmutableList.Builder<LocalProperty<VariableReferenceExpression>> newLocalProperties = ImmutableList.builder();
+                for (LocalProperty<VariableReferenceExpression> property : this.localProperties) {
+                    if (!property.isOrderSensitive()) {
+                        newLocalProperties.add(property);
+                    }
+                    else {
+                        break;
+                    }
+                }
+                localProperties = newLocalProperties.build();
             }
             return new ActualProperties(global, localProperties, constants);
         }

presto-tests/src/test/java/com/facebook/presto/tests/TestDistributedSpilledQueries.java

@@ -18,6 +18,7 @@
 import com.facebook.presto.testing.QueryRunner;
 import com.facebook.presto.tpch.TpchPlugin;
 import com.google.common.collect.ImmutableMap;
+import org.intellij.lang.annotations.Language;
 import org.testng.annotations.Test;
 
 import java.nio.file.Paths;
@@ -75,4 +76,21 @@ public void testAssignUniqueId()
         // TODO: disabled until https://github.com/prestodb/presto/issues/8926 is resolved
         //       due to long running query test created many spill files on disk.
     }
+
+    @Test
+    public void testQueriesWithSpill()
+    {
+        // Test double filtered left, right, full and inner joins with right constant equality.
+        @Language("SQL") String query = "with t1 as (select max(totalprice) maxprice, min(totalprice) minprice, custkey ckey from orders group by custkey), " +
+                "t2 as (select custkey, totalprice, (select maxprice from t1 where ckey = custkey) maxprice, " +
+                "(select minprice from t1 where ckey=custkey) minprice from orders) select custkey from t2 where " +
+                "(totalprice between minprice and maxprice) group by custkey";
+        Session enableSpill = Session.builder(getSession())
+                .setSystemProperty("spill_enabled", "true")
+                .build();
+        Session disableSpill = Session.builder(getSession())
+                .setSystemProperty("spill_enabled", "true")
+                .build();
+        assertQueryWithSameQueryRunner(disableSpill, query, enableSpill);
+    }
 }