Ignore sub-fields with warninigs and keep only parent.

Author: mashhurs

docs/index.asciidoc

@@ -304,8 +304,12 @@ NOTE: If your index has a mapping with sub-objects where `status.code` and `stat
 ===== Conflict on multi-fields
 
 ES|QL query fetches all parent and sub-fields fields if your {es} index has https://www.elastic.co/docs/reference/elasticsearch/mapping-reference/multi-fields[multi-fields] or https://www.elastic.co/docs/reference/elasticsearch/mapping-reference/subobjects[subobjects].
-Since {ls} events cannot contain parent field's concrete value and sub-field values together, the plugin cannot map the result to {ls} event and produces `_elasticsearch_input_failure` tagged failed event.
-We recommend using the `RENAME` (or `DROP`) keyword in your ES|QL query explicitly rename the fields to overcome this issue.
+Since {ls} events cannot contain parent field's concrete value and sub-field values together, the plugin ignores sub-fields with warning and includes parent.
+We recommend using the `RENAME` (or `DROP` to avoid warnings) keyword in your ES|QL query explicitly rename the fields to include sub-fields into the event.
+
+This a common occurrence if your template or mapping follows the pattern of always indexing strings as "text" (`field`) + " keyword" (`field.keyword`) multi-field.
+In this case it's recommended to do `KEEP field` if the string is identical and there is only one subfield as the engine will optimize and retrieve the keyword, otherwise you can do `KEEP field.keyword | RENAME field.keyword as field`.
+
 To illustrate the situation with example, assuming your mapping has a time `time` field with `time.min` and `time.max` sub-fields as following:
 [source, ruby]
     "properties": {
@@ -315,8 +319,6 @@ To illustrate the situation with example, assuming your mapping has a time `time
     }
 
 The ES|QL result will contain all three fields but the plugin cannot map them into {ls} event.
-
-This a common occurence if your template or mapping follows the pattern of always indexing strings as "text" (`field`) + " keyword" (`field.keyword`) multi-field. In this case it's recommended to do `KEEP field` if the string is identical and there is only one subfield as the engine will optimize and retrieve the keyword, otherwise you can do `KEEP field.keyword | RENAME field.keyword as field` .
 To avoid this, you can use the `RENAME` keyword to rename the `time` parent field to get all three fields with unique fields.
 [source, ruby]
     ...

lib/logstash/inputs/elasticsearch/esql.rb

@@ -22,9 +22,9 @@ def initialize(client, plugin)
 
           target_field = plugin.params["target"]
           if target_field
-            def self.apply_target(path) = "[#{target_field}][#{path}]"
+            def self.apply_target(path); "[#{target_field}][#{path}]"; end
           else
-            def self.apply_target(path) = path
+            def self.apply_target(path); path; end
           end
 
           @query = plugin.params["query"]
@@ -77,11 +77,16 @@ def retryable(job_name, &block)
         # @param output_queue [Queue] The queue to push processed events to
         def process_response(columns, values, output_queue)
           column_specs = columns.map { |column| ColumnSpec.new(column) }
+          sub_element_mark_map = mark_sub_elements(column_specs)
+          multi_fields = sub_element_mark_map.filter_map { |key, val| key.name if val == true }
+          logger.warn("Multi-fields found in ES|QL result and they will not be available in the event. Please use `RENAME` command if you want to include them.", { :detected_multi_fields => multi_fields }) if multi_fields.any?
+
           values.each do |row|
             event = column_specs.zip(row).each_with_object(LogStash::Event.new) do |(column, value), event|
               # `unless value.nil?` is a part of `drop_null_columns` that if some of columns' values are not `nil`, `nil` values appear
               # we should continuously filter out them to achieve full `drop_null_columns` on each individual row (ideal `LIMIT 1` result)
-              unless value.nil?
+              # we also exclude sub-elements of main field
+              if value && sub_element_mark_map[column] == false
                 field_reference = apply_target(column.field_reference)
                 event.set(field_reference, ESQL_PARSERS_BY_TYPE[column.type].call(value))
               end
@@ -95,6 +100,30 @@ def process_response(columns, values, output_queue)
             output_queue << failed_event
           end
         end
+
+        # Determines whether each column in a collection is a nested sub-element (example "user.age")
+        # of another column in the same collection (example "user").
+        #
+        # @param columns [Array<ColumnSpec>] An array of objects with a `name` attribute representing field paths.
+        # @return [Hash<ColumnSpec, Boolean>] A hash mapping each column to `true` if it is a sub-element of another field, `false` otherwise.
+        # Time complexity: (O(NlogN+N*K)) where K is the number of conflict depth
+        #   without (`prefix_set`) memoization, it would be O(N^2)
+        def mark_sub_elements(columns)
+          # Sort columns by name length (ascending)
+          sorted_columns = columns.sort_by { |c| c.name.length }
+          prefix_set = Set.new # memoization set
+
+          sorted_columns.each_with_object({}) do |column, memo|
+            # Split the column name into parts (e.g., "user.profile.age" → ["user", "profile", "age"])
+            parts = column.name.split('.')
+
+            # Generate all possible parent prefixes (e.g., "user", "user.profile")
+            # and check if any parent prefix exists in the set
+            parent_prefixes = (0...parts.size - 1).map { |i| parts[0..i].join('.') }
+            memo[column] = parent_prefixes.any? { |prefix| prefix_set.include?(prefix) }
+            prefix_set.add(column.name)
+          end
+        end
       end
 
       # Class representing a column specification in the ESQL response['columns']

spec/inputs/elasticsearch_esql_spec.rb

@@ -54,7 +54,6 @@
       before do
         allow(esql_executor).to receive(:retryable).and_yield
         allow(client).to receive_message_chain(:esql, :query).and_return(response)
-        allow(plugin).to receive(:decorate_event)
       end
 
       it "executes the ESQL query and processes the results" do
@@ -87,7 +86,6 @@
       before do
         allow(esql_executor).to receive(:retryable).and_yield
         allow(client).to receive_message_chain(:esql, :query).and_return(response)
-        allow(plugin).to receive(:decorate_event)
         allow(response).to receive(:headers).and_return({})
       end
 
@@ -125,6 +123,36 @@
         end
       end
     end
+
+    context "when sub-elements occur in the result" do
+      let(:response) { {
+        'values' => [[50, 1, 100], [50, 0, 1000], [50, 9, 99999]],
+        'columns' =>
+          [
+            { 'name' => 'time', 'type' => 'long' },
+            { 'name' => 'time.min', 'type' => 'long' },
+            { 'name' => 'time.max', 'type' => 'long' },
+          ]
+      } }
+
+      before do
+        allow(esql_executor).to receive(:retryable).and_yield
+        allow(client).to receive_message_chain(:esql, :query).and_return(response)
+        allow(response).to receive(:headers).and_return({})
+      end
+
+      it "includes 1st depth elements into event" do
+        esql_executor.do_run(output_queue, plugin_config["query"])
+
+        expect(output_queue.size).to eq(3)
+        3.times do
+          event = output_queue.pop
+          expect(event.get('time')).to eq(50)
+          expect(event.get('[time][min]')).to eq(nil)
+          expect(event.get('[time][max]')).to eq(nil)
+        end
+      end
+    end
   end
 
   describe "#column spec" do