Keep cache of column types

Author: alexowens90

cpp/arcticdb/entity/stream_descriptor.hpp

@@ -14,6 +14,8 @@
 #include <arcticdb/util/variant.hpp>
 #include <arcticdb/entity/types_proto.hpp>
 
+#include <ankerl/unordered_dense.h>
+
 namespace arcticdb::entity {
 
 struct SegmentDescriptorImpl : public SegmentDescriptor {
@@ -274,6 +276,28 @@ struct StreamDescriptor {
 struct OutputSchema {
     StreamDescriptor stream_descriptor_;
     arcticdb::proto::descriptors::NormalizationMetadata norm_metadata_;
+
+    OutputSchema(StreamDescriptor stream_descriptor,
+                 arcticdb::proto::descriptors::NormalizationMetadata norm_metadata):
+            stream_descriptor_(std::move(stream_descriptor)),
+            norm_metadata_(std::move(norm_metadata)) {};
+
+    ankerl::unordered_dense::map<std::string, DataType>& column_types() {
+        if (!column_types_.has_value()) {
+            column_types_ = ankerl::unordered_dense::map<std::string, DataType>();
+            column_types_->reserve(stream_descriptor_.field_count());
+            for (const auto& field: stream_descriptor_.fields()) {
+                column_types_->emplace(field.name(), field.type().data_type());
+            }
+        }
+        return *column_types_;
+    }
+
+    void clear_column_types() {
+        column_types_ = std::nullopt;
+    }
+private:
+    std::optional<ankerl::unordered_dense::map<std::string, DataType>> column_types_;
 };
 
 template <class IndexType>

cpp/arcticdb/processing/clause.cpp

@@ -134,21 +134,14 @@ std::vector<EntityId> FilterClause::process(std::vector<EntityId>&& entity_ids)
 }
 
 OutputSchema FilterClause::modify_schema(OutputSchema&& output_schema) const {
-    // TODO: Factor out checking against clause_info_.input_columns_ into separate function and call from all clauses
-    // TODO: Consider adding (optional, lazily constructed?) unordered map from column names to data types in
-    //  output_schema to make this sort of operation more efficient
+    const auto& column_types = output_schema.column_types();
     for (const auto& input_column: *clause_info_.input_columns_) {
         schema::check<ErrorCode::E_COLUMN_DOESNT_EXIST>(
-                output_schema.stream_descriptor_.find_field(input_column).has_value(),
+                column_types.contains(input_column),
                 "FilterClause requires column '{}' to exist in input data",
                 input_column
                 );
     }
-    // TODO: Factor this out with same code in ProjectClause
-    std::unordered_map<std::string, DataType> column_types;
-    for (const auto& field: output_schema.stream_descriptor_.fields()) {
-        column_types.emplace(field.name(), field.type().data_type());
-    }
     auto expr = expression_context_->expression_nodes_.get_value(expression_context_->root_node_name_.value);
     auto opt_datatype = expr->compute(*expression_context_, column_types);
     user_input::check<ErrorCode::E_INVALID_USER_ARGUMENT>(!opt_datatype.has_value(), "FilterClause AST produces a column, not a bitset");
@@ -193,24 +186,19 @@ std::vector<EntityId> ProjectClause::process(std::vector<EntityId>&& entity_ids)
 }
 
 OutputSchema ProjectClause::modify_schema(OutputSchema&& output_schema) const {
-    // TODO: Factor out checking against clause_info_.input_columns_ into separate function and call from all clauses
-    // TODO: Consider adding (optional, lazily constructed?) unordered map from column names to data types in
-    //  output_schema to make this sort of operation more efficient
+    auto& column_types = output_schema.column_types();
     for (const auto& input_column: *clause_info_.input_columns_) {
         schema::check<ErrorCode::E_COLUMN_DOESNT_EXIST>(
-                output_schema.stream_descriptor_.find_field(input_column).has_value(),
-                "ProjectClause requires column '{}' to exist in input data",
+                column_types.contains(input_column),
+                "FilterClause requires column '{}' to exist in input data",
                 input_column
         );
     }
-    std::unordered_map<std::string, DataType> column_types;
-    for (const auto& field: output_schema.stream_descriptor_.fields()) {
-        column_types.emplace(field.name(), field.type().data_type());
-    }
     auto expr = expression_context_->expression_nodes_.get_value(expression_context_->root_node_name_.value);
     auto opt_datatype = expr->compute(*expression_context_, column_types);
     user_input::check<ErrorCode::E_INVALID_USER_ARGUMENT>(opt_datatype.has_value(), "ProjectClause AST produces a bitset, not a column");
     output_schema.stream_descriptor_.add_scalar_field(*opt_datatype, output_column_);
+    column_types.emplace(output_column_, *opt_datatype);
     return output_schema;
 }
 
@@ -477,12 +465,10 @@ std::vector<EntityId> AggregationClause::process(std::vector<EntityId>&& entity_
 }
 
 OutputSchema AggregationClause::modify_schema(OutputSchema&& output_schema) const {
-    // TODO: Factor out checking against clause_info_.input_columns_ into separate function and call from all clauses
-    // TODO: Consider adding (optional, lazily constructed?) unordered map from column names to data types in
-    //  output_schema to make this sort of operation more efficient
+    const auto& column_types = output_schema.column_types();
     for (const auto& input_column: *clause_info_.input_columns_) {
         schema::check<ErrorCode::E_COLUMN_DOESNT_EXIST>(
-                output_schema.stream_descriptor_.find_field(input_column).has_value(),
+                column_types.contains(input_column),
                 "AggregationClause requires column '{}' to exist in input data",
                 input_column
         );
@@ -491,7 +477,6 @@ OutputSchema AggregationClause::modify_schema(OutputSchema&& output_schema) cons
     stream_desc.add_field(output_schema.stream_descriptor_.field(*output_schema.stream_descriptor_.find_field(grouping_column_)));
     stream_desc.set_index({0, IndexDescriptorImpl::Type::ROWCOUNT});
 
-    // TODO: Similar to process method, consider refactoring
     for (const auto& agg: aggregators_){
         const auto& input_column_name = agg.get_input_column_name().value;
         const auto& output_column_name = agg.get_output_column_name().value;
@@ -503,6 +488,7 @@ OutputSchema AggregationClause::modify_schema(OutputSchema&& output_schema) cons
     }
 
     output_schema.stream_descriptor_ = std::move(stream_desc);
+    output_schema.clear_column_types();
     auto mutable_index = output_schema.norm_metadata_.mutable_df()->mutable_common()->mutable_index();
     mutable_index->set_name(grouping_column_);
     mutable_index->clear_fake_name();
@@ -549,12 +535,10 @@ OutputSchema ResampleClause<closed_boundary>::modify_schema(OutputSchema&& outpu
             output_schema.stream_descriptor_.field(0).type() == make_scalar_type(DataType::NANOSECONDS_UTC64),
             "ResampleClause can only be applied to timeseries"
     );
-    // TODO: Factor out checking against clause_info_.input_columns_ into separate function and call from all clauses
-    // TODO: Consider adding (optional, lazily constructed?) unordered map from column names to data types in
-    //  output_schema to make this sort of operation more efficient
+    const auto& column_types = output_schema.column_types();
     for (const auto& input_column: *clause_info_.input_columns_) {
         schema::check<ErrorCode::E_COLUMN_DOESNT_EXIST>(
-                output_schema.stream_descriptor_.find_field(input_column).has_value(),
+                column_types.contains(input_column),
                 "ResampleClause requires column '{}' to exist in input data",
                 input_column
         );
@@ -572,6 +556,7 @@ OutputSchema ResampleClause<closed_boundary>::modify_schema(OutputSchema&& outpu
         stream_desc.add_scalar_field(output_column_type, output_column_name);
     }
     output_schema.stream_descriptor_ = std::move(stream_desc);
+    output_schema.clear_column_types();
 
     if (output_schema.norm_metadata_.df().common().has_multi_index()) {
         const auto& multi_index = output_schema.norm_metadata_.mutable_df()->mutable_common()->multi_index();

cpp/arcticdb/processing/expression_node.cpp

@@ -71,7 +71,7 @@ VariantData ExpressionNode::compute(ProcessingUnit& seg) const {
 }
 
 std::optional<DataType> ExpressionNode::compute(ExpressionContext& expression_context,
-                                                const std::unordered_map<std::string, DataType>& column_types) const {
+                                                const ankerl::unordered_dense::map<std::string, DataType>& column_types) const {
     std::optional<DataType> res;
     std::optional<DataType> left_type = util::variant_match(
             left_,

cpp/arcticdb/processing/expression_node.hpp

@@ -92,7 +92,7 @@ struct ExpressionNode {
     // TODO: better return type
     // TODO: Make expression_context const
     std::optional<DataType> compute(ExpressionContext& expression_context,
-                                    const std::unordered_map<std::string, DataType>& column_types) const;
+                                    const ankerl::unordered_dense::map<std::string, DataType>& column_types) const;
 };
 
 } //namespace arcticdb