Folder: vector

main changes:

1. Make varchar and varbinary compatible.
2. Add ValueStreamNode operator

velox/vector/BaseVector.cpp

@@ -489,7 +489,7 @@ std::string BaseVector::toString(
 }
 
 void BaseVector::ensureWritable(const SelectivityVector& rows) {
-  auto newSize = std::max<vector_size_t>(rows.end(), length_);
+  auto newSize = std::max<vector_size_t>(rows.size(), length_);
   if (nulls_ && !(nulls_->unique() && nulls_->isMutable())) {
     BufferPtr newNulls = AlignedBuffer::allocate<bool>(newSize, pool_);
     auto rawNewNulls = newNulls->asMutable<uint64_t>();
@@ -511,9 +511,9 @@ void BaseVector::ensureWritable(
     VectorPool* vectorPool) {
   if (!result) {
     if (vectorPool) {
-      result = vectorPool->get(type, rows.end());
+      result = vectorPool->get(type, rows.size());
     } else {
-      result = BaseVector::create(type, rows.end(), pool);
+      result = BaseVector::create(type, rows.size(), pool);
     }
     return;
   }
@@ -542,7 +542,7 @@ void BaseVector::ensureWritable(
 
   // The copy-on-write size is the max of the writable row set and the
   // vector.
-  auto targetSize = std::max<vector_size_t>(rows.end(), result->size());
+  auto targetSize = std::max<vector_size_t>(rows.size(), result->size());
 
   VectorPtr copy;
   if (vectorPool) {
@@ -605,7 +605,7 @@ VectorPtr BaseVector::createConstant(
     variant value,
     vector_size_t size,
     velox::memory::MemoryPool* pool) {
-  VELOX_CHECK_EQ(type->kind(), value.kind());
+  VELOX_CHECK(compatibleKind(type->kind(), value.kind()));
   return VELOX_DYNAMIC_SCALAR_TYPE_DISPATCH_ALL(
       newConstant, value.kind(), type, value, size, pool);
 }

velox/vector/BaseVector.h

@@ -673,7 +673,11 @@ class BaseVector {
   // two unknowns but values cannot be assigned into an unknown 'left'
   // from a not-unknown 'right'.
   static bool compatibleKind(TypeKind left, TypeKind right) {
-    return left == right || right == TypeKind::UNKNOWN;
+    // Vectors of VARCHAR and VARBINARY are compatible with each other.
+    bool varcharAndBinary =
+        (left == TypeKind::VARCHAR && right == TypeKind::VARBINARY) ||
+        (left == TypeKind::VARBINARY && right == TypeKind::VARCHAR);
+    return left == right || right == TypeKind::UNKNOWN || varcharAndBinary;
   }
 
   /// Returns a brief summary of the vector. If 'recursive' is true, includes a

velox/vector/ComplexVector.cpp

@@ -209,7 +209,8 @@ void RowVector::copy(
     BufferPtr mappedIndices;
     vector_size_t* rawMappedIndices = nullptr;
     if (toSourceRow) {
-      mappedIndices = AlignedBuffer::allocate<vector_size_t>(rows.end(), pool_);
+      mappedIndices =
+          AlignedBuffer::allocate<vector_size_t>(rows.size(), pool_);
       rawMappedIndices = mappedIndices->asMutable<vector_size_t>();
       nonNullRows.applyToSelected(
           [&](auto row) { rawMappedIndices[row] = indices[toSourceRow[row]]; });
@@ -687,7 +688,7 @@ std::string ArrayVector::toString(vector_size_t index) const {
 }
 
 void ArrayVector::ensureWritable(const SelectivityVector& rows) {
-  auto newSize = std::max<vector_size_t>(rows.end(), BaseVector::length_);
+  auto newSize = std::max<vector_size_t>(rows.size(), BaseVector::length_);
   if (offsets_ && !offsets_->unique()) {
     BufferPtr newOffsets =
         AlignedBuffer::allocate<vector_size_t>(newSize, BaseVector::pool_);
@@ -949,7 +950,7 @@ std::string MapVector::toString(vector_size_t index) const {
 }
 
 void MapVector::ensureWritable(const SelectivityVector& rows) {
-  auto newSize = std::max<vector_size_t>(rows.end(), BaseVector::length_);
+  auto newSize = std::max<vector_size_t>(rows.size(), BaseVector::length_);
   if (offsets_ && !offsets_->unique()) {
     BufferPtr newOffsets =
         AlignedBuffer::allocate<vector_size_t>(newSize, BaseVector::pool_);

velox/vector/ComplexVectorStream.h

@@ -0,0 +1,34 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include "velox/vector/ComplexVector.h"
+
+namespace facebook::velox {
+
+class RowVectorStream {
+ public:
+  RowVectorStream() {}
+
+  virtual ~RowVectorStream() = default;
+
+  virtual bool hasNext() = 0;
+
+  virtual RowVectorPtr next() = 0;
+};
+
+} // namespace facebook::velox

velox/vector/FlatVector-inl.h

@@ -349,7 +349,7 @@ void FlatVector<T>::resize(vector_size_t newSize, bool setNotNull) {
 
 template <typename T>
 void FlatVector<T>::ensureWritable(const SelectivityVector& rows) {
-  auto newSize = std::max<vector_size_t>(rows.end(), BaseVector::length_);
+  auto newSize = std::max<vector_size_t>(rows.size(), BaseVector::length_);
   if (values_ && !(values_->unique() && values_->isMutable())) {
     BufferPtr newValues;
     if constexpr (std::is_same_v<T, StringView>) {

velox/vector/arrow/Bridge.cpp

@@ -954,6 +954,36 @@ VectorPtr importFromArrowImpl(
     memory::MemoryPool* pool,
     bool isViewer);
 
+VectorPtr createDecimalVector(
+    memory::MemoryPool* pool,
+    const TypePtr& type,
+    BufferPtr nulls,
+    const ArrowSchema& arrowSchema,
+    const ArrowArray& arrowArray,
+    WrapInBufferViewFunc wrapInBufferView) {
+  auto valueBuf = wrapInBufferView(
+      arrowArray.buffers[1], arrowArray.length * sizeof(int128_t));
+
+  auto dst = valueBuf->as<uint8_t>();
+
+  VectorPtr base = BaseVector::create(type, arrowArray.length, pool);
+  base->setNulls(nulls);
+
+  auto flatVector =
+      std::dynamic_pointer_cast<FlatVector<UnscaledShortDecimal>>(base);
+
+  for (int i = 0; i < arrowArray.length; i++) {
+    int128_t result;
+    memcpy(&result, dst + i * sizeof(int128_t), sizeof(int128_t));
+    int64_t value = static_cast<int64_t>(result);
+    if (!base->isNullAt(i)) {
+      flatVector->set(i, UnscaledShortDecimal(static_cast<int64_t>(result)));
+    }
+  }
+
+  return flatVector;
+}
+
 RowVectorPtr createRowVector(
     memory::MemoryPool* pool,
     const RowTypePtr& rowType,
@@ -1154,6 +1184,11 @@ VectorPtr importFromArrowImpl(
     return createMapVector(
         pool, type, nulls, arrowSchema, arrowArray, isViewer, wrapInBufferView);
   }
+  if (type->isShortDecimal()) {
+    return createDecimalVector(
+        pool, type, nulls, arrowSchema, arrowArray, wrapInBufferView);
+  }
+
   // Other primitive types.
   VELOX_CHECK(
       type->isPrimitiveType(),
@@ -1166,7 +1201,7 @@ VectorPtr importFromArrowImpl(
   auto values = wrapInBufferView(
       arrowArray.buffers[1], arrowArray.length * type->cppSizeInBytes());
 
-  return VELOX_DYNAMIC_SCALAR_TYPE_DISPATCH(
+  return VELOX_DYNAMIC_SCALAR_TYPE_DISPATCH_ALL(
       createFlatVector,
       type->kind(),
       pool,

velox/vector/tests/EnsureWritableVectorTest.cpp

@@ -421,17 +421,18 @@ TEST_F(EnsureWritableVectorTest, constant) {
   }
 
   // If constant has smaller size, check that we follow the selectivity vector
-  // max seleced row size.
+  // size.
   {
     const vector_size_t selectivityVectorSize = 100;
     auto constant = BaseVector::createConstant(
         BIGINT(), variant::create<TypeKind::BIGINT>(123), 1, pool_.get());
-    SelectivityVector rows(selectivityVectorSize);
-    rows.setValid(99, false);
-    rows.updateBounds();
-    BaseVector::ensureWritable(rows, BIGINT(), pool_.get(), constant);
+    BaseVector::ensureWritable(
+        SelectivityVector::empty(selectivityVectorSize),
+        BIGINT(),
+        pool_.get(),
+        constant);
     EXPECT_EQ(VectorEncoding::Simple::FLAT, constant->encoding());
-    EXPECT_EQ(99, constant->size());
+    EXPECT_EQ(selectivityVectorSize, constant->size());
   }
 
   // If constant has larger size, check that we follow the constant vector