Allow omitted JSON schema versions

Author: jwils

elasticgraph-indexer/README.md

@@ -74,3 +74,7 @@ def decode(payload)
   # return an array of ElasticGraph indexing event hashes
 end
 ```
+
+Decoded event hashes do not need to provide a schema version. When a version is omitted, the latest
+available JSON schema artifact version is used for validation and record preparation. JSON ingestion
+payloads may still include `json_schema_version` to request a specific JSON schema version.

elasticgraph-indexer/lib/elastic_graph/indexer/indexing_event_decoder.rb

@@ -24,7 +24,8 @@ def initialize(config:, schema_artifacts:, logger:)
         end
 
         # @param payload [String] a raw payload from the transport
-        # @return [Array<Hash<String, Object>>] the decoded ElasticGraph indexing events
+        # @return [Array<Hash<String, Object>>] the decoded ElasticGraph indexing events. Events do not
+        #   need to include a schema version; when omitted, the latest available JSON schema version is used.
         def decode(payload)
           # :nocov: -- must return an array to satisfy Steep type checking but never called
           []

elasticgraph-indexer/lib/elastic_graph/indexer/operation/factory.rb

@@ -28,9 +28,11 @@ class Factory < Support::MemoizableData.define(
         def build(event)
           event = prepare_event(event)
 
-          selected_json_schema_version = select_json_schema_version(event) { |failure| return failure }
+          requested_json_schema_version = json_schema_version_from(event)
+          selected_json_schema_version = select_json_schema_version(event, requested_json_schema_version) { |failure| return failure }
+          event = event.merge(JSON_SCHEMA_VERSION_KEY => requested_json_schema_version)
 
-          # Because the `select_json_schema_version` picks the closest-matching json schema version, the incoming
+          # Because `select_json_schema_version` picks the closest-matching json schema version, the incoming
           # event might not match the expected json_schema_version value in the json schema (which is a `const` field).
           # This is by design, since we're picking a schema based on best-effort, so to avoid that by-design validation error,
           # performing the envelope validation on a "patched" version of the event.
@@ -49,15 +51,10 @@ def build(event)
 
         private
 
-        def select_json_schema_version(event)
+        def select_json_schema_version(event, requested_json_schema_version)
           available_json_schema_versions = schema_artifacts.available_json_schema_versions
 
-          requested_json_schema_version = event[JSON_SCHEMA_VERSION_KEY]
-
-          # First check that a valid value has been requested (a positive integer)
-          if !event.key?(JSON_SCHEMA_VERSION_KEY)
-            yield build_failed_result(event, JSON_SCHEMA_VERSION_KEY, "Event lacks a `#{JSON_SCHEMA_VERSION_KEY}`")
-          elsif !requested_json_schema_version.is_a?(Integer) || requested_json_schema_version < 1
+          unless requested_json_schema_version.is_a?(Integer) && requested_json_schema_version >= 1
             yield build_failed_result(event, JSON_SCHEMA_VERSION_KEY, "#{JSON_SCHEMA_VERSION_KEY} (#{requested_json_schema_version}) must be a positive integer.")
           end
 
@@ -86,7 +83,7 @@ def select_json_schema_version(event)
           if selected_json_schema_version.nil?
             yield build_failed_result(
               event, JSON_SCHEMA_VERSION_KEY,
-              "Failed to select json schema version. Requested version: #{event[JSON_SCHEMA_VERSION_KEY]}. \
+              "Failed to select json schema version. Requested version: #{requested_json_schema_version}. \
               Available json schema versions: #{available_json_schema_versions.sort.join(", ")}"
             )
           end
@@ -117,6 +114,10 @@ def prepare_event(event)
           event.merge("record" => event["record"].merge("id" => event.fetch("id")))
         end
 
+        def json_schema_version_from(event)
+          event.fetch(JSON_SCHEMA_VERSION_KEY) { schema_artifacts.available_json_schema_versions.max }
+        end
+
         def validate_record_returning_failure(event, selected_json_schema_version)
           record = event.fetch("record")
           graphql_type_name = event.fetch("type")

elasticgraph-indexer/sig/elastic_graph/indexer/operation/factory.rbs

@@ -42,8 +42,9 @@ module ElasticGraph
         @validator_factories_by_version: ::Hash[::Integer, Support::JSONSchema::ValidatorFactory]?
         def validator_factories_by_version: () -> ::Hash[::Integer, Support::JSONSchema::ValidatorFactory]
 
-        def select_json_schema_version: (event) { (BuildResult) -> bot } -> (::Integer | bot)
+        def select_json_schema_version: (event, untyped) { (BuildResult) -> bot } -> (::Integer | bot)
         def prepare_event: (event) -> event
+        def json_schema_version_from: (event) -> untyped
         def validate_record_returning_failure: (event, ::Integer) -> BuildResult?
         def build_failed_result: (event, ::String, ::String) -> BuildResult
         def build_all_operations_for: (event, _RecordPreparer) -> ::Array[_Operation]

elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb

@@ -193,10 +193,22 @@ module Operation
             expect_failed_event_error(event, "missing_keys", "version")
           end
 
-          it "notifies an error on missing `#{JSON_SCHEMA_VERSION_KEY}`" do
+          it "defaults to the latest json schema version when no json schema version is specified" do
             event = build_upsert_event(:component).except(JSON_SCHEMA_VERSION_KEY)
 
-            expect_failed_event_error(event, JSON_SCHEMA_VERSION_KEY)
+            expect(build_expecting_success(event)).to contain_exactly(new_primary_indexing_operation(
+              event.merge(JSON_SCHEMA_VERSION_KEY => 1)
+            ))
+          end
+
+          it "accepts json_schema_version as an input alias for JSON ingestion events" do
+            event = build_upsert_event(:component, id: "1", __version: 1)
+              .except(JSON_SCHEMA_VERSION_KEY)
+              .merge(JSON_SCHEMA_VERSION_KEY => 1)
+
+            expect(build_expecting_success(event)).to eq([new_primary_indexing_operation(
+              event.except(JSON_SCHEMA_VERSION_KEY).merge(JSON_SCHEMA_VERSION_KEY => 1)
+            )])
           end
 
           it "notifies an error on wrong field types" do
@@ -249,7 +261,7 @@ module Operation
           context "when the indexer has json schemas v2 and v4 (v4 adds yellow color)" do
             before do
               # With the "real" version one as a baseline, create a separate version with a small schema change.
-              # Tests will then specify the desired json_schema_version in the event payload to test the schema-choosing
+              # Tests will then specify the desired json schema version in the event payload to test the schema-choosing
               # behavior of the `factory` class.
               schemas = {
                 2 => indexer.schema_artifacts.json_schemas_for(1),
@@ -269,7 +281,7 @@ module Operation
             end
 
             it "validates against an older version of a json schema if specified" do
-              # YELLOW doesn't exist in schema version 2. So expect an error when json_schema_version is set to 2.
+              # YELLOW doesn't exist in json schema version 2. So expect an error when json_schema_version is set to 2.
               event = build_upsert_event(:widget, id: "1", __version: 1, __json_schema_version: 2)
               event["record"]["options"]["color"] = "YELLOW"
 
@@ -355,13 +367,13 @@ module Operation
               )
             end
 
-            it "notifies an error if an invalid (e.g. negative) json_schema_version is specified" do
+            it "notifies an error if an invalid (e.g. negative) json schema version is specified" do
               event = build_upsert_event(:widget, id: "1", __version: 1, __json_schema_version: -1)
 
               expect_failed_event_error(event, "must be a positive integer", "(-1)")
             end
 
-            it "notifies an error if it's unable to select a json_schema_version" do
+            it "notifies an error if it's unable to select a json schema version" do
               event = build_upsert_event(:component, id: "1", __version: 1)
               event["record"]["name"] = 123
 

elasticgraph-warehouse_lambda/spec/unit/elastic_graph/warehouse_lambda/warehouse_dumper_spec.rb

@@ -30,22 +30,22 @@ class WarehouseLambda
           "type" => "Widget",
           "id" => "1",
           "version" => 3,
-          "json_schema_version" => 1,
+          JSON_SCHEMA_VERSION_KEY => 1,
           "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}
         })
       end
 
       it "writes operations to S3 as gzipped JSONL files and returns success results" do
-        op1 = new_primary_indexing_operation({"type" => "Widget", "id" => "1", "version" => 3, "json_schema_version" => 1, "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}})
-        op2 = new_primary_indexing_operation({"type" => "Widget", "id" => "2", "version" => 5, "json_schema_version" => 2, "record" => {"id" => "2", "dayOfWeek" => "TUE", "created_at" => "2024-09-15T13:30:12Z", "workspace_id" => "ws-2"}})
+        op1 = new_primary_indexing_operation({"type" => "Widget", "id" => "1", "version" => 3, JSON_SCHEMA_VERSION_KEY => 1, "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}})
+        op2 = new_primary_indexing_operation({"type" => "Widget", "id" => "2", "version" => 5, JSON_SCHEMA_VERSION_KEY => 2, "record" => {"id" => "2", "dayOfWeek" => "TUE", "created_at" => "2024-09-15T13:30:12Z", "workspace_id" => "ws-2"}})
         operations = [op1, op2]
 
         results = warehouse_dumper.bulk(operations)
 
-        # Verify S3 uploads - should have 2 files (one for json_schema_version 1, one for json_schema_version 2)
+        # Verify S3 uploads - should have 2 files (one for json schema version 1, one for json schema version 2)
         expect(s3_client.api_requests.map { |req| req[:operation_name] }).to eq [:put_object, :put_object]
 
-        # Verify first file (json_schema_version 1)
+        # Verify first file (json schema version 1)
         params1 = s3_client.api_requests[0].fetch(:params)
         expect(params1[:bucket]).to eq s3_bucket_name
         expect(params1[:key]).to match %r{Data0001/Widget/v1/2024-09-15/[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}\.jsonl\.gz}
@@ -61,7 +61,7 @@ class WarehouseLambda
         lines1 = jsonl_content1.split("\n")
         expect(lines1.size).to eq 1
 
-        # Verify second file (json_schema_version 2)
+        # Verify second file (json schema version 2)
         params2 = s3_client.api_requests[1].fetch(:params)
         expect(params2[:key]).to match %r{Data0001/Widget/v2/2024-09-15/[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}\.jsonl\.gz}
 
@@ -99,8 +99,8 @@ class WarehouseLambda
       end
 
       it "writes operations of different types to separate S3 files" do
-        widget_op = new_primary_indexing_operation({"type" => "Widget", "id" => "1", "version" => 3, "json_schema_version" => 1, "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}})
-        component_op = new_primary_indexing_operation({"type" => "Component", "id" => "c1", "version" => 2, "json_schema_version" => 1, "record" => {"id" => "c1", "created_at" => "2024-09-15T12:30:12Z"}})
+        widget_op = new_primary_indexing_operation({"type" => "Widget", "id" => "1", "version" => 3, JSON_SCHEMA_VERSION_KEY => 1, "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}})
+        component_op = new_primary_indexing_operation({"type" => "Component", "id" => "c1", "version" => 2, JSON_SCHEMA_VERSION_KEY => 1, "record" => {"id" => "c1", "created_at" => "2024-09-15T12:30:12Z"}})
         operations = [widget_op, component_op]
 
         warehouse_dumper.bulk(operations)
@@ -113,8 +113,8 @@ class WarehouseLambda
       end
 
       it "logs structured information about received batch and dumped files" do
-        widget_op = new_primary_indexing_operation({"type" => "Widget", "id" => "1", "version" => 3, "json_schema_version" => 1, "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}})
-        component_op = new_primary_indexing_operation({"type" => "Component", "id" => "c1", "version" => 2, "json_schema_version" => 1, "record" => {"id" => "c1", "created_at" => "2024-09-15T12:30:12Z"}})
+        widget_op = new_primary_indexing_operation({"type" => "Widget", "id" => "1", "version" => 3, JSON_SCHEMA_VERSION_KEY => 1, "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}})
+        component_op = new_primary_indexing_operation({"type" => "Component", "id" => "c1", "version" => 2, JSON_SCHEMA_VERSION_KEY => 1, "record" => {"id" => "c1", "created_at" => "2024-09-15T12:30:12Z"}})
         operations = [widget_op, component_op]
 
         warehouse_dumper.bulk(operations)
@@ -127,13 +127,13 @@ class WarehouseLambda
           a_hash_including({
             "s3_bucket" => s3_bucket_name,
             "type" => "Widget",
-            "json_schema_version" => 1,
+            JSON_SCHEMA_VERSION_KEY => 1,
             "record_count" => 1
           }),
           a_hash_including({
             "s3_bucket" => s3_bucket_name,
             "type" => "Component",
-            "json_schema_version" => 1,
+            JSON_SCHEMA_VERSION_KEY => 1,
             "record_count" => 1
           })
         ]
@@ -183,7 +183,7 @@ class WarehouseLambda
           "type" => "Widget",
           "id" => "1",
           "version" => 3,
-          "json_schema_version" => 1,
+          JSON_SCHEMA_VERSION_KEY => 1,
           "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}
         })