fix: Back out "[velox][PR] feat: Optimize deserialize UnsafeRows to RowVector" (facebookincubator#12978)

xiaoxmeng · facebook-github-bot · commit 7668c98c1cc9 · 2025-04-09T14:25:25.000-07:00
Summary: Pull Request resolved: facebookincubator#12978 Original commit changeset: 329fefe9edc1 Original Phabricator Diff: D71754296 Reviewed By: Yuhta Differential Revision: D72728731 fbshipit-source-id: d76756b84b00a663c118588f0327b5bfdab636ef
diff --git a/velox/row/UnsafeRowDeserializers.h b/velox/row/UnsafeRowDeserializers.h
diff --git a/velox/row/UnsafeRowFast.cpp b/velox/row/UnsafeRowFast.cpp
diff --git a/velox/row/UnsafeRowFast.h b/velox/row/UnsafeRowFast.h
@@ -43,14 +43,6 @@ 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);
 
diff --git a/velox/row/benchmarks/DynamicRowVectorDeserializeBenchmark.cpp b/velox/row/benchmarks/DynamicRowVectorDeserializeBenchmark.cpp
@@ -0,0 +1,180 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <folly/Benchmark.h>
+#include <folly/init/Init.h>
+#include <random>
+
+#include "velox/row/UnsafeRowDeserializers.h"
+#include "velox/row/UnsafeRowFast.h"
+#include "velox/type/Type.h"
+#include "velox/vector/fuzzer/VectorFuzzer.h"
+#include "velox/vector/tests/utils/VectorMaker.h"
+
+namespace facebook::spark::benchmarks {
+namespace {
+using namespace facebook::velox;
+using namespace facebook::velox::row;
+using facebook::velox::test::VectorMaker;
+
+class Deserializer {
+ public:
+  virtual ~Deserializer() = default;
+  virtual void deserialize(
+      const std::vector<std::optional<std::string_view>>& data,
+      const TypePtr& type) = 0;
+};
+
+class UnsaferowBatchDeserializer : public Deserializer {
+ public:
+  UnsaferowBatchDeserializer() {}
+
+  void deserialize(
+      const std::vector<std::optional<std::string_view>>& data,
+      const TypePtr& type) override {
+    UnsafeRowDeserializer::deserialize(data, type, pool_.get());
+  }
+
+ private:
+  std::shared_ptr<memory::MemoryPool> pool_{
+      memory::memoryManager()->addLeafPool()};
+};
+
+class BenchmarkHelper {
+ public:
+  std::tuple<std::vector<std::optional<std::string_view>>, TypePtr>
+  randomUnsaferows(int nFields, int nRows, bool stringOnly) {
+    RowTypePtr rowType;
+    std::vector<std::string> names;
+    std::vector<TypePtr> types;
+    names.reserve(nFields);
+    types.reserve(nFields);
+    for (int32_t i = 0; i < nFields; ++i) {
+      names.push_back("");
+      if (stringOnly) {
+        types.push_back(VARCHAR());
+      } else {
+        auto idx = folly::Random::rand32() % allTypes_.size();
+        types.push_back(allTypes_[idx]);
+      }
+    }
+    rowType =
+        TypeFactory<TypeKind::ROW>::create(std::move(names), std::move(types));
+
+    VectorFuzzer::Options opts;
+    opts.vectorSize = 1;
+    opts.nullRatio = 0.1;
+    opts.stringVariableLength = true;
+    opts.stringLength = 20;
+    // Spark uses microseconds to store timestamp
+    opts.timestampPrecision =
+        VectorFuzzer::Options::TimestampPrecision::kMicroSeconds;
+
+    auto seed = folly::Random::rand32();
+    VectorFuzzer fuzzer(opts, pool_.get(), seed);
+    const auto inputVector = fuzzer.fuzzInputRow(rowType);
+    std::vector<std::optional<std::string_view>> results;
+    results.reserve(nRows);
+    // Serialize rowVector into bytes.
+    UnsafeRowFast unsafeRow(inputVector);
+    for (int32_t i = 0; i < nRows; ++i) {
+      BufferPtr bufferPtr =
+          AlignedBuffer::allocate<char>(1024, pool_.get(), true);
+      char* buffer = bufferPtr->asMutable<char>();
+      auto rowSize = unsafeRow.serialize(0, buffer);
+      results.push_back(std::string_view(buffer, rowSize));
+    }
+    return {results, rowType};
+  }
+
+ private:
+  std::vector<TypePtr> allTypes_{
+      BOOLEAN(),
+      TINYINT(),
+      SMALLINT(),
+      INTEGER(),
+      BIGINT(),
+      REAL(),
+      DOUBLE(),
+      VARCHAR(),
+      TIMESTAMP(),
+      ARRAY(INTEGER()),
+      MAP(VARCHAR(), ARRAY(INTEGER())),
+      ROW({INTEGER()})};
+
+  std::shared_ptr<memory::MemoryPool> pool_{
+      memory::memoryManager()->addLeafPool()};
+};
+
+int deserialize(
+    int nIters,
+    int nFields,
+    int nRows,
+    bool variable,
+    std::unique_ptr<Deserializer> deserializer) {
+  folly::BenchmarkSuspender suspender;
+  BenchmarkHelper helper;
+  auto [data, rowType] = helper.randomUnsaferows(nFields, nRows, variable);
+  suspender.dismiss();
+
+  for (int i = 0; i < nIters; i++) {
+    deserializer->deserialize(data, rowType);
+  }
+
+  return nIters * nFields * nRows;
+}
+
+BENCHMARK_NAMED_PARAM_MULTI(
+    deserialize,
+    batch_10_100k_string_only,
+    10,
+    100000,
+    true,
+    std::make_unique<UnsaferowBatchDeserializer>());
+
+BENCHMARK_NAMED_PARAM_MULTI(
+    deserialize,
+    batch_100_100k_string_only,
+    100,
+    100000,
+    true,
+    std::make_unique<UnsaferowBatchDeserializer>());
+
+BENCHMARK_NAMED_PARAM_MULTI(
+    deserialize,
+    batch_10_100k_all_types,
+    10,
+    100000,
+    false,
+    std::make_unique<UnsaferowBatchDeserializer>());
+
+BENCHMARK_NAMED_PARAM_MULTI(
+    deserialize,
+    batch_100_100k_all_types,
+    100,
+    100000,
+    false,
+    std::make_unique<UnsaferowBatchDeserializer>());
+
+} // namespace
+} // namespace facebook::spark::benchmarks
+
+int main(int argc, char** argv) {
+  folly::Init init{&argc, &argv};
+  facebook::velox::memory::MemoryManager::initialize({});
+  folly::runBenchmarks();
+  return 0;
+}
diff --git a/velox/row/benchmarks/UnsafeRowSerializeBenchmark.cpp b/velox/row/benchmarks/UnsafeRowSerializeBenchmark.cpp
@@ -19,6 +19,7 @@
 #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"
 
@@ -48,7 +49,7 @@ class SerializeBenchmark {
     auto serialized = serialize(fast, data->size(), buffer);
     suspender.dismiss();
 
-    auto copy = UnsafeRowFast::deserialize(serialized, rowType, pool());
+    auto copy = UnsafeRowDeserializer::deserialize(serialized, rowType, pool());
     VELOX_CHECK_EQ(copy->size(), data->size());
   }
 
@@ -144,17 +145,17 @@ class SerializeBenchmark {
     return totalSize;
   }
 
-  std::vector<char*> serialize(
+  std::vector<std::optional<std::string_view>> serialize(
       UnsafeRowFast& unsafeRow,
       vector_size_t numRows,
       BufferPtr& buffer) {
-    std::vector<char*> serialized;
+    std::vector<std::optional<std::string_view>> 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(rawBuffer + offset);
+      serialized.push_back(std::string_view(rawBuffer + offset, rowSize));
       offset += rowSize;
     }
 
diff --git a/velox/row/tests/UnsafeRowFuzzTest.cpp b/velox/row/tests/UnsafeRowFuzzTest.cpp
@@ -45,10 +45,10 @@ class UnsafeRowFuzzTests : public ::testing::Test {
     }
   }
 
-  template <typename T>
   void doTest(
       const RowTypePtr& rowType,
-      std::function<std::vector<T>(const RowVectorPtr& data)> serializeFunc) {
+      std::function<std::vector<std::optional<std::string_view>>(
+          const RowVectorPtr& data)> serializeFunc) {
     VectorFuzzer::Options opts;
     opts.vectorSize = kNumBuffers;
     opts.nullRatio = 0.1;
@@ -79,18 +79,12 @@ class UnsafeRowFuzzTests : public ::testing::Test {
 
       // Serialize rowVector into bytes.
       auto serialized = serializeFunc(inputVector);
-      if constexpr (std::is_same_v<T, std::optional<std::string_view>>) {
-        // Deserialize previous bytes back to row vector
-        VectorPtr outputVector = UnsafeRowDeserializer::deserialize(
-            serialized, rowType, pool_.get());
-
-        assertEqualVectors(inputVector, outputVector);
-      } else {
-        VectorPtr outputVector =
-            UnsafeRowFast::deserialize(serialized, rowType, pool_.get());
-
-        assertEqualVectors(inputVector, outputVector);
-      }
+
+      // Deserialize previous bytes back to row vector
+      VectorPtr outputVector =
+          UnsafeRowDeserializer::deserialize(serialized, rowType, pool_.get());
+
+      assertEqualVectors(inputVector, outputVector);
     }
   }
 
@@ -168,9 +162,9 @@ TEST_F(UnsafeRowFuzzTests, fast) {
       MAP(BIGINT(), ROW({BOOLEAN(), TINYINT(), REAL()})),
   });
 
-  doTest<char*>(rowType, [&](const RowVectorPtr& data) {
+  doTest(rowType, [&](const RowVectorPtr& data) {
     const auto numRows = data->size();
-    std::vector<char*> serialized;
+    std::vector<std::optional<std::string_view>> serialized;
     serialized.reserve(numRows);
 
     UnsafeRowFast fast(data);
@@ -195,44 +189,10 @@ TEST_F(UnsafeRowFuzzTests, fast) {
 
       EXPECT_EQ(rowSize, fast.rowSize(i)) << i << ", " << data->toString(i);
 
-      serialized.push_back(buffers_[i]);
+      serialized.push_back(std::string_view(buffers_[i], rowSize));
     }
     return serialized;
   });
-
-  doTest<std::optional<std::string_view>>(
-      rowType, [&](const RowVectorPtr& data) {
-        const auto numRows = data->size();
-        std::vector<std::optional<std::string_view>> serialized;
-        serialized.reserve(numRows);
-
-        UnsafeRowFast fast(data);
-
-        std::vector<vector_size_t> rows(numRows);
-        std::iota(rows.begin(), rows.end(), 0);
-        std::vector<vector_size_t> serializedRowSizes(numRows);
-        std::vector<vector_size_t*> serializedRowSizesPtr(numRows);
-        for (auto i = 0; i < numRows; ++i) {
-          serializedRowSizesPtr[i] = &serializedRowSizes[i];
-        }
-        fast.serializedRowSizes(
-            folly::Range(rows.data(), numRows), serializedRowSizesPtr.data());
-        for (auto i = 0; i < numRows; ++i) {
-          // The serialized row includes the size of the row.
-          VELOX_CHECK_EQ(
-              serializedRowSizes[i], fast.rowSize(i) + sizeof(uint32_t));
-        }
-
-        for (auto i = 0; i < data->size(); ++i) {
-          auto rowSize = fast.serialize(i, buffers_[i]);
-          VELOX_CHECK_LE(rowSize, kBufferSize);
-
-          EXPECT_EQ(rowSize, fast.rowSize(i)) << i << ", " << data->toString(i);
-
-          serialized.push_back(std::string_view(buffers_[i], rowSize));
-        }
-        return serialized;
-      });
 }
 
 } // namespace
diff --git a/velox/serializers/RowSerializer.h b/velox/serializers/RowSerializer.h
@@ -414,13 +414,9 @@ class RowDeserializer {
           uncompressedSource, header.uncompressedSize + initialSize);
       while (rowIterator.hasNext()) {
         serializedBuffers.emplace_back(std::move(rowIterator.next()));
-        if constexpr (std::is_same_v<SerializeView, std::string_view>) {
-          serializedRows.push_back(std::string_view(
-              serializedBuffers.back()->data(),
-              serializedBuffers.back()->size()));
-        } else {
-          serializedRows.push_back(serializedBuffers.back()->data());
-        }
+        serializedRows.push_back(std::string_view(
+            serializedBuffers.back()->data(),
+            serializedBuffers.back()->size()));
       }
     }
   }
diff --git a/velox/serializers/UnsafeRowSerializer.cpp b/velox/serializers/UnsafeRowSerializer.cpp
@@ -15,6 +15,7 @@
  */
 #include "velox/serializers/UnsafeRowSerializer.h"
 #include <folly/lang/Bits.h>
+#include "velox/row/UnsafeRowDeserializers.h"
 #include "velox/row/UnsafeRowFast.h"
 #include "velox/serializers/RowSerializer.h"
 
@@ -43,18 +44,19 @@ void UnsafeRowVectorSerde::deserialize(
     RowTypePtr type,
     RowVectorPtr* result,
     const Options* options) {
-  std::vector<char*> serializedRows;
+  std::vector<std::optional<std::string_view>> serializedRows;
   std::vector<std::unique_ptr<std::string>> serializedBuffers;
-  RowDeserializer<char*>::deserialize<RowIteratorImpl>(
-      source, serializedRows, serializedBuffers, options);
+  RowDeserializer<std::optional<std::string_view>>::deserialize<
+      RowIteratorImpl>(source, serializedRows, serializedBuffers, options);
 
   if (serializedRows.empty()) {
     *result = BaseVector::create<RowVector>(type, 0, pool);
     return;
   }
 
   *result = std::dynamic_pointer_cast<RowVector>(
-      row::UnsafeRowFast::deserialize(serializedRows, type, pool));
+      velox::row::UnsafeRowDeserializer::deserialize(
+          serializedRows, type, pool));
 }
 
 // static

Original file line number	Diff line number	Diff line change
`@@ -414,13 +414,9 @@ class RowDeserializer {`
`414`	`414`	`uncompressedSource, header.uncompressedSize + initialSize);`
`415`	`415`	`while (rowIterator.hasNext()) {`
`416`	`416`	`serializedBuffers.emplace_back(std::move(rowIterator.next()));`
`417`		`- if constexpr (std::is_same_v<SerializeView, std::string_view>) {`
`418`		`- serializedRows.push_back(std::string_view(`
`419`		`- serializedBuffers.back()->data(),`
`420`		`- serializedBuffers.back()->size()));`
`421`		`- } else {`
`422`		`- serializedRows.push_back(serializedBuffers.back()->data());`
`423`		`- }`
	`417`	`+ serializedRows.push_back(std::string_view(`
	`418`	`+ serializedBuffers.back()->data(),`
	`419`	`+ serializedBuffers.back()->size()));`
`424`	`420`	`}`
`425`	`421`	`}`
`426`	`422`	`}`