Limit Dictionary to Single Level

- Modifies wrapInDictionary to flatten indices of a wrapped dictionary iwth the wrapping indices.
- Makes lazy loading of a dictionary encoded column to combine the indices with a dictionary wrapper if loading with lazy wrapped in a dictionary.
- Adds functions to transpose dictionaries with and without nulls.
- Changes  NestedLoopJoin and MergeJoin so  that they wrap their input only after the wrapping indices are known. Previously these would wrap first and only then fill in the indices.
- Checks that we do not come across multiple nested dictionaries.

CMakeLists.txt

@@ -508,8 +508,7 @@ if(${VELOX_BUILD_TESTING})
   set_source(c-ares)
   resolve_dependency(c-ares)
 
-  set_source(gRPC)
-  resolve_dependency(gRPC)
+  # set_source(gRPC) resolve_dependency(gRPC)
 endif()
 
 if(VELOX_ENABLE_REMOTE_FUNCTIONS)

velox/core/QueryConfig.h

@@ -768,7 +768,7 @@ class QueryConfig {
   }
 
   bool validateOutputFromOperators() const {
-    return get<bool>(kValidateOutputFromOperators, false);
+    return get<bool>(kValidateOutputFromOperators, true);
   }
 
   bool isExpressionEvaluationCacheEnabled() const {

velox/exec/HashProbe.cpp

@@ -770,6 +770,7 @@ void HashProbe::fillLeftSemiProjectMatchColumn(vector_size_t size) {
 
 void HashProbe::fillOutput(vector_size_t size) {
   prepareOutput(size);
+  WrapState state;
 
   for (auto projection : identityProjections_) {
     // Load input vector if it is being split into multiple batches. It is not
@@ -778,7 +779,7 @@ void HashProbe::fillOutput(vector_size_t size) {
     auto inputChild = input_->childAt(projection.inputChannel);
 
     output_->childAt(projection.outputChannel) =
-        wrapChild(size, outputRowMapping_, inputChild);
+        wrapOne(size, outputRowMapping_, inputChild, nullptr, state);
   }
 
   if (isLeftSemiProjectJoin(joinType_)) {
@@ -1115,6 +1116,7 @@ bool HashProbe::maybeReadSpillOutput() {
 
 RowVectorPtr HashProbe::createFilterInput(vector_size_t size) {
   std::vector<VectorPtr> filterColumns(filterInputType_->size());
+  WrapState state;
   for (auto projection : filterInputProjections_) {
     if (projectedInputColumns_.find(projection.inputChannel) !=
         projectedInputColumns_.end()) {
@@ -1129,8 +1131,12 @@ RowVectorPtr HashProbe::createFilterInput(vector_size_t size) {
       ensureLoadedIfNotAtEnd(projection.inputChannel);
     }
 
-    filterColumns[projection.outputChannel] = wrapChild(
-        size, outputRowMapping_, input_->childAt(projection.inputChannel));
+    filterColumns[projection.outputChannel] = wrapOne(
+        size,
+        outputRowMapping_,
+        input_->childAt(projection.inputChannel),
+        nullptr,
+        state);
   }
 
   extractColumns(

velox/exec/HashProbe.h

@@ -19,6 +19,7 @@
 #include "velox/exec/HashPartitionFunction.h"
 #include "velox/exec/HashTable.h"
 #include "velox/exec/Operator.h"
+#include "velox/exec/OperatorUtils.h"
 #include "velox/exec/ProbeOperatorState.h"
 #include "velox/exec/VectorHasher.h"
 

velox/exec/LocalPartition.cpp

@@ -15,6 +15,7 @@
  */
 
 #include "velox/exec/LocalPartition.h"
+#include "velox/exec/OperatorUtils.h"
 #include "velox/exec/Task.h"
 
 namespace facebook::velox::exec {
@@ -284,9 +285,10 @@ RowVectorPtr
 wrapChildren(const RowVectorPtr& input, vector_size_t size, BufferPtr indices) {
   std::vector<VectorPtr> wrappedChildren;
   wrappedChildren.reserve(input->type()->size());
+  WrapState state;
   for (auto i = 0; i < input->type()->size(); i++) {
-    wrappedChildren.emplace_back(BaseVector::wrapInDictionary(
-        BufferPtr(nullptr), indices, size, input->childAt(i)));
+    wrappedChildren.emplace_back(
+        wrapOne(size, indices, input->childAt(i), nullptr, state));
   }
 
   return std::make_shared<RowVector>(

velox/exec/LocalPlanner.cpp

@@ -47,6 +47,8 @@
 #include "velox/exec/Values.h"
 #include "velox/exec/Window.h"
 
+DEFINE_bool(merge_project, true, "Merge consecutive filter and project nodes");
+
 namespace facebook::velox::exec {
 
 namespace detail {
@@ -431,8 +433,10 @@ std::shared_ptr<Driver> DriverFactory::createDriver(
             std::dynamic_pointer_cast<const core::FilterNode>(planNode)) {
       if (i < planNodes.size() - 1) {
         auto next = planNodes[i + 1];
-        if (auto projectNode =
-                std::dynamic_pointer_cast<const core::ProjectNode>(next)) {
+        std::shared_ptr<const core::ProjectNode> projectNode;
+        if (FLAGS_merge_project &&
+            (projectNode =
+                 std::dynamic_pointer_cast<const core::ProjectNode>(next))) {
           operators.push_back(std::make_unique<FilterProject>(
               id, ctx.get(), filterNode, projectNode));
           i++;