Add skip_record_validation_for indexer config for sampled backfill validation

elasticgraph-indexer/lib/elastic_graph/indexer.rb

@@ -85,6 +85,7 @@ def operation_factory
           record_preparer_factory: record_preparer_factory,
           logger: datastore_core.logger,
           skip_derived_indexing_type_updates: config.skip_derived_indexing_type_updates,
+          skip_record_validation_for: config.skip_record_validation_for,
           configure_record_validator: nil
         )
       end

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

@@ -11,7 +11,7 @@
 
 module ElasticGraph
   class Indexer
-    class Config < Support::Config.define(:latency_slo_thresholds_by_timestamp_in_ms, :skip_derived_indexing_type_updates, :extension_modules)
+    class Config < Support::Config.define(:latency_slo_thresholds_by_timestamp_in_ms, :skip_derived_indexing_type_updates, :skip_record_validation_for, :extension_modules)
       json_schema at: "indexer",
         optional: false,
         description: "Configuration for indexing operations and metrics used by `elasticgraph-indexer`.",
@@ -43,15 +43,36 @@ class Config < Support::Config.define(:latency_slo_thresholds_by_timestamp_in_ms
               {"WidgetWorkspace" => ["ABC12345678"]}
             ]
           },
+          skip_record_validation_for: {
+            description: "Map of GraphQL type names to the fraction of records (in `[0.0, 1.0]`) whose " \
+              "per-type JSON schema validation should be skipped. `0.0` (or an absent key) validates every record; " \
+              "`1.0` skips every record; values in between sample. The decision is deterministic per event id " \
+              "(`type:id@vversion`) so the same event makes the same skip decision on retry across indexer pods. " \
+              "The event envelope (op, id, type, version, json_schema_version, latency_timestamps) is always " \
+              "validated. Intended for backfills of trusted, pre-validated data where skipping the per-record " \
+              "schema walk yields meaningful ingest speedups, with a sampled fraction left validated as a canary " \
+              "for schema drift. Leave empty for live-traffic ingestion: the datastore mappings will not catch all " \
+              "the constraints (regex, enum, min/max, format, abstract-type discriminators) that the JSON schema " \
+              "enforces.",
+            type: "object",
+            patternProperties: {/^[A-Z]\w*$/.source => {type: "number", minimum: 0, maximum: 1}},
+            additionalProperties: false,
+            default: {}, # : untyped
+            examples: [
+              {}, # : untyped
+              {"Widget" => 0.9, "Component" => 1.0}
+            ]
+          },
           extension_modules: Support::Config::EXTENSION_MODULE_SCHEMA
         }
 
       private
 
-      def convert_values(skip_derived_indexing_type_updates:, latency_slo_thresholds_by_timestamp_in_ms:, extension_modules:)
+      def convert_values(skip_derived_indexing_type_updates:, latency_slo_thresholds_by_timestamp_in_ms:, skip_record_validation_for:, extension_modules:)
         {
           skip_derived_indexing_type_updates: skip_derived_indexing_type_updates.transform_values(&:to_set),
           latency_slo_thresholds_by_timestamp_in_ms: latency_slo_thresholds_by_timestamp_in_ms,
+          skip_record_validation_for: skip_record_validation_for.transform_values(&:to_f),
           extension_modules: SchemaArtifacts::RuntimeMetadata::ExtensionLoader.load_component_extensions(extension_modules)
         }
       end

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

@@ -13,6 +13,7 @@
 require "elastic_graph/indexer/record_preparer"
 require "elastic_graph/support/json_schema/validator_factory"
 require "elastic_graph/support/memoizable_data"
+require "zlib"
 
 module ElasticGraph
   class Indexer
@@ -23,6 +24,7 @@ class Factory < Support::MemoizableData.define(
         :record_preparer_factory,
         :logger,
         :skip_derived_indexing_type_updates,
+        :skip_record_validation_for,
         :configure_record_validator
       )
         def build(event)
@@ -40,11 +42,27 @@ def build(event)
             return build_failed_result(event, "event payload", error_message)
           end
 
-          failed_result = validate_record_returning_failure(event, selected_json_schema_version)
-          failed_result || BuildResult.success(build_all_operations_for(
-            event,
-            record_preparer_factory.for_json_schema_version(selected_json_schema_version)
-          ))
+          graphql_type_name = event.fetch("type")
+          validation_skipped = skip_validation?(graphql_type_name, event)
+
+          unless validation_skipped
+            failed_result = validate_record_returning_failure(event, graphql_type_name, selected_json_schema_version)
+            return failed_result if failed_result
+          end
+
+          begin
+            BuildResult.success(
+              build_all_operations_for(event, record_preparer_factory.for_json_schema_version(selected_json_schema_version)),
+              validation_skipped_for: validation_skipped ? graphql_type_name : nil
+            )
+          rescue RecordPreparer::UnknownTypeError
+            # Safety net for `skip_record_validation_for`: when record validation is skipped, an
+            # event with a missing or unknown abstract-type `__typename` reaches `RecordPreparer`
+            # and raises. Convert it to a `FailedEventError` so callers see a structured failure
+            # rather than an exception leaking out of `Factory#build`. The message intentionally
+            # omits the offending value to avoid leaking record data.
+            build_failed_result(event, "#{graphql_type_name} record", "Missing or unknown `__typename` for an abstract-type field.")
+          end
         end
 
         private
@@ -117,16 +135,27 @@ def prepare_event(event)
           event.merge("record" => event["record"].merge("id" => event.fetch("id")))
         end
 
-        def validate_record_returning_failure(event, selected_json_schema_version)
+        def validate_record_returning_failure(event, graphql_type_name, selected_json_schema_version)
           record = event.fetch("record")
-          graphql_type_name = event.fetch("type")
           validator = validator(graphql_type_name, selected_json_schema_version)
 
           if (error_message = validator.validate_with_error_message(record))
             build_failed_result(event, "#{graphql_type_name} record", error_message)
           end
         end
 
+        # Decides whether to skip per-record validation for `event` of `type`. The decision is
+        # deterministic per event id: a stable `Zlib.crc32` of `EventID#to_s` puts each event in a
+        # bucket in `[0.0, 1.0)` that is compared against the configured skip rate. Same event id =>
+        # same decision across pods and retries, so retries do not flip records between
+        # validated/skipped. `String#hash` is unsuitable here: `RUBY_HASH_SEED` is per-process.
+        def skip_validation?(type, event)
+          rate = skip_record_validation_for[type]
+          return false if rate.nil? || rate <= 0.0
+          return true if rate >= 1.0
+          ::Zlib.crc32(EventID.from_event(event).to_s).fdiv(2**32) < rate
+        end
+
         def build_failed_result(event, payload_description, validation_message)
           message = "Malformed #{payload_description}. #{validation_message}"
 
@@ -192,14 +221,16 @@ def raise(*args)
         # Return value from `build` that indicates what happened.
         # - If it was successful, `operations` will be a non-empty array of operations and `failed_event_error` will be nil.
         # - If there was a validation issue, `operations` will be an empty array and `failed_event_error` will be non-nil.
-        BuildResult = ::Data.define(:operations, :failed_event_error) do
+        # - `validation_skipped_for` names the event's GraphQL type when per-record validation was skipped
+        #   (via `skip_record_validation_for`), and is nil otherwise. `Processor` aggregates this for observability.
+        BuildResult = ::Data.define(:operations, :failed_event_error, :validation_skipped_for) do
           # @implements BuildResult
-          def self.success(operations)
-            new(operations, nil)
+          def self.success(operations, validation_skipped_for: nil)
+            new(operations, nil, validation_skipped_for)
           end
 
           def self.failure(failed_event_error)
-            new([], failed_event_error)
+            new([], failed_event_error, nil)
           end
         end
       end

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

@@ -45,6 +45,8 @@ def process_returning_failures(events, refresh_indices: false)
 
         factory_results = factory_results_by_event.values
 
+        log_skipped_record_validations(factory_results)
+
         bulk_result = @datastore_router.bulk(factory_results.flat_map(&:operations), refresh: refresh_indices)
         successful_operations = bulk_result.successful_operations(check_failures: false)
 
@@ -62,6 +64,21 @@ def process_returning_failures(events, refresh_indices: false)
 
       private
 
+      # Emits a single aggregate log line per batch when any records had their per-record validation
+      # skipped (via `skip_record_validation_for`). Skipping a safety check should never be silent, but
+      # per-record logging would be untenable at backfill scale (a `1.0` skip rate is one line per record),
+      # so we tally by type and log once. Mirrors the batch-level `ElasticGraphIndexingLatencies` log.
+      def log_skipped_record_validations(factory_results)
+        counts_by_type = factory_results.filter_map(&:validation_skipped_for).tally
+        return if counts_by_type.empty?
+
+        @logger.info({
+          "message_type" => "RecordValidationSkipped",
+          "count" => counts_by_type.values.sum,
+          "counts_by_type" => counts_by_type
+        })
+      end
+
       def categorize_failures(failures, events)
         source_event_versions_by_cluster_by_op = @datastore_router.source_event_versions_in_index(
           failures.flat_map { |f| f.versioned_operations.to_a }

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

@@ -11,6 +11,13 @@
 module ElasticGraph
   class Indexer
     class RecordPreparer
+      # Raised when an abstract-type record is missing a `__typename` (or carries one that does not
+      # name a known concrete subtype). Normally JSON-schema validation prevents reaching here, but
+      # `Indexer::Config#skip_record_validation_for` lets that validation be skipped -- so this error
+      # exists as a typed, catchable signal that callers can convert into a structured failure.
+      class UnknownTypeError < ::KeyError
+      end
+
       # Provides the ability to get a `RecordPreparer` for a specific JSON schema version.
       class Factory
         def initialize(schema_artifacts)
@@ -116,7 +123,10 @@ def prepare_for_index(type_name, value, mapping_properties)
           #
           # If `type_name` is an abstract type, we need to look at the `__typename` field to see
           # what the concrete subtype is. `__typename` is required on abstract types and indicates that.
-          eg_meta_by_field_name = @eg_meta_by_field_name_by_concrete_type.fetch(value["__typename"] || type_name)
+          concrete_type = value["__typename"] || type_name
+          eg_meta_by_field_name = @eg_meta_by_field_name_by_concrete_type.fetch(concrete_type) do
+            raise UnknownTypeError, "No concrete type named `#{concrete_type}` is known to the record preparer."
+          end
 
           # We only want to consider __typename if it's in the per-record mapping in order to determine
           # whether __typename is required on records. When it's a constant_keyword it exists at the index

elasticgraph-indexer/sig/elastic_graph/indexer/config.rbs

@@ -5,16 +5,19 @@ module ElasticGraph
 
       attr_reader latency_slo_thresholds_by_timestamp_in_ms: ::Hash[::String, ::Integer]
       attr_reader skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]]
+      attr_reader skip_record_validation_for: ::Hash[::String, ::Float]
       attr_reader extension_modules: ::Array[::Module]
 
       def initialize: (
         ?latency_slo_thresholds_by_timestamp_in_ms: ::Hash[::String, ::Integer],
         ?skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]],
+        ?skip_record_validation_for: ::Hash[::String, ::Float],
         ?extension_modules: ::Array[::Module]) -> void
 
       def with: (
         ?latency_slo_thresholds_by_timestamp_in_ms: ::Hash[::String, ::Integer],
         ?skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]],
+        ?skip_record_validation_for: ::Hash[::String, ::Float],
         ?extension_modules: ::Array[::Module]) -> Config
 
       def self.members: () -> ::Array[::Symbol]
@@ -26,6 +29,7 @@ module ElasticGraph
       def convert_values: (
         latency_slo_thresholds_by_timestamp_in_ms: untyped,
         skip_derived_indexing_type_updates: untyped,
+        skip_record_validation_for: untyped,
         extension_modules: untyped
       ) -> ::Hash[::Symbol, untyped]
 

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

@@ -11,6 +11,7 @@ module ElasticGraph
         attr_reader record_preparer_factory: RecordPreparer::Factory
         attr_reader logger: ::Logger
         attr_reader skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]]
+        attr_reader skip_record_validation_for: ::Hash[::String, ::Float]
         attr_reader configure_record_validator: (^(validatorFactory) -> validatorFactory)?
 
         def initialize: (
@@ -19,6 +20,7 @@ module ElasticGraph
           record_preparer_factory: RecordPreparer::Factory,
           logger: ::Logger,
           skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]],
+          skip_record_validation_for: ::Hash[::String, ::Float],
           configure_record_validator: (^(validatorFactory) -> validatorFactory)?
         ) -> void
 
@@ -28,6 +30,7 @@ module ElasticGraph
           ?record_preparer_factory: RecordPreparer::Factory,
           ?logger: ::Logger,
           ?skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]],
+          ?skip_record_validation_for: ::Hash[::String, ::Float],
           ?configure_record_validator: (^(validatorFactory) -> validatorFactory)?
         ) -> instance
       end
@@ -44,7 +47,8 @@ module ElasticGraph
 
         def select_json_schema_version: (event) { (BuildResult) -> bot } -> (::Integer | bot)
         def prepare_event: (event) -> event
-        def validate_record_returning_failure: (event, ::Integer) -> BuildResult?
+        def validate_record_returning_failure: (event, ::String, ::Integer) -> BuildResult?
+        def skip_validation?: (::String, event) -> bool
         def build_failed_result: (event, ::String, ::String) -> BuildResult
         def build_all_operations_for: (event, _RecordPreparer) -> ::Array[_Operation]
         def index_definitions_for: (::String) -> ::Array[DatastoreCore::_IndexDefinition]
@@ -53,15 +57,17 @@ module ElasticGraph
         class BuildResult
           attr_reader operations: ::Array[_Operation]
           attr_reader failed_event_error: FailedEventError?
+          attr_reader validation_skipped_for: ::String?
 
-          def initialize: (::Array[_Operation], FailedEventError?) -> void
+          def initialize: (::Array[_Operation], FailedEventError?, ::String?) -> void
 
           def with: (
             ?operations: ::Array[_Operation],
-            ?failed_event_error: FailedEventError?
+            ?failed_event_error: FailedEventError?,
+            ?validation_skipped_for: ::String?
           ) -> BuildResult
 
-          def self.success: (::Array[_Operation]) -> BuildResult
+          def self.success: (::Array[_Operation], ?validation_skipped_for: ::String?) -> BuildResult
           def self.failure: (FailedEventError) -> BuildResult
         end
       end

elasticgraph-indexer/sig/elastic_graph/indexer/processor.rbs

@@ -20,6 +20,7 @@ module ElasticGraph
       @indexing_latency_slo_thresholds_by_timestamp_in_ms: ::Hash[::String, ::Integer]
       @clock: singleton(::Time)
 
+      def log_skipped_record_validations: (::Array[Operation::Factory::BuildResult]) -> void
       def categorize_failures: (::Array[FailedEventError], ::Array[event]) -> ::Array[FailedEventError]
       def calculate_latency_metrics: (::Array[_Operation], ::Array[Operation::Result]) -> void
     end

elasticgraph-indexer/sig/elastic_graph/indexer/record_preparer.rbs

@@ -8,6 +8,9 @@ module ElasticGraph
       type egMetaByFieldHash = ::Hash[::String, {"type" => ::String, "nameInIndex" => ::String}]
       type egMetaByFieldByTypeHash = ::Hash[::String, egMetaByFieldHash]
 
+      class UnknownTypeError < ::KeyError[untyped, untyped]
+      end
+
       class Factory
         def initialize: (schemaArtifacts) -> void
 

elasticgraph-indexer/spec/unit/elastic_graph/indexer/config_spec.rb

@@ -32,6 +32,33 @@ class Indexer
         expect(config.skip_derived_indexing_type_updates).to eq("WidgetCurrency" => ["USD"].to_set)
       end
 
+      it "coerces `skip_record_validation_for` rates to floats (so integer YAML values like `1` become `1.0`)" do
+        config = Config.from_parsed_yaml("indexer" => {
+          "latency_slo_thresholds_by_timestamp_in_ms" => {},
+          "skip_record_validation_for" => {
+            "Widget" => 1,
+            "Component" => 0.5
+          }
+        })
+
+        expect(config.skip_record_validation_for).to eq("Widget" => 1.0, "Component" => 0.5)
+        expect(config.skip_record_validation_for.values).to all(be_a(::Float))
+      end
+
+      it "rejects `skip_record_validation_for` rates outside `[0.0, 1.0]`" do
+        expect {
+          Config.from_parsed_yaml("indexer" => {
+            "skip_record_validation_for" => {"Widget" => 1.5}
+          })
+        }.to raise_error Errors::ConfigError
+
+        expect {
+          Config.from_parsed_yaml("indexer" => {
+            "skip_record_validation_for" => {"Widget" => -0.1}
+          })
+        }.to raise_error Errors::ConfigError
+      end
+
       describe "#extension_modules", :in_temp_dir do
         it "loads the extension modules from disk" do
           File.write("eg_extension_module1.rb", <<~EOS)