Make resampling work using dynamic schema

Author: vasil-pashov

cpp/arcticdb/processing/clause.cpp

@@ -671,8 +671,11 @@ std::vector<EntityId> ResampleClause<closed_boundary>::process(std::vector<Entit
                                 }
             );
         }
-        auto aggregated_column = std::make_shared<Column>(aggregator.aggregate(input_index_columns, input_agg_columns, bucket_boundaries, *output_index_column, string_pool));
-        seg.add_column(scalar_field(aggregated_column->type().data_type(), aggregator.get_output_column_name().value), aggregated_column);
+        std::optional<Column> aggregated_column = aggregator.aggregate(input_index_columns, input_agg_columns, bucket_boundaries, *output_index_column, string_pool);
+        if (aggregated_column) {
+            auto aggregated_column_ptr = std::make_shared<Column>(std::move(*aggregated_column));
+            seg.add_column(scalar_field(aggregated_column_ptr->type().data_type(), aggregator.get_output_column_name().value), std::move(aggregated_column_ptr));
+        }
     }
     seg.set_row_data(output_index_column->row_count() - 1);
     return push_entities(*component_manager_, ProcessingUnit(std::move(seg), std::move(output_row_range)));

cpp/arcticdb/processing/sorted_aggregation.cpp

@@ -12,14 +12,17 @@
 namespace arcticdb {
 
 template<AggregationOperator aggregation_operator, ResampleBoundary closed_boundary>
-Column SortedAggregator<aggregation_operator, closed_boundary>::aggregate(const std::vector<std::shared_ptr<Column>>& input_index_columns,
+std::optional<Column> SortedAggregator<aggregation_operator, closed_boundary>::aggregate(const std::vector<std::shared_ptr<Column>>& input_index_columns,
                                                                           const std::vector<std::optional<ColumnWithStrings>>& input_agg_columns,
                                                                           const std::vector<timestamp>& bucket_boundaries,
                                                                           const Column& output_index_column,
                                                                           StringPool& string_pool) const {
     using IndexTDT = ScalarTagType<DataTypeTag<DataType::NANOSECONDS_UTC64>>;
-    auto common_input_type = generate_common_input_type(input_agg_columns);
-    Column res(TypeDescriptor(generate_output_data_type(common_input_type), Dimension::Dim0), output_index_column.row_count(), AllocationType::PRESIZED, Sparsity::NOT_PERMITTED);
+    const std::optional<DataType> common_input_type = generate_common_input_type(input_agg_columns);
+    if (!common_input_type.has_value()) {
+        return std::nullopt;
+    }
+    Column res(TypeDescriptor(generate_output_data_type(*common_input_type), Dimension::Dim0), output_index_column.row_count(), AllocationType::PRESIZED, Sparsity::NOT_PERMITTED);
     details::visit_type(
         res.type().data_type(),
         [this,
@@ -128,11 +131,11 @@ Column SortedAggregator<aggregation_operator, closed_boundary>::aggregate(const
             }
         }
     );
-    return res;
+    return {std::move(res)};
 }
 
 template<AggregationOperator aggregation_operator, ResampleBoundary closed_boundary>
-DataType SortedAggregator<aggregation_operator, closed_boundary>::generate_common_input_type(
+std::optional<DataType> SortedAggregator<aggregation_operator, closed_boundary>::generate_common_input_type(
         const std::vector<std::optional<ColumnWithStrings>>& input_agg_columns
         ) const {
     std::optional<DataType> common_input_type;
@@ -141,17 +144,9 @@ DataType SortedAggregator<aggregation_operator, closed_boundary>::generate_commo
             auto input_data_type = opt_input_agg_column->column_->type().data_type();
             check_aggregator_supported_with_data_type(input_data_type);
             add_data_type_impl(input_data_type, common_input_type);
-        } else {
-            // Column is missing from this row-slice due to dynamic schema, currently unsupported
-            schema::raise<ErrorCode::E_UNSUPPORTED_COLUMN_TYPE>("Resample: Cannot aggregate column '{}' as it is missing from some row slices",
-                                                                get_input_column_name().value);
         }
     }
-    // Column is missing from all row-slices due to dynamic schema, currently unsupported
-    schema::check<ErrorCode::E_UNSUPPORTED_COLUMN_TYPE>(common_input_type.has_value(),
-                                                        "Resample: Cannot aggregate column '{}' as it is missing from some row slices",
-                                                        get_input_column_name().value);
-    return *common_input_type;
+    return common_input_type;
 }
 
 template<AggregationOperator aggregation_operator, ResampleBoundary closed_boundary>

cpp/arcticdb/processing/sorted_aggregation.hpp

@@ -28,7 +28,7 @@ struct ISortedAggregator {
     struct Interface : Base {
         [[nodiscard]] ColumnName get_input_column_name() const { return folly::poly_call<0>(*this); };
         [[nodiscard]] ColumnName get_output_column_name() const { return folly::poly_call<1>(*this); };
-        [[nodiscard]] Column aggregate(const std::vector<std::shared_ptr<Column>>& input_index_columns,
+        [[nodiscard]] std::optional<Column> aggregate(const std::vector<std::shared_ptr<Column>>& input_index_columns,
                                        const std::vector<std::optional<ColumnWithStrings>>& input_agg_columns,
                                        const std::vector<timestamp>& bucket_boundaries,
                                        const Column& output_index_column,
@@ -354,13 +354,13 @@ class SortedAggregator
     [[nodiscard]] ColumnName get_input_column_name() const { return input_column_name_; }
     [[nodiscard]] ColumnName get_output_column_name() const { return output_column_name_; }
 
-    [[nodiscard]] Column aggregate(const std::vector<std::shared_ptr<Column>>& input_index_columns,
+    [[nodiscard]] std::optional<Column> aggregate(const std::vector<std::shared_ptr<Column>>& input_index_columns,
                                    const std::vector<std::optional<ColumnWithStrings>>& input_agg_columns,
                                    const std::vector<timestamp>& bucket_boundaries,
                                    const Column& output_index_column,
                                    StringPool& string_pool) const;
 private:
-    [[nodiscard]] DataType generate_common_input_type(const std::vector<std::optional<ColumnWithStrings>>& input_agg_columns) const;
+    [[nodiscard]] std::optional<DataType> generate_common_input_type(const std::vector<std::optional<ColumnWithStrings>>& input_agg_columns) const;
     void check_aggregator_supported_with_data_type(DataType data_type) const;
     [[nodiscard]] DataType generate_output_data_type(DataType common_input_data_type) const;
     [[nodiscard]] bool index_value_past_end_of_bucket(timestamp index_value, timestamp bucket_end) const;

cpp/arcticdb/storage/s3/s3_client_wrapper.cpp

@@ -135,10 +135,7 @@ folly::Future<S3Result<std::monostate>> S3ClientTestWrapper::delete_object(
 
     return actual_client_->delete_object(s3_object_name, bucket_name);
 }
-
-// Using a fixed page size since it's only being used for simple tests.
-// If we ever need to configure it we should move it to the s3 proto config instead.
-constexpr auto page_size = 10;
+    
 S3Result<ListObjectsOutput> S3ClientTestWrapper::list_objects(
         const std::string& name_prefix,
         const std::string& bucket_name,