WIP: Join schema index construction

alexowens90 · alexowens90 · commit a4d4fe1adc13 · 2025-03-18T17:29:15.000Z
diff --git a/cpp/arcticdb/CMakeLists.txt b/cpp/arcticdb/CMakeLists.txt
@@ -910,92 +910,93 @@ if(${TEST})
     python_utils_dump_vars_if_enabled("Python for test compilation")
 
     set(unit_test_srcs
-            async/test/test_async.cpp
-            codec/test/test_codec.cpp
-            codec/test/test_encode_field_collection.cpp
-            codec/test/test_segment_header.cpp
-            codec/test/test_encoded_field.cpp
-            column_store/test/ingestion_stress_test.cpp
-            column_store/test/test_column.cpp
-            column_store/test/test_column_data_random_accessor.cpp
-            column_store/test/test_index_filtering.cpp
-            column_store/test/test_memory_segment.cpp
-            column_store/test/test_statistics.cpp
-            entity/test/test_atom_key.cpp
-            entity/test/test_key_serialization.cpp
-            entity/test/test_metrics.cpp
-            entity/test/test_ref_key.cpp
-            entity/test/test_tensor.cpp
-            log/test/test_log.cpp
-            pipeline/test/test_container.hpp
-            pipeline/test/test_pipeline.cpp
-            pipeline/test/test_query.cpp
-            pipeline/test/test_frame_allocation.cpp
-            util/test/test_regex.cpp
-            processing/test/test_arithmetic_type_promotion.cpp
-            processing/test/test_clause.cpp
-            processing/test/test_component_manager.cpp
-            processing/test/test_expression.cpp
-            processing/test/test_filter_and_project_sparse.cpp
-            processing/test/test_type_promotion.cpp
-            processing/test/test_operation_dispatch.cpp
-            processing/test/test_output_schema_aggregator_types.cpp
-            processing/test/test_output_schema_ast_validity.cpp
-            processing/test/test_output_schema_basic.cpp
-            processing/test/test_parallel_processing.cpp
-            processing/test/test_resample.cpp
-            processing/test/test_set_membership.cpp
-            processing/test/test_signed_unsigned_comparison.cpp
-            processing/test/test_type_comparison.cpp
-            storage/test/test_local_storages.cpp
-            storage/test/test_memory_storage.cpp
-            storage/test/test_s3_storage.cpp
-            storage/test/test_storage_factory.cpp
-            storage/test/common.hpp
-            storage/test/test_storage_exceptions.cpp
-            storage/test/test_azure_storage.cpp
-            storage/test/common.hpp
-            storage/test/test_storage_operations.cpp
-            stream/test/stream_test_common.cpp
-            stream/test/test_aggregator.cpp
-            stream/test/test_incompletes.cpp
-            stream/test/test_protobuf_mappings.cpp
-            stream/test/test_row_builder.cpp
-            stream/test/test_segment_aggregator.cpp
-            stream/test/test_types.cpp
-            util/memory_tracing.hpp
+#            async/test/test_async.cpp
+#            codec/test/test_codec.cpp
+#            codec/test/test_encode_field_collection.cpp
+#            codec/test/test_segment_header.cpp
+#            codec/test/test_encoded_field.cpp
+#            column_store/test/ingestion_stress_test.cpp
+#            column_store/test/test_column.cpp
+#            column_store/test/test_column_data_random_accessor.cpp
+#            column_store/test/test_index_filtering.cpp
+#            column_store/test/test_memory_segment.cpp
+#            column_store/test/test_statistics.cpp
+#            entity/test/test_atom_key.cpp
+#            entity/test/test_key_serialization.cpp
+#            entity/test/test_metrics.cpp
+#            entity/test/test_ref_key.cpp
+#            entity/test/test_tensor.cpp
+#            log/test/test_log.cpp
+#            pipeline/test/test_container.hpp
+#            pipeline/test/test_pipeline.cpp
+#            pipeline/test/test_query.cpp
+#            pipeline/test/test_frame_allocation.cpp
+#            util/test/test_regex.cpp
+#            processing/test/test_arithmetic_type_promotion.cpp
+#            processing/test/test_clause.cpp
+#            processing/test/test_component_manager.cpp
+#            processing/test/test_expression.cpp
+#            processing/test/test_filter_and_project_sparse.cpp
+            processing/test/test_join_schemas.cpp
+#            processing/test/test_type_promotion.cpp
+#            processing/test/test_operation_dispatch.cpp
+#            processing/test/test_output_schema_aggregator_types.cpp
+#            processing/test/test_output_schema_ast_validity.cpp
+#            processing/test/test_output_schema_basic.cpp
+#            processing/test/test_parallel_processing.cpp
+#            processing/test/test_resample.cpp
+#            processing/test/test_set_membership.cpp
+#            processing/test/test_signed_unsigned_comparison.cpp
+#            processing/test/test_type_comparison.cpp
+#            storage/test/test_local_storages.cpp
+#            storage/test/test_memory_storage.cpp
+#            storage/test/test_s3_storage.cpp
+#            storage/test/test_storage_factory.cpp
+#            storage/test/common.hpp
+#            storage/test/test_storage_exceptions.cpp
+#            storage/test/test_azure_storage.cpp
+#            storage/test/common.hpp
+#            storage/test/test_storage_operations.cpp
+#            stream/test/stream_test_common.cpp
+#            stream/test/test_aggregator.cpp
+#            stream/test/test_incompletes.cpp
+#            stream/test/test_protobuf_mappings.cpp
+#            stream/test/test_row_builder.cpp
+#            stream/test/test_segment_aggregator.cpp
+#            stream/test/test_types.cpp
+#            util/memory_tracing.hpp
             util/test/gtest_main.cpp
-            util/test/random_throw.hpp
-            util/test/test_bitmagic.cpp
-            util/test/test_buffer_pool.cpp
-            util/test/test_composite.cpp
-            util/test/test_cursor.cpp
-            util/test/test_decimal.cpp
-            util/test/test_exponential_backoff.cpp
-            util/test/test_folly.cpp
-            util/test/test_format_date.cpp
-            util/test/test_hash.cpp
-            util/test/test_id_transformation.cpp
-            util/test/test_key_utils.cpp
-            util/test/test_ranges_from_future.cpp
-            util/test/test_reliable_storage_lock.cpp
-            util/test/test_slab_allocator.cpp
-            util/test/test_storage_lock.cpp
-            util/test/test_string_pool.cpp
-            util/test/test_string_utils.cpp
-            util/test/test_tracing_allocator.cpp
-            version/test/test_append.cpp
-            version/test/test_key_block.cpp
-            version/test/test_sort_index.cpp
-            version/test/test_sorting_info_state_machine.cpp
-            version/test/test_sparse.cpp
-            version/test/test_stream_version_data.cpp
-            version/test/test_symbol_list.cpp
-            version/test/test_version_map.cpp
-            version/test/test_version_map_batch.cpp
-            version/test/test_version_store.cpp
-            version/test/version_map_model.hpp
-            python/python_handlers.cpp
+#            util/test/random_throw.hpp
+#            util/test/test_bitmagic.cpp
+#            util/test/test_buffer_pool.cpp
+#            util/test/test_composite.cpp
+#            util/test/test_cursor.cpp
+#            util/test/test_decimal.cpp
+#            util/test/test_exponential_backoff.cpp
+#            util/test/test_folly.cpp
+#            util/test/test_format_date.cpp
+#            util/test/test_hash.cpp
+#            util/test/test_id_transformation.cpp
+#            util/test/test_key_utils.cpp
+#            util/test/test_ranges_from_future.cpp
+#            util/test/test_reliable_storage_lock.cpp
+#            util/test/test_slab_allocator.cpp
+#            util/test/test_storage_lock.cpp
+#            util/test/test_string_pool.cpp
+#            util/test/test_string_utils.cpp
+#            util/test/test_tracing_allocator.cpp
+#            version/test/test_append.cpp
+#            version/test/test_key_block.cpp
+#            version/test/test_sort_index.cpp
+#            version/test/test_sorting_info_state_machine.cpp
+#            version/test/test_sparse.cpp
+#            version/test/test_stream_version_data.cpp
+#            version/test/test_symbol_list.cpp
+#            version/test/test_version_map.cpp
+#            version/test/test_version_map_batch.cpp
+#            version/test/test_version_store.cpp
+#            version/test/version_map_model.hpp
+#            python/python_handlers.cpp
 )
 
     set(EXECUTABLE_PERMS OWNER_WRITE OWNER_READ OWNER_EXECUTE GROUP_READ GROUP_EXECUTE WORLD_READ WORLD_EXECUTE) # 755
diff --git a/cpp/arcticdb/processing/clause.cpp b/cpp/arcticdb/processing/clause.cpp
@@ -1381,7 +1381,120 @@ std::vector<EntityId> ConcatClause::process(std::vector<EntityId>&& entity_ids)
     return std::move(entity_ids);
 }
 
+// TODO: Move somewhere else
+// Verifies that all of the index information specified in the input schemas are compatible with each other, and that
+// the requisite fields are present in the StreamDescriptors if necessary
+std::pair<IndexDescriptorImpl, arcticdb::proto::descriptors::NormalizationMetadata> generate_index_schema(
+        const std::vector<OutputSchema>& input_schemas) {
+    util::check<ErrorCode::E_ASSERTION_FAILURE>(!input_schemas.empty(), "Cannot join empty list of schemas");
+    std::optional<bool> has_multi_index;
+    std::optional<std::string> name;
+    std::optional<bool> is_int;
+    std::optional<std::string> tz;
+    std::optional<bool> is_physically_stored;
+    std::optional<int> field_count;
+    std::optional<int> start;
+    std::optional<int> step;
+    for (const auto& schema: input_schemas) {
+        const auto& common = schema.norm_metadata_.df().common();
+        if (!has_multi_index.has_value()) {
+            has_multi_index = common.has_multi_index();
+        } else {
+            user_input::check<ErrorCode::E_INVALID_USER_ARGUMENT>(
+                    schema.norm_metadata_.df().common().has_multi_index() == has_multi_index,
+                    "Mismatching norm metadata in schema join");
+        }
+        if (has_multi_index) {
+            const auto& index = common.multi_index();
+            if (!name.has_value()) {
+                name = index.name();
+                is_int = index.is_int();
+            } else {
+                if ((index.name() != *name) || (index.is_int() != *is_int)) {
+                    name = "";
+                    is_int = false;
+                }
+            }
+            if (!tz.has_value()) {
+                tz = index.tz();
+            } else {
+                if (index.tz() != *tz) {
+                    tz = "";
+                }
+            }
+            if (!field_count.has_value()) {
+                field_count = index.field_count();
+            } else {
+                user_input::check<ErrorCode::E_INVALID_USER_ARGUMENT>(
+                        index.field_count() == *field_count,
+                        "Mismatching norm metadata in schema join");
+            }
+        } else {
+            const auto& index = common.index();
+            if (!name.has_value()) {
+                name = index.name();
+                is_int = index.is_int();
+            } else {
+                if ((index.name() != *name) || (index.is_int() != *is_int)) {
+                    name = "";
+                    is_int = false;
+                }
+            }
+            if (!tz.has_value()) {
+                tz = index.tz();
+            } else {
+                if (index.tz() != *tz) {
+                    tz = "";
+                }
+            }
+            if (!is_physically_stored.has_value()) {
+                is_physically_stored = index.is_physically_stored();
+            } else {
+                user_input::check<ErrorCode::E_INVALID_USER_ARGUMENT>(
+                        index.is_physically_stored() == *is_physically_stored,
+                        "Mismatching norm metadata in schema join");
+            }
+            if (!start.has_value()) {
+                start = index.start();
+                step = index.step();
+            } else {
+
+            }
+        }
+    }
+}
+
 OutputSchema ConcatClause::join_schemas(std::vector<OutputSchema>&& output_schemas) const {
+    // Norm meta ensure:
+    // All have PandasIndex OR PandasMultiIndex
+    // If PandasIndex:
+    //  - name - if all the same maintain, otherwise empty string
+    //  - is_int - means name is a string representation of an integer, so fold into name logic
+    //  - tz - same as name
+    //  - is_physically stored must all be the same
+    //  - RangeIndex
+    //    - start==0/step==1 - maintain
+    //    - All steps the same, use start from first schema and maintain step
+    //    - Otherwise, log warning, pick  set start==0/step==1
+    // PandasMultiIndex same, minus is_physically stored, plus field_count matching
+
+    // StreamDescriptor index ensure:
+    //  - Type is the same (empty matches anything)
+    //  - Field count is the same
+
+    // Indexing ensure:
+    //  - norm meta version and stream descriptor version semantically match
+    //  - (check that this the case for all of the modify_schema methods on single-symbol clauses)
+    //  - all descriptors have requisite index columns
+
+    // Inner join:
+    //  - Create map from column name to type for first input schema
+    //  - Iterate remaining schema, reducing key set for any column names not in each schema
+    //  - For columns being retained, identify common type
+
+    // Outer join:
+    //  - As above, but step 2 is union rather than intersection
+
     // TODO: Implement this properly for inner/outer options
     // For now, just check they are all the same
     const auto& reference_schema = output_schemas.front();
diff --git a/cpp/arcticdb/processing/clause.hpp b/cpp/arcticdb/processing/clause.hpp
@@ -787,7 +787,7 @@ struct ConcatClause {
     std::shared_ptr<ComponentManager> component_manager_;
     JoinType join_type_;
 
-    ConcatClause(JoinType join_type);
+    explicit ConcatClause(JoinType join_type);
 
     ARCTICDB_MOVE_COPY_DEFAULT(ConcatClause)
 
diff --git a/cpp/arcticdb/processing/test/test_join_schemas.cpp b/cpp/arcticdb/processing/test/test_join_schemas.cpp
@@ -0,0 +1,16 @@
+/* Copyright 2025 Man Group Operations Limited
+ *
+ * Use of this software is governed by the Business Source License 1.1 included in the file licenses/BSL.txt.
+ *
+ * As of the Change Date specified in that file, in accordance with the Business Source License, use of this software will be governed by the Apache License, version 2.0.
+ */
+
+#include <google/protobuf/util/message_differencer.h>
+
+#include <gtest/gtest.h>
+#include <arcticdb/processing/clause.hpp>
+
+using namespace arcticdb;
+using namespace arcticdb::pipelines;
+using namespace google::protobuf::util;
+
diff --git a/cpp/arcticdb/processing/test/test_output_schema_basic.cpp b/cpp/arcticdb/processing/test/test_output_schema_basic.cpp
@@ -59,6 +59,11 @@ TEST(OutputSchema, NonModifyingClauses) {
     output_schema = merge_clause.modify_schema({stream_desc.clone(), norm_meta});
     ASSERT_EQ(output_schema.stream_descriptor(), stream_desc);
     ASSERT_TRUE(MessageDifferencer::Equals(output_schema.norm_metadata_, norm_meta));
+
+    ConcatClause concat_clause{{}};
+    output_schema = concat_clause.modify_schema({stream_desc.clone(), norm_meta});
+    ASSERT_EQ(output_schema.stream_descriptor(), stream_desc);
+    ASSERT_TRUE(MessageDifferencer::Equals(output_schema.norm_metadata_, norm_meta));
 }
 
 TEST(OutputSchema, DateRangeClauseRequiresTimeseries) {
