storage/CollisionFreeVectorTable.cpp - incubator-retired-quickstep - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  **/

 #include "storage/CollisionFreeVectorTable.hpp"

 #include <atomic>
 #include <cstddef>
 #include <cstdint>
 #include <cstdlib>
 #include <memory>
 #include <vector>

 #include "expressions/aggregation/AggregationHandle.hpp"
 #include "expressions/aggregation/AggregationID.hpp"
 #include "storage/StorageBlockInfo.hpp"
 #include "storage/StorageManager.hpp"
 #include "storage/ValueAccessor.hpp"
 #include "storage/ValueAccessorMultiplexer.hpp"
 #include "storage/ValueAccessorUtil.hpp"
 #include "types/containers/ColumnVectorsValueAccessor.hpp"
 #include "utility/BarrieredReadWriteConcurrentBitVector.hpp"

 #include "glog/logging.h"

 namespace quickstep {

 CollisionFreeVectorTable::CollisionFreeVectorTable(
     const Type *key_type,
     const std::size_t num_entries,
     const std::size_t memory_size,
     const std::size_t num_init_partitions,
     const std::size_t num_finalize_partitions,
     const std::vector<std::size_t> &state_offsets,
     const std::vector<AggregationHandle *> &handles,
     StorageManager *storage_manager)
     : key_type_(key_type),
       num_entries_(num_entries),
       num_handles_(handles.size()),
       handles_(handles),
       memory_size_(memory_size),
       num_init_partitions_(num_init_partitions),
       num_finalize_partitions_(num_finalize_partitions),
       storage_manager_(storage_manager) {
   DCHECK_GT(num_entries, 0u);
   DCHECK_GT(num_finalize_partitions_, 0u);
   DCHECK_EQ(num_handles_, state_offsets.size());

   const std::size_t num_storage_slots =
       storage_manager_->SlotsNeededForBytes(memory_size_);

   const block_id blob_id = storage_manager_->createBlob(num_storage_slots);
   blob_ = storage_manager_->getBlobMutable(blob_id);

   void *memory_start = blob_->getMemoryMutable();
   existence_map_.reset(new BarrieredReadWriteConcurrentBitVector(
       reinterpret_cast<char *>(memory_start),
       num_entries,
       false /* initialize */));

   for (std::size_t i = 0; i < num_handles_; ++i) {
     // Columnwise layout.
     vec_tables_.emplace_back(
         reinterpret_cast<char *>(memory_start) + state_offsets.at(i));
   }
 }

 CollisionFreeVectorTable::~CollisionFreeVectorTable() {
   const block_id blob_id = blob_->getID();
   blob_.release();
   storage_manager_->deleteBlockOrBlobFile(blob_id);
 }

 void CollisionFreeVectorTable::destroyPayload() {
 }

 bool CollisionFreeVectorTable::upsertValueAccessorCompositeKey(
     const std::vector<std::vector<MultiSourceAttributeId>> &argument_ids,
     const std::vector<MultiSourceAttributeId> &key_ids,
     const ValueAccessorMultiplexer &accessor_mux) {
   DCHECK_EQ(1u, key_ids.size());

   if (handles_.empty()) {
     InvokeOnValueAccessorMaybeTupleIdSequenceAdapter(
         accessor_mux.getValueAccessorBySource(key_ids.front().source),
         [&key_ids, this](auto *accessor) -> void {  // NOLINT(build/c++11)
       this->upsertValueAccessorKeyOnlyHelper(key_type_->isNullable(),
                                              key_type_,
                                              key_ids.front().attr_id,
                                              accessor);
     });
     return true;
   }

   DCHECK(accessor_mux.getDerivedAccessor() == nullptr ||
          accessor_mux.getDerivedAccessor()->getImplementationType()
              == ValueAccessor::Implementation::kColumnVectors);

   ValueAccessor *base_accessor = accessor_mux.getBaseAccessor();
   ColumnVectorsValueAccessor *derived_accesor =
       static_cast<ColumnVectorsValueAccessor *>(accessor_mux.getDerivedAccessor());

   // Dispatch to specialized implementations to achieve maximum performance.
   InvokeOnValueAccessorMaybeTupleIdSequenceAdapter(
       base_accessor,
       [&argument_ids, &key_ids, &derived_accesor, this](auto *accessor) -> void {  // NOLINT(build/c++11)
     const ValueAccessorSource key_source = key_ids.front().source;
     const attribute_id key_id = key_ids.front().attr_id;
     const bool is_key_nullable = key_type_->isNullable();

     for (std::size_t i = 0; i < num_handles_; ++i) {
       DCHECK_LE(argument_ids[i].size(), 1u);

       const AggregationHandle *handle = handles_[i];
       const auto &argument_types = handle->getArgumentTypes();
       const auto &argument_ids_i = argument_ids[i];

       ValueAccessorSource argument_source;
       attribute_id argument_id;
       const Type *argument_type;
       bool is_argument_nullable;

       if (argument_ids_i.empty()) {
         argument_source = ValueAccessorSource::kInvalid;
         argument_id = kInvalidAttributeID;

         DCHECK(argument_types.empty());
         argument_type = nullptr;
         is_argument_nullable = false;
       } else {
         DCHECK_EQ(1u, argument_ids_i.size());
         argument_source = argument_ids_i.front().source;
         argument_id = argument_ids_i.front().attr_id;

         DCHECK_EQ(1u, argument_types.size());
         argument_type = argument_types.front();
         is_argument_nullable = argument_type->isNullable();
       }

       if (key_source == ValueAccessorSource::kBase) {
         if (argument_source == ValueAccessorSource::kBase) {
           this->upsertValueAccessorDispatchHelper<false>(is_key_nullable,
                                                          is_argument_nullable,
                                                          key_type_,
                                                          argument_type,
                                                          handle->getAggregationID(),
                                                          key_id,
                                                          argument_id,
                                                          vec_tables_[i],
                                                          accessor,
                                                          accessor);
         } else {
           this->upsertValueAccessorDispatchHelper<true>(is_key_nullable,
                                                         is_argument_nullable,
                                                         key_type_,
                                                         argument_type,
                                                         handle->getAggregationID(),
                                                         key_id,
                                                         argument_id,
                                                         vec_tables_[i],
                                                         accessor,
                                                         derived_accesor);
         }
       } else {
         if (argument_source == ValueAccessorSource::kBase) {
           this->upsertValueAccessorDispatchHelper<true>(is_key_nullable,
                                                         is_argument_nullable,
                                                         key_type_,
                                                         argument_type,
                                                         handle->getAggregationID(),
                                                         key_id,
                                                         argument_id,
                                                         vec_tables_[i],
                                                         derived_accesor,
                                                         accessor);
         } else {
           this->upsertValueAccessorDispatchHelper<false>(is_key_nullable,
                                                          is_argument_nullable,
                                                          key_type_,
                                                          argument_type,
                                                          handle->getAggregationID(),
                                                          key_id,
                                                          argument_id,
                                                          vec_tables_[i],
                                                          derived_accesor,
                                                          derived_accesor);
         }
       }
     }
   });
   return true;
 }

 void CollisionFreeVectorTable::finalizeKey(const std::size_t partition_id,
                                            NativeColumnVector *output_cv) const {
   const std::size_t start_position =
       calculatePartitionStartPosition(partition_id);
   const std::size_t end_position =
       calculatePartitionEndPosition(partition_id);

   switch (key_type_->getTypeID()) {
     case TypeID::kInt:
       finalizeKeyInternal<int>(start_position, end_position, output_cv);
       return;
     case TypeID::kLong:
       finalizeKeyInternal<std::int64_t>(start_position, end_position, output_cv);
       return;
     default:
       LOG(FATAL) << "Not supported";
   }
 }

 void CollisionFreeVectorTable::finalizeState(const std::size_t partition_id,
                                              const std::size_t handle_id,
                                              NativeColumnVector *output_cv) const {
   const std::size_t start_position =
       calculatePartitionStartPosition(partition_id);
   const std::size_t end_position =
       calculatePartitionEndPosition(partition_id);

   const AggregationHandle *handle = handles_[handle_id];
   const auto &argument_types = handle->getArgumentTypes();
   const Type *argument_type =
       argument_types.empty() ? nullptr : argument_types.front();

   finalizeStateDispatchHelper(handle->getAggregationID(),
                               argument_type,
                               vec_tables_[handle_id],
                               start_position,
                               end_position,
                               output_cv);
 }

 }  // namespace quickstep
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	**/

	#include "storage/CollisionFreeVectorTable.hpp"

	#include <atomic>
	#include <cstddef>
	#include <cstdint>
	#include <cstdlib>
	#include <memory>
	#include <vector>

	#include "expressions/aggregation/AggregationHandle.hpp"
	#include "expressions/aggregation/AggregationID.hpp"
	#include "storage/StorageBlockInfo.hpp"
	#include "storage/StorageManager.hpp"
	#include "storage/ValueAccessor.hpp"
	#include "storage/ValueAccessorMultiplexer.hpp"
	#include "storage/ValueAccessorUtil.hpp"
	#include "types/containers/ColumnVectorsValueAccessor.hpp"
	#include "utility/BarrieredReadWriteConcurrentBitVector.hpp"

	#include "glog/logging.h"

	namespace quickstep {

	CollisionFreeVectorTable::CollisionFreeVectorTable(
	const Type *key_type,
	const std::size_t num_entries,
	const std::size_t memory_size,
	const std::size_t num_init_partitions,
	const std::size_t num_finalize_partitions,
	const std::vector<std::size_t> &state_offsets,
	const std::vector<AggregationHandle *> &handles,
	StorageManager *storage_manager)
	: key_type_(key_type),
	num_entries_(num_entries),
	num_handles_(handles.size()),
	handles_(handles),
	memory_size_(memory_size),
	num_init_partitions_(num_init_partitions),
	num_finalize_partitions_(num_finalize_partitions),
	storage_manager_(storage_manager) {
	DCHECK_GT(num_entries, 0u);
	DCHECK_GT(num_finalize_partitions_, 0u);
	DCHECK_EQ(num_handles_, state_offsets.size());

	const std::size_t num_storage_slots =
	storage_manager_->SlotsNeededForBytes(memory_size_);

	const block_id blob_id = storage_manager_->createBlob(num_storage_slots);
	blob_ = storage_manager_->getBlobMutable(blob_id);

	void *memory_start = blob_->getMemoryMutable();
	existence_map_.reset(new BarrieredReadWriteConcurrentBitVector(
	reinterpret_cast<char *>(memory_start),
	num_entries,
	false /* initialize */));

	for (std::size_t i = 0; i < num_handles_; ++i) {
	// Columnwise layout.
	vec_tables_.emplace_back(
	reinterpret_cast<char *>(memory_start) + state_offsets.at(i));
	}
	}

	CollisionFreeVectorTable::~CollisionFreeVectorTable() {
	const block_id blob_id = blob_->getID();
	blob_.release();
	storage_manager_->deleteBlockOrBlobFile(blob_id);
	}

	void CollisionFreeVectorTable::destroyPayload() {
	}

	bool CollisionFreeVectorTable::upsertValueAccessorCompositeKey(
	const std::vector<std::vector<MultiSourceAttributeId>> &argument_ids,
	const std::vector<MultiSourceAttributeId> &key_ids,
	const ValueAccessorMultiplexer &accessor_mux) {
	DCHECK_EQ(1u, key_ids.size());

	if (handles_.empty()) {
	InvokeOnValueAccessorMaybeTupleIdSequenceAdapter(
	accessor_mux.getValueAccessorBySource(key_ids.front().source),
	[&key_ids, this](auto *accessor) -> void { // NOLINT(build/c++11)
	this->upsertValueAccessorKeyOnlyHelper(key_type_->isNullable(),
	key_type_,
	key_ids.front().attr_id,
	accessor);
	});
	return true;
	}

	DCHECK(accessor_mux.getDerivedAccessor() == nullptr \|\|
	accessor_mux.getDerivedAccessor()->getImplementationType()
	== ValueAccessor::Implementation::kColumnVectors);

	ValueAccessor *base_accessor = accessor_mux.getBaseAccessor();
	ColumnVectorsValueAccessor *derived_accesor =
	static_cast<ColumnVectorsValueAccessor *>(accessor_mux.getDerivedAccessor());

	// Dispatch to specialized implementations to achieve maximum performance.
	InvokeOnValueAccessorMaybeTupleIdSequenceAdapter(
	base_accessor,
	[&argument_ids, &key_ids, &derived_accesor, this](auto *accessor) -> void { // NOLINT(build/c++11)
	const ValueAccessorSource key_source = key_ids.front().source;
	const attribute_id key_id = key_ids.front().attr_id;
	const bool is_key_nullable = key_type_->isNullable();

	for (std::size_t i = 0; i < num_handles_; ++i) {
	DCHECK_LE(argument_ids[i].size(), 1u);

	const AggregationHandle *handle = handles_[i];
	const auto &argument_types = handle->getArgumentTypes();
	const auto &argument_ids_i = argument_ids[i];

	ValueAccessorSource argument_source;
	attribute_id argument_id;
	const Type *argument_type;
	bool is_argument_nullable;

	if (argument_ids_i.empty()) {
	argument_source = ValueAccessorSource::kInvalid;
	argument_id = kInvalidAttributeID;

	DCHECK(argument_types.empty());
	argument_type = nullptr;
	is_argument_nullable = false;
	} else {
	DCHECK_EQ(1u, argument_ids_i.size());
	argument_source = argument_ids_i.front().source;
	argument_id = argument_ids_i.front().attr_id;

	DCHECK_EQ(1u, argument_types.size());
	argument_type = argument_types.front();
	is_argument_nullable = argument_type->isNullable();
	}

	if (key_source == ValueAccessorSource::kBase) {
	if (argument_source == ValueAccessorSource::kBase) {
	this->upsertValueAccessorDispatchHelper<false>(is_key_nullable,
	is_argument_nullable,
	key_type_,
	argument_type,
	handle->getAggregationID(),
	key_id,
	argument_id,
	vec_tables_[i],
	accessor,
	accessor);
	} else {
	this->upsertValueAccessorDispatchHelper<true>(is_key_nullable,
	is_argument_nullable,
	key_type_,
	argument_type,
	handle->getAggregationID(),
	key_id,
	argument_id,
	vec_tables_[i],
	accessor,
	derived_accesor);
	}
	} else {
	if (argument_source == ValueAccessorSource::kBase) {
	this->upsertValueAccessorDispatchHelper<true>(is_key_nullable,
	is_argument_nullable,
	key_type_,
	argument_type,
	handle->getAggregationID(),
	key_id,
	argument_id,
	vec_tables_[i],
	derived_accesor,
	accessor);
	} else {
	this->upsertValueAccessorDispatchHelper<false>(is_key_nullable,
	is_argument_nullable,
	key_type_,
	argument_type,
	handle->getAggregationID(),
	key_id,
	argument_id,
	vec_tables_[i],
	derived_accesor,
	derived_accesor);
	}
	}
	}
	});
	return true;
	}

	void CollisionFreeVectorTable::finalizeKey(const std::size_t partition_id,
	NativeColumnVector *output_cv) const {
	const std::size_t start_position =
	calculatePartitionStartPosition(partition_id);
	const std::size_t end_position =
	calculatePartitionEndPosition(partition_id);

	switch (key_type_->getTypeID()) {
	case TypeID::kInt:
	finalizeKeyInternal<int>(start_position, end_position, output_cv);
	return;
	case TypeID::kLong:
	finalizeKeyInternal<std::int64_t>(start_position, end_position, output_cv);
	return;
	default:
	LOG(FATAL) << "Not supported";
	}
	}

	void CollisionFreeVectorTable::finalizeState(const std::size_t partition_id,
	const std::size_t handle_id,
	NativeColumnVector *output_cv) const {
	const std::size_t start_position =
	calculatePartitionStartPosition(partition_id);
	const std::size_t end_position =
	calculatePartitionEndPosition(partition_id);

	const AggregationHandle *handle = handles_[handle_id];
	const auto &argument_types = handle->getArgumentTypes();
	const Type *argument_type =
	argument_types.empty() ? nullptr : argument_types.front();

	finalizeStateDispatchHelper(handle->getAggregationID(),
	argument_type,
	vec_tables_[handle_id],
	start_position,
	end_position,
	output_cv);
	}

	} // namespace quickstep