Add protobuf ingestion adapter and indexing parity

elasticgraph-indexer/spec/acceptance/schema_evolution_spec.rb

@@ -12,7 +12,7 @@
 
 module ElasticGraph
   RSpec.describe "Indexing schema evolution", :factories, :capture_logs, :in_temp_dir, :rake_task do
-    for_each_ingestion_format do
+    for_each_ingestion_format do |format|
       let(:path_to_schema) { "config/schema.rb" }
 
       before do
@@ -104,7 +104,7 @@ def build_widget(json_schema_version:)
       end
 
       context "when a schema has evolved to lose a new field and we are running on a new datastore cluster that doesn't have the removed field in its mapping" do
-        it "can still ingest old schema version events that have the removed field" do
+        it "can still ingest old schema version events that have the removed field", pending: format == :proto && "Protobuf dumps do not require explicit field deletions yet." do
           write_address_schema_def(json_schema_version: 1, address_extras: "t.field 'deprecated', 'String'")
           dump_artifacts
 
@@ -212,7 +212,7 @@ def get_address_payload(id)
         end
 
         context "when a public field is renamed and `name_in_index` is used to make the new field keep reading and writing the existing index field" do
-          it "can index events for either JSON schema version" do
+          it "can index events for either JSON schema version", pending: format == :proto && "Protobuf dumps do not require explicit rename or deletion declarations yet." do
             # Write a v1 schema with a `full_name` field that is called `name` in the index
             write_teams_schema_def(json_schema_version: 1) do |team_def|
               safe_replace(
@@ -257,7 +257,7 @@ def get_address_payload(id)
         end
 
         context "when an embedded type that was in an old JSON schema has been dropped" do
-          it "correctly ignores that type's data when ingesting old events" do
+          it "correctly ignores that type's data when ingesting old events", pending: format == :proto && "Protobuf dumps do not require explicit field and type deletions yet." do
             write_teams_schema_def(json_schema_version: 1) do |team_def|
               # V1 has a `team_details: TeamDetails` field.
               expect(team_def).to include('schema.object_type "TeamDetails"', 't.field "details", "TeamDetails"')
@@ -299,7 +299,7 @@ def get_address_payload(id)
         end
 
         context "when an indexed type that was in an old JSON schema has been dropped" do
-          it "ignores an old event attempting to upsert that type" do
+          it "ignores an old event attempting to upsert that type", pending: format == :proto && "Protobuf needs type-deletion checks and handling of historical envelopes." do
             write_address_schema_def(json_schema_version: 1, schema_extras: <<~EOS)
               schema.object_type "Team" do |t|
                 t.field "id", "ID!"
@@ -334,6 +334,7 @@ def get_address_payload(id)
       end
 
       def write_teams_schema_def(json_schema_version:)
+        @publisher_version = json_schema_version
         # Comment out some lines that lead to schema dump warnings (we don't want the warnings in the output).
         schema_def_contents = existing_team_schema_def.gsub(/^\s+t\.field schema\.state.schema_elements.count/, "#")
         schema_def_contents = yield schema_def_contents
@@ -354,6 +355,7 @@ def write_teams_schema_def(json_schema_version:)
       end
 
       def write_address_schema_def(json_schema_version:, address_extras: "", schema_extras: "")
+        @publisher_version = json_schema_version
         # This is a pared down schema definition of our normal test schema `Address` type.
         ::File.write(path_to_schema, <<~EOS)
           ElasticGraph.define_schema do |schema|
@@ -372,6 +374,7 @@ def write_address_schema_def(json_schema_version:, address_extras: "", schema_ex
       end
 
       def write_widget_schema_def(json_schema_version:, widget_extras: "", widgets_index_config: "")
+        @publisher_version = json_schema_version
         # This is a pared down schema definition of our normal test schema `Address` type.
         ::File.write(path_to_schema, <<~EOS)
           ElasticGraph.define_schema do |schema|
@@ -410,6 +413,9 @@ def dump_artifacts
             output: output
           )
         end
+        if respond_to?(:capture_proto_publisher)
+          capture_proto_publisher(@publisher_version, SchemaArtifacts::FromDisk.new("config/schema/artifacts"))
+        end
       end
 
       def boot_indexer

elasticgraph-indexer/spec/support/ingestion_formats.rb

@@ -8,6 +8,7 @@
 
 require "elastic_graph/indexer/test_support/converters"
 require "elastic_graph/proto_ingestion/schema_definition/api_extension"
+require "elastic_graph/spec_support/proto_publisher"
 
 module ElasticGraph
   # Runs specs against each ingestion format, so that everything we verify for JSON ingestion is also
@@ -39,7 +40,7 @@ def build_indexer(schema_artifacts: ingestion_schema_artifacts, **options, &cust
     end
 
     def process_indexing_events(events, via: indexer)
-      via.processor.process(round_trip_indexing_events(events), refresh_indices: true)
+      via.processor.process(round_trip_indexing_events(events, schema_artifacts: via.schema_artifacts), refresh_indices: true)
     end
 
     # Overrides the `UsesDatastore` helper so that records are indexed via the ingestion format under test.
@@ -65,7 +66,7 @@ def ingestion_schema_definition_extension_modules
       end
 
       # Publishes the given events in this ingestion format, returning the events the indexer receives.
-      def round_trip_indexing_events(events)
+      def round_trip_indexing_events(events, schema_artifacts: ingestion_schema_artifacts)
         events
       end
     end
@@ -74,10 +75,7 @@ module ProtoFormat
       include IngestionFormatSupport
 
       FORMAT = :proto
-      # Protobuf examples are pending until the indexer can ingest protobuf events. RSpec still runs pending
-      # examples and fails any that pass, so once protobuf ingestion works, this blanket `pending` gives way
-      # to `pending` metadata on the examples that still expose gaps.
-      METADATA = {ingests_proto_data: true, pending: "`elasticgraph-indexer` cannot ingest protobuf events yet."}
+      METADATA = {ingests_proto_data: true}
 
       # The stock schema artifacts are dumped with the protobuf extension. Using them avoids re-loading `config/schema.rb`:
       # Ruby resets a file's coverage each time it is loaded, so a re-load would discard coverage from earlier loads.
@@ -89,11 +87,19 @@ def ingestion_schema_definition_extension_modules
         [ProtoIngestion::SchemaDefinition::APIExtension]
       end
 
-      # simplecov:disable -- the indexer has no protobuf ingestion adapter yet, so it fails before it calls this.
-      def round_trip_indexing_events(events)
-        raise ::NotImplementedError, "Protobuf indexing events cannot be decoded yet."
+      # Capture publishers before another dump overwrites their schema and metadata.
+      # These versions select test fixtures only; no version is placed on the protobuf wire.
+      def capture_proto_publisher(version, schema_artifacts)
+        (@proto_publishers_by_version ||= {})[version] = SpecSupport::ProtoPublisher.new(schema_artifacts)
+      end
+
+      def round_trip_indexing_events(events, schema_artifacts: ingestion_schema_artifacts)
+        publisher = (@proto_publishers ||= {}.compare_by_identity)[schema_artifacts] ||= SpecSupport::ProtoPublisher.new(schema_artifacts)
+        events.map do |event|
+          historical_publisher = @proto_publishers_by_version&.fetch(event.schema_version) || publisher
+          publisher.round_trip(event, publisher: historical_publisher)
+        end
       end
-      # simplecov:enable
     end
   end
 

elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/indexer_extension_spec.rb

@@ -7,12 +7,24 @@
 # frozen_string_literal: true
 
 require "elastic_graph/json_ingestion/indexer_extension"
+require "elastic_graph/json_ingestion/schema_definition/api_extension"
 
 module ElasticGraph
   module JSONIngestion
     RSpec.describe IndexerExtension, :builds_indexer, :factories do
+      let(:schema_artifacts) do
+        generate_schema_artifacts(extension_modules: [SchemaDefinition::APIExtension]) do |schema|
+          schema.json_schema_version 1
+          schema.object_type "Component" do |type|
+            type.field "id", "ID!"
+            type.field "name", "String"
+            type.index "components"
+          end
+        end
+      end
+
       it "makes the JSON ingestion adapter available to indexers built for schemas defined with JSON ingestion support, with no configuration needed" do
-        indexer = build_indexer
+        indexer = build_indexer(schema_artifacts: schema_artifacts)
 
         expect(indexer.ingestion_adapters_by_format).to match("json" => an_instance_of(IngestionAdapter))
         expect(indexer.ingestion_adapters_by_format).to be(indexer.ingestion_adapters_by_format), "expected the adapters to be memoized"
@@ -31,7 +43,7 @@ module JSONIngestion
           end
         end
 
-        indexer = build_indexer(extension_modules: [extension])
+        indexer = build_indexer(schema_artifacts: schema_artifacts, extension_modules: [extension])
 
         expect(indexer.ingestion_adapters_by_format).to match(
           "json" => an_instance_of(IngestionAdapter),
@@ -40,14 +52,14 @@ module JSONIngestion
       end
 
       it "adds the JSON adapter to an explicitly empty ingestion adapter registry" do
-        indexer = build_indexer(ingestion_adapters_by_format: {})
+        indexer = build_indexer(schema_artifacts: schema_artifacts, ingestion_adapters_by_format: {})
 
         expect(indexer.ingestion_adapters_by_format).to match("json" => an_instance_of(IngestionAdapter))
       end
 
       it "adds the JSON adapter to an explicitly injected ingestion adapter registry" do
         adapters = {"custom" => Object.new}
-        indexer = build_indexer(ingestion_adapters_by_format: adapters)
+        indexer = build_indexer(schema_artifacts: schema_artifacts, ingestion_adapters_by_format: adapters)
 
         expect(indexer.ingestion_adapters_by_format).to match(
           "json" => an_instance_of(IngestionAdapter),

elasticgraph-proto_ingestion/README.md

@@ -1,7 +1,7 @@
 # ElasticGraph::ProtoIngestion
 
 An ElasticGraph extension that supports ingesting Protocol Buffer data into ElasticGraph.
-Currently it generates Protocol Buffers schema artifacts from ElasticGraph schemas: it emits
+It generates Protocol Buffers schema artifacts and provides a native-record adapter for the shared indexer. It emits
 `proto3` by default and can emit `proto2`, and supports arbitrary file-level header lines (such
 as `option` declarations).
 
@@ -84,6 +84,23 @@ end
 After running `bundle exec rake schema_artifacts:dump`, ElasticGraph will generate a `schema.proto`
 schema artifact, and will maintain a `proto_field_numbers.yaml` file alongside your schema definition.
 
+## Indexing Adapter
+
+The schema extension registers a `proto` ingestion adapter with the shared indexer. The adapter
+validates native protobuf records against ElasticGraph scalar constraints and prepares the
+records for normal, derived, and multi-source indexing. It uses the typed protobuf runtime metadata
+from the schema dump for private index field names and scalar and enum rules.
+
+Transport integrations must validate the event envelope and decode records with descriptors from
+the same schema dump before passing `ElasticGraph::Indexer::Event` instances to the processor.
+The events use `ingestion_format: "proto"`, `schema_version: nil`, and a native protobuf message
+as `record`. Transport payload decoding and a public protobuf indexer wrapper will follow separately.
+
+Generated optional fields preserve explicit zero, false, and empty strings. Protobuf repeated fields
+represent an absent list as an empty list. The adapter does not enforce GraphQL non-null fields,
+because events contain no publisher schema version. Numeric range checks remain active when general
+record validation is sampled out. Rollover timestamps are required when an event targets a rollover index.
+
 ## Schema Definition API
 
 ### Protobuf Syntax (`proto2` / `proto3`)

elasticgraph-proto_ingestion/elasticgraph-proto_ingestion.gemspec

@@ -38,6 +38,9 @@ Gem::Specification.new do |spec|
   spec.add_dependency "elasticgraph-support", ElasticGraph::VERSION
   spec.add_dependency "google-protobuf", "~> 4.31"
 
+  spec.add_development_dependency "elasticgraph-elasticsearch", ElasticGraph::VERSION
+  spec.add_development_dependency "elasticgraph-json_ingestion", ElasticGraph::VERSION
+  spec.add_development_dependency "elasticgraph-opensearch", ElasticGraph::VERSION
   # This gem's schema-definition extension code references `elasticgraph-schema_definition`, but
   # applications load it through schema-definition tasks after `elasticgraph-schema_definition` is already
   # available. Keeping this as a development dependency avoids a runtime dependency cycle.

elasticgraph-proto_ingestion/lib/elastic_graph/proto_ingestion.rb

@@ -7,7 +7,7 @@
 # frozen_string_literal: true
 
 module ElasticGraph
-  # Namespace for Protocol Buffers schema artifact generation extensions.
+  # Namespace for Protocol Buffers schema generation and ingestion.
   module ProtoIngestion
     # The name of the generated Protocol Buffers schema file.
     PROTO_SCHEMA_FILE = "schema.proto"

elasticgraph-proto_ingestion/lib/elastic_graph/proto_ingestion/indexer_extension.rb

@@ -6,12 +6,16 @@
 #
 # frozen_string_literal: true
 
+require "elastic_graph/proto_ingestion/ingestion_adapter"
+
 module ElasticGraph
   module ProtoIngestion
-    # Indexer extension that schemas using {SchemaDefinition::APIExtension} register in their runtime
-    # metadata, configured with the ingestion facts that protobuf descriptors cannot express. It does
-    # not yet install a protobuf ingestion adapter.
+    # Enables protobuf ingestion for schemas using the protobuf schema definition extension.
     module IndexerExtension
+      # @return [Hash<String, ElasticGraph::Indexer::IngestionAdapter::Interface>] the installed ingestion adapters
+      def ingestion_adapters_by_format
+        @ingestion_adapters_by_format ||= super.merge("proto" => IngestionAdapter.new(schema_artifacts: schema_artifacts))
+      end
     end
   end
 end

elasticgraph-proto_ingestion/lib/elastic_graph/proto_ingestion/ingestion_adapter.rb

@@ -0,0 +1,113 @@
+# 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/indexer/event"
+require "elastic_graph/indexer/ingestion_adapter"
+require "elastic_graph/proto_ingestion/record_preparer"
+require "elastic_graph/proto_ingestion/record_validator"
+require "elastic_graph/proto_ingestion/runtime_metadata/schema"
+
+module ElasticGraph
+  module ProtoIngestion
+    # Validates native protobuf events and supplies a native record preparer to the shared indexing pipeline.
+    # This adapter does not load or select JSON schema versions.
+    class IngestionAdapter
+      ValidationResult = ElasticGraph::Indexer::IngestionAdapter::ValidationResult
+      private_constant :ValidationResult
+
+      # @param schema_artifacts [SchemaArtifacts::FromDisk] generated schema artifacts
+      def initialize(schema_artifacts:)
+        @metadata = RuntimeMetadata::Schema.from_schema_artifacts(schema_artifacts)
+        @record_preparer = RecordPreparer.new(schema_artifacts)
+        @validator = RecordValidator.new(@metadata)
+        object_types = schema_artifacts.runtime_metadata.object_types_by_name
+        @update_targets = object_types.transform_values(&:update_targets)
+        @ingestible_types = object_types.filter_map { |name, type| name unless type.update_targets.empty? }.to_set
+      end
+
+      # Validates the native protobuf record and prepares the event for the shared indexing pipeline.
+      # @param event [ElasticGraph::Indexer::Event] a protobuf indexing event with a validated envelope
+      # @param skip_record_validation [Boolean] whether to omit semantic record validation
+      # @return [::ElasticGraph::Indexer::IngestionAdapter::ValidationResult] validation and normalized event
+      def validate_event(event, skip_record_validation: false)
+        record = event.record
+        expected_name = "#{@metadata.package_name}.#{event.type}"
+        unless record.class.respond_to?(:descriptor) && record.class.descriptor.name == expected_name
+          return invalid("record", "Expected a protobuf message of type #{expected_name}.")
+        end
+
+        type = event.type
+        if record.class.descriptor.lookup_oneof("value")
+          alternative = record.value
+          return invalid("record", "An abstract record must select a concrete subtype.") unless alternative
+          record = record.public_send(alternative)
+          type = record.class.descriptor.name.delete_prefix("#{@metadata.package_name}.")
+        end
+
+        return invalid("record", "The selected subtype is not ingestible.") unless @ingestible_types.include?(type)
+        if (error = @validator.error_for(record, wire_contract_only: skip_record_validation))
+          return invalid("#{event.type} record", error)
+        end
+
+        record = record.dup
+        record.id = event.id
+        if (error = rollover_error_for(type, record))
+          return invalid("#{event.type} record", error)
+        end
+
+        ValidationResult.valid(event.with(record: record, type: type), @record_preparer)
+      end
+
+      private
+
+      # Rollover timestamps select the destination index, rather than imposing GraphQL nullability.
+      # A target with no derived IDs creates no operations and therefore needs no timestamp.
+      def rollover_error_for(type, record)
+        @update_targets.fetch(type).each do |target|
+          next unless (path = target.rollover_timestamp_value_source)
+          _, id = resolve_path(record, target.id_source)
+          next if id.to_s.strip.empty?
+          public_path, value = resolve_path(record, path)
+          return "record.#{public_path.join(".")} is required to select a rollover index." if value.nil?
+        end
+        nil
+      end
+
+      def resolve_path(record, path)
+        descriptor = record.class.descriptor
+        current = record # : untyped
+        public_path = path.split(".").map do |part|
+          if (oneof = descriptor.lookup_oneof("value"))
+            if (alternative = current&.value)
+              current = current.public_send(alternative)
+              descriptor = current.class.descriptor
+            else
+              oneof_fields = oneof.to_a # : Array[Google::Protobuf::FieldDescriptor]
+              descriptor = oneof_fields.fetch(0).subtype # : Google::Protobuf::Descriptor
+            end
+          end
+
+          type_name = descriptor.name.delete_prefix("#{@metadata.package_name}.")
+          fields = @metadata.types_by_name[type_name]&.fields_by_name
+          public_name = fields&.find { |_, meta| meta.name_in_index == part }&.first || part
+          field = descriptor.lookup(public_name)
+          current = if current&.public_send(:"has_#{public_name}?")
+            current[public_name]
+          end
+          descriptor = field.subtype if field.type == :message
+          public_name
+        end
+        [public_path, current]
+      end
+
+      def invalid(description, error)
+        ValidationResult.invalid(validation_target: description, message: error)
+      end
+    end
+  end
+end