| /** |
| * 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/WindowAggregationOperationState.hpp" |
| |
| #include <cstddef> |
| #include <cstdio> |
| #include <memory> |
| #include <string> |
| #include <utility> |
| #include <vector> |
| |
| #include "catalog/CatalogDatabaseLite.hpp" |
| #include "catalog/CatalogRelationSchema.hpp" |
| #include "catalog/CatalogTypedefs.hpp" |
| #include "expressions/ExpressionFactories.hpp" |
| #include "expressions/Expressions.pb.h" |
| #include "expressions/scalar/Scalar.hpp" |
| #include "expressions/scalar/ScalarAttribute.hpp" |
| #include "expressions/window_aggregation/WindowAggregateFunction.hpp" |
| #include "expressions/window_aggregation/WindowAggregateFunctionFactory.hpp" |
| #include "expressions/window_aggregation/WindowAggregationHandle.hpp" |
| #include "expressions/window_aggregation/WindowAggregationID.hpp" |
| #include "storage/InsertDestination.hpp" |
| #include "storage/StorageManager.hpp" |
| #include "storage/SubBlocksReference.hpp" |
| #include "storage/ValueAccessor.hpp" |
| #include "storage/ValueAccessorUtil.hpp" |
| #include "storage/WindowAggregationOperationState.pb.h" |
| #include "types/containers/ColumnVector.hpp" |
| #include "types/containers/ColumnVectorsValueAccessor.hpp" |
| #include "types/containers/ColumnVectorUtil.hpp" |
| #include "utility/ColumnVectorCache.hpp" |
| |
| #include "glog/logging.h" |
| |
| namespace quickstep { |
| |
| WindowAggregationOperationState::WindowAggregationOperationState( |
| const CatalogRelationSchema &input_relation, |
| const WindowAggregateFunction *window_aggregate_function, |
| std::vector<std::unique_ptr<const Scalar>> &&arguments, |
| const std::vector<std::unique_ptr<const Scalar>> &partition_by_attributes, |
| const std::vector<std::unique_ptr<const Scalar>> &order_by_attributes, |
| const bool is_row, |
| const std::int64_t num_preceding, |
| const std::int64_t num_following, |
| StorageManager *storage_manager) |
| : input_relation_(input_relation), |
| arguments_(std::move(arguments)), |
| storage_manager_(storage_manager) { |
| // Get the Types of this window aggregate's arguments so that we can create an |
| // AggregationHandle. |
| std::vector<const Type*> argument_types; |
| for (const std::unique_ptr<const Scalar> &argument : arguments_) { |
| argument_types.emplace_back(&argument->getType()); |
| } |
| |
| // Check if window aggregate function could apply to the arguments. |
| DCHECK(window_aggregate_function->canApplyToTypes(argument_types)); |
| |
| // Create the handle and initial state. |
| window_aggregation_handle_.reset( |
| window_aggregate_function->createHandle(argument_types, |
| partition_by_attributes, |
| order_by_attributes, |
| is_row, |
| num_preceding, |
| num_following)); |
| } |
| |
| WindowAggregationOperationState* WindowAggregationOperationState::ReconstructFromProto( |
| const serialization::WindowAggregationOperationState &proto, |
| const CatalogDatabaseLite &database, |
| StorageManager *storage_manager) { |
| DCHECK(ProtoIsValid(proto, database)); |
| |
| std::vector<std::unique_ptr<const Scalar>> arguments; |
| arguments.reserve(proto.arguments_size()); |
| for (int argument_idx = 0; argument_idx < proto.arguments_size(); ++argument_idx) { |
| arguments.emplace_back(ScalarFactory::ReconstructFromProto( |
| proto.arguments(argument_idx), |
| database)); |
| } |
| |
| std::vector<std::unique_ptr<const Scalar>> partition_by_attributes; |
| for (int attribute_idx = 0; |
| attribute_idx < proto.partition_by_attributes_size(); |
| ++attribute_idx) { |
| partition_by_attributes.emplace_back(ScalarFactory::ReconstructFromProto( |
| proto.partition_by_attributes(attribute_idx), |
| database)); |
| } |
| |
| std::vector<std::unique_ptr<const Scalar>> order_by_attributes; |
| for (int attribute_idx = 0; |
| attribute_idx < proto.order_by_attributes_size(); |
| ++attribute_idx) { |
| order_by_attributes.emplace_back(ScalarFactory::ReconstructFromProto( |
| proto.order_by_attributes(attribute_idx), |
| database)); |
| } |
| |
| const bool is_row = proto.is_row(); |
| const std::int64_t num_preceding = proto.num_preceding(); |
| const std::int64_t num_following = proto.num_following(); |
| |
| return new WindowAggregationOperationState(database.getRelationSchemaById(proto.input_relation_id()), |
| &WindowAggregateFunctionFactory::ReconstructFromProto(proto.function()), |
| std::move(arguments), |
| partition_by_attributes, |
| order_by_attributes, |
| is_row, |
| num_preceding, |
| num_following, |
| storage_manager); |
| } |
| |
| bool WindowAggregationOperationState::ProtoIsValid(const serialization::WindowAggregationOperationState &proto, |
| const CatalogDatabaseLite &database) { |
| if (!proto.IsInitialized() || |
| !database.hasRelationWithId(proto.input_relation_id())) { |
| return false; |
| } |
| |
| if (!WindowAggregateFunctionFactory::ProtoIsValid(proto.function())) { |
| return false; |
| } |
| |
| // TODO(chasseur): We may also want to check that the specified |
| // AggregateFunction is applicable to the specified arguments, but that |
| // requires partial deserialization and may be too heavyweight for this |
| // method. |
| // TODO(Shixuan): The TODO for AggregateFunction could also be applied here. |
| for (int argument_idx = 0; |
| argument_idx < proto.arguments_size(); |
| ++argument_idx) { |
| if (!ScalarFactory::ProtoIsValid(proto.arguments(argument_idx), database)) { |
| return false; |
| } |
| } |
| |
| for (int attribute_idx = 0; |
| attribute_idx < proto.partition_by_attributes_size(); |
| ++attribute_idx) { |
| if (!ScalarFactory::ProtoIsValid(proto.partition_by_attributes(attribute_idx), |
| database)) { |
| return false; |
| } |
| } |
| |
| for (int attribute_idx = 0; |
| attribute_idx < proto.order_by_attributes_size(); |
| ++attribute_idx) { |
| if (!ScalarFactory::ProtoIsValid(proto.order_by_attributes(attribute_idx), |
| database)) { |
| return false; |
| } |
| } |
| |
| if (proto.num_preceding() < -1 || proto.num_following() < -1) { |
| return false; |
| } |
| |
| return true; |
| } |
| |
| void WindowAggregationOperationState::windowAggregateBlocks( |
| InsertDestination *output_destination, |
| const std::vector<block_id> &block_ids) { |
| // TODO(Shixuan): This is a quick fix for currently unsupported functions in |
| // order to pass the query_optimizer test. |
| if (window_aggregation_handle_.get() == nullptr) { |
| std::cout << "The function will be supported in the near future :)\n"; |
| return; |
| } |
| |
| // Get the total number of tuples. |
| int num_tuples = 0; |
| for (const block_id block_idx : block_ids) { |
| num_tuples += |
| storage_manager_->getBlock(block_idx, input_relation_)->getNumTuples(); |
| } |
| |
| // Construct column vectors for attributes. |
| std::vector<ColumnVector*> attribute_vecs; |
| for (std::size_t attr_id = 0; attr_id < input_relation_.size(); ++attr_id) { |
| const CatalogAttribute *attr = input_relation_.getAttributeById(attr_id); |
| const Type &type = attr->getType(); |
| |
| if (NativeColumnVector::UsableForType(type)) { |
| attribute_vecs.push_back(new NativeColumnVector(type, num_tuples)); |
| } else { |
| attribute_vecs.push_back(new IndirectColumnVector(type, num_tuples)); |
| } |
| } |
| |
| // Construct column vectors for arguments. |
| std::vector<ColumnVector*> argument_vecs; |
| for (const std::unique_ptr<const Scalar> &argument : arguments_) { |
| const Type &type = argument->getType(); |
| |
| if (NativeColumnVector::UsableForType(type)) { |
| argument_vecs.push_back(new NativeColumnVector(type, num_tuples)); |
| } else { |
| argument_vecs.push_back(new IndirectColumnVector(type, num_tuples)); |
| } |
| } |
| |
| // TODO(Shixuan): Add Support for Vector Copy Elision Selection. |
| // Add tuples and arguments into ColumnVectors. |
| for (const block_id block_idx : block_ids) { |
| BlockReference block = storage_manager_->getBlock(block_idx, input_relation_); |
| const TupleStorageSubBlock &tuple_block = block->getTupleStorageSubBlock(); |
| SubBlocksReference sub_block_ref(tuple_block, |
| block->getIndices(), |
| block->getIndicesConsistent()); |
| ValueAccessor *tuple_accessor = tuple_block.createValueAccessor(); |
| ColumnVectorsValueAccessor *argument_accessor = nullptr; |
| if (!arguments_.empty()) { |
| argument_accessor = new ColumnVectorsValueAccessor(); |
| } |
| |
| std::unique_ptr<ColumnVectorCache> cv_cache = std::make_unique<ColumnVectorCache>(); |
| for (const std::unique_ptr<const Scalar> &argument : arguments_) { |
| argument_accessor->addColumn(argument->getAllValues(tuple_accessor, |
| &sub_block_ref, |
| cv_cache.get())); |
| } |
| |
| // Release common subexpression cache as early as possible. |
| cv_cache.reset(); |
| |
| InvokeOnAnyValueAccessor(tuple_accessor, |
| [&] (auto *tuple_accessor) -> void { // NOLINT(build/c++11) |
| tuple_accessor->beginIteration(); |
| if (argument_accessor != nullptr) { |
| argument_accessor->beginIteration(); |
| } |
| |
| while (tuple_accessor->next()) { |
| if (argument_accessor != nullptr) { |
| argument_accessor->next(); |
| } |
| |
| for (std::size_t attr_id = 0; attr_id < attribute_vecs.size(); ++attr_id) { |
| ColumnVector *attr_vec = attribute_vecs[attr_id]; |
| if (attr_vec->isNative()) { |
| static_cast<NativeColumnVector*>(attr_vec)->appendTypedValue( |
| tuple_accessor->getTypedValue(attr_id)); |
| } else { |
| static_cast<IndirectColumnVector*>(attr_vec)->appendTypedValue( |
| tuple_accessor->getTypedValue(attr_id)); |
| } |
| } |
| |
| for (std::size_t argument_idx = 0; |
| argument_idx < argument_vecs.size(); |
| ++argument_idx) { |
| ColumnVector *argument = argument_vecs[argument_idx]; |
| if (argument->isNative()) { |
| static_cast<NativeColumnVector*>(argument)->appendTypedValue( |
| argument_accessor->getTypedValue(argument_idx)); |
| } else { |
| static_cast<IndirectColumnVector*>(argument)->appendTypedValue( |
| argument_accessor->getTypedValue(argument_idx)); |
| } |
| } |
| } |
| }); |
| } |
| |
| // Construct the value accessor for tuples in all blocks |
| ColumnVectorsValueAccessor *all_blocks_accessor |
| = new ColumnVectorsValueAccessor(); |
| for (ColumnVector *attr_vec : attribute_vecs) { |
| all_blocks_accessor->addColumn(attr_vec); |
| } |
| |
| // Do actual calculation in handle. |
| ColumnVector *window_aggregates = |
| window_aggregation_handle_->calculate(all_blocks_accessor, |
| argument_vecs); |
| |
| all_blocks_accessor->addColumn(window_aggregates); |
| output_destination->bulkInsertTuples(all_blocks_accessor); |
| } |
| |
| } // namespace quickstep |
