Prepare datastore_core and schema_artifacts for nested sourced_from fields. (#1255)

This is the parts of #1252 which are ready to merge.

Co-authored-by: ellisandrews-toast <ellis.andrews@toasttab.com>

Author: myronmarston

elasticgraph-datastore_core/lib/elastic_graph/datastore_core/index_definition.rb

@@ -33,7 +33,8 @@ def self.with(name:, runtime_metadata:, config:, datastore_clients_by_name:, sch
           env_index_config: env_index_config,
           defined_clusters: config.clusters.keys.to_set,
           datastore_clients_by_name: datastore_clients_by_name,
-          has_had_multiple_sources: runtime_metadata.has_had_multiple_sources
+          has_had_multiple_sources: runtime_metadata.has_had_multiple_sources,
+          sourced_from_nested_paths_by_qualified_relationship: runtime_metadata.sourced_from_nested_paths_by_qualified_relationship
         }
 
         if (rollover = runtime_metadata.rollover)

elasticgraph-datastore_core/lib/elastic_graph/datastore_core/index_definition/base.rb

@@ -130,6 +130,14 @@ def list_counts_field_paths_for_source(source)
           @list_counts_field_paths_for_source[source] ||= identify_list_counts_field_paths_for_source(source)
         end
 
+        # The value of the painless `index_data` script's `sourcedFromNestedPaths` param: nested `sourced_from`
+        # paths keyed by qualified relationship. Empty when the index has no nested sourced fields.
+        def sourced_from_nested_paths_as_painless_param
+          @sourced_from_nested_paths_as_painless_param ||= sourced_from_nested_paths_by_qualified_relationship.transform_values do |segments|
+            segments.map(&:to_painless_hash)
+          end
+        end
+
         def to_s
           "#<#{self.class.name} #{name}>"
         end

elasticgraph-datastore_core/lib/elastic_graph/datastore_core/index_definition/index.rb

@@ -15,17 +15,20 @@ class DatastoreCore
     module IndexDefinition
       class Index < Support::MemoizableData.define(
         :name, :route_with, :default_sort_clauses, :current_sources, :fields_by_path,
-        :env_index_config, :defined_clusters, :datastore_clients_by_name, :env_agnostic_settings, :has_had_multiple_sources
+        :env_index_config, :defined_clusters, :datastore_clients_by_name, :env_agnostic_settings, :has_had_multiple_sources,
+        :sourced_from_nested_paths_by_qualified_relationship
       )
         # `Data.define` provides all these methods:
         # @dynamic name, route_with, default_sort_clauses, current_sources, fields_by_path, env_index_config, env_agnostic_settings
         # @dynamic defined_clusters, datastore_clients_by_name, initialize, has_had_multiple_sources
+        # @dynamic sourced_from_nested_paths_by_qualified_relationship
 
         # `include IndexDefinition::Base` provides all these methods. Steep should be able to detect it
         # but can't for some reason so we have to declare them with `@dynamic`.
         # @dynamic flattened_env_setting_overrides, routing_value_for_prepared_record, has_custom_routing?, cluster_to_query
         # @dynamic clusters_to_index_into, all_accessible_cluster_names, ignored_values_for_routing, searches_could_hit_incomplete_docs?, max_result_window
         # @dynamic accessible_cluster_names_to_index_into, accessible_from_queries?, known_related_query_rollover_indices, list_counts_field_paths_for_source
+        # @dynamic sourced_from_nested_paths_as_painless_param
         include IndexDefinition::Base
 
         def mappings_in_datastore(datastore_client)

elasticgraph-datastore_core/lib/elastic_graph/datastore_core/index_definition/rollover_index_template.rb

@@ -23,17 +23,19 @@ module IndexDefinition
       class RolloverIndexTemplate < Support::MemoizableData.define(
         :name, :route_with, :default_sort_clauses, :current_sources, :fields_by_path, :env_index_config,
         :index_args, :defined_clusters, :datastore_clients_by_name, :timestamp_field_path, :frequency,
-        :env_agnostic_settings, :has_had_multiple_sources
+        :env_agnostic_settings, :has_had_multiple_sources, :sourced_from_nested_paths_by_qualified_relationship
       )
         # `Data.define` provides all these methods:
         # @dynamic name, route_with, default_sort_clauses, current_sources, fields_by_path, env_index_config, env_agnostic_settings
         # @dynamic index_args, defined_clusters, datastore_clients_by_name, timestamp_field_path, frequency, initialize, has_had_multiple_sources
+        # @dynamic sourced_from_nested_paths_by_qualified_relationship
 
         # `include IndexDefinition::Base` provides all these methods. Steep should be able to detect it
         # but can't for some reason so we have to declare them with `@dynamic`.
         # @dynamic flattened_env_setting_overrides, routing_value_for_prepared_record, has_custom_routing?, cluster_to_query
         # @dynamic clusters_to_index_into, all_accessible_cluster_names, ignored_values_for_routing, searches_could_hit_incomplete_docs?, max_result_window
         # @dynamic accessible_cluster_names_to_index_into, accessible_from_queries?, known_related_query_rollover_indices, list_counts_field_paths_for_source
+        # @dynamic sourced_from_nested_paths_as_painless_param
         include IndexDefinition::Base
 
         def mappings_in_datastore(datastore_client)

elasticgraph-datastore_core/sig/elastic_graph/datastore_core/index_definition.rbs

@@ -16,6 +16,7 @@ module ElasticGraph
       def accessible_from_queries?: () -> bool
       def known_related_query_rollover_indices: () -> ::Array[IndexDefinition::RolloverIndex]
       def list_counts_field_paths_for_source: (::String) -> ::Set[::String]
+      def sourced_from_nested_paths_as_painless_param: () -> ::Hash[::String, ::Array[::Hash[::String, ::String]]]
     end
 
     # Defines methods of the _IndexDefinition interface that each specific implementation must provide.
@@ -26,6 +27,7 @@ module ElasticGraph
       def current_sources: () -> ::Set[::String]
       def fields_by_path: () -> ::Hash[::String, SchemaArtifacts::RuntimeMetadata::IndexField]
       def has_had_multiple_sources: () -> bool
+      def sourced_from_nested_paths_by_qualified_relationship: () -> ::Hash[::String, ::Array[SchemaArtifacts::RuntimeMetadata::sourcedFromNestedPathSegment]]
       def env_index_config: () -> Configuration::IndexDefinition
       def env_agnostic_settings: () -> ::Hash[::String, untyped]
       def defined_clusters: () -> ::Set[::String]

elasticgraph-datastore_core/sig/elastic_graph/datastore_core/index_definition/base.rbs

@@ -10,6 +10,7 @@ module ElasticGraph
         @known_related_query_rollover_indices: ::Array[RolloverIndex]?
         @searches_could_hit_incomplete_docs: bool
         @list_counts_field_paths_for_source: ::Hash[::String, ::Set[::String]]
+        @sourced_from_nested_paths_as_painless_param: ::Hash[::String, ::Array[::Hash[::String, ::String]]]?
         @max_result_window: ::Integer?
 
         def to_s: () -> ::String

elasticgraph-datastore_core/spec/unit/elastic_graph/datastore_core/index_definition/implementation_shared_examples.rb

@@ -544,16 +544,62 @@ def accessible_cluster_names_to_index_into_for_config(clusters:, index_definitio
               index.list_counts_field_paths_for_source(SELF_RELATIONSHIP_NAME)
             )
           end
+        end
+
+        describe "#sourced_from_nested_paths_as_painless_param" do
+          it "returns an empty hash for an index with no nested `sourced_from` fields" do
+            index = define_index
 
-          def update_type_for_index(schema)
-            yield schema.state.object_types_by_name.fetch("MyType")
+            expect(index.sourced_from_nested_paths_as_painless_param).to eq({})
+          end
+
+          it "converts the registered nested paths into the painless hash form, keyed by qualified relationship" do
+            index = define_index(schema_def: lambda do |schema|
+              schema.object_type "NestedThing" do |t|
+                t.field "id", "ID!"
+                t.field "name", "String" do |f|
+                  f.sourced_from "related_thing", "name"
+                end
+                t.relates_to_one "related_thing", "RelatedThing", via: "nested_thing_id", dir: :in, indexing_only: true do |r|
+                  r.parent_relationship "MyType", "related_things"
+                end
+              end
+
+              schema.object_type "RelatedThing" do |t|
+                t.field "id", "ID!"
+                t.field "my_type_id", "ID"
+                t.field "nested_thing_id", "ID"
+                t.field "name", "String"
+                t.field "created_at", "DateTime!"
+                t.index "related_things"
+              end
+
+              update_type_for_index(schema) do |t|
+                t.field "nested_things", "[NestedThing!]!" do |f|
+                  f.mapping type: "object"
+                end
+                # `equivalent_field` is required because `MyType` may be a rollover index, so the indexer
+                # needs to know which `RelatedThing` timestamp selects the `MyType` index to update.
+                t.relates_to_many "related_things", "RelatedThing", via: "my_type_id", dir: :in, indexing_only: true, singular: "related_thing" do |r|
+                  r.equivalent_field "created_at"
+                end
+              end
+            end) { |i| i.has_had_multiple_sources! }
+
+            expect(index.sourced_from_nested_paths_as_painless_param).to eq(
+              "nested_things.related_thing" => [{"field" => "nested_things", "sourceField" => "nested_thing_id"}]
+            )
           end
         end
 
         def define_index(name = "my_type", **options, &block)
           define_datastore_core_with_index(name, **options, &block)
             .index_definitions_by_name.fetch(name)
         end
+
+        def update_type_for_index(schema)
+          yield schema.state.object_types_by_name.fetch("MyType")
+        end
       end
     end
   end

elasticgraph-schema_artifacts/lib/elastic_graph/schema_artifacts/runtime_metadata/sourced_from_nested_path_segment.rb

@@ -25,9 +25,6 @@ def self.from_hash(hash)
       # always on `id` (relationships join on `id` via foreign keys), so it's implicit rather
       # than stored here.
       #
-      # A future PR will add `to_painless_param` to convert these segments into the
-      # camelCase hash format expected by the painless script (with a "type" discriminator).
-      #
       # @private
       class ListPathSegment < ::Data.define(:field, :source_field)
         FIELD = "field"
@@ -41,10 +38,14 @@ def to_dumpable_hash
         def self.from_hash(hash)
           new(field: hash[FIELD], source_field: hash[SOURCE_FIELD])
         end
+
+        # The painless script expects camelCase and discriminates list segments by the presence of `sourceField`.
+        def to_painless_hash
+          {"field" => field, "sourceField" => source_field}
+        end
       end
 
       # Represents a segment in a nested sourced path that navigates into an object field.
-      # See `ListPathSegment` for notes on `to_painless_param`.
       #
       # @private
       class ObjectPathSegment < ::Data.define(:field)
@@ -58,6 +59,11 @@ def to_dumpable_hash
         def self.from_hash(hash)
           new(field: hash[FIELD])
         end
+
+        # No `sourceField`, which is how the painless script tells object segments from list segments.
+        def to_painless_hash
+          {"field" => field}
+        end
       end
     end
   end

elasticgraph-schema_artifacts/sig/elastic_graph/schema_artifacts/runtime_metadata/sourced_from_nested_path_segment.rbs

@@ -14,6 +14,7 @@ module ElasticGraph
 
         def self.from_hash: (::Hash[::String, untyped]) -> ListPathSegment
         def to_dumpable_hash: () -> ::Hash[::String, ::String]
+        def to_painless_hash: () -> ::Hash[::String, ::String]
       end
 
       class ObjectPathSegmentSuperType
@@ -27,6 +28,7 @@ module ElasticGraph
 
         def self.from_hash: (::Hash[::String, untyped]) -> ObjectPathSegment
         def to_dumpable_hash: () -> ::Hash[::String, ::String]
+        def to_painless_hash: () -> ::Hash[::String, ::String]
       end
 
       type sourcedFromNestedPathSegment = ListPathSegment | ObjectPathSegment

elasticgraph-schema_artifacts/spec/unit/elastic_graph/schema_artifacts/runtime_metadata/sourced_from_nested_path_segment_spec.rb

@@ -0,0 +1,29 @@
+# Copyright 2024 - 2026 Block, Inc.
+#
+# Use of this source code is governed by an MIT-style
+# license that can be found in the LICENSE file or at
+# https://opensource.org/licenses/MIT.
+#
+# frozen_string_literal: true
+
+require "elastic_graph/schema_artifacts/runtime_metadata/sourced_from_nested_path_segment"
+
+module ElasticGraph
+  module SchemaArtifacts
+    module RuntimeMetadata
+      RSpec.describe SourcedFromNestedPathSegment do
+        it "converts a list segment to a painless hash with a camelCased `sourceField`" do
+          segment = ListPathSegment.new(field: "players", source_field: "playerId")
+
+          expect(segment.to_painless_hash).to eq("field" => "players", "sourceField" => "playerId")
+        end
+
+        it "converts an object segment to a painless hash without a `sourceField` (which marks it an object segment)" do
+          segment = ObjectPathSegment.new(field: "roster")
+
+          expect(segment.to_painless_hash).to eq("field" => "roster")
+        end
+      end
+    end
+  end
+end