feat: Optimize UnsafeRow to RowVector (facebookincubator#12841)

Summary:
A follow up of facebookincubator#11936
The above PR which has been reverted introduces a core dump when the schema is `Row(MAP(INTEGER(), ROW({ARRAY(INTEGER()), INTEGER()})))`
and the nested row is null, we should check the null buffer first, and then slice the
ARRAY's buffer.

Pull Request resolved: facebookincubator#12841

Reviewed By: pedroerp

Differential Revision: D72997008

Pulled By: kKPulla

fbshipit-source-id: d187393217792bd70a2970569a8075890960071a

Author: jinchengchenghh

velox/row/UnsafeRowDeserializers.h

@@ -43,6 +43,14 @@ class UnsafeRowFast {
   /// 'buffer' must have sufficient capacity and set to all zeros.
   int32_t serialize(vector_size_t index, char* buffer) const;
 
+  /// Deserializes multiple rows into a RowVector of specified type. The type
+  /// must match the contents of the serialized rows.
+  /// @param data The start memory address of each row.
+  static RowVectorPtr deserialize(
+      const std::vector<char*>& data,
+      const RowTypePtr& rowType,
+      memory::MemoryPool* pool);
+
  protected:
   explicit UnsafeRowFast(const VectorPtr& vector);
 

velox/row/UnsafeRowFast.cpp

@@ -19,7 +19,6 @@
 #include "velox/common/memory/HashStringAllocator.h"
 #include "velox/exec/ContainerRowSerde.h"
 #include "velox/row/CompactRow.h"
-#include "velox/row/UnsafeRowDeserializers.h"
 #include "velox/row/UnsafeRowFast.h"
 #include "velox/vector/fuzzer/VectorFuzzer.h"
 
@@ -49,7 +48,7 @@ class SerializeBenchmark {
     auto serialized = serialize(fast, data->size(), buffer);
     suspender.dismiss();
 
-    auto copy = UnsafeRowDeserializer::deserialize(serialized, rowType, pool());
+    auto copy = UnsafeRowFast::deserialize(serialized, rowType, pool());
     VELOX_CHECK_EQ(copy->size(), data->size());
   }
 
@@ -145,17 +144,17 @@ class SerializeBenchmark {
     return totalSize;
   }
 
-  std::vector<std::optional<std::string_view>> serialize(
+  std::vector<char*> serialize(
       UnsafeRowFast& unsafeRow,
       vector_size_t numRows,
       BufferPtr& buffer) {
-    std::vector<std::optional<std::string_view>> serialized;
+    std::vector<char*> serialized;
     auto rawBuffer = buffer->asMutable<char>();
 
     size_t offset = 0;
     for (auto i = 0; i < numRows; ++i) {
       auto rowSize = unsafeRow.serialize(i, rawBuffer + offset);
-      serialized.push_back(std::string_view(rawBuffer + offset, rowSize));
+      serialized.push_back(rawBuffer + offset);
       offset += rowSize;
     }
 

velox/row/UnsafeRowFast.h

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-add_executable(velox_row_test CompactRowTest.cpp UnsafeRowFuzzTest.cpp)
+add_executable(velox_row_test CompactRowTest.cpp UnsafeRowTest.cpp)
 
 add_test(velox_row_test velox_row_test)