enable clear of hashtable

internal/core/src/exec/HashTable.cpp

@@ -202,7 +202,7 @@ void HashTable<nullableKeys>::allocateTables(uint64_t size) {
     // The total size is 8 bytes per slot, in groups of 16 slots with 16 bytes of
     // tags and 16 * 6 bytes of pointers and a padding of 16 bytes to round up the
     // cache line.
-    // TODO must support memory pool here to avoid OOM
+    // TODO support memory pool here to avoid OOM
     table_ = new char*[capacity_];
     memset(table_, 0, capacity_ * sizeof(char*));
 }
@@ -330,7 +330,12 @@ void HashTable<nullableKeys>::setHashMode(HashMode mode, int32_t numNew) {
 
 template <bool nullable>
 void HashTable<nullable>::clear(bool freeTable) {
-
+    if(table_) {
+        delete[] table_;
+        table_ = nullptr;
+    }
+    rows_->clear();
+    numDistinct_ = 0;
 }
 
 template class HashTable<true>;

internal/core/src/exec/HashTable.h

@@ -141,7 +141,6 @@ virtual void clear(bool freeTable = false) = 0;
 protected:
   std::vector<std::unique_ptr<VectorHasher>> hashers_;
   std::unique_ptr<RowContainer> rows_;
-  char** table_ = nullptr;
 };
 
 class ProbeState;
@@ -158,7 +157,7 @@ class HashTable : public BaseHashTable {
             keyTypes.push_back(hasher->ChannelDataType());
         }
         hashMode_ = HashMode::kHash;
-        rows_ = std::make_unique<RowContainer>(keyTypes, accumulators, nullableKeys, hashMode_ != HashMode::kHash);
+        rows_ = std::make_unique<RowContainer>(keyTypes, accumulators, nullableKeys);
     };
 
     void setHashMode(HashMode mode, int32_t numNew) override;
@@ -242,6 +241,8 @@ class HashTable : public BaseHashTable {
     // a power of 2.
     void allocateTables(uint64_t size);
 
+    void extractGroups(char** output_groups, size_t group_count);
+
     template<bool isJoin, bool isNormalizedKey = false>
     void fullProbe(HashLookup& lookup, ProbeState& state, bool extraCheck);
 
@@ -287,6 +288,7 @@ class HashTable : public BaseHashTable {
   int8_t sizeBits_;
 
   int64_t numRehashes_{0};
+  char** table_ = nullptr;
 
   HashMode hashMode() const override {
     return hashMode_;

internal/core/src/exec/operator/AggregationNode.cpp

@@ -36,9 +36,8 @@ RowVectorPtr PhyAggregationNode::GetOutput() {
   auto batch_size = queryConfig->get_expr_batch_size();
   const auto outputRowCount = isGlobal_? 1: batch_size;
   prepareOutput(outputRowCount);
-  const bool hasData = grouping_set_->getOutput(outputRowCount, outputRowCount, resultIterator_, output_);
+  const bool hasData = grouping_set_->getOutput(output_);
   if (!hasData) {
-      resultIterator_.reset();
       if (no_more_input_) {
         finished_ = true;
       }

internal/core/src/exec/operator/AggregationNode.h

@@ -76,8 +76,6 @@ class PhyAggregationNode: public Operator {
     // flush.
     int64_t numOutputRows_ = 0;
     bool finished_ = false;
-
-    RowContainerIterator resultIterator_;
 };
 }
 }

internal/core/src/exec/operator/query-agg/GroupingSet.cpp

@@ -95,37 +95,30 @@ void GroupingSet::addGlobalAggregationInput(const milvus::RowVectorPtr& input) {
     tempVectors_.clear();
 }
 
-bool GroupingSet::getGlobalAggregationOutput(milvus::exec::RowContainerIterator &iterator,
-                                             milvus::RowVectorPtr &result) {
-    if (iterator.allocationIndex != 0) {
-        return false;
-    }
+bool GroupingSet::getGlobalAggregationOutput(milvus::RowVectorPtr &result) {
     initializeGlobalAggregation();
     auto groups = lookup_->hits_.data();
     for(auto i = 0; i < aggregates_.size(); i++) {
         auto& function = aggregates_[i].function_;
         auto resultVector = result->child(aggregates_[i].output_);
         function->extractValues(groups, 1, &resultVector);
     }
-    iterator.allocationIndex = std::numeric_limits<int32_t>::max();
     return true;
 }
 
-bool GroupingSet::getOutput(int32_t maxOutputRows, int32_t maxOutputBytes, milvus::exec::RowContainerIterator &iterator,
-                            milvus::RowVectorPtr &result) {
+bool GroupingSet::getOutput(milvus::RowVectorPtr &result) {
+    if (hash_table_ == nullptr) {
+        return false;
+    }
     if (isGlobal_) {
-        return getGlobalAggregationOutput(iterator, result);
-    }
-    char* groups[maxOutputRows];
-    const int32_t numGroups = hash_table_?hash_table_->rows()
-            ->listRows(&iterator, maxOutputRows, maxOutputBytes, groups):0;
-    if(numGroups == 0) {
-        if (hash_table_ != nullptr) {
-            hash_table_->clear();
-        }
+        return getGlobalAggregationOutput(result);
+    }
+    const auto& all_rows = hash_table_->rows()->allRows();
+    if(all_rows.empty()) {
+        hash_table_->clear();
         return false;
     }
-    extractGroups(folly::Range<char**>(groups, numGroups), result);
+    extractGroups(folly::Range<char**>(const_cast<char**>(all_rows.data()), all_rows.size()), result);
     return true;
 }
 

internal/core/src/exec/operator/query-agg/GroupingSet.h

@@ -58,10 +58,7 @@ class GroupingSet {
     // fit.
     void ensureInputFits(const RowVectorPtr& input);
 
-    bool getOutput(int32_t maxOutputRows,
-                   int32_t maxOutputBytes,
-                   RowContainerIterator& iterator,
-                   RowVectorPtr& result);
+    bool getOutput(RowVectorPtr& result);
 
     bool hasOutput() {
         return noMoreInput_;
@@ -71,7 +68,7 @@ class GroupingSet {
 
     void populateTempVectors(int32_t aggregateIndex, const RowVectorPtr& input);
 
-    bool getGlobalAggregationOutput(RowContainerIterator& iterator, RowVectorPtr& result);
+    bool getGlobalAggregationOutput(RowVectorPtr& result);
 
 private:
     const bool isGlobal_;

internal/core/src/exec/operator/query-agg/RowContainer.cpp

@@ -23,12 +23,10 @@ namespace exec {
 
 RowContainer::RowContainer(const std::vector<DataType> &keyTypes,
                            const std::vector<Accumulator>& accumulators,
-                           bool nullableKeys,
-                           bool hasNormalizedKeys):
+                           bool nullableKeys):
                            keyTypes_(keyTypes),
                            accumulators_(accumulators),
-                           nullableKeys_(nullableKeys),
-                           hasNormalizedKeys_(hasNormalizedKeys){
+                           nullableKeys_(nullableKeys){
     int32_t offset = 0;
     int32_t nullOffset = 0;
     bool isVariableWidth = false;
@@ -109,17 +107,13 @@ RowContainer::RowContainer(const std::vector<DataType> &keyTypes,
 }
 
 char* RowContainer::newRow() {
-    ++numRows_;
-    char* row;
-    if (firstFreeRow_) {
-        row = firstFreeRow_;
-        AssertInfo(milvus::bits::isBitSet(row, freeFlagOffset_), "freeRow must be freed before inserted into the linked list");
-        firstFreeRow_ = nextFree(row);
-        --numFreeRows_;
-    } else {
-        row = new char[fixedRowSize_ + alignment_];
+    char* row = new char[fixedRowSize_ + alignment_];
+    if (rows_.size() < numRows_ + 1) {
+        rows_.resize(numRows_ + 1024);
     }
-    return nullptr;
+    rows_[numRows_] = row;
+    ++numRows_;
+    return row;
 }
 
 void RowContainer::store(const milvus::ColumnVectorPtr &column_data, milvus::vector_size_t index, char *row,

internal/core/src/exec/operator/query-agg/RowContainer.h

@@ -107,28 +107,11 @@ class RowColumn {
    const uint64_t packedOffsets_;
 };
 
-using normalized_key_t = uint64_t;
-
-struct RowContainerIterator {
-    int32_t allocationIndex = 0;
-    int32_t rowOffset = 0;
-
-    char* rowBegin_{nullptr};
-    inline char* currentRow() const {
-        return rowBegin_;
-    }
-
-    void reset() {
-        *this = {};
-    }
-};
-
 class RowContainer {
 public:
     RowContainer(const std::vector<DataType>& keyTypes,
                  const std::vector<Accumulator>& accumulators,
-                 bool nullableKeys,
-                 bool hasNormalizedKeys);
+                 bool nullableKeys);
 
     // The number of flags (bits) per accumulator, one for null and one for
     // initialized.
@@ -155,13 +138,6 @@ class RowContainer {
         return rowSizeOffset_;
     }
 
-    int32_t listRows(RowContainerIterator* iter,
-                     int32_t maxRows,
-                     uint64_t maxBytes,
-                     char** rows) {
-        return 0;
-    }
-
     static inline bool isNullAt(const char* row, int32_t nullByte, uint8_t nullMask) {
         return (row[nullByte] & nullMask) != 0;
     }
@@ -291,22 +267,23 @@ class RowContainer {
             const VectorPtr& result){
         auto maxRows = numRows + resultOffset;
         AssertInfo(maxRows == result->size(), "extracted rows number should be equal to the size of result vector");
-        if constexpr (std::is_same_v<T, std::string> || std::is_same_v<T, std::string_view>) {
-            PanicInfo(DataTypeInvalid, "Not support extract string values for now");
-        } else {
-            auto result_column_vec = std::dynamic_pointer_cast<ColumnVector>(result);
-            AssertInfo(result_column_vec != nullptr, "Input column to extract result must be of ColumnVector type");
-            for (auto i = 0; i < numRows; i++) {
-                const char *row;
-                if constexpr (useRowNumber) {
-                    auto rowNumber = rowNumbers[i];
-                    row = rowNumber >= 0 ? rows[rowNumber] : nullptr;
-                } else {
-                    row = rows[i];
-                }
-                auto resultIndex = resultOffset + i;
-                if (row == nullptr || isNullAt(row, nullByte, nullMask)) {
-                    result_column_vec->nullAt(resultIndex);
+        auto result_column_vec = std::dynamic_pointer_cast<ColumnVector>(result);
+        AssertInfo(result_column_vec != nullptr, "Input column to extract result must be of ColumnVector type");
+        for (auto i = 0; i < numRows; i++) {
+            const char *row;
+            if constexpr (useRowNumber) {
+                auto rowNumber = rowNumbers[i];
+                row = rowNumber >= 0 ? rows[rowNumber] : nullptr;
+            } else {
+                row = rows[i];
+            }
+            auto resultIndex = resultOffset + i;
+            if (row == nullptr || isNullAt(row, nullByte, nullMask)) {
+                result_column_vec->nullAt(resultIndex);
+            } else {
+                if constexpr (std::is_same_v<T, std::string> || std::is_same_v<T, std::string_view>) {
+                    auto* str_ptr = valueAt<T*>(row, offset);
+                    result_column_vec->SetValueAt<T>(resultIndex, *str_ptr);
                 } else {
                     result_column_vec->SetValueAt<T>(resultIndex, valueAt<T>(row, offset));
                 }
@@ -360,7 +337,6 @@ class RowContainer {
             PanicInfo(DataTypeInvalid, "Not Support Extract types:[ROW/JSON/ARRAY/NONE]");
         } else {
             using T = typename milvus::TypeTraits<Type>::NativeType;
-
             auto nullMask = column.nullMask();
             auto offset = column.offset();
             if (nullMask) {
@@ -403,6 +379,10 @@ class RowContainer {
         extractColumn(rows, numRows, columnAt(column_idx), 0, result);
     }
 
+    const std::vector<char*>& allRows() const {
+        return rows_;
+    }
+
     static inline int32_t nullByte(int32_t nullOffset) {
         return nullOffset / 8;
     }
@@ -426,15 +406,20 @@ class RowContainer {
         return nullMask(accumulatorFlagsOffset) << 1;
     }
 
-
+    void clear() {
+        for (auto row: rows_) {
+            delete row;
+        }
+        numRows_ = 0;
+        numFreeRows_ = 0;
+    }
 
 private:
      // Offset of the pointer to the next free row on a free row.
     static constexpr uint32_t kNextFreeOffset_ = 0;
 
     const std::vector<DataType> keyTypes_;
     const bool nullableKeys_;
-    const bool hasNormalizedKeys_;
     std::vector<uint32_t> offsets_;
     std::vector<uint32_t> nullOffsets_;
 
@@ -456,10 +441,9 @@ class RowContainer {
 
     std::vector<Accumulator> accumulators_;
 
-    // Head of linked list of free rows.
-    char* firstFreeRow_ = nullptr;
     uint64_t numRows_ = 0;
     uint64_t numFreeRows_ = 0;
+    std::vector<char*> rows_{};
 };
 
 inline void RowContainer::extractColumn(const char *const *rows, int32_t num_rows, milvus::exec::RowColumn column,