Optimize remaining filter to not eagerly materialize multi-referenced fields only directly referenced in AND clauses (facebookincubator#10645)

Summary:
Pull Request resolved: facebookincubator#10645

When a field is referenced in both remaining filter and output vector,
we eagerly materialize it to avoid missing rows loaded.  The condition here can
be relaxed: if the field is only referenced in AND clauses without any other
conditional expressions, we should consider it safe to be lazy loaded because we
will not need more rows than any of the AND clause needs.  This helps us
improving some queries from 181 hours to 7.53 hours, comparing to Java using
38.8 hours.

Reviewed By: kevinwilfong

Differential Revision: D60561537

fbshipit-source-id: 92e509f75f71c2baaf732d7c44cc64902eb350b6

Author: Yuhta

velox/connectors/hive/HiveDataSource.cpp

@@ -32,6 +32,30 @@ namespace facebook::velox::connector::hive {
 class HiveTableHandle;
 class HiveColumnHandle;
 
+namespace {
+
+bool isMember(
+    const std::vector<exec::FieldReference*>& fields,
+    const exec::FieldReference& field) {
+  return std::find(fields.begin(), fields.end(), &field) != fields.end();
+}
+
+bool shouldEagerlyMaterialize(
+    const exec::Expr& remainingFilter,
+    const exec::FieldReference& field) {
+  if (!remainingFilter.evaluatesArgumentsOnNonIncreasingSelection()) {
+    return true;
+  }
+  for (auto& input : remainingFilter.inputs()) {
+    if (isMember(input->distinctFields(), field) && input->hasConditionals()) {
+      return true;
+    }
+  }
+  return false;
+}
+
+} // namespace
+
 HiveDataSource::HiveDataSource(
     const RowTypePtr& outputType,
     const std::shared_ptr<connector::ConnectorTableHandle>& tableHandle,
@@ -126,7 +150,9 @@ HiveDataSource::HiveDataSource(
     for (auto& input : remainingFilterExpr->distinctFields()) {
       auto it = columnNames.find(input->field());
       if (it != columnNames.end()) {
-        multiReferencedFields_.push_back(it->second);
+        if (shouldEagerlyMaterialize(*remainingFilterExpr, *input)) {
+          multiReferencedFields_.push_back(it->second);
+        }
         continue;
       }
       // Remaining filter may reference columns that are not used otherwise,

velox/exec/tests/TableScanTest.cpp

@@ -3277,6 +3277,49 @@ TEST_F(TableScanTest, remainingFilterLazyWithMultiReferences) {
   ASSERT_TRUE(waitForTaskCompletion(cursor->task().get()));
 }
 
+// When the multi-referenced fields are in AND clauses without any other
+// conditionals, they should not be eagerly materialized.
+TEST_F(
+    TableScanTest,
+    remainingFilterLazyWithMultiReferencesDirectlyInAndClause) {
+  constexpr int kSize = 10;
+  auto vector = makeRowVector({
+      makeFlatVector<int64_t>(kSize, folly::identity),
+      makeFlatVector<int64_t>(kSize, folly::identity),
+  });
+  auto schema = asRowType(vector->type());
+  auto file = TempFilePath::create();
+  writeToFile(file->getPath(), {vector});
+  CursorParameters params;
+  params.copyResult = false;
+  params.singleThreaded = true;
+  params.planNode = PlanBuilder()
+                        .tableScan(schema, {}, "c0 % 7 == 0 AND c1 % 2 == 0")
+                        .planNode();
+  auto cursor = TaskCursor::create(params);
+  cursor->task()->addSplit(
+      "0", exec::Split(makeHiveConnectorSplit(file->getPath())));
+  cursor->task()->noMoreSplits("0");
+  ASSERT_TRUE(cursor->moveNext());
+  auto* result = cursor->current()->asUnchecked<RowVector>();
+  ASSERT_EQ(result->size(), 1);
+  auto* c0 =
+      result->childAt(0)->loadedVector()->asUnchecked<SimpleVector<int64_t>>();
+  ASSERT_FALSE(c0->isNullAt(0));
+  ASSERT_EQ(c0->valueAt(0), 0);
+  auto* c1 = result->childAt(1)->loadedVector();
+  ASSERT_EQ(c1->encoding(), VectorEncoding::Simple::DICTIONARY);
+  auto* c1Dict = c1->asUnchecked<DictionaryVector<int64_t>>();
+  ASSERT_FALSE(c1Dict->isNullAt(0));
+  ASSERT_EQ(c1Dict->valueAt(0), 0);
+  ASSERT_EQ(
+      c1Dict->valueVector()->encoding(), VectorEncoding::Simple::DICTIONARY);
+  c1Dict = c1Dict->valueVector()->asUnchecked<DictionaryVector<int64_t>>();
+  ASSERT_EQ(c1Dict->valueVector()->size(), 2);
+  ASSERT_FALSE(cursor->moveNext());
+  ASSERT_TRUE(waitForTaskCompletion(cursor->task().get()));
+}
+
 TEST_F(TableScanTest, remainingFilterSkippedStrides) {
   auto rowType = ROW({{"c0", BIGINT()}, {"c1", BIGINT()}});
   std::vector<RowVectorPtr> vectors(3);

velox/expression/Expr.cpp

@@ -299,7 +299,7 @@ void Expr::computeMetadata() {
   }
 
   // (5) Compute hasConditionals_.
-  hasConditionals_ = hasConditionals(this);
+  hasConditionals_ = exec::hasConditionals(this);
 
   metaDataComputed_ = true;
 }

velox/expression/Expr.h

@@ -279,6 +279,10 @@ class Expr {
     return false;
   }
 
+  bool hasConditionals() const {
+    return hasConditionals_;
+  }
+
   bool isDeterministic() const {
     return deterministic_;
   }
@@ -297,6 +301,16 @@ class Expr {
     isMultiplyReferenced_ = true;
   }
 
+  /// True if this is a special form where the next argument will always be
+  /// evaluated on a subset of the rows for which the previous one was
+  /// evaluated.  This is true of AND and no other at this time.  This implies
+  /// that lazies can be loaded on first use and not before starting evaluating
+  /// the form.  This is so because a subsequent use will never access rows that
+  /// were not in scope for the previous one.
+  virtual bool evaluatesArgumentsOnNonIncreasingSelection() const {
+    return false;
+  }
+
   std::vector<common::Subfield> extractSubfields() const;
 
   virtual void extractSubfieldsImpl(
@@ -556,16 +570,6 @@ class Expr {
   // referenced fields.
   virtual void computeDistinctFields();
 
-  // True if this is a spcial form where the next argument will always be
-  // evaluated on a subset of the rows for which the previous one was evaluated.
-  // This is true of AND and no other at this time.  This implies that lazies
-  // can be loaded on first use and not before starting evaluating the form.
-  // This is so because a subsequent use will never access rows that were not in
-  // scope for the previous one.
-  virtual bool evaluatesArgumentsOnNonIncreasingSelection() const {
-    return false;
-  }
-
   const TypePtr type_;
   const std::vector<std::shared_ptr<Expr>> inputs_;
   const std::string name_;