refactor(nimble): Preserve MAP spec in flat map column readers

Summary:
A production issue S602695 VeloxRuntimeError blocking feature injection pipeline exposed an issue in the Nimble flat map reader.

For context, earlier this half ROO feature injection was enabled in production and supported by both Velox map and flat map vector encodings. A particular pipeline was blocked this past weekend due to flat map reader crashing. A temporary mitigation was to use regular map instead of the flat map encoding as it appeared flat map reader was attempting to set values that were uninitialized by the reader.

The culprit was how scan spec is constructed for flat map. Rather than appending all features to keys and values children, we should follow map scan spec construction by preserving the keys and values and create child columns using the preserved values child spec.

Differential Revision: D89326920

Author: peterenescu

velox/dwio/common/SelectiveFlatMapColumnReader.cpp

@@ -57,42 +57,16 @@ void SelectiveFlatMapColumnReader::getValues(
   auto* resultFlatMap = prepareResult(*result, keysVector_, rows.size());
   setComplexNulls(rows, *result);
 
-  for (const auto& childSpec : scanSpec_->children()) {
-    VELOX_TRACE_HISTORY_PUSH("getValues %s", childSpec->fieldName().c_str());
-    if (!childSpec->keepValues()) {
-      continue;
-    }
-
-    VELOX_CHECK(
-        childSpec->readFromFile(),
-        "Flatmap children must always be read from file.");
-
-    if (childSpec->subscript() == kConstantChildSpecSubscript) {
-      continue;
-    }
-
-    const auto channel = childSpec->channel();
-    const auto index = childSpec->subscript();
-    auto& childResult = resultFlatMap->mapValuesAt(channel);
-
-    VELOX_CHECK(
-        !childSpec->deltaUpdate(),
-        "Delta update not supported in flat map yet");
-    VELOX_CHECK(
-        !childSpec->isConstant(),
-        "Flat map values cannot be constant in scanSpec.");
-    VELOX_CHECK_EQ(
-        childSpec->columnType(),
-        velox::common::ScanSpec::ColumnType::kRegular,
-        "Flat map only supports regular column types in scan spec.");
-
-    children_[index]->getValues(rows, &childResult);
-
-    for (size_t i = 0; i < children_.size(); ++i) {
-      const auto& inMap = inMapBuffer(i);
-      if (inMap) {
-        resultFlatMap->inMapsAt(i, true) = inMap;
-      }
+  // Loop over column readers
+  for (int i = 0; i < children_.size(); ++i) {
+    auto& child = children_[i];
+    VectorPtr values;
+    child->getValues(rows, &values);
+    resultFlatMap->mapValuesAt(i) = values;
+
+    const auto& inMap = inMapBuffer(i);
+    if (inMap) {
+      resultFlatMap->inMapsAt(i, true) = inMap;
     }
   }
 }

velox/vector/FlatMapVector.cpp

@@ -123,12 +123,15 @@ void FlatMapVector::resize(vector_size_t newSize, bool setNotNull) {
       // to skip uniqueness check since effectively we are just changing
       // the length.
       if (newSize > oldSize) {
-        VELOX_CHECK_EQ(
-            values.use_count(), 1, "Resizing shared map values vector");
+        if (values.use_count() > 1) {
+          LOG(WARNING) << "Resizing shared map values vector";
+        }
         values->resize(newSize, setNotNull);
 
         if (i < inMaps_.size() && inMaps_[i] != nullptr) {
-          VELOX_CHECK(inMaps_[i]->unique(), "Resizing shared in map vector");
+          if (!inMaps_[i]->unique()) {
+            LOG(WARNING) << "Resizing shared in map vector";
+          }
           AlignedBuffer::reallocate<bool>(&inMaps_[i], newSize, 0);
         }
       }