feat: Optimize UnsafeRow deserialize for scalar type

velox/row/CMakeLists.txt

@@ -12,7 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-velox_add_library(velox_row_fast CompactRow.cpp UnsafeRowFast.cpp)
+velox_add_library(velox_row_fast CompactRow.cpp UnsafeRowDeserializers.cpp
+                  UnsafeRowFast.cpp)
 
 velox_link_libraries(velox_row_fast PUBLIC velox_vector)
 

velox/row/UnsafeRowDeserializers.cpp

@@ -0,0 +1,281 @@
+/*
+ * 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 "velox/row/UnsafeRowDeserializers.h"
+
+namespace facebook::velox::row {
+namespace {
+// Returns the offset of a column to starting memory address of one row.
+// @param nullBitsetWidthInBytes The null-tracking bit set is aligned to 8-byte
+// word boundaries. It stores one bit per field.
+// @param columnIdx column index.
+inline int64_t getFieldOffset(
+    int64_t nullBitsetWidthInBytes,
+    column_index_t columnIdx) {
+  return nullBitsetWidthInBytes + UnsafeRow::kFieldWidthBytes * columnIdx;
+}
+
+inline bool isNullAt(const uint8_t* memoryAddress, vector_size_t row) {
+  return bits::isBitSet(memoryAddress, row);
+}
+
+size_t getTotalStringSize(
+    column_index_t columnIdx,
+    vector_size_t numRows,
+    int64_t fieldOffset,
+    const std::vector<int64_t>& offsets,
+    const uint8_t* memoryAddress) {
+  size_t size = 0;
+  for (auto row = 0; row < numRows; row++) {
+    if (isNullAt(memoryAddress + offsets[row], columnIdx)) {
+      continue;
+    }
+
+    int64_t offsetAndSize =
+        *(int64_t*)(memoryAddress + offsets[row] + fieldOffset);
+    int32_t length = static_cast<int32_t>(offsetAndSize);
+    if (!StringView::isInline(length)) {
+      size += length;
+    }
+  }
+  return size;
+}
+
+template <TypeKind Kind>
+VectorPtr createFlatVectorFast(
+    const TypePtr& type,
+    column_index_t columnIdx,
+    vector_size_t numRows,
+    int64_t fieldOffset,
+    const std::vector<int64_t>& offsets,
+    const uint8_t* memoryAddress,
+    memory::MemoryPool* pool) {
+  using T = typename TypeTraits<Kind>::NativeType;
+  constexpr uint32_t typeWidth = sizeof(T);
+  auto column = BaseVector::create<FlatVector<T>>(type, numRows, pool);
+  auto rawValues = column->template mutableRawValues<uint8_t>();
+  const auto shift = __builtin_ctz(typeWidth);
+  for (auto row = 0; row < numRows; row++) {
+    if (!isNullAt(memoryAddress + offsets[row], columnIdx)) {
+      const uint8_t* srcPtr = (memoryAddress + offsets[row] + fieldOffset);
+      uint8_t* destPtr = rawValues + (row << shift);
+      memcpy(destPtr, srcPtr, typeWidth);
+    } else {
+      column->setNull(row, true);
+    }
+  }
+  return column;
+}
+
+template <>
+VectorPtr createFlatVectorFast<TypeKind::HUGEINT>(
+    const TypePtr& type,
+    column_index_t columnIdx,
+    vector_size_t numRows,
+    int64_t fieldOffset,
+    const std::vector<int64_t>& offsets,
+    const uint8_t* memoryAddress,
+    memory::MemoryPool* pool) {
+  auto column = BaseVector::create<FlatVector<int128_t>>(type, numRows, pool);
+  auto rawValues = column->mutableRawValues<uint8_t>();
+  for (auto row = 0; row < numRows; row++) {
+    if (!isNullAt(memoryAddress + offsets[row], columnIdx)) {
+      const int64_t offsetAndSize =
+          *(int64_t*)(memoryAddress + offsets[row] + fieldOffset);
+      const int32_t length = static_cast<int32_t>(offsetAndSize);
+      const int32_t wordOffset = static_cast<int32_t>(offsetAndSize >> 32);
+      rawValues[row] =
+          UnsafeRowPrimitiveBatchDeserializer::deserializeLongDecimal(
+              std::string_view(reinterpret_cast<const char*>(
+                  memoryAddress + offsets[row] + wordOffset, length)));
+    } else {
+      column->setNull(row, true);
+    }
+  }
+  return column;
+}
+
+template <>
+VectorPtr createFlatVectorFast<TypeKind::BOOLEAN>(
+    const TypePtr& type,
+    column_index_t columnIdx,
+    vector_size_t numRows,
+    int64_t fieldOffset,
+    const std::vector<int64_t>& offsets,
+    const uint8_t* memoryAddress,
+    memory::MemoryPool* pool) {
+  auto column = BaseVector::create<FlatVector<bool>>(type, numRows, pool);
+  auto rawValues = column->mutableRawValues<uint64_t>();
+  for (auto row = 0; row < numRows; row++) {
+    if (!isNullAt(memoryAddress + offsets[row], columnIdx)) {
+      const bool value = *(bool*)(memoryAddress + offsets[row] + fieldOffset);
+      bits::setBit(rawValues, row, value);
+    } else {
+      column->setNull(row, true);
+    }
+  }
+  return column;
+}
+
+template <>
+VectorPtr createFlatVectorFast<TypeKind::TIMESTAMP>(
+    const TypePtr& type,
+    column_index_t columnIdx,
+    vector_size_t numRows,
+    int64_t fieldOffset,
+    const std::vector<int64_t>& offsets,
+    const uint8_t* memoryAddress,
+    memory::MemoryPool* pool) {
+  auto column = BaseVector::create<FlatVector<Timestamp>>(type, numRows, pool);
+  for (auto row = 0; row < numRows; row++) {
+    if (!isNullAt(memoryAddress + offsets[row], columnIdx)) {
+      const int64_t value =
+          *(int64_t*)(memoryAddress + offsets[row] + fieldOffset);
+      column->set(row, Timestamp::fromMicros(value));
+    } else {
+      column->setNull(row, true);
+    }
+  }
+  return column;
+}
+
+template <>
+VectorPtr createFlatVectorFast<TypeKind::VARCHAR>(
+    const TypePtr& type,
+    column_index_t columnIdx,
+    vector_size_t numRows,
+    int64_t fieldOffset,
+    const std::vector<int64_t>& offsets,
+    const uint8_t* memoryAddress,
+    memory::MemoryPool* pool) {
+  auto column = BaseVector::create<FlatVector<StringView>>(type, numRows, pool);
+  const auto totalSize = getTotalStringSize(
+      columnIdx, numRows, fieldOffset, offsets, memoryAddress);
+  char* rawBuffer = column->getRawStringBufferWithSpace(totalSize, true);
+  for (auto row = 0; row < numRows; row++) {
+    if (!isNullAt(memoryAddress + offsets[row], columnIdx)) {
+      const int64_t offsetAndSize =
+          *(int64_t*)(memoryAddress + offsets[row] + fieldOffset);
+      const int32_t length = static_cast<int32_t>(offsetAndSize);
+      const int32_t wordOffset = static_cast<int32_t>(offsetAndSize >> 32);
+      auto* valueSrc = memoryAddress + offsets[row] + wordOffset;
+      if (StringView::isInline(length)) {
+        column->set(
+            row, StringView(reinterpret_cast<const char*>(valueSrc), length));
+      } else {
+        memcpy(rawBuffer, valueSrc, length);
+        column->setNoCopy(row, StringView(rawBuffer, length));
+        rawBuffer += length;
+      }
+    } else {
+      column->setNull(row, true);
+    }
+  }
+  return column;
+}
+
+template <>
+VectorPtr createFlatVectorFast<TypeKind::VARBINARY>(
+    const TypePtr& type,
+    column_index_t columnIdx,
+    vector_size_t numRows,
+    int64_t fieldOffset,
+    const std::vector<int64_t>& offsets,
+    const uint8_t* memoryAddress,
+    memory::MemoryPool* pool) {
+  return createFlatVectorFast<TypeKind::VARCHAR>(
+      type, columnIdx, numRows, fieldOffset, offsets, memoryAddress, pool);
+}
+
+VectorPtr createUnknownFlatVector(
+    vector_size_t numRows,
+    memory::MemoryPool* pool) {
+  const auto nulls = allocateNulls(numRows, pool, bits::kNull);
+  return std::make_shared<FlatVector<UnknownValue>>(
+      pool,
+      UNKNOWN(),
+      nulls,
+      numRows,
+      nullptr, // values
+      std::vector<BufferPtr>{}); // stringBuffers
+}
+
+bool fastSupported(const RowTypePtr& type) {
+  for (auto i = 0; i < type->size(); i++) {
+    const auto kind = type->childAt(i)->kind();
+    switch (kind) {
+      case TypeKind::ARRAY:
+      case TypeKind::MAP:
+      case TypeKind::ROW:
+        return false;
+      default:
+        break;
+    }
+  }
+  return true;
+}
+
+VectorPtr deserializeFast(
+    const uint8_t* memoryAddress,
+    const RowTypePtr& type,
+    const std::vector<int64_t>& offsets,
+    vector_size_t numRows,
+    memory::MemoryPool* pool) {
+  const auto numFields = type->size();
+  const int64_t nullBitsetWidthInBytes = UnsafeRow::getNullLength(numFields);
+  std::vector<VectorPtr> columns(numFields);
+  for (auto i = 0; i < numFields; i++) {
+    const auto fieldOffset = getFieldOffset(nullBitsetWidthInBytes, i);
+    const auto& colType = type->childAt(i);
+    if (colType->kind() == TypeKind::UNKNOWN) {
+      columns[i] = createUnknownFlatVector(numRows, pool);
+    } else {
+      columns[i] = VELOX_DYNAMIC_SCALAR_TYPE_DISPATCH(
+          createFlatVectorFast,
+          colType->kind(),
+          colType,
+          i,
+          numRows,
+          fieldOffset,
+          offsets,
+          memoryAddress,
+          pool);
+    }
+  }
+
+  return std::make_shared<RowVector>(
+      pool, type, BufferPtr(nullptr), numRows, std::move(columns));
+}
+} // namespace
+
+// static
+VectorPtr UnsafeRowDeserializer::deserialize(
+    const uint8_t* memoryAddress,
+    const RowTypePtr& type,
+    const std::vector<int64_t>& offsets,
+    memory::MemoryPool* pool) {
+  const vector_size_t numRows = offsets.size() - 1;
+  if (fastSupported(type)) {
+    return deserializeFast(memoryAddress, type, offsets, numRows, pool);
+  }
+  std::vector<std::optional<std::string_view>> data;
+  for (auto i = 0; i < numRows; i++) {
+    const auto length = offsets[i + 1] - offsets[i];
+    data.emplace_back(std::string_view(
+        reinterpret_cast<const char*>(memoryAddress + offsets[i]), length));
+  }
+  return deserialize(data, type, pool);
+}
+} // namespace facebook::velox::row

velox/row/UnsafeRowDeserializers.h

@@ -552,6 +552,50 @@ struct UnsafeRowPrimitiveBatchDeserializer {
  * a Vector.
  */
 struct UnsafeRowDeserializer {
+ public:
+  /// Deserializes rows that are stored in contiguous memory.
+  /// If all column types are primitive, fast deserialization is used.
+  /// Otherwise, rows are deserialized one by one as before.
+  /// Fast deserialization calculates the starting memory address of data
+  /// based on the column index and row number, setting all data for one column
+  /// at once.
+  /// The null-tracking bit set is aligned to 8-byte word boundaries. It stores
+  /// one bit per field.
+  /// Each row has three parts: [null-tracking bit set] [values] [variable
+  /// length portion] In the `values` region, we store one 8-byte word per
+  /// field. For fields that hold fixed-length primitive types, such as long,
+  /// double, or int, we store the value directly in the word. For fields with
+  /// non-primitive or variable-length values, we store a relative offset
+  /// (w.r.t. the base address of the row) that points to the beginning of the
+  /// variable-length field, and length (they are combined into a long).
+  /// So we can get the starting memory address of each row by its offset, and
+  /// get the data of one column by field offset which is computed by fixed
+  /// width null bit set and fixed width column word.
+  /// @param memoryAddress The starting memory address of the serialized rows.
+  /// @param type The element type.
+  /// @param offsets Offset of each row serialized data. It's size should be
+  /// equal to the number of deserialized rows + 1. First offset is 0.
+  /// @param pool The memory pool used to allocate vector.
+  static VectorPtr deserialize(
+      const uint8_t* memoryAddress,
+      const RowTypePtr& type,
+      const std::vector<int64_t>& offsets,
+      memory::MemoryPool* pool);
+
+  /// Deserializes a complex element type to its Vector representation.
+  /// @param data A vector of string_view over a given element in the
+  /// UnsafeRow.
+  /// @param type the element type.
+  /// @param pool the memory pool to allocate Vectors from data to a array.
+  /// @return a VectorPtr
+  static VectorPtr deserialize(
+      const std::vector<std::optional<std::string_view>>& data,
+      const TypePtr& type,
+      memory::MemoryPool* pool) {
+    return convertToVectors(getBatchIteratorPtr(data, type), pool);
+  }
+
+ private:
   /**
    * Allocate and populate the metadata Vectors in ArrayVector or MapVector.
    * @param dataIterator iterator that points to whole column batch of data.
@@ -823,38 +867,6 @@ struct UnsafeRowDeserializer {
       VELOX_NYI("Unsupported data iterators type");
     }
   }
-
-  /**
-   * Deserializes a complex element type to its Vector representation.
-   * @param data A string_view over a given element in the UnsafeRow.
-   * @param type the element type.
-   * @param pool the memory pool to allocate Vectors from
-   *data to a array.
-   * @return a VectorPtr
-   */
-  static VectorPtr deserializeOne(
-      std::optional<std::string_view> data,
-      const TypePtr& type,
-      memory::MemoryPool* pool) {
-    std::vector<std::optional<std::string_view>> vectors{data};
-    return deserialize(vectors, type, pool);
-  }
-
-  /**
-   * Deserializes a complex element type to its Vector representation.
-   * @param data A vector of string_view over a given element in the
-   *UnsafeRow.
-   * @param type the element type.
-   * @param pool the memory pool to allocate Vectors from
-   *data to a array.
-   * @return a VectorPtr
-   */
-  static VectorPtr deserialize(
-      const std::vector<std::optional<std::string_view>>& data,
-      const TypePtr& type,
-      memory::MemoryPool* pool) {
-    return convertToVectors(getBatchIteratorPtr(data, type), pool);
-  }
 };
 
 } // namespace facebook::velox::row

velox/row/benchmarks/CMakeLists.txt

@@ -22,3 +22,14 @@ target_link_libraries(
     velox_vector_fuzzer
     Folly::folly
     ${FOLLY_BENCHMARK})
+
+add_executable(velox_row_deserialize_benchmark
+               DynamicRowVectorDeserializeBenchmark.cpp)
+target_link_libraries(
+  velox_row_deserialize_benchmark
+  PRIVATE
+    velox_exec
+    velox_row_fast
+    velox_vector_fuzzer
+    Folly::folly
+    ${FOLLY_BENCHMARK})