From f81cbf8579ecdc3e4d4b6ef9d6a05a69c5943183 Mon Sep 17 00:00:00 2001 From: Josh Wilson Date: Mon, 7 Sep 2026 21:33:20 -0500 Subject: [PATCH] Complete protobuf schemas for ingestion Preserve singular field presence in proto2 and proto3, represent nested lists with wrapper messages, and include types used only as indexing sources. Compile both syntaxes and verify their binary round trips. --- .github/workflows/ci.yaml | 3 + elasticgraph-proto_ingestion/README.md | 8 ++- .../schema_definition/results_extension.rb | 3 +- .../schema_definition/schema.rb | 20 ++---- .../object_interface_and_union_extension.rb | 43 +++++++++--- .../schema_definition/schema.rbs | 5 +- .../object_interface_and_union_extension.rbs | 2 + .../schema_compilation_spec.rb | 67 +++++++++++++++++++ .../schema_definition/api_extension_spec.rb | 2 +- .../scalar_proto_types_spec.rb | 24 +++---- .../schema_definition/schema_spec.rb | 37 ++++++---- 11 files changed, 163 insertions(+), 51 deletions(-) create mode 100644 elasticgraph-proto_ingestion/spec/integration/elastic_graph/proto_ingestion/schema_definition/schema_compilation_spec.rb diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 08d95e009..41e25e53a 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -91,6 +91,9 @@ jobs: ruby-version: ${{ matrix.ruby }} bundler-cache: true # runs 'bundle install' and caches installed gems automatically + - name: Install protobuf compiler + run: sudo apt-get update && sudo apt-get install -y protobuf-compiler + - name: Setup Docker Compose uses: KengoTODA/actions-setup-docker-compose@477353946803dd64eaa44008b865b6bfc88cab4e # v1.2.4 env: diff --git a/elasticgraph-proto_ingestion/README.md b/elasticgraph-proto_ingestion/README.md index b7182f035..c566f4118 100644 --- a/elasticgraph-proto_ingestion/README.md +++ b/elasticgraph-proto_ingestion/README.md @@ -77,6 +77,8 @@ 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. +Compile generated schemas with a standard `protoc` version supporting proto3 optional fields (3.15 or newer). Publishers must regenerate code to distinguish omitted scalar fields from explicitly supplied default values. + ## Schema Definition API ### Protobuf Syntax (`proto2` / `proto3`) @@ -247,8 +249,10 @@ Additionally: in a comment above the field (e.g. `// Must be formatted as an ISO 8601 date, e.g. "2024-11-25".`). Values are validated when events are ingested, just as with JSON ingestion. - List types become `repeated` fields. -- Lists of lists (e.g. `[[Float!]!]!`) are not supported because Protocol Buffers cannot represent - them directly. Schema artifact generation raises an error identifying the unsupported field. +- Nested lists use generated wrapper messages with a repeated `values` field at each inner level. +- Singular fields use `optional` in proto2 and proto3, preserving explicit zero, false, and empty string values. +- Repeated fields cannot distinguish a null list from an empty list or represent null elements. An empty wrapper represents an empty inner list. +- Types that supply `sourced_from` fields are included even when they do not have their own index. - Enum types generate `enum` definitions whose values are prefixed with the enum type name in `UPPER_SNAKE_CASE`, including a zero-valued `*_UNSPECIFIED` entry. ## Stable Field Numbers diff --git a/elasticgraph-proto_ingestion/lib/elastic_graph/proto_ingestion/schema_definition/results_extension.rb b/elasticgraph-proto_ingestion/lib/elastic_graph/proto_ingestion/schema_definition/results_extension.rb index ec8b1becb..a3b9491fc 100644 --- a/elasticgraph-proto_ingestion/lib/elastic_graph/proto_ingestion/schema_definition/results_extension.rb +++ b/elasticgraph-proto_ingestion/lib/elastic_graph/proto_ingestion/schema_definition/results_extension.rb @@ -40,7 +40,8 @@ def protobuf_schema_generator Schema.new( state: extension_state, all_types: all_types, - ingestion_state: extension_state.proto_ingestion_state + ingestion_state: extension_state.proto_ingestion_state, + sourced_type_names: sourced_update_targets_by_source_type_name.keys ) end end diff --git a/elasticgraph-proto_ingestion/lib/elastic_graph/proto_ingestion/schema_definition/schema.rb b/elasticgraph-proto_ingestion/lib/elastic_graph/proto_ingestion/schema_definition/schema.rb index 4f3a62abc..9533195d7 100644 --- a/elasticgraph-proto_ingestion/lib/elastic_graph/proto_ingestion/schema_definition/schema.rb +++ b/elasticgraph-proto_ingestion/lib/elastic_graph/proto_ingestion/schema_definition/schema.rb @@ -64,9 +64,10 @@ def self.validate_header_lines(header_lines) # @param state [ElasticGraph::SchemaDefinition::State] # @param all_types [Array] # @param ingestion_state [ProtoIngestionState] this extension's configured schema definition state - def initialize(state:, all_types:, ingestion_state:) + def initialize(state:, all_types:, ingestion_state:, sourced_type_names: []) @state = state @all_types = all_types + @sourced_type_names = sourced_type_names @package_name = ingestion_state.package_name @syntax = self.class.validate_syntax(ingestion_state.syntax) @header_lines = self.class.validate_header_lines(ingestion_state.header_lines) @@ -123,30 +124,23 @@ def field_number_for(message_name:, type_name:, public_field_name:) # Returns the label prefix (including its trailing space) that a field declaration needs # under the configured syntax, or an empty string when the field takes no label. # - # `proto2` requires an explicit label on every field, so non-repeated fields get - # `optional `; `proto3` labels repeated fields only. Note that `oneof` alternatives never + # Non-repeated fields use `optional` in both syntaxes so that absence is distinct from + # an explicitly supplied zero, false, or empty string. `oneof` alternatives never # get a label under either syntax -- protoc rejects one -- so the `oneof` renderer in # `ObjectInterfaceAndUnionExtension` does not call this. # # @api private def field_label_prefix(repeated:) return "repeated " if repeated - proto2? ? "optional " : "" - end - - # Indicates whether the generator emits `proto2` rather than `proto3`. - # - # @api private - def proto2? - @syntax == "proto2" + "optional " end private - # Selects the indexed root types and every type transitively referenced by their protobuf + # Selects indexed and source-only root types and every type referenced by their protobuf # representations. All traversal state is local so repeated calls are independent. def proto_types - types_to_visit = _ = @state.indexed_types_by_index_name.values.dup + types_to_visit = _ = @state.indexed_types_by_index_name.values + @sourced_type_names.map { |name| @state.types_by_name.fetch(name) } type_names_to_render = ::Set.new while (type = types_to_visit.shift) diff --git a/elasticgraph-proto_ingestion/lib/elastic_graph/proto_ingestion/schema_definition/schema_elements/object_interface_and_union_extension.rb b/elasticgraph-proto_ingestion/lib/elastic_graph/proto_ingestion/schema_definition/schema_elements/object_interface_and_union_extension.rb index c6e34aaae..d1869fc26 100644 --- a/elasticgraph-proto_ingestion/lib/elastic_graph/proto_ingestion/schema_definition/schema_elements/object_interface_and_union_extension.rb +++ b/elasticgraph-proto_ingestion/lib/elastic_graph/proto_ingestion/schema_definition/schema_elements/object_interface_and_union_extension.rb @@ -111,11 +111,12 @@ def render_proto_message(schema, message_name, package_name) end field_definitions << " // Next field number: #{schema.next_field_number_for(message_name)}" - <<~PROTO.chomp + definition = <<~PROTO.chomp #{documentation}message #{message_name} { #{field_definitions.join("\n")} } PROTO + [definition, *proto_list_wrapper_definitions(package_name)].join("\n\n") end def render_proto_oneof(schema, message_name, package_name) @@ -175,17 +176,43 @@ def field_comment_lines_for(doc_comment, field_comment) doc_lines + [" //"] + format_lines end - def proto_field_type_for(type_ref, package_name:, context_field_name:) - list_depth, base_type_ref = ObjectInterfaceAndUnionExtension.list_depth_and_base_type(type_ref) + def proto_list_wrapper_name(field_name, level) + wrapper_name = "#{name}_#{field_name}_List#{level}" + if schema_def_state.types_by_name.key?(wrapper_name) + raise Errors::SchemaError, "Generated protobuf list wrapper `#{wrapper_name}` conflicts with a schema type. Rename that type." + end + wrapper_name + end - if list_depth > 1 - raise Errors::SchemaError, "Field `#{name}.#{context_field_name}` has type `#{type_ref.name}`, " \ - "but Protocol Buffers cannot represent lists of lists directly. " \ - "`elasticgraph-proto_ingestion` supports fields with at most one list level." + def proto_list_wrapper_definitions(package_name) + proto_fields.flat_map do |schema_field, field| + depth, base_type = ObjectInterfaceAndUnionExtension.list_depth_and_base_type(field.type) + (1...depth).map do |level| + proto_base_type = _ = base_type.resolved + element_type = if level == depth - 1 + proto_base_type.proto_type_reference(package_name) + else + ".#{package_name}.#{proto_list_wrapper_name(schema_field.name, level + 1)}" + end + <<~PROTO.chomp + message #{proto_list_wrapper_name(schema_field.name, level)} { + repeated #{element_type} values = 1; + } + PROTO + end end + end + + def proto_field_type_for(type_ref, package_name:, context_field_name:) + list_depth, base_type_ref = ObjectInterfaceAndUnionExtension.list_depth_and_base_type(type_ref) proto_type = _ = base_type_ref.resolved - [list_depth == 1, proto_type.proto_type_reference(package_name), proto_type.protobuf_field_comment] + field_type = if list_depth > 1 + ".#{package_name}.#{proto_list_wrapper_name(context_field_name, 1)}" + else + proto_type.proto_type_reference(package_name) + end + [list_depth >= 1, field_type, proto_type.protobuf_field_comment] end end end diff --git a/elasticgraph-proto_ingestion/sig/elastic_graph/proto_ingestion/schema_definition/schema.rbs b/elasticgraph-proto_ingestion/sig/elastic_graph/proto_ingestion/schema_definition/schema.rbs index 7559cf446..f5573fcd7 100644 --- a/elasticgraph-proto_ingestion/sig/elastic_graph/proto_ingestion/schema_definition/schema.rbs +++ b/elasticgraph-proto_ingestion/sig/elastic_graph/proto_ingestion/schema_definition/schema.rbs @@ -13,6 +13,7 @@ module ElasticGraph @syntax: ::String @header_lines: ::Array[::String] @state: ::ElasticGraph::SchemaDefinition::State + @sourced_type_names: Array[String] @all_types: ::Array[::ElasticGraph::SchemaDefinition::SchemaElements::graphQLType] @package_name: ::String @field_number_mappings: FieldNumberMappings @@ -21,7 +22,8 @@ module ElasticGraph def initialize: ( state: ::ElasticGraph::SchemaDefinition::State, all_types: ::Array[::ElasticGraph::SchemaDefinition::SchemaElements::graphQLType], - ingestion_state: ProtoIngestionState + ingestion_state: ProtoIngestionState, + ?sourced_type_names: Array[String] ) -> void def to_proto: () -> ::String @@ -37,7 +39,6 @@ module ElasticGraph def next_enum_value_number_for: (::String) -> ::Integer def reserved_enum_value_numbers_for: (::String, ::Array[::String]) -> ::Hash[::String, ::Integer] def field_label_prefix: (repeated: bool) -> ::String - def proto2?: () -> bool private diff --git a/elasticgraph-proto_ingestion/sig/elastic_graph/proto_ingestion/schema_definition/schema_elements/object_interface_and_union_extension.rbs b/elasticgraph-proto_ingestion/sig/elastic_graph/proto_ingestion/schema_definition/schema_elements/object_interface_and_union_extension.rbs index 8c2535187..c0d7186fa 100644 --- a/elasticgraph-proto_ingestion/sig/elastic_graph/proto_ingestion/schema_definition/schema_elements/object_interface_and_union_extension.rbs +++ b/elasticgraph-proto_ingestion/sig/elastic_graph/proto_ingestion/schema_definition/schema_elements/object_interface_and_union_extension.rbs @@ -27,6 +27,8 @@ module ElasticGraph ::ElasticGraph::SchemaDefinition::Indexing::Field ]] def field_comment_lines_for: (::String? doc_comment, ::String? field_comment) -> ::Array[::String] + def proto_list_wrapper_name: (String, Integer) -> String + def proto_list_wrapper_definitions: (String) -> Array[String] def proto_field_type_for: ( ::ElasticGraph::SchemaDefinition::SchemaElements::TypeReference, package_name: ::String, diff --git a/elasticgraph-proto_ingestion/spec/integration/elastic_graph/proto_ingestion/schema_definition/schema_compilation_spec.rb b/elasticgraph-proto_ingestion/spec/integration/elastic_graph/proto_ingestion/schema_definition/schema_compilation_spec.rb new file mode 100644 index 000000000..27b97e75a --- /dev/null +++ b/elasticgraph-proto_ingestion/spec/integration/elastic_graph/proto_ingestion/schema_definition/schema_compilation_spec.rb @@ -0,0 +1,67 @@ +# 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/proto_ingestion/schema_definition/api_extension" +require "open3" + +module ElasticGraph + module ProtoIngestion + module SchemaDefinition + RSpec.describe "Generated protobuf schemas", :in_temp_dir do + [:proto2, :proto3].each do |syntax| + it "compiles #{syntax} source-only types and preserves defaults and nested lists on the wire" do + proto = define_proto_schema do |s| + s.proto_schema_artifacts package_name: "elasticgraph", syntax: syntax + s.object_type "Source" do |t| + t.field "id", "ID!" + t.field "product_id", "ID!" + t.field "name", "String" + t.field "enabled", "Boolean" + t.field "quantity", "Int" + t.field "matrix", "[[Int!]!]!" + t.field "cube", "[[[Int!]!]!]!" + t.field "details", "SourceDetails" + end + s.object_type("SourceDetails") { |t| t.field "label", "String" } + s.object_type "Product" do |t| + t.field "id", "ID!" + t.relates_to_one "source", "Source", via: "product_id", dir: :in, indexing_only: true + t.field("source_name", "String") { |f| f.sourced_from "source", "name" } + t.index("products") { |i| i.has_had_multiple_sources! } + end + end + File.write("schema.proto", proto) + source = <<~PROTO + id: "source" + product_id: "product" + name: "" + enabled: false + quantity: 0 + matrix { values: 1 values: 2 } + matrix {} + cube { values { values: 3 } values {} } + details { label: "nested source" } + PROTO + encoded = run_protoc("--encode=elasticgraph.Source", source) + decoded = run_protoc("--decode=elasticgraph.Source", encoded) + expect(decoded).to include('name: ""', "enabled: false", "quantity: 0", 'label: "nested source"') + expect(decoded).to include("matrix {\n values: 1\n values: 2\n}\nmatrix {\n}") + expect(decoded).to include("cube {\n values {\n values: 3\n }\n values {\n }\n}") + expect(run_protoc("--decode=elasticgraph.Source", run_protoc("--encode=elasticgraph.Source", ""))).to eq("") + end + end + + def run_protoc(operation, input) + output, errors, status = Open3.capture3(ENV.fetch("PROTOC", "protoc"), "--proto_path=.", operation, "schema.proto", stdin_data: input, binmode: true) + expect(status.success?).to be(true), errors + output + end + end + end + end +end diff --git a/elasticgraph-proto_ingestion/spec/unit/elastic_graph/proto_ingestion/schema_definition/api_extension_spec.rb b/elasticgraph-proto_ingestion/spec/unit/elastic_graph/proto_ingestion/schema_definition/api_extension_spec.rb index 1fd9ed70b..4597eaa45 100644 --- a/elasticgraph-proto_ingestion/spec/unit/elastic_graph/proto_ingestion/schema_definition/api_extension_spec.rb +++ b/elasticgraph-proto_ingestion/spec/unit/elastic_graph/proto_ingestion/schema_definition/api_extension_spec.rb @@ -48,7 +48,7 @@ module SchemaDefinition aggregate_failures do built_in_scalar_options.each.with_index(1) do |(type_name, options), field_number| - field_line = " #{options.fetch(:type)} #{type_name.downcase} = #{field_number};" + field_line = " optional #{options.fetch(:type)} #{type_name.downcase} = #{field_number};" field_comment = options[:field_comment] expect(proto).to include(field_comment ? " // #{field_comment}\n#{field_line}" : field_line) end diff --git a/elasticgraph-proto_ingestion/spec/unit/elastic_graph/proto_ingestion/schema_definition/scalar_proto_types_spec.rb b/elasticgraph-proto_ingestion/spec/unit/elastic_graph/proto_ingestion/schema_definition/scalar_proto_types_spec.rb index 613f3d419..386fa1e91 100644 --- a/elasticgraph-proto_ingestion/spec/unit/elastic_graph/proto_ingestion/schema_definition/scalar_proto_types_spec.rb +++ b/elasticgraph-proto_ingestion/spec/unit/elastic_graph/proto_ingestion/schema_definition/scalar_proto_types_spec.rb @@ -32,9 +32,9 @@ module SchemaDefinition import "google/protobuf/timestamp.proto"; message Event { - string id = 1; - google.protobuf.Timestamp created_at = 2; - google.protobuf.Timestamp updated_at = 3; + optional string id = 1; + optional google.protobuf.Timestamp created_at = 2; + optional google.protobuf.Timestamp updated_at = 3; // Next field number: 4 } PROTO @@ -61,9 +61,9 @@ module SchemaDefinition package elasticgraph; message Event { - string id = 1; + optional string id = 1; // Must be formatted as an ISO 8601 timestamp. - string created_at = 2; + optional string created_at = 2; // Next field number: 3 } PROTO @@ -91,8 +91,8 @@ module SchemaDefinition import "google/type/date.proto"; message Event { - string id = 1; - google.type.Date occurred_on = 2; + optional string id = 1; + optional google.type.Date occurred_on = 2; // Next field number: 3 } PROTO @@ -172,13 +172,13 @@ module SchemaDefinition expect(proto_type_def_from(proto, "Person")).to eq(<<~PROTO.strip) message Person { - string id = 1; + optional string id = 1; // The dates that matter to this person. // // Must be formatted as an ISO 8601 date, e.g. "2024-11-25". repeated string important_dates = 2; // Must be an IANA time zone identifier, e.g. "America/Los_Angeles". - string time_zone = 3; + optional string time_zone = 3; // Next field number: 4 } PROTO @@ -199,7 +199,7 @@ module SchemaDefinition end expect(proto).to include('import "my-app/types/v1.money.proto";') - expect(proto).to include(" // Amount and currency.\n myapp.types.Money total = 2;") + expect(proto).to include(" // Amount and currency.\n optional myapp.types.Money total = 2;") end it "rejects an `import:` that is not the path of a `.proto` file" do @@ -246,13 +246,13 @@ module SchemaDefinition expect(proto_type_def_from(proto, "Order")).to eq(<<~PROTO.strip) message Order { - string id = 1; + optional string id = 1; // What the customer owes. // // Must be an amount and a currency. // // The amount is in minor units. - string total = 2; + optional string total = 2; // Next field number: 3 } PROTO diff --git a/elasticgraph-proto_ingestion/spec/unit/elastic_graph/proto_ingestion/schema_definition/schema_spec.rb b/elasticgraph-proto_ingestion/spec/unit/elastic_graph/proto_ingestion/schema_definition/schema_spec.rb index 497ae0b13..ed2bd86bd 100644 --- a/elasticgraph-proto_ingestion/spec/unit/elastic_graph/proto_ingestion/schema_definition/schema_spec.rb +++ b/elasticgraph-proto_ingestion/spec/unit/elastic_graph/proto_ingestion/schema_definition/schema_spec.rb @@ -48,16 +48,16 @@ module SchemaDefinition // An account in the system. message Account { // The account's unique identifier. - string id = 1; - .elasticgraph.Status status = 2; - .elasticgraph.Address address = 3; + optional string id = 1; + optional .elasticgraph.Status status = 2; + optional .elasticgraph.Address address = 3; repeated string tags = 4; // Next field number: 5 } message Address { - string street = 1; - string city = 2; + optional string street = 1; + optional string city = 2; // Next field number: 3 } @@ -301,20 +301,33 @@ module SchemaDefinition PROTO end - it "rejects lists of lists" do + it "represents lists of lists with generated wrapper messages" do + proto = define_proto_schema do |s| + s.object_type "Matrix" do |t| + t.field "id", "ID" + t.field "values", "[[Float!]!]!" + t.index "matrices" + end + end + expect(proto).to include("repeated .elasticgraph.Matrix_values_List1 values = 2;") + expect(proto_type_def_from(proto, "Matrix_values_List1")).to eq(<<~PROTO.strip) + message Matrix_values_List1 { + repeated double values = 1; + } + PROTO + end + + it "rejects collisions between generated list wrappers and user-defined types" do expect { define_proto_schema do |s| + s.object_type("Matrix_values_List1") { |t| t.field "value", "Int" } s.object_type "Matrix" do |t| t.field "id", "ID" - t.field "values", "[[Float!]!]!" + t.field "values", "[[Int!]!]!" t.index "matrices" end end - }.to raise_error(Errors::SchemaError, a_string_including( - "Field `Matrix.values` has type `[[Float!]!]!`", - "Protocol Buffers cannot represent lists of lists directly", - "at most one list level" - )) + }.to raise_error(Errors::SchemaError, /Generated protobuf list wrapper `Matrix_values_List1` conflicts/) end it "uses custom proto scalar mappings" do