Fix IdentityProjection condition for HashProbe (facebookincubator#1463)

Summary:
Fix the isIdentityProjection_ condition for HashProbe to make sure the input and output have the same fields in the same order. When isIdentityProjection_ is set, HashProbe::getOutput() directly returns input_ in the case of replacedWithDynamicFilter_. So the input and output should have the same RowType.

Pull Request resolved: facebookincubator#1463

Reviewed By: mbasmanova

Differential Revision: D35836201

Pulled By: gggrace14

fbshipit-source-id: 645037d59d0a999403f2eea7552c869624673216

velox/exec/HashProbe.cpp

@@ -81,14 +81,14 @@ HashProbe::HashProbe(
     initializeFilter(joinNode->filter(), probeType, tableType);
   }
 
-  bool isIdentityProjection = true;
+  size_t countIdentityProjection = 0;
   for (auto i = 0; i < probeType->size(); ++i) {
     auto name = probeType->nameOf(i);
     auto outIndex = outputType_->getChildIdxIfExists(name);
     if (outIndex.has_value()) {
       identityProjections_.emplace_back(i, outIndex.value());
-      if (outIndex.value() != i) {
-        isIdentityProjection = false;
+      if (outIndex.value() == i) {
+        countIdentityProjection++;
       }
     }
   }
@@ -100,7 +100,8 @@ HashProbe::HashProbe(
     }
   }
 
-  if (isIdentityProjection && tableResultProjections_.empty()) {
+  if (countIdentityProjection == probeType->size() &&
+      tableResultProjections_.empty()) {
     isIdentityProjection_ = true;
   }
 }

velox/exec/tests/HashJoinTest.cpp

@@ -765,7 +765,27 @@ TEST_F(HashJoinTest, dynamicFilters) {
         "SELECT t.c0, t.c1 + 1 FROM t, u WHERE t.c0 = u.c0");
     EXPECT_EQ(1, getFiltersProduced(task, 1).sum);
     EXPECT_EQ(1, getFiltersAccepted(task, 0).sum);
-    EXPECT_GT(getReplacedWithFilterRows(task, 1).sum, 0);
+    EXPECT_EQ(getReplacedWithFilterRows(task, 1).sum, numRowsBuild * numSplits);
+    EXPECT_LT(getInputPositions(task, 1), numRowsProbe * numSplits);
+  }
+
+  // Push-down that turns join into a no-op with output having a different
+  // number of columns than the input.
+  {
+    core::PlanNodeId probeScanId;
+    auto op = PlanBuilder(planNodeIdGenerator)
+                  .tableScan(probeType)
+                  .capturePlanNodeId(probeScanId)
+                  .hashJoin({"c0"}, {"u_c0"}, keyOnlyBuildSide, "", {"c0"})
+                  .planNode();
+
+    auto task = assertQuery(
+        op,
+        {{probeScanId, leftFiles}},
+        "SELECT t.c0 FROM t JOIN u ON (t.c0 = u.c0)");
+    EXPECT_EQ(1, getFiltersProduced(task, 1).sum);
+    EXPECT_EQ(1, getFiltersAccepted(task, 0).sum);
+    EXPECT_EQ(getReplacedWithFilterRows(task, 1).sum, numRowsBuild * numSplits);
     EXPECT_LT(getInputPositions(task, 1), numRowsProbe * numSplits);
   }