quickstep-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jianq...@apache.org
Subject [2/2] incubator-quickstep git commit: Initial update
Date Mon, 31 Oct 2016 20:13:59 GMT
Initial update


Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/c1baa015
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/c1baa015
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/c1baa015

Branch: refs/heads/collision-free-agg
Commit: c1baa0150bd949b3bff3af39c262106836c4a168
Parents: bc3d8a5
Author: Jianqiao Zhu <jianqiao@cs.wisc.edu>
Authored: Thu Oct 27 12:23:03 2016 -0500
Committer: Jianqiao Zhu <jianqiao@cs.wisc.edu>
Committed: Mon Oct 31 15:12:53 2016 -0500

----------------------------------------------------------------------
 .../aggregation/AggregationConcreteHandle.cpp   |  22 +-
 expressions/aggregation/AggregationHandle.hpp   |  32 +-
 .../aggregation/AggregationHandleAvg.cpp        |  32 +-
 .../aggregation/AggregationHandleAvg.hpp        |  11 +-
 .../aggregation/AggregationHandleCount.cpp      |  57 +--
 .../aggregation/AggregationHandleCount.hpp      |  11 +-
 .../aggregation/AggregationHandleDistinct.hpp   |  16 +-
 .../aggregation/AggregationHandleMax.cpp        |  30 +-
 .../aggregation/AggregationHandleMax.hpp        |  11 +-
 .../aggregation/AggregationHandleMin.cpp        |  30 +-
 .../aggregation/AggregationHandleMin.hpp        |  11 +-
 .../aggregation/AggregationHandleSum.cpp        |  31 +-
 .../aggregation/AggregationHandleSum.hpp        |  12 +-
 query_execution/QueryContext.hpp                |  14 -
 .../DestroyAggregationStateOperator.cpp         |   7 -
 storage/AggregationOperationState.cpp           | 395 +++++++++----------
 storage/AggregationOperationState.hpp           |  68 ++--
 storage/FastHashTable.hpp                       | 364 ++++++-----------
 storage/HashTableBase.hpp                       |   7 +-
 storage/StorageBlock.cpp                        | 198 ----------
 storage/StorageBlock.hpp                        | 115 ------
 21 files changed, 460 insertions(+), 1014 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c1baa015/expressions/aggregation/AggregationConcreteHandle.cpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationConcreteHandle.cpp b/expressions/aggregation/AggregationConcreteHandle.cpp
index e3fb520..fd23c7e 100644
--- a/expressions/aggregation/AggregationConcreteHandle.cpp
+++ b/expressions/aggregation/AggregationConcreteHandle.cpp
@@ -52,17 +52,17 @@ void AggregationConcreteHandle::insertValueAccessorIntoDistinctifyHashTable(
     AggregationStateHashTableBase *distinctify_hash_table) const {
   // If the key-value pair is already there, we don't need to update the value,
   // which should always be "true". I.e. the value is just a placeholder.
-
-  AggregationStateFastHashTable *hash_table =
-      static_cast<AggregationStateFastHashTable *>(distinctify_hash_table);
-  if (key_ids.size() == 1) {
-    hash_table->upsertValueAccessorFast(
-        key_ids, accessor, key_ids[0], true /* check_for_null_keys */);
-  } else {
-    std::vector<attribute_id> empty_args {kInvalidAttributeID};
-    hash_table->upsertValueAccessorCompositeKeyFast(
-        empty_args, accessor, key_ids, true /* check_for_null_keys */);
-  }
+  LOG(FATAL) << "Not supported";
+//  AggregationStateFastHashTable *hash_table =
+//      static_cast<AggregationStateFastHashTable *>(distinctify_hash_table);
+//  if (key_ids.size() == 1) {
+//    hash_table->upsertValueAccessorFast(
+//        key_ids, accessor, key_ids[0], true /* check_for_null_keys */);
+//  } else {
+//    std::vector<attribute_id> empty_args {kInvalidAttributeID};
+//    hash_table->upsertValueAccessorCompositeKeyFast(
+//        empty_args, accessor, key_ids, true /* check_for_null_keys */);
+//  }
 }
 
 }  // namespace quickstep

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c1baa015/expressions/aggregation/AggregationHandle.hpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandle.hpp b/expressions/aggregation/AggregationHandle.hpp
index 4b51179..751b390 100644
--- a/expressions/aggregation/AggregationHandle.hpp
+++ b/expressions/aggregation/AggregationHandle.hpp
@@ -32,6 +32,7 @@
 namespace quickstep {
 
 class ColumnVector;
+class ColumnVectorsValueAccessor;
 class StorageManager;
 class Type;
 class ValueAccessor;
@@ -153,39 +154,16 @@ class AggregationHandle {
       const std::size_t num_tuples) const = 0;
 
   /**
-   * @brief Accumulate (iterate over) all values in one or more ColumnVectors
-   *        and return a new AggregationState which can be merged with other
-   *        states or finalized.
+   * @brief TODO
    *
-   * @param column_vectors One or more ColumnVectors that the aggregate will be
-   *        applied to. These correspond to the aggregate function's arguments,
-   *        in order.
    * @return A new AggregationState which contains the accumulated results from
    *         applying the aggregate to column_vectors. Caller is responsible
    *         for deleting the returned AggregationState.
    **/
-  virtual AggregationState* accumulateColumnVectors(
-      const std::vector<std::unique_ptr<ColumnVector>> &column_vectors) const = 0;
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-  /**
-   * @brief Accumulate (iterate over) all values in columns accessible through
-   *        a ValueAccessor and return a new AggregationState which can be
-   *        merged with other states or finalized.
-   *
-   * @param accessor A ValueAccessor that the columns to be aggregated can be
-   *        accessed through.
-   * @param accessor_ids The attribute_ids that correspond to the columns in
-   *        accessor to aggeregate. These correspond to the aggregate
-   *        function's arguments, in order.
-   * @return A new AggregationState which contains the accumulated results from
-   *         applying the aggregate to the specified columns in accessor.
-   *         Caller is responsible for deleting the returned AggregationState.
-   **/
-  virtual AggregationState* accumulateValueAccessor(
+  virtual AggregationState* accumulate(
       ValueAccessor *accessor,
-      const std::vector<attribute_id> &accessor_ids) const = 0;
-#endif
+      ColumnVectorsValueAccessor *aux_accessor,
+      const std::vector<attribute_id> &argument_ids) const = 0;
 
   /**
    * @brief Perform an aggregation with GROUP BY over all the tuples accessible

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c1baa015/expressions/aggregation/AggregationHandleAvg.cpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleAvg.cpp b/expressions/aggregation/AggregationHandleAvg.cpp
index 2481092..672db1f 100644
--- a/expressions/aggregation/AggregationHandleAvg.cpp
+++ b/expressions/aggregation/AggregationHandleAvg.cpp
@@ -96,34 +96,28 @@ AggregationStateHashTableBase* AggregationHandleAvg::createGroupByHashTable(
       hash_table_impl, group_by_types, estimated_num_groups, storage_manager);
 }
 
-AggregationState* AggregationHandleAvg::accumulateColumnVectors(
-    const std::vector<std::unique_ptr<ColumnVector>> &column_vectors) const {
-  DCHECK_EQ(1u, column_vectors.size())
-      << "Got wrong number of ColumnVectors for AVG: " << column_vectors.size();
+AggregationState* AggregationHandleAvg::accumulate(
+    ValueAccessor *accessor,
+    ColumnVectorsValueAccessor *aux_accessor,
+    const std::vector<attribute_id> &argument_ids) const {
+  DCHECK_EQ(1u, argument_ids.size())
+      << "Got wrong number of attributes for AVG: " << argument_ids.size();
 
-  AggregationStateAvg *state = new AggregationStateAvg(blank_state_);
-  std::size_t count = 0;
-  state->sum_ = fast_add_operator_->accumulateColumnVector(
-      state->sum_, *column_vectors.front(), &count);
-  state->count_ = count;
-  return state;
-}
+  const attribute_id argument_id = argument_ids.front();
+  DCHECK_NE(argument_id, kInvalidAttributeID);
 
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-AggregationState* AggregationHandleAvg::accumulateValueAccessor(
-    ValueAccessor *accessor,
-    const std::vector<attribute_id> &accessor_ids) const {
-  DCHECK_EQ(1u, accessor_ids.size())
-      << "Got wrong number of attributes for AVG: " << accessor_ids.size();
+  ValueAccessor *target_accessor =
+      argument_id >= 0 ? accessor : aux_accessor;
+  const attribute_id target_argument_id =
+      argument_id >= 0 ? argument_id : -(argument_id+2);
 
   AggregationStateAvg *state = new AggregationStateAvg(blank_state_);
   std::size_t count = 0;
   state->sum_ = fast_add_operator_->accumulateValueAccessor(
-      state->sum_, accessor, accessor_ids.front(), &count);
+      state->sum_, target_accessor, target_argument_id, &count);
   state->count_ = count;
   return state;
 }
-#endif
 
 void AggregationHandleAvg::aggregateValueAccessorIntoHashTable(
     ValueAccessor *accessor,

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c1baa015/expressions/aggregation/AggregationHandleAvg.hpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleAvg.hpp b/expressions/aggregation/AggregationHandleAvg.hpp
index 47132c6..46983be 100644
--- a/expressions/aggregation/AggregationHandleAvg.hpp
+++ b/expressions/aggregation/AggregationHandleAvg.hpp
@@ -169,15 +169,10 @@ class AggregationHandleAvg : public AggregationConcreteHandle {
     }
   }
 
-  AggregationState* accumulateColumnVectors(
-      const std::vector<std::unique_ptr<ColumnVector>> &column_vectors)
-      const override;
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-  AggregationState* accumulateValueAccessor(
+  AggregationState* accumulate(
       ValueAccessor *accessor,
-      const std::vector<attribute_id> &accessor_id) const override;
-#endif
+      ColumnVectorsValueAccessor *aux_accessor,
+      const std::vector<attribute_id> &argument_ids) const override;
 
   void aggregateValueAccessorIntoHashTable(
       ValueAccessor *accessor,

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c1baa015/expressions/aggregation/AggregationHandleCount.cpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleCount.cpp b/expressions/aggregation/AggregationHandleCount.cpp
index 034c942..9c00449 100644
--- a/expressions/aggregation/AggregationHandleCount.cpp
+++ b/expressions/aggregation/AggregationHandleCount.cpp
@@ -63,58 +63,32 @@ AggregationHandleCount<count_star, nullable_type>::createGroupByHashTable(
 
 template <bool count_star, bool nullable_type>
 AggregationState*
-AggregationHandleCount<count_star, nullable_type>::accumulateColumnVectors(
-    const std::vector<std::unique_ptr<ColumnVector>> &column_vectors) const {
+AggregationHandleCount<count_star, nullable_type>::accumulate(
+    ValueAccessor *accessor,
+    ColumnVectorsValueAccessor *aux_accessor,
+    const std::vector<attribute_id> &argument_ids) const {
   DCHECK(!count_star)
       << "Called non-nullary accumulation method on an AggregationHandleCount "
       << "set up for nullary COUNT(*)";
 
-  DCHECK_EQ(1u, column_vectors.size())
-      << "Got wrong number of ColumnVectors for COUNT: "
-      << column_vectors.size();
+  DCHECK_EQ(1u, argument_ids.size())
+      << "Got wrong number of attributes for COUNT: " << argument_ids.size();
 
-  std::size_t count = 0;
-  InvokeOnColumnVector(
-      *column_vectors.front(),
-      [&](const auto &column_vector) -> void {  // NOLINT(build/c++11)
-        if (nullable_type) {
-          // TODO(shoban): Iterating over the ColumnVector is a rather slow way
-          // to do this. We should look at extending the ColumnVector interface
-          // to do a quick count of the non-null values (i.e. the length minus
-          // the population count of the null bitmap). We should do something
-          // similar for ValueAccessor too.
-          for (std::size_t pos = 0; pos < column_vector.size(); ++pos) {
-            count += !column_vector.getTypedValue(pos).isNull();
-          }
-        } else {
-          count = column_vector.size();
-        }
-      });
+  const attribute_id argument_id = argument_ids.front();
+  DCHECK_NE(argument_id, kInvalidAttributeID);
 
-  return new AggregationStateCount(count);
-}
+  ValueAccessor *target_accessor =
+      argument_id >= 0 ? accessor : aux_accessor;
+  const attribute_id target_argument_id =
+      argument_id >= 0 ? argument_id : -(argument_id+2);
 
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-template <bool count_star, bool nullable_type>
-AggregationState*
-AggregationHandleCount<count_star, nullable_type>::accumulateValueAccessor(
-    ValueAccessor *accessor,
-    const std::vector<attribute_id> &accessor_ids) const {
-  DCHECK(!count_star)
-      << "Called non-nullary accumulation method on an AggregationHandleCount "
-      << "set up for nullary COUNT(*)";
-
-  DCHECK_EQ(1u, accessor_ids.size())
-      << "Got wrong number of attributes for COUNT: " << accessor_ids.size();
-
-  const attribute_id accessor_id = accessor_ids.front();
   std::size_t count = 0;
   InvokeOnValueAccessorMaybeTupleIdSequenceAdapter(
-      accessor,
-      [&accessor_id, &count](auto *accessor) -> void {  // NOLINT(build/c++11)
+      target_accessor,
+      [&target_argument_id, &count](auto *accessor) -> void {  // NOLINT(build/c++11)
         if (nullable_type) {
           while (accessor->next()) {
-            count += !accessor->getTypedValue(accessor_id).isNull();
+            count += !accessor->getTypedValue(target_argument_id).isNull();
           }
         } else {
           count = accessor->getNumTuples();
@@ -123,7 +97,6 @@ AggregationHandleCount<count_star, nullable_type>::accumulateValueAccessor(
 
   return new AggregationStateCount(count);
 }
-#endif
 
 template <bool count_star, bool nullable_type>
 void AggregationHandleCount<count_star, nullable_type>::

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c1baa015/expressions/aggregation/AggregationHandleCount.hpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleCount.hpp b/expressions/aggregation/AggregationHandleCount.hpp
index 6aab0cd..52e63f5 100644
--- a/expressions/aggregation/AggregationHandleCount.hpp
+++ b/expressions/aggregation/AggregationHandleCount.hpp
@@ -162,15 +162,10 @@ class AggregationHandleCount : public AggregationConcreteHandle {
     return new AggregationStateCount(num_tuples);
   }
 
-  AggregationState* accumulateColumnVectors(
-      const std::vector<std::unique_ptr<ColumnVector>> &column_vectors)
-      const override;
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-  AggregationState* accumulateValueAccessor(
+  AggregationState* accumulate(
       ValueAccessor *accessor,
-      const std::vector<attribute_id> &accessor_id) const override;
-#endif
+      ColumnVectorsValueAccessor *aux_accessor,
+      const std::vector<attribute_id> &argument_ids) const override;
 
   void aggregateValueAccessorIntoHashTable(
       ValueAccessor *accessor,

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c1baa015/expressions/aggregation/AggregationHandleDistinct.hpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleDistinct.hpp b/expressions/aggregation/AggregationHandleDistinct.hpp
index 838bfdd..5d45219 100644
--- a/expressions/aggregation/AggregationHandleDistinct.hpp
+++ b/expressions/aggregation/AggregationHandleDistinct.hpp
@@ -62,21 +62,13 @@ class AggregationHandleDistinct : public AggregationConcreteHandle {
         << "AggregationHandleDistinct does not support accumulateNullary().";
   }
 
-  AggregationState* accumulateColumnVectors(
-      const std::vector<std::unique_ptr<ColumnVector>> &column_vectors)
-      const override {
-    LOG(FATAL) << "AggregationHandleDistinct does not support "
-                  "accumulateColumnVectors().";
-  }
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-  AggregationState* accumulateValueAccessor(
+  AggregationState* accumulate(
       ValueAccessor *accessor,
-      const std::vector<attribute_id> &accessor_ids) const override {
+      ColumnVectorsValueAccessor *aux_accessor,
+      const std::vector<attribute_id> &argument_ids) const override {
     LOG(FATAL) << "AggregationHandleDistinct does not support "
-                  "accumulateValueAccessor().";
+                  "accumulate().";
   }
-#endif
 
   void mergeStates(const AggregationState &source,
                    AggregationState *destination) const override {

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c1baa015/expressions/aggregation/AggregationHandleMax.cpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleMax.cpp b/expressions/aggregation/AggregationHandleMax.cpp
index c2d571b..aa12ed2 100644
--- a/expressions/aggregation/AggregationHandleMax.cpp
+++ b/expressions/aggregation/AggregationHandleMax.cpp
@@ -54,28 +54,26 @@ AggregationStateHashTableBase* AggregationHandleMax::createGroupByHashTable(
       hash_table_impl, group_by_types, estimated_num_groups, storage_manager);
 }
 
-AggregationState* AggregationHandleMax::accumulateColumnVectors(
-    const std::vector<std::unique_ptr<ColumnVector>> &column_vectors) const {
-  DCHECK_EQ(1u, column_vectors.size())
-      << "Got wrong number of ColumnVectors for MAX: " << column_vectors.size();
+AggregationState* AggregationHandleMax::accumulate(
+    ValueAccessor *accessor,
+    ColumnVectorsValueAccessor *aux_accessor,
+    const std::vector<attribute_id> &argument_ids) const {
+  DCHECK_EQ(1u, argument_ids.size())
+      << "Got wrong number of attributes for MAX: " << argument_ids.size();
 
-  return new AggregationStateMax(fast_comparator_->accumulateColumnVector(
-      type_.getNullableVersion().makeNullValue(), *column_vectors.front()));
-}
+  const attribute_id argument_id = argument_ids.front();
+  DCHECK_NE(argument_id, kInvalidAttributeID);
 
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-AggregationState* AggregationHandleMax::accumulateValueAccessor(
-    ValueAccessor *accessor,
-    const std::vector<attribute_id> &accessor_ids) const {
-  DCHECK_EQ(1u, accessor_ids.size())
-      << "Got wrong number of attributes for MAX: " << accessor_ids.size();
+  ValueAccessor *target_accessor =
+      argument_id >= 0 ? accessor : aux_accessor;
+  const attribute_id target_argument_id =
+      argument_id >= 0 ? argument_id : -(argument_id+2);
 
   return new AggregationStateMax(fast_comparator_->accumulateValueAccessor(
       type_.getNullableVersion().makeNullValue(),
-      accessor,
-      accessor_ids.front()));
+      target_accessor,
+      target_argument_id));
 }
-#endif  // QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
 
 void AggregationHandleMax::aggregateValueAccessorIntoHashTable(
     ValueAccessor *accessor,

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c1baa015/expressions/aggregation/AggregationHandleMax.hpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleMax.hpp b/expressions/aggregation/AggregationHandleMax.hpp
index d851a0c..ab7da0a 100644
--- a/expressions/aggregation/AggregationHandleMax.hpp
+++ b/expressions/aggregation/AggregationHandleMax.hpp
@@ -136,15 +136,10 @@ class AggregationHandleMax : public AggregationConcreteHandle {
     }
   }
 
-  AggregationState* accumulateColumnVectors(
-      const std::vector<std::unique_ptr<ColumnVector>> &column_vectors)
-      const override;
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-  AggregationState* accumulateValueAccessor(
+  AggregationState* accumulate(
       ValueAccessor *accessor,
-      const std::vector<attribute_id> &accessor_ids) const override;
-#endif
+      ColumnVectorsValueAccessor *aux_accessor,
+      const std::vector<attribute_id> &argument_ids) const override;
 
   void aggregateValueAccessorIntoHashTable(
       ValueAccessor *accessor,

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c1baa015/expressions/aggregation/AggregationHandleMin.cpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleMin.cpp b/expressions/aggregation/AggregationHandleMin.cpp
index a07f299..12701f2 100644
--- a/expressions/aggregation/AggregationHandleMin.cpp
+++ b/expressions/aggregation/AggregationHandleMin.cpp
@@ -54,28 +54,26 @@ AggregationStateHashTableBase* AggregationHandleMin::createGroupByHashTable(
       hash_table_impl, group_by_types, estimated_num_groups, storage_manager);
 }
 
-AggregationState* AggregationHandleMin::accumulateColumnVectors(
-    const std::vector<std::unique_ptr<ColumnVector>> &column_vectors) const {
-  DCHECK_EQ(1u, column_vectors.size())
-      << "Got wrong number of ColumnVectors for MIN: " << column_vectors.size();
+AggregationState* AggregationHandleMin::accumulate(
+    ValueAccessor *accessor,
+    ColumnVectorsValueAccessor *aux_accessor,
+    const std::vector<attribute_id> &argument_ids) const {
+  DCHECK_EQ(1u, argument_ids.size())
+      << "Got wrong number of attributes for MIN: " << argument_ids.size();
 
-  return new AggregationStateMin(fast_comparator_->accumulateColumnVector(
-      type_.getNullableVersion().makeNullValue(), *column_vectors.front()));
-}
+  const attribute_id argument_id = argument_ids.front();
+  DCHECK_NE(argument_id, kInvalidAttributeID);
 
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-AggregationState* AggregationHandleMin::accumulateValueAccessor(
-    ValueAccessor *accessor,
-    const std::vector<attribute_id> &accessor_ids) const {
-  DCHECK_EQ(1u, accessor_ids.size())
-      << "Got wrong number of attributes for MIN: " << accessor_ids.size();
+  ValueAccessor *target_accessor =
+      argument_id >= 0 ? accessor : aux_accessor;
+  const attribute_id target_argument_id =
+      argument_id >= 0 ? argument_id : -(argument_id+2);
 
   return new AggregationStateMin(fast_comparator_->accumulateValueAccessor(
       type_.getNullableVersion().makeNullValue(),
-      accessor,
-      accessor_ids.front()));
+      target_accessor,
+      target_argument_id));
 }
-#endif  // QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
 
 void AggregationHandleMin::aggregateValueAccessorIntoHashTable(
     ValueAccessor *accessor,

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c1baa015/expressions/aggregation/AggregationHandleMin.hpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleMin.hpp b/expressions/aggregation/AggregationHandleMin.hpp
index e3472ec..dcadb80 100644
--- a/expressions/aggregation/AggregationHandleMin.hpp
+++ b/expressions/aggregation/AggregationHandleMin.hpp
@@ -138,15 +138,10 @@ class AggregationHandleMin : public AggregationConcreteHandle {
     }
   }
 
-  AggregationState* accumulateColumnVectors(
-      const std::vector<std::unique_ptr<ColumnVector>> &column_vectors)
-      const override;
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-  AggregationState* accumulateValueAccessor(
+  AggregationState* accumulate(
       ValueAccessor *accessor,
-      const std::vector<attribute_id> &accessor_ids) const override;
-#endif
+      ColumnVectorsValueAccessor *aux_accessor,
+      const std::vector<attribute_id> &argument_ids) const override;
 
   void aggregateValueAccessorIntoHashTable(
       ValueAccessor *accessor,

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c1baa015/expressions/aggregation/AggregationHandleSum.cpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleSum.cpp b/expressions/aggregation/AggregationHandleSum.cpp
index 642d88d..f00e00c 100644
--- a/expressions/aggregation/AggregationHandleSum.cpp
+++ b/expressions/aggregation/AggregationHandleSum.cpp
@@ -88,29 +88,26 @@ AggregationStateHashTableBase* AggregationHandleSum::createGroupByHashTable(
       hash_table_impl, group_by_types, estimated_num_groups, storage_manager);
 }
 
-AggregationState* AggregationHandleSum::accumulateColumnVectors(
-    const std::vector<std::unique_ptr<ColumnVector>> &column_vectors) const {
-  DCHECK_EQ(1u, column_vectors.size())
-      << "Got wrong number of ColumnVectors for SUM: " << column_vectors.size();
-  std::size_t num_tuples = 0;
-  TypedValue cv_sum = fast_operator_->accumulateColumnVector(
-      blank_state_.sum_, *column_vectors.front(), &num_tuples);
-  return new AggregationStateSum(std::move(cv_sum), num_tuples == 0);
-}
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-AggregationState* AggregationHandleSum::accumulateValueAccessor(
+AggregationState* AggregationHandleSum::accumulate(
     ValueAccessor *accessor,
-    const std::vector<attribute_id> &accessor_ids) const {
-  DCHECK_EQ(1u, accessor_ids.size())
-      << "Got wrong number of attributes for SUM: " << accessor_ids.size();
+    ColumnVectorsValueAccessor *aux_accessor,
+    const std::vector<attribute_id> &argument_ids) const {
+  DCHECK_EQ(1u, argument_ids.size())
+      << "Got wrong number of attributes for SUM: " << argument_ids.size();
+
+  const attribute_id argument_id = argument_ids.front();
+  DCHECK_NE(argument_id, kInvalidAttributeID);
+
+  ValueAccessor *target_accessor =
+      argument_id >= 0 ? accessor : aux_accessor;
+  const attribute_id target_argument_id =
+      argument_id >= 0 ? argument_id : -(argument_id+2);
 
   std::size_t num_tuples = 0;
   TypedValue va_sum = fast_operator_->accumulateValueAccessor(
-      blank_state_.sum_, accessor, accessor_ids.front(), &num_tuples);
+        blank_state_.sum_, target_accessor, target_argument_id, &num_tuples);
   return new AggregationStateSum(std::move(va_sum), num_tuples == 0);
 }
-#endif
 
 void AggregationHandleSum::aggregateValueAccessorIntoHashTable(
     ValueAccessor *accessor,

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c1baa015/expressions/aggregation/AggregationHandleSum.hpp
----------------------------------------------------------------------
diff --git a/expressions/aggregation/AggregationHandleSum.hpp b/expressions/aggregation/AggregationHandleSum.hpp
index f0d23e1..cc24335 100644
--- a/expressions/aggregation/AggregationHandleSum.hpp
+++ b/expressions/aggregation/AggregationHandleSum.hpp
@@ -41,6 +41,7 @@
 namespace quickstep {
 
 class ColumnVector;
+class ColumnVectorsValueAccessor;
 class StorageManager;
 class ValueAccessor;
 
@@ -161,15 +162,10 @@ class AggregationHandleSum : public AggregationConcreteHandle {
     }
   }
 
-  AggregationState* accumulateColumnVectors(
-      const std::vector<std::unique_ptr<ColumnVector>> &column_vectors)
-      const override;
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-  AggregationState* accumulateValueAccessor(
+  AggregationState* accumulate(
       ValueAccessor *accessor,
-      const std::vector<attribute_id> &accessor_id) const override;
-#endif
+      ColumnVectorsValueAccessor *aux_accessor,
+      const std::vector<attribute_id> &argument_ids) const override;
 
   void aggregateValueAccessorIntoHashTable(
       ValueAccessor *accessor,

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c1baa015/query_execution/QueryContext.hpp
----------------------------------------------------------------------
diff --git a/query_execution/QueryContext.hpp b/query_execution/QueryContext.hpp
index 7ad8fa1..8733093 100644
--- a/query_execution/QueryContext.hpp
+++ b/query_execution/QueryContext.hpp
@@ -200,20 +200,6 @@ class QueryContext {
   }
 
   /**
-   * @brief Destroy the payloads from the aggregation hash tables.
-   *
-   * @warning After calling these methods, the hash table will be in an invalid
-   *          state. No other operation should be performed on them.
-   *
-   * @param id The ID of the AggregationOperationState.
-   **/
-  inline void destroyAggregationHashTablePayload(const aggregation_state_id id) {
-    DCHECK_LT(id, aggregation_states_.size());
-    DCHECK(aggregation_states_[id]);
-    aggregation_states_[id]->destroyAggregationHashTablePayload();
-  }
-
-  /**
    * @brief Whether the given GeneratorFunctionHandle id is valid.
    *
    * @param id The GeneratorFunctionHandle id.

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c1baa015/relational_operators/DestroyAggregationStateOperator.cpp
----------------------------------------------------------------------
diff --git a/relational_operators/DestroyAggregationStateOperator.cpp b/relational_operators/DestroyAggregationStateOperator.cpp
index 49be43d..62ca9e7 100644
--- a/relational_operators/DestroyAggregationStateOperator.cpp
+++ b/relational_operators/DestroyAggregationStateOperator.cpp
@@ -58,13 +58,6 @@ bool DestroyAggregationStateOperator::getAllWorkOrderProtos(WorkOrderProtosConta
 }
 
 void DestroyAggregationStateWorkOrder::execute() {
-  // NOTE(harshad) : The destroyAggregationHashTablePayload call is separate
-  // from the destroyAggregationState call. The reason is that the aggregation
-  // hash tables don't own the AggregationHandle objects. However the hash table
-  // class requires the handles for destroying the payload (see the
-  // destroyPayload methods in AggregationHandle classes). Therefore, we first
-  // destroy the payloads in the hash table and then destroy the hash table.
-  query_context_->destroyAggregationHashTablePayload(aggr_state_index_);
   query_context_->destroyAggregationState(aggr_state_index_);
 }
 

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c1baa015/storage/AggregationOperationState.cpp
----------------------------------------------------------------------
diff --git a/storage/AggregationOperationState.cpp b/storage/AggregationOperationState.cpp
index b942c1b..6470f33 100644
--- a/storage/AggregationOperationState.cpp
+++ b/storage/AggregationOperationState.cpp
@@ -39,13 +39,14 @@
 #include "expressions/predicate/Predicate.hpp"
 #include "expressions/scalar/Scalar.hpp"
 #include "storage/AggregationOperationState.pb.h"
-#include "storage/HashTable.hpp"
+#include "storage/FastHashTable.hpp"
+#include "storage/FastHashTableFactory.hpp"
 #include "storage/HashTableBase.hpp"
-#include "storage/HashTableFactory.hpp"
 #include "storage/InsertDestination.hpp"
 #include "storage/StorageBlock.hpp"
 #include "storage/StorageBlockInfo.hpp"
 #include "storage/StorageManager.hpp"
+#include "storage/SubBlocksReference.hpp"
 #include "storage/TupleIdSequence.hpp"
 #include "storage/ValueAccessor.hpp"
 #include "types/TypedValue.hpp"
@@ -83,33 +84,42 @@ AggregationOperationState::AggregationOperationState(
       is_aggregate_partitioned_(checkAggregatePartitioned(
           estimated_num_entries, is_distinct, group_by, aggregate_functions)),
       predicate_(predicate),
-      group_by_list_(std::move(group_by)),
-      arguments_(std::move(arguments)),
       is_distinct_(std::move(is_distinct)),
       storage_manager_(storage_manager) {
   // Sanity checks: each aggregate has a corresponding list of arguments.
-  DCHECK(aggregate_functions.size() == arguments_.size());
+  DCHECK(aggregate_functions.size() == arguments.size());
 
   // Get the types of GROUP BY expressions for creating HashTables below.
-  std::vector<const Type *> group_by_types;
-  for (const std::unique_ptr<const Scalar> &group_by_element : group_by_list_) {
-    group_by_types.emplace_back(&group_by_element->getType());
+  for (const std::unique_ptr<const Scalar> &group_by_element : group_by) {
+    group_by_types_.emplace_back(&group_by_element->getType());
+  }
+
+  // Prepare group-by element attribute ids and non-trivial expressions.
+  for (std::unique_ptr<const Scalar> &group_by_element : group_by) {
+    const attribute_id attr_id =
+        group_by_element->getAttributeIdForValueAccessor();
+    if (attr_id == kInvalidAttributeID) {
+      const attribute_id non_trivial_attr_id =
+          -(static_cast<attribute_id>(non_trivial_expressions_.size()) + 2);
+      non_trivial_expressions_.emplace_back(group_by_element.release());
+      group_by_key_ids_.emplace_back(non_trivial_attr_id);
+    } else {
+      group_by_key_ids_.emplace_back(attr_id);
+    }
   }
 
   std::vector<AggregationHandle *> group_by_handles;
-  group_by_handles.clear();
 
   if (aggregate_functions.size() == 0) {
     // If there is no aggregation function, then it is a distinctify operation
     // on the group-by expressions.
-    DCHECK_GT(group_by_list_.size(), 0u);
+    DCHECK_GT(group_by_key_ids_.size(), 0u);
 
     handles_.emplace_back(new AggregationHandleDistinct());
-    arguments_.push_back({});
     is_distinct_.emplace_back(false);
     group_by_hashtable_pool_.reset(new HashTablePool(estimated_num_entries,
                                                      hash_table_impl_type,
-                                                     group_by_types,
+                                                     group_by_types_,
                                                      {1},
                                                      handles_,
                                                      storage_manager));
@@ -117,8 +127,8 @@ AggregationOperationState::AggregationOperationState(
     // Set up each individual aggregate in this operation.
     std::vector<const AggregateFunction *>::const_iterator agg_func_it =
         aggregate_functions.begin();
-    std::vector<std::vector<std::unique_ptr<const Scalar>>>::const_iterator
-        args_it = arguments_.begin();
+    std::vector<std::vector<std::unique_ptr<const Scalar>>>::iterator
+        args_it = arguments.begin();
     std::vector<bool>::const_iterator is_distinct_it = is_distinct_.begin();
     std::vector<HashTableImplType>::const_iterator
         distinctify_hash_table_impl_types_it =
@@ -133,6 +143,22 @@ AggregationOperationState::AggregationOperationState(
         argument_types.emplace_back(&argument->getType());
       }
 
+      // Prepare argument attribute ids and non-trivial expressions.
+      std::vector<attribute_id> argument_ids;
+      for (std::unique_ptr<const Scalar> &argument : *args_it) {
+        const attribute_id attr_id =
+            argument->getAttributeIdForValueAccessor();
+        if (attr_id == kInvalidAttributeID) {
+          const attribute_id non_trivial_attr_id =
+              -(static_cast<attribute_id>(non_trivial_expressions_.size()) + 2);
+          non_trivial_expressions_.emplace_back(argument.release());
+          argument_ids.emplace_back(non_trivial_attr_id);
+        } else {
+          argument_ids.emplace_back(attr_id);
+        }
+      }
+      argument_ids_.emplace_back(std::move(argument_ids));
+
       // Sanity checks: aggregate function exists and can apply to the specified
       // arguments.
       DCHECK(*agg_func_it != nullptr);
@@ -142,7 +168,7 @@ AggregationOperationState::AggregationOperationState(
       // to do actual aggregate computation.
       handles_.emplace_back((*agg_func_it)->createHandle(argument_types));
 
-      if (!group_by_list_.empty()) {
+      if (!group_by_key_ids_.empty()) {
         // Aggregation with GROUP BY: combined payload is partially updated in
         // the presence of DISTINCT.
         if (*is_distinct_it) {
@@ -153,35 +179,12 @@ AggregationOperationState::AggregationOperationState(
       } else {
         // Aggregation without GROUP BY: create a single global state.
         single_states_.emplace_back(handles_.back()->createInitialState());
-
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-        // See if all of this aggregate's arguments are attributes in the input
-        // relation. If so, remember the attribute IDs so that we can do copy
-        // elision when actually performing the aggregation.
-        std::vector<attribute_id> local_arguments_as_attributes;
-        local_arguments_as_attributes.reserve(args_it->size());
-        for (const std::unique_ptr<const Scalar> &argument : *args_it) {
-          const attribute_id argument_id =
-              argument->getAttributeIdForValueAccessor();
-          if (argument_id == -1) {
-            local_arguments_as_attributes.clear();
-            break;
-          } else {
-            DCHECK_EQ(input_relation_.getID(),
-                      argument->getRelationIdForValueAccessor());
-            local_arguments_as_attributes.push_back(argument_id);
-          }
-        }
-
-        arguments_as_attributes_.emplace_back(
-            std::move(local_arguments_as_attributes));
-#endif
       }
 
       // Initialize the corresponding distinctify hash table if this is a
       // DISTINCT aggregation.
       if (*is_distinct_it) {
-        std::vector<const Type *> key_types(group_by_types);
+        std::vector<const Type *> key_types(group_by_types_);
         key_types.insert(
             key_types.end(), argument_types.begin(), argument_types.end());
         // TODO(jianqiao): estimated_num_entries is quite inaccurate for
@@ -203,12 +206,12 @@ AggregationOperationState::AggregationOperationState(
       }
     }
 
-    if (!group_by_handles.empty()) {
-      // Aggregation with GROUP BY: create a HashTable pool.
+    // Aggregation with GROUP BY: create a HashTable pool.
+    if (!group_by_key_ids_.empty()) {
       if (!is_aggregate_partitioned_) {
         group_by_hashtable_pool_.reset(new HashTablePool(estimated_num_entries,
                                                          hash_table_impl_type,
-                                                         group_by_types,
+                                                         group_by_types_,
                                                          payload_sizes,
                                                          group_by_handles,
                                                          storage_manager));
@@ -217,7 +220,7 @@ AggregationOperationState::AggregationOperationState(
             new PartitionedHashTablePool(estimated_num_entries,
                                          FLAGS_num_aggregation_partitions,
                                          hash_table_impl_type,
-                                         group_by_types,
+                                         group_by_types_,
                                          payload_sizes,
                                          group_by_handles,
                                          storage_manager));
@@ -355,7 +358,7 @@ bool AggregationOperationState::ProtoIsValid(
 
 void AggregationOperationState::aggregateBlock(const block_id input_block,
                                                LIPFilterAdaptiveProber *lip_filter_adaptive_prober) {
-  if (group_by_list_.empty()) {
+  if (group_by_key_ids_.empty()) {
     aggregateBlockSingleState(input_block);
   } else {
     aggregateBlockHashTable(input_block, lip_filter_adaptive_prober);
@@ -364,7 +367,7 @@ void AggregationOperationState::aggregateBlock(const block_id input_block,
 
 void AggregationOperationState::finalizeAggregate(
     InsertDestination *output_destination) {
-  if (group_by_list_.empty()) {
+  if (group_by_key_ids_.empty()) {
     finalizeSingleState(output_destination);
   } else {
     finalizeHashTable(output_destination);
@@ -392,37 +395,53 @@ void AggregationOperationState::aggregateBlockSingleState(
 
   std::unique_ptr<TupleIdSequence> matches;
   if (predicate_ != nullptr) {
-    std::unique_ptr<ValueAccessor> accessor(
-        block->getTupleStorageSubBlock().createValueAccessor());
     matches.reset(block->getMatchesForPredicate(predicate_.get(), matches.get()));
   }
 
-  for (std::size_t agg_idx = 0; agg_idx < handles_.size(); ++agg_idx) {
-    const std::vector<attribute_id> *local_arguments_as_attributes = nullptr;
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-    // If all arguments are attributes of the input relation, elide a copy.
-    if (!arguments_as_attributes_[agg_idx].empty()) {
-      local_arguments_as_attributes = &(arguments_as_attributes_[agg_idx]);
+  const auto &tuple_store = block->getTupleStorageSubBlock();
+  std::unique_ptr<ValueAccessor> accessor(
+      tuple_store.createValueAccessor(matches.get()));
+
+  ColumnVectorsValueAccessor non_trivial_results;
+  if (!non_trivial_expressions_.empty()) {
+    SubBlocksReference sub_blocks_ref(tuple_store,
+                                      block->getIndices(),
+                                      block->getIndicesConsistent());
+    for (const auto &expression : non_trivial_expressions_) {
+      non_trivial_results.addColumn(
+          expression->getAllValues(accessor.get(), &sub_blocks_ref));
     }
-#endif
+  }
+
+  for (std::size_t agg_idx = 0; agg_idx < handles_.size(); ++agg_idx) {
     if (is_distinct_[agg_idx]) {
+      LOG(FATAL) << "Distinct aggregation not supported";
       // Call StorageBlock::aggregateDistinct() to put the arguments as keys
       // directly into the (threadsafe) shared global distinctify HashTable
       // for this aggregate.
-      block->aggregateDistinct(*handles_[agg_idx],
-                               arguments_[agg_idx],
-                               local_arguments_as_attributes,
-                               {}, /* group_by */
-                               matches.get(),
-                               distinctify_hashtables_[agg_idx].get(),
-                               nullptr /* reuse_group_by_vectors */);
+//      block->aggregateDistinct(*handles_[agg_idx],
+//                               arguments_[agg_idx],
+//                               local_arguments_as_attributes,
+//                               {}, /* group_by */
+//                               matches.get(),
+//                               distinctify_hashtables_[agg_idx].get(),
+//                               nullptr /* reuse_group_by_vectors */);
+      // TODO(jianqiao): handle distinct
       local_state.emplace_back(nullptr);
     } else {
-      // Call StorageBlock::aggregate() to actually do the aggregation.
-      local_state.emplace_back(block->aggregate(*handles_[agg_idx],
-                                                arguments_[agg_idx],
-                                                local_arguments_as_attributes,
-                                                matches.get()));
+      const auto &argument_ids = argument_ids_[agg_idx];
+      const auto *handle = handles_[agg_idx];
+
+      AggregationState *state;
+      if (argument_ids.empty()) {
+        // Special case. This is a nullary aggregate (i.e. COUNT(*)).
+        state = handle->accumulateNullary(matches == nullptr ? tuple_store.numTuples()
+                                                             : matches->size());
+      } else {
+        // Have the AggregationHandle actually do the aggregation.
+        state = handle->accumulate(accessor.get(), &non_trivial_results, argument_ids);
+      }
+      local_state.emplace_back(state);
     }
   }
 
@@ -435,104 +454,107 @@ void AggregationOperationState::aggregateBlockHashTable(
     LIPFilterAdaptiveProber *lip_filter_adaptive_prober) {
   BlockReference block(
       storage_manager_->getBlock(input_block, input_relation_));
+  const auto &tuple_store = block->getTupleStorageSubBlock();
+  std::unique_ptr<ValueAccessor> base_accessor(tuple_store.createValueAccessor());
+  std::unique_ptr<ValueAccessor> shared_accessor;
+  ValueAccessor *accessor = base_accessor.get();
 
   // Apply the predicate first, then the LIPFilters, to generate a TupleIdSequence
   // as the existence map for the tuples.
   std::unique_ptr<TupleIdSequence> matches;
   if (predicate_ != nullptr) {
     matches.reset(block->getMatchesForPredicate(predicate_.get()));
+    shared_accessor.reset(
+        base_accessor->createSharedTupleIdSequenceAdapterVirtual(*matches));
+    accessor = shared_accessor.get();
   }
   if (lip_filter_adaptive_prober != nullptr) {
-    std::unique_ptr<ValueAccessor> accessor(
-        block->getTupleStorageSubBlock().createValueAccessor(matches.get()));
-    matches.reset(lip_filter_adaptive_prober->filterValueAccessor(accessor.get()));
+    matches.reset(lip_filter_adaptive_prober->filterValueAccessor(accessor));
+    shared_accessor.reset(
+        base_accessor->createSharedTupleIdSequenceAdapterVirtual(*matches));
+    accessor = shared_accessor.get();
   }
 
-  // This holds values of all the GROUP BY attributes so that the can be reused
-  // across multiple aggregates (i.e. we only pay the cost of evaluatin the
-  // GROUP BY expressions once).
-  std::vector<std::unique_ptr<ColumnVector>> reuse_group_by_vectors;
+  ColumnVectorsValueAccessor non_trivial_results;
+  if (!non_trivial_expressions_.empty()) {
+    SubBlocksReference sub_blocks_ref(tuple_store,
+                                      block->getIndices(),
+                                      block->getIndicesConsistent());
+    for (const auto &expression : non_trivial_expressions_) {
+      non_trivial_results.addColumn(
+          expression->getAllValues(accessor, &sub_blocks_ref));
+    }
+  }
 
   for (std::size_t agg_idx = 0; agg_idx < handles_.size(); ++agg_idx) {
     if (is_distinct_[agg_idx]) {
+      LOG(FATAL) << "Distinct aggregation not supported";
       // Call StorageBlock::aggregateDistinct() to insert the GROUP BY
       // expression
       // values and the aggregation arguments together as keys directly into the
       // (threadsafe) shared global distinctify HashTable for this aggregate.
-      block->aggregateDistinct(*handles_[agg_idx],
-                               arguments_[agg_idx],
-                               nullptr, /* arguments_as_attributes */
-                               group_by_list_,
-                               matches.get(),
-                               distinctify_hashtables_[agg_idx].get(),
-                               &reuse_group_by_vectors);
+//      block->aggregateDistinct(*handles_[agg_idx],
+//                               arguments_[agg_idx],
+//                               nullptr, /* arguments_as_attributes */
+//                               group_by_list_,
+//                               matches.get(),
+//                               distinctify_hashtables_[agg_idx].get(),
+//                               &reuse_group_by_vectors);
+      // TODO(jianqiao): handle distinct
     }
   }
 
   if (!is_aggregate_partitioned_) {
-    // Call StorageBlock::aggregateGroupBy() to aggregate this block's values
-    // directly into the (threadsafe) shared global HashTable for this
-    // aggregate.
     DCHECK(group_by_hashtable_pool_ != nullptr);
+
     AggregationStateHashTableBase *agg_hash_table =
-      group_by_hashtable_pool_->getHashTableFast();
+        group_by_hashtable_pool_->getHashTableFast();
     DCHECK(agg_hash_table != nullptr);
-    block->aggregateGroupBy(arguments_,
-                            group_by_list_,
-                            matches.get(),
-                            agg_hash_table,
-                            &reuse_group_by_vectors);
+
+    accessor->beginIterationVirtual();
+    agg_hash_table->upsertValueAccessorCompositeKeyFast(accessor,
+                                                        &non_trivial_results,
+                                                        argument_ids_,
+                                                        group_by_key_ids_,
+                                                        true /* check_for_null_keys */);
     group_by_hashtable_pool_->returnHashTable(agg_hash_table);
   } else {
-    ColumnVectorsValueAccessor temp_result;
-    // IDs of 'arguments' as attributes in the ValueAccessor we create below.
-    std::vector<attribute_id> argument_ids;
-
-    // IDs of GROUP BY key element(s) in the ValueAccessor we create below.
-    std::vector<attribute_id> key_ids;
-    const std::size_t num_partitions = partitioned_group_by_hashtable_pool_->getNumPartitions();
-    block->aggregateGroupByPartitioned(
-        arguments_,
-        group_by_list_,
-        matches.get(),
-        num_partitions,
-        &temp_result,
-        &argument_ids,
-        &key_ids,
-        &reuse_group_by_vectors);
-    // Compute the partitions for the tuple formed by group by values.
-    std::vector<std::unique_ptr<TupleIdSequence>> partition_membership;
-    partition_membership.resize(num_partitions);
-
-    // Create a tuple-id sequence for each partition.
-    for (std::size_t partition = 0; partition < num_partitions; ++partition) {
-      partition_membership[partition].reset(
-          new TupleIdSequence(temp_result.getEndPosition()));
-    }
-
-    // Iterate over ValueAccessor for each tuple,
-    // set a bit in the appropriate TupleIdSequence.
-    temp_result.beginIteration();
-    while (temp_result.next()) {
-      // We need a unique_ptr because getTupleWithAttributes() uses "new".
-      std::unique_ptr<Tuple> curr_tuple(temp_result.getTupleWithAttributes(key_ids));
-      const std::size_t curr_tuple_partition_id =
-          curr_tuple->getTupleHash() % num_partitions;
-      partition_membership[curr_tuple_partition_id]->set(
-          temp_result.getCurrentPosition(), true);
-    }
-    // For each partition, create an adapter around Value Accessor and
-    // TupleIdSequence.
-    std::vector<std::unique_ptr<
-        TupleIdSequenceAdapterValueAccessor<ColumnVectorsValueAccessor>>> adapter;
-    adapter.resize(num_partitions);
-    for (std::size_t partition = 0; partition < num_partitions; ++partition) {
-      adapter[partition].reset(temp_result.createSharedTupleIdSequenceAdapter(
-          *(partition_membership)[partition]));
-      partitioned_group_by_hashtable_pool_->getHashTable(partition)
-          ->upsertValueAccessorCompositeKeyFast(
-              argument_ids, adapter[partition].get(), key_ids, true);
-    }
+    LOG(FATAL) << "Partitioned aggregation not supported";
+//    const std::size_t num_partitions = partitioned_group_by_hashtable_pool_->getNumPartitions();
+//
+//    // Compute the partitions for the tuple formed by group by values.
+//    std::vector<std::unique_ptr<TupleIdSequence>> partition_membership;
+//    partition_membership.resize(num_partitions);
+//
+//    // Create a tuple-id sequence for each partition.
+//    for (std::size_t partition = 0; partition < num_partitions; ++partition) {
+//      partition_membership[partition].reset(
+//          new TupleIdSequence(temp_result.getEndPosition()));
+//    }
+//
+//    // Iterate over ValueAccessor for each tuple,
+//    // set a bit in the appropriate TupleIdSequence.
+//    temp_result.beginIteration();
+//    while (temp_result.next()) {
+//      // We need a unique_ptr because getTupleWithAttributes() uses "new".
+//      std::unique_ptr<Tuple> curr_tuple(temp_result.getTupleWithAttributes(key_ids));
+//      const std::size_t curr_tuple_partition_id =
+//          curr_tuple->getTupleHash() % num_partitions;
+//      partition_membership[curr_tuple_partition_id]->set(
+//          temp_result.getCurrentPosition(), true);
+//    }
+//    // For each partition, create an adapter around Value Accessor and
+//    // TupleIdSequence.
+//    std::vector<std::unique_ptr<
+//        TupleIdSequenceAdapterValueAccessor<ColumnVectorsValueAccessor>>> adapter;
+//    adapter.resize(num_partitions);
+//    for (std::size_t partition = 0; partition < num_partitions; ++partition) {
+//      adapter[partition].reset(temp_result.createSharedTupleIdSequenceAdapter(
+//          *(partition_membership)[partition]));
+//      partitioned_group_by_hashtable_pool_->getHashTable(partition)
+//          ->upsertValueAccessorCompositeKeyFast(
+//              argument_ids, adapter[partition].get(), key_ids, true);
+//    }
   }
 }
 
@@ -577,59 +599,36 @@ void AggregationOperationState::finalizeHashTable(
   // e.g. Keep merging entries from smaller hash tables to larger.
 
   auto *hash_tables = group_by_hashtable_pool_->getAllHashTables();
-  if (hash_tables->size() > 1) {
-    for (int hash_table_index = 0;
-         hash_table_index < static_cast<int>(hash_tables->size() - 1);
-         ++hash_table_index) {
-      // Merge each hash table to the last hash table.
-      mergeGroupByHashTables((*hash_tables)[hash_table_index].get(),
-                             hash_tables->back().get());
-    }
+  DCHECK(hash_tables != nullptr);
+  if (hash_tables->empty()) {
+    return;
+  }
+
+  std::unique_ptr<AggregationStateHashTableBase> final_hash_table(
+      hash_tables->back().release());
+  for (std::size_t i = 0; i < hash_tables->size() - 1; ++i) {
+    std::unique_ptr<AggregationStateHashTableBase> hash_table(
+        hash_tables->at(i).release());
+    mergeGroupByHashTables(hash_table.get(), final_hash_table.get());
+    hash_table->destroyPayload();
   }
 
   // Collect per-aggregate finalized values.
   std::vector<std::unique_ptr<ColumnVector>> final_values;
   for (std::size_t agg_idx = 0; agg_idx < handles_.size(); ++agg_idx) {
     if (is_distinct_[agg_idx]) {
-      DCHECK(group_by_hashtable_pool_ != nullptr);
-      auto *hash_tables = group_by_hashtable_pool_->getAllHashTables();
-      DCHECK(hash_tables != nullptr);
-      if (hash_tables->empty()) {
-        // We may have a case where hash_tables is empty, e.g. no input blocks.
-        // However for aggregateOnDistinctifyHashTableForGroupBy to work
-        // correctly, we should create an empty group by hash table.
-        AggregationStateHashTableBase *new_hash_table =
-            group_by_hashtable_pool_->getHashTableFast();
-        group_by_hashtable_pool_->returnHashTable(new_hash_table);
-        hash_tables = group_by_hashtable_pool_->getAllHashTables();
-      }
-      DCHECK(hash_tables->back() != nullptr);
-      AggregationStateHashTableBase *agg_hash_table = hash_tables->back().get();
-      DCHECK(agg_hash_table != nullptr);
       handles_[agg_idx]->allowUpdate();
       handles_[agg_idx]->aggregateOnDistinctifyHashTableForGroupBy(
-          *distinctify_hashtables_[agg_idx], agg_hash_table, agg_idx);
+          *distinctify_hashtables_[agg_idx], final_hash_table.get(), agg_idx);
     }
 
-    auto *hash_tables = group_by_hashtable_pool_->getAllHashTables();
-    DCHECK(hash_tables != nullptr);
-    if (hash_tables->empty()) {
-      // We may have a case where hash_tables is empty, e.g. no input blocks.
-      // However for aggregateOnDistinctifyHashTableForGroupBy to work
-      // correctly, we should create an empty group by hash table.
-      AggregationStateHashTableBase *new_hash_table =
-          group_by_hashtable_pool_->getHashTableFast();
-      group_by_hashtable_pool_->returnHashTable(new_hash_table);
-      hash_tables = group_by_hashtable_pool_->getAllHashTables();
-    }
-    AggregationStateHashTableBase *agg_hash_table = hash_tables->back().get();
-    DCHECK(agg_hash_table != nullptr);
     ColumnVector *agg_result_col = handles_[agg_idx]->finalizeHashTable(
-        *agg_hash_table, &group_by_keys, agg_idx);
+        *final_hash_table, &group_by_keys, agg_idx);
     if (agg_result_col != nullptr) {
       final_values.emplace_back(agg_result_col);
     }
   }
+  final_hash_table->destroyPayload();
 
   // Reorganize 'group_by_keys' in column-major order so that we can make a
   // ColumnVectorsValueAccessor to bulk-insert results.
@@ -640,11 +639,10 @@ void AggregationOperationState::finalizeHashTable(
   // in a single HashTable.
   std::vector<std::unique_ptr<ColumnVector>> group_by_cvs;
   std::size_t group_by_element_idx = 0;
-  for (const std::unique_ptr<const Scalar> &group_by_element : group_by_list_) {
-    const Type &group_by_type = group_by_element->getType();
-    if (NativeColumnVector::UsableForType(group_by_type)) {
+  for (const Type *group_by_type : group_by_types_) {
+    if (NativeColumnVector::UsableForType(*group_by_type)) {
       NativeColumnVector *element_cv =
-          new NativeColumnVector(group_by_type, group_by_keys.size());
+          new NativeColumnVector(*group_by_type, group_by_keys.size());
       group_by_cvs.emplace_back(element_cv);
       for (std::vector<TypedValue> &group_key : group_by_keys) {
         element_cv->appendTypedValue(
@@ -652,7 +650,7 @@ void AggregationOperationState::finalizeHashTable(
       }
     } else {
       IndirectColumnVector *element_cv =
-          new IndirectColumnVector(group_by_type, group_by_keys.size());
+          new IndirectColumnVector(*group_by_type, group_by_keys.size());
       group_by_cvs.emplace_back(element_cv);
       for (std::vector<TypedValue> &group_key : group_by_keys) {
         element_cv->appendTypedValue(
@@ -676,25 +674,6 @@ void AggregationOperationState::finalizeHashTable(
   output_destination->bulkInsertTuples(&complete_result);
 }
 
-void AggregationOperationState::destroyAggregationHashTablePayload() {
-  std::vector<std::unique_ptr<AggregationStateHashTableBase>> *all_hash_tables =
-      nullptr;
-  if (!is_aggregate_partitioned_) {
-    if (group_by_hashtable_pool_ != nullptr) {
-      all_hash_tables = group_by_hashtable_pool_->getAllHashTables();
-    }
-  } else {
-    if (partitioned_group_by_hashtable_pool_ != nullptr) {
-      all_hash_tables = partitioned_group_by_hashtable_pool_->getAllHashTables();
-    }
-  }
-  if (all_hash_tables != nullptr) {
-    for (std::size_t ht_index = 0; ht_index < all_hash_tables->size(); ++ht_index) {
-      (*all_hash_tables)[ht_index]->destroyPayload();
-    }
-  }
-}
-
 void AggregationOperationState::finalizeAggregatePartitioned(
     const std::size_t partition_id, InsertDestination *output_destination) {
   // Each element of 'group_by_keys' is a vector of values for a particular
@@ -703,15 +682,16 @@ void AggregationOperationState::finalizeAggregatePartitioned(
 
   // Collect per-aggregate finalized values.
   std::vector<std::unique_ptr<ColumnVector>> final_values;
+  AggregationStateHashTableBase *hash_table =
+      partitioned_group_by_hashtable_pool_->getHashTable(partition_id);
   for (std::size_t agg_idx = 0; agg_idx < handles_.size(); ++agg_idx) {
-    AggregationStateHashTableBase *hash_table =
-        partitioned_group_by_hashtable_pool_->getHashTable(partition_id);
     ColumnVector *agg_result_col = handles_[agg_idx]->finalizeHashTable(
         *hash_table, &group_by_keys, agg_idx);
     if (agg_result_col != nullptr) {
       final_values.emplace_back(agg_result_col);
     }
   }
+  hash_table->destroyPayload();
 
   // Reorganize 'group_by_keys' in column-major order so that we can make a
   // ColumnVectorsValueAccessor to bulk-insert results.
@@ -722,16 +702,17 @@ void AggregationOperationState::finalizeAggregatePartitioned(
   // in a single HashTable.
   std::vector<std::unique_ptr<ColumnVector>> group_by_cvs;
   std::size_t group_by_element_idx = 0;
-  for (const std::unique_ptr<const Scalar> &group_by_element : group_by_list_) {
-    const Type &group_by_type = group_by_element->getType();
-    if (NativeColumnVector::UsableForType(group_by_type)) {
-      NativeColumnVector *element_cv = new NativeColumnVector(group_by_type, group_by_keys.size());
+  for (const Type *group_by_type : group_by_types_) {
+    if (NativeColumnVector::UsableForType(*group_by_type)) {
+      NativeColumnVector *element_cv =
+          new NativeColumnVector(*group_by_type, group_by_keys.size());
       group_by_cvs.emplace_back(element_cv);
       for (std::vector<TypedValue> &group_key : group_by_keys) {
         element_cv->appendTypedValue(std::move(group_key[group_by_element_idx]));
       }
     } else {
-      IndirectColumnVector *element_cv = new IndirectColumnVector(group_by_type, group_by_keys.size());
+      IndirectColumnVector *element_cv =
+          new IndirectColumnVector(*group_by_type, group_by_keys.size());
       group_by_cvs.emplace_back(element_cv);
       for (std::vector<TypedValue> &group_key : group_by_keys) {
         element_cv->appendTypedValue(std::move(group_key[group_by_element_idx]));

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c1baa015/storage/AggregationOperationState.hpp
----------------------------------------------------------------------
diff --git a/storage/AggregationOperationState.hpp b/storage/AggregationOperationState.hpp
index e0826b0..9f2bd92 100644
--- a/storage/AggregationOperationState.hpp
+++ b/storage/AggregationOperationState.hpp
@@ -33,6 +33,7 @@
 #include "storage/HashTableBase.hpp"
 #include "storage/HashTablePool.hpp"
 #include "storage/PartitionedHashTablePool.hpp"
+#include "storage/StorageBlock.hpp"
 #include "storage/StorageBlockInfo.hpp"
 #include "utility/Macros.hpp"
 
@@ -46,6 +47,7 @@ class CatalogRelationSchema;
 class InsertDestination;
 class LIPFilterAdaptiveProber;
 class StorageManager;
+class TupleIdSequence;
 
 DECLARE_int32(num_aggregation_partitions);
 DECLARE_int32(partition_aggregation_num_groups_threshold);
@@ -178,11 +180,6 @@ class AggregationOperationState {
   void finalizeAggregate(InsertDestination *output_destination);
 
   /**
-   * @brief Destroy the payloads in the aggregation hash tables.
-   **/
-  void destroyAggregationHashTablePayload();
-
-  /**
    * @brief Generate the final results for the aggregates managed by this
    *        AggregationOperationState and write them out to StorageBlock(s).
    *        In this implementation, each thread picks a hash table belonging to
@@ -236,27 +233,28 @@ class AggregationOperationState {
       const std::vector<bool> &is_distinct,
       const std::vector<std::unique_ptr<const Scalar>> &group_by,
       const std::vector<const AggregateFunction *> &aggregate_functions) const {
-    // If there's no aggregation, return false.
-    if (aggregate_functions.empty()) {
-      return false;
-    }
-    // Check if there's a distinct operation involved in any aggregate, if so
-    // the aggregate can't be partitioned.
-    for (auto distinct : is_distinct) {
-      if (distinct) {
-        return false;
-      }
-    }
-    // There's no distinct aggregation involved, Check if there's at least one
-    // GROUP BY operation.
-    if (group_by.empty()) {
-      return false;
-    }
-    // There are GROUP BYs without DISTINCT. Check if the estimated number of
-    // groups is large enough to warrant a partitioned aggregation.
-    return estimated_num_groups >
-           static_cast<std::size_t>(
-               FLAGS_partition_aggregation_num_groups_threshold);
+//    // If there's no aggregation, return false.
+//    if (aggregate_functions.empty()) {
+//      return false;
+//    }
+//    // Check if there's a distinct operation involved in any aggregate, if so
+//    // the aggregate can't be partitioned.
+//    for (auto distinct : is_distinct) {
+//      if (distinct) {
+//        return false;
+//      }
+//    }
+//    // There's no distinct aggregation involved, Check if there's at least one
+//    // GROUP BY operation.
+//    if (group_by.empty()) {
+//      return false;
+//    }
+//    // There are GROUP BYs without DISTINCT. Check if the estimated number of
+//    // groups is large enough to warrant a partitioned aggregation.
+//    return estimated_num_groups >
+//           static_cast<std::size_t>(
+//               FLAGS_partition_aggregation_num_groups_threshold);
+    return false;
   }
 
   // Common state for all aggregates in this operation: the input relation, the
@@ -267,27 +265,27 @@ class AggregationOperationState {
   const bool is_aggregate_partitioned_;
 
   std::unique_ptr<const Predicate> predicate_;
-  std::vector<std::unique_ptr<const Scalar>> group_by_list_;
 
   // Each individual aggregate in this operation has an AggregationHandle and
-  // some number of Scalar arguments.
+  // zero (indicated by -1) or one argument.
   std::vector<AggregationHandle *> handles_;
-  std::vector<std::vector<std::unique_ptr<const Scalar>>> arguments_;
 
   // For each aggregate, whether DISTINCT should be applied to the aggregate's
   // arguments.
   std::vector<bool> is_distinct_;
 
+  // Non-trivial group-by/argument expressions that need to be evaluated.
+  std::vector<std::unique_ptr<const Scalar>> non_trivial_expressions_;
+
+  std::vector<attribute_id> group_by_key_ids_;
+  std::vector<std::vector<attribute_id>> argument_ids_;
+
+  std::vector<const Type *> group_by_types_;
+
   // Hash table for obtaining distinct (i.e. unique) arguments.
   std::vector<std::unique_ptr<AggregationStateHashTableBase>>
       distinctify_hashtables_;
 
-#ifdef QUICKSTEP_ENABLE_VECTOR_COPY_ELISION_SELECTION
-  // If all an aggregate's argument expressions are simply attributes in
-  // 'input_relation_', then this caches the attribute IDs of those arguments.
-  std::vector<std::vector<attribute_id>> arguments_as_attributes_;
-#endif
-
   // Per-aggregate global states for aggregation without GROUP BY.
   std::vector<std::unique_ptr<AggregationState>> single_states_;
 

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c1baa015/storage/FastHashTable.hpp
----------------------------------------------------------------------
diff --git a/storage/FastHashTable.hpp b/storage/FastHashTable.hpp
index 4a82a62..04e33bd 100644
--- a/storage/FastHashTable.hpp
+++ b/storage/FastHashTable.hpp
@@ -39,6 +39,7 @@
 #include "threading/SpinSharedMutex.hpp"
 #include "types/Type.hpp"
 #include "types/TypedValue.hpp"
+#include "types/containers/ColumnVectorsValueAccessor.hpp"
 #include "utility/HashPair.hpp"
 #include "utility/Macros.hpp"
 
@@ -407,109 +408,17 @@ class FastHashTable : public HashTableBase<resizable,
                               const std::uint8_t *init_value_ptr,
                               const std::uint8_t *source_state);
 
-  /**
-   * @brief Apply a functor to (multiple) entries in this hash table, with keys
-   *        drawn from a ValueAccessor. New values are first inserted if not
-   *        already present.
-   *
-   * @warning This method is only usable if allow_duplicate_keys is false.
-   * @warning This method is threadsafe with regard to other calls to upsert(),
-   *          upsertCompositeKey(), upsertValueAccessor(), and
-   *          upsertValueAccessorCompositeKey(), but should not be used
-   *          simultaneously with put(), putCompositeKey(), putValueAccessor(),
-   *          or putValueAccessorCompositeKey().
-   * @warning The ValueAccessor reference and ValueT* pointer passed to
-   *          functor's call operator are only guaranteed to be valid for the
-   *          duration of the call. The functor should not store a copy of
-   *          these pointers and assume that they remain valid.
-   * @warning Although this method itself is threadsafe, the ValueT object
-   *          accessed by functor is not guaranteed to be (although it is
-   *          guaranteed that its initial insertion will be atomic). If it is
-   *          possible for multiple threads to call upsertValueAccessor() with
-   *          the same key at the same time, then their access to ValueT should
-   *          be made threadsafe (e.g. with the use of atomic types, mutexes,
-   *          or some other external synchronization).
-   * @note This version is for single scalar keys, see also
-   *       upsertValueAccessorCompositeKey().
-   * @note If the hash table is (close to) full and resizable is true, this
-   *       routine might result in rebuilding the entire hash table.
-   *
-   * @param accessor A ValueAccessor which will be used to access keys.
-   *        beginIteration() should be called on accessor before calling this
-   *        method.
-   * @param key_attr_id The attribute ID of the keys to be read from accessor.
-   * @param check_for_null_keys If true, each key will be checked to see if it
-   *        is null before upserting it (null keys are skipped). This must be
-   *        set to true if some of the keys that will be read from accessor may
-   *        be null.
-   * @param functor A pointer to a functor, which should provide a call
-   *        operator that takes two arguments: const ValueAccessor& (or better
-   *        yet, a templated call operator which takes a const reference to
-   *        some subclass of ValueAccessor as its first argument) and ValueT*.
-   *        The call operator will be invoked once for every tuple with a
-   *        non-null key in accessor.
-   * @return True on success, false if upsert failed because there was not
-   *         enough space to insert new entries for all the keys in accessor
-   *         (note that some entries may still have been upserted, and
-   *         accessor's iteration will be left on the first tuple which could
-   *         not be inserted).
-   **/
   bool upsertValueAccessorFast(
-      const std::vector<attribute_id> &argument_ids,
       ValueAccessor *accessor,
+      ColumnVectorsValueAccessor *aux_accessor,
+      const std::vector<attribute_id> &argument_ids,
       const attribute_id key_attr_id,
       const bool check_for_null_keys);
 
-  /**
-   * @brief Apply a functor to (multiple) entries in this hash table, with keys
-   *        drawn from a ValueAccessor. New values are first inserted if not
-   *        already present. Composite key version.
-   *
-   * @warning This method is only usable if allow_duplicate_keys is false.
-   * @warning This method is threadsafe with regard to other calls to upsert(),
-   *          upsertCompositeKey(), upsertValueAccessor(), and
-   *          upsertValueAccessorCompositeKey(), but should not be used
-   *          simultaneously with put(), putCompositeKey(), putValueAccessor(),
-   *          or putValueAccessorCompositeKey().
-   * @warning The ValueAccessor reference and ValueT* pointer passed to
-   *          functor's call operator are only guaranteed to be valid for the
-   *          duration of the call. The functor should not store a copy of
-   *          these pointers and assume that they remain valid.
-   * @warning Although this method itself is threadsafe, the ValueT object
-   *          accessed by functor is not guaranteed to be (although it is
-   *          guaranteed that its initial insertion will be atomic). If it is
-   *          possible for multiple threads to call upsertValueAccessor() with
-   *          the same key at the same time, then their access to ValueT should
-   *          be made threadsafe (e.g. with the use of atomic types, mutexes,
-   *          or some other external synchronization).
-   * @note This version is for composite keys, see also upsertValueAccessor().
-   * @note If the hash table is (close to) full and resizable is true, this
-   *       routine might result in rebuilding the entire hash table.
-   *
-   * @param accessor A ValueAccessor which will be used to access keys.
-   *        beginIteration() should be called on accessor before calling this
-   *        method.
-   * @param key_attr_ids The attribute IDs of each key component to be read
-   *        from accessor.
-   * @param check_for_null_keys If true, each key will be checked to see if it
-   *        is null before upserting it (null keys are skipped). This must be
-   *        set to true if some of the keys that will be read from accessor may
-   *        be null.
-   * @param functor A pointer to a functor, which should provide a call
-   *        operator that takes two arguments: const ValueAccessor& (or better
-   *        yet, a templated call operator which takes a const reference to
-   *        some subclass of ValueAccessor as its first argument) and ValueT*.
-   *        The call operator will be invoked once for every tuple with a
-   *        non-null key in accessor.
-   * @return True on success, false if upsert failed because there was not
-   *         enough space to insert new entries for all the keys in accessor
-   *         (note that some entries may still have been upserted, and
-   *         accessor's iteration will be left on the first tuple which could
-   *         not be inserted).
-   **/
   bool upsertValueAccessorCompositeKeyFast(
-      const std::vector<attribute_id> &argument,
       ValueAccessor *accessor,
+      ColumnVectorsValueAccessor *aux_accessor,
+      const std::vector<std::vector<attribute_id>> &argument_ids,
       const std::vector<attribute_id> &key_attr_ids,
       const bool check_for_null_keys) override;
 
@@ -1779,84 +1688,87 @@ bool FastHashTable<resizable,
                    force_key_copy,
                    allow_duplicate_keys>::
     upsertValueAccessorFast(
-        const std::vector<attribute_id> &argument_ids,
         ValueAccessor *accessor,
+        ColumnVectorsValueAccessor *aux_accessor,
+        const std::vector<attribute_id> &argument_ids,
         const attribute_id key_attr_id,
         const bool check_for_null_keys) {
-  DEBUG_ASSERT(!allow_duplicate_keys);
-  std::size_t variable_size;
-  return InvokeOnAnyValueAccessor(
-      accessor,
-      [&](auto *accessor) -> bool {  // NOLINT(build/c++11)
-        if (resizable) {
-          bool continuing = true;
-          while (continuing) {
-            {
-              continuing = false;
-              SpinSharedMutexSharedLock<true> lock(resize_shared_mutex_);
-              while (accessor->next()) {
-                TypedValue key = accessor->getTypedValue(key_attr_id);
-                if (check_for_null_keys && key.isNull()) {
-                  continue;
-                }
-                variable_size = (force_key_copy && !scalar_key_inline_)
-                                    ? key.getDataSize()
-                                    : 0;
-                std::uint8_t *value =
-                    this->upsertInternalFast(key, variable_size, nullptr);
-                if (value == nullptr) {
-                  continuing = true;
-                  break;
-                } else {
-                  SpinMutexLock lock(*(reinterpret_cast<SpinMutex *>(value)));
-                  for (unsigned int k = 0; k < num_handles_; ++k) {
-                    if (argument_ids[k] != kInvalidAttributeID) {
-                      handles_[k]->updateStateUnary(
-                          accessor->getTypedValue(argument_ids[k]),
-                          value + payload_offsets_[k]);
-                    } else {
-                      handles_[k]->updateStateNullary(value +
-                                                      payload_offsets_[k]);
-                    }
-                  }
-                }
-              }
-            }
-            if (continuing) {
-              this->resize(0, variable_size);
-              accessor->previous();
-            }
-          }
-        } else {
-          while (accessor->next()) {
-            TypedValue key = accessor->getTypedValue(key_attr_id);
-            if (check_for_null_keys && key.isNull()) {
-              continue;
-            }
-            variable_size =
-                (force_key_copy && !scalar_key_inline_) ? key.getDataSize() : 0;
-            std::uint8_t *value =
-                this->upsertInternalFast(key, variable_size, nullptr);
-            if (value == nullptr) {
-              return false;
-            } else {
-              SpinMutexLock lock(*(reinterpret_cast<SpinMutex *>(value)));
-              for (unsigned int k = 0; k < num_handles_; ++k) {
-                if (argument_ids[k] != kInvalidAttributeID) {
-                  handles_[k]->updateStateUnary(
-                      accessor->getTypedValue(argument_ids[k]),
-                      value + payload_offsets_[k]);
-                } else {
-                  handles_[k]->updateStateNullary(value +
-                                                  payload_offsets_[k]);
-                }
-              }
-            }
-          }
-        }
-
-        return true;
-      });
+  LOG(FATAL) << "Not implemented";
+  return true;
+//  DEBUG_ASSERT(!allow_duplicate_keys);
+//  std::size_t variable_size;
+//  return InvokeOnAnyValueAccessor(
+//      accessor,
+//      [&](auto *accessor) -> bool {  // NOLINT(build/c++11)
+//        if (resizable) {
+//          bool continuing = true;
+//          while (continuing) {
+//            {
+//              continuing = false;
+//              SpinSharedMutexSharedLock<true> lock(resize_shared_mutex_);
+//              while (accessor->next()) {
+//                TypedValue key = accessor->getTypedValue(key_attr_id);
+//                if (check_for_null_keys && key.isNull()) {
+//                  continue;
+//                }
+//                variable_size = (force_key_copy && !scalar_key_inline_)
+//                                    ? key.getDataSize()
+//                                    : 0;
+//                std::uint8_t *value =
+//                    this->upsertInternalFast(key, variable_size, nullptr);
+//                if (value == nullptr) {
+//                  continuing = true;
+//                  break;
+//                } else {
+//                  SpinMutexLock lock(*(reinterpret_cast<SpinMutex *>(value)));
+//                  for (unsigned int k = 0; k < num_handles_; ++k) {
+//                    if (argument_ids[k] != kInvalidAttributeID) {
+//                      handles_[k]->updateStateUnary(
+//                          accessor->getTypedValue(argument_ids[k]),
+//                          value + payload_offsets_[k]);
+//                    } else {
+//                      handles_[k]->updateStateNullary(value +
+//                                                      payload_offsets_[k]);
+//                    }
+//                  }
+//                }
+//              }
+//            }
+//            if (continuing) {
+//              this->resize(0, variable_size);
+//              accessor->previous();
+//            }
+//          }
+//        } else {
+//          while (accessor->next()) {
+//            TypedValue key = accessor->getTypedValue(key_attr_id);
+//            if (check_for_null_keys && key.isNull()) {
+//              continue;
+//            }
+//            variable_size =
+//                (force_key_copy && !scalar_key_inline_) ? key.getDataSize() : 0;
+//            std::uint8_t *value =
+//                this->upsertInternalFast(key, variable_size, nullptr);
+//            if (value == nullptr) {
+//              return false;
+//            } else {
+//              SpinMutexLock lock(*(reinterpret_cast<SpinMutex *>(value)));
+//              for (unsigned int k = 0; k < num_handles_; ++k) {
+//                if (argument_ids[k] != kInvalidAttributeID) {
+//                  handles_[k]->updateStateUnary(
+//                      accessor->getTypedValue(argument_ids[k]),
+//                      value + payload_offsets_[k]);
+//                } else {
+//                  handles_[k]->updateStateNullary(value +
+//                                                  payload_offsets_[k]);
+//                }
+//              }
+//            }
+//          }
+//        }
+//
+//        return true;
+//      });
 }
 
 template <bool resizable,
@@ -1868,90 +1780,70 @@ bool FastHashTable<resizable,
                    force_key_copy,
                    allow_duplicate_keys>::
     upsertValueAccessorCompositeKeyFast(
-        const std::vector<attribute_id> &argument_ids,
         ValueAccessor *accessor,
+        ColumnVectorsValueAccessor *aux_accessor,
+        const std::vector<std::vector<attribute_id>> &argument_ids,
         const std::vector<attribute_id> &key_attr_ids,
         const bool check_for_null_keys) {
+  DEBUG_ASSERT(resizable);
   DEBUG_ASSERT(!allow_duplicate_keys);
+
   std::size_t variable_size;
   std::vector<TypedValue> key_vector;
   key_vector.resize(key_attr_ids.size());
   return InvokeOnAnyValueAccessor(
       accessor,
       [&](auto *accessor) -> bool {  // NOLINT(build/c++11)
-        if (resizable) {
-          bool continuing = true;
-          while (continuing) {
-            {
-              continuing = false;
-              SpinSharedMutexSharedLock<true> lock(resize_shared_mutex_);
-              while (accessor->next()) {
-                if (this->GetCompositeKeyFromValueAccessor(*accessor,
-                                                           key_attr_ids,
-                                                           check_for_null_keys,
-                                                           &key_vector)) {
-                  continue;
-                }
-                variable_size =
-                    this->calculateVariableLengthCompositeKeyCopySize(
-                        key_vector);
-                std::uint8_t *value = this->upsertCompositeKeyInternalFast(
-                    key_vector, nullptr, variable_size);
-                if (value == nullptr) {
-                  continuing = true;
-                  break;
-                } else {
-                  SpinMutexLock lock(*(reinterpret_cast<SpinMutex *>(value)));
-                  for (unsigned int k = 0; k < num_handles_; ++k) {
-                    if (argument_ids[k] != kInvalidAttributeID) {
-                      handles_[k]->updateStateUnary(
-                          accessor->getTypedValue(argument_ids[k]),
-                          value + payload_offsets_[k]);
-                    } else {
-                      handles_[k]->updateStateNullary(value +
-                                                      payload_offsets_[k]);
-                    }
-                  }
-                }
-              }
-            }
-            if (continuing) {
-              this->resize(0, variable_size);
-              accessor->previous();
-            }
+    bool continuing = true;
+    while (continuing) {
+      {
+        continuing = false;
+        SpinSharedMutexSharedLock<true> lock(resize_shared_mutex_);
+        while (accessor->next()) {
+          aux_accessor->next();
+          // TODO(jianqiao): templatize to involve aux_accessor
+          if (this->GetCompositeKeyFromValueAccessor(*accessor,
+                                                     key_attr_ids,
+                                                     check_for_null_keys,
+                                                     &key_vector)) {
+            continue;
           }
-        } else {
-          while (accessor->next()) {
-            if (this->GetCompositeKeyFromValueAccessor(*accessor,
-                                                       key_attr_ids,
-                                                       check_for_null_keys,
-                                                       &key_vector)) {
-              continue;
-            }
-            variable_size =
-                this->calculateVariableLengthCompositeKeyCopySize(key_vector);
-            std::uint8_t *value = this->upsertCompositeKeyInternalFast(
-                key_vector, nullptr, variable_size);
-            if (value == nullptr) {
-              return false;
-            } else {
-              SpinMutexLock lock(*(reinterpret_cast<SpinMutex *>(value)));
-              for (unsigned int k = 0; k < num_handles_; ++k) {
-                if (argument_ids[k] != kInvalidAttributeID) {
+          variable_size = this->calculateVariableLengthCompositeKeyCopySize(key_vector);
+          std::uint8_t *value = this->upsertCompositeKeyInternalFast(
+              key_vector, nullptr, variable_size);
+          if (value == nullptr) {
+            continuing = true;
+            break;
+          } else {
+            SpinMutexLock lock(*(reinterpret_cast<SpinMutex *>(value)));
+            for (unsigned int k = 0; k < num_handles_; ++k) {
+              const auto &ids = argument_ids[k];
+              if (ids.empty()) {
+                handles_[k]->updateStateNullary(value +
+                                                payload_offsets_[k]);
+              } else {
+                const attribute_id argument_id = ids.front();
+                if (argument_id >= 0) {
                   handles_[k]->updateStateUnary(
-                      accessor->getTypedValue(argument_ids[k]),
+                      accessor->getTypedValue(argument_id),
                       value + payload_offsets_[k]);
                 } else {
-                  handles_[k]->updateStateNullary(value +
-                                                  payload_offsets_[k]);
+                  handles_[k]->updateStateUnary(
+                      aux_accessor->getTypedValue(-(argument_id+2)),
+                      value + payload_offsets_[k]);
                 }
               }
             }
           }
         }
-
-        return true;
-      });
+      }
+      if (continuing) {
+        this->resize(0, variable_size);
+        accessor->previous();
+      }
+    }
+    return true;
+  });
 }
 
 template <bool resizable,

http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/c1baa015/storage/HashTableBase.hpp
----------------------------------------------------------------------
diff --git a/storage/HashTableBase.hpp b/storage/HashTableBase.hpp
index a3180bb..e3b8dbd 100644
--- a/storage/HashTableBase.hpp
+++ b/storage/HashTableBase.hpp
@@ -23,11 +23,13 @@
 #include <cstddef>
 #include <vector>
 
-#include "ValueAccessor.hpp"
 #include "utility/Macros.hpp"
 
 namespace quickstep {
 
+class ColumnVectorsValueAccessor;
+class ValueAccessor;
+
 /** \addtogroup Storage
  *  @{
  */
@@ -92,8 +94,9 @@ class HashTableBase {
    * specialization from this file.
    **/
   virtual bool upsertValueAccessorCompositeKeyFast(
-      const std::vector<attribute_id> &argument,
       ValueAccessor *accessor,
+      ColumnVectorsValueAccessor *aux_accessor,
+      const std::vector<std::vector<attribute_id>> &argument_ids,
       const std::vector<attribute_id> &key_attr_ids,
       const bool check_for_null_keys) {
     return false;


Mime
View raw message