Require drop null columns by default. Set limit to 1 by defaul if FROM query used. Add more debug, ward logs.

mashhurs · mashhurs · commit cac36808f2e5 · 2025-05-05T19:45:27.000-07:00
diff --git a/docs/index.asciidoc b/docs/index.asciidoc
@@ -135,7 +135,7 @@ To configure ES|QL query in the plugin, set your ES|QL query in the `query` para
 
 IMPORTANT: We recommend understanding https://www.elastic.co/guide/en/elasticsearch/reference/current/esql-limitations.html[ES|QL current limitations] before using it in production environments.
 
-The following is a basic ES|QL query that sets food name to transaction event based on event's food ID:
+The following is a basic ES|QL query that sets food name to transaction event based on upstream event's food ID:
 [source, ruby]
     filter {
       elasticsearch {
@@ -145,27 +145,20 @@ The following is a basic ES|QL query that sets food name to transaction event ba
           FROM food-index
             | WHERE id = "?food_id"
         '
-        fields => { "name" => "food_name" }
         query_params => {
-          named_params => ["food_id" => "%{[food_id]}"]
-          drop_null_columns => true
+          named_params => ["food_id" => "[food][id]"]
         }
+        fields => { "food.name" => "food_name" }
       }
     }
 
 Set `config.support_escapes: true` in `logstash.yml` if you need to escape special chars in the query.
 
-NOTE: With ES|QL query, {ls} doesn't generate `event.original`.
+In the result event, the plugin sets total result size in `[@metadata][total_hits]` field. It also limits the result size to 1 when `FROM` query is used.
 
-In the result event, the plugin sets total result size in `[@metadata][total_values]` field. It also limits the result size to 1 when `FROM` query is used.
+NOTE: If `FROM` execution command used and not `LIMIT` is set, the plugin attaches `| LIMIT 1`.
 
-Consider the following caveat scenarios:
-
-- ES|QL by default returns entire columns even if their values are `null`. The plugin provides a `drop_null_columns` option via <<plugins-{type}s-{plugin}-query_params>>. Enabling this parameter instructs {es} to automatically exclude columns with null values from query results.
-- If your {es} index uses https://www.elastic.co/docs/reference/elasticsearch/mapping-reference/multi-fields[multi-fields] mapping(s), ES|QL query fetches all parent and sub-fields fields. Since {ls} events cannot contain parent field's concrete value and sub-field values together, we recommend using the `DROP` keyword in your ES|QL query explicitly remove sub-fields.
-- If your {es} index contains top level `tags` field, this will conflict with {ls} event's reserved `tags` field. {ls} moves `tags` field values to the `_tags` and populates `tags` with `["_tagsparsefailure"]`.
-
-For comprehensive ES|QL syntax reference and best practices, see the https://www.elastic.co/guide/en/elasticsearch/reference/current/esql-syntax.html[official {es} documentation].
+For comprehensive ES|QL syntax reference and best practices, see the https://www.elastic.co/guide/en/elasticsearch/reference/current/esql-syntax.html[{es} ES|QL documentation].
 
 [id="plugins-{type}s-{plugin}-options"]
 ==== Elasticsearch Filter Configuration Options
diff --git a/lib/logstash/filters/elasticsearch.rb b/lib/logstash/filters/elasticsearch.rb
@@ -136,7 +136,6 @@ class LogStash::Filters::Elasticsearch < LogStash::Filters::Base
   #   named params can be applied as following via query_params:
   #   query_params => {
   #     "named_params" => [ {"type" => "%{[type]}"}]
-  #     "drop_null_columns" => true
   #   }
   config :query_params, :validate => :hash, :default => {}
 
@@ -177,8 +176,11 @@ def register
     query_type = resolve_query_type
     case query_type
     when "esql"
+      invalid_params_with_esql = original_params.keys & %w(index query_template sort docinfo_fields aggregation_fields enable_sort result_size)
+      raise LogStash::ConfigurationError, "Configured #{invalid_params_with_esql} params cannot be used with ES|QL query" if invalid_params_with_esql.any?
+
       validate_ls_version_for_esql_support!
-      validate_params_with_esql_query!
+      validate_esql_query_and_params!
       @esql_executor ||= LogStash::Filters::Elasticsearch::EsqlExecutor.new(self, @logger)
     else # dsl
       validate_dsl_query_settings!
@@ -448,19 +450,25 @@ def validate_ls_version_for_esql_support!
     end
   end
 
-  def validate_params_with_esql_query!
-    invalid_params_with_esql = original_params.keys & %w(index query_template sort docinfo_fields aggregation_fields enable_sort result_size)
-    fail("Configured #{invalid_params_with_esql} params cannot be used with ES|QL query") if invalid_params_with_esql.any?
-
-    accepted_query_params = %w(named_params drop_null_columns)
+  def validate_esql_query_and_params!
+    accepted_query_params = %w(named_params)
     original_query_params = original_params["query_params"] ||= {}
     invalid_query_params = original_query_params.keys - accepted_query_params
-    fail("#{accepted_query_params} options are accepted in `query_params`, but found #{invalid_query_params} invalid option(s)") if invalid_query_params.any?
+    raise LogStash::ConfigurationError, "#{accepted_query_params} option(s) accepted in `query_params`, but found #{invalid_query_params} invalid option(s)" if invalid_query_params.any?
 
     is_named_params_array = original_query_params["named_params"] ? original_query_params["named_params"].class.eql?(Array) : true
-    fail("`query_params => named_params` is required to be array") unless is_named_params_array
+    raise LogStash::ConfigurationError, "`query_params => named_params` is required to be array" unless is_named_params_array
+
+    named_params = original_query_params["named_params"] ||= []
+    named_params_keys = named_params.map(&:keys).flatten
 
-    # TODO: validate that placeholders in query should match the named_params
+    placeholders = @query.scan(/\?(\w+)/).flatten
+    raise LogStash::ConfigurationError, "Number of placeholders in `query` and `named_params` do not match" unless placeholders.size == named_params_keys.size
+
+    placeholders.each do |placeholder|
+      placeholder.delete_prefix!("?")
+      raise LogStash::ConfigurationError, "Placeholder #{placeholder} not found in query" unless named_params_keys.include?(placeholder)
+    end
   end
 
   def validate_es_for_esql_support!
diff --git a/lib/logstash/filters/elasticsearch/esql_executor.rb b/lib/logstash/filters/elasticsearch/esql_executor.rb
@@ -7,14 +7,19 @@ class EsqlExecutor
 
         def initialize(plugin, logger)
           @plugin = plugin
+          @logger = logger
 
-          params = plugin.params["query_params"] || {}
-          @drop_null_columns = params["drop_null_columns"] || false
-          @named_params = params["named_params"] || []
           @query = plugin.params["query"]
+          if @query.strip.start_with?("FROM") && !@query.match?(/\|\s*LIMIT/)
+            @logger.warn("ES|QL query doesn't contain LIMIT, adding `| LIMIT 1` to optimize the performance")
+            @query.concat(' | LIMIT 1')
+          end
+
+          query_params = plugin.params["query_params"] || {}
+          @named_params = query_params["named_params"] || []
           @fields = plugin.params["fields"]
           @tag_on_failure = plugin.params["tag_on_failure"]
-          @logger = logger
+          @logger.debug("ES|QL query executor initialized with ", query: @query, named_params: @named_params)
         end
 
         def process(client, event)
@@ -34,8 +39,11 @@ def resolve_parameters(event)
           @named_params.map do |entry|
             entry.each_with_object({}) do |(key, value), new_entry|
               begin
-                new_entry[key] = event.sprintf(value)
+                resolved_value = event.get(value)
+                @logger.debug("Resolved value for #{key}: #{resolved_value}, its class: #{resolved_value.class}")
+                new_entry[key] = resolved_value
               rescue => e
+                # catches invalid field reference
                 @logger.error("Failed to resolve parameter", key: key, value: value, error: e.message)
                 raise
               end
@@ -44,30 +52,45 @@ def resolve_parameters(event)
         end
 
         def execute_query(client, params)
+          # debug logs  may help to check what query shape the plugin is sending to ES
           @logger.debug("Executing ES|QL query", query: @query, params: params)
-          client.search({ body: { query: @query, params: params }, format: 'json', drop_null_columns: @drop_null_columns }, 'esql')
+          client.search({ body: { query: @query, params: params }, format: 'json', drop_null_columns: true }, 'esql')
         end
 
         def process_response(event, response)
-          return unless response['values'] && response['columns']
+          columns = response['columns'].freeze
+          values = response['values'].freeze
+          if values.nil? || values.size == 0
+            @logger.debug("Empty ES|QL query result", columns: columns, values: values)
+            return
+          end
+
+          # this shouldn't never happen but just in case not crash the plugin
+          if columns.nil? || columns.size == 0
+            @logger.error("No columns exist but received values", columns: columns, values: values)
+            return
+          end
 
-          # TODO: set to the target field once target support is added
-          event.set("[@metadata][total_values]", response['values'].size)
-          add_requested_fields(event, response)
+          # TODO: do we need to set `total_hits` to target?
+          #   if not, how do we resolve conflict with existing es-input total_hits field?
+          #   FYI: with DSL it stores in `[@metadata][total_hits]`
+          event.set("[@metadata][total_hits]", values.size)
+          add_requested_fields(event, columns, values)
         end
 
         def inform_warning(response)
           return unless (warning = response&.headers&.dig('warning'))
           @logger.warn("ES|QL executor received warning", { message: warning })
         end
 
-        def add_requested_fields(event, response)
+        def add_requested_fields(event, columns, values)
           @fields.each do |old_key, new_key|
-            column_index = response['columns'].find_index { |col| col['name'] == old_key }
+            column_index = columns.find_index { |col| col['name'] == old_key }
             next unless column_index
 
-            values = response['values'].map { |entry| entry[column_index] }
-            event.set(new_key, values.one? ? values.first : values) if values&.size > 0
+            row_values = values.map { |entry| entry[column_index] }&.compact # remove non-exist field values with compact
+            # TODO: set to the target field once target support is added
+            event.set(new_key, row_values.one? ? row_values.first : row_values) if row_values&.size > 0
           end
         end
       end
diff --git a/spec/filters/elasticsearch_esql_spec.rb b/spec/filters/elasticsearch_esql_spec.rb
@@ -13,43 +13,44 @@
   end
   let(:esql_executor) { described_class.new(plugin, logger) }
 
-  describe "when initializes" do
+  context "when initializes" do
     it "sets up the ESQL client with correct parameters" do
+      allow(logger).to receive(:debug)
       expect(esql_executor.instance_variable_get(:@query)).to eq(plugin_config["query"])
       expect(esql_executor.instance_variable_get(:@named_params)).to eq([])
-      expect(esql_executor.instance_variable_get(:@drop_null_columns)).to eq(false)
       expect(esql_executor.instance_variable_get(:@fields)).to eq({})
       expect(esql_executor.instance_variable_get(:@tag_on_failure)).to eq(["_elasticsearch_lookup_failure"])
     end
   end
 
-  describe "when processes" do
+  context "when processes" do
     let(:plugin_config) {
       super()
         .merge(
           {
             "query" => "FROM my-index | WHERE field = ?foo | LIMIT 5",
-            "query_params" => { "named_params" => [{ "foo" => "%{bar}" }] },
-            "fields" => { "val" => "val_new" }
+            "query_params" => { "named_params" => [{ "foo" => "[bar]" }] },
+            "fields" => { "val" => "val_new", "odd" => "new_odd" }
           })
     }
     let(:event) { LogStash::Event.new({}) }
-    let(:response) { { 'values' => [%w[foo bar]], 'columns' => [{ 'name' => 'id' }, { 'name' => 'val' }] } }
+    let(:response) { { 'values' => [["foo", "bar", nil]], 'columns' => [{ 'name' => 'id' }, { 'name' => 'val' }, { 'name' => 'odd' }] } }
 
     before do
-      allow(event).to receive(:sprintf).and_return("resolved_value")
+      allow(logger).to receive(:debug)
     end
 
     it "resolves parameters" do
-      expect(event).to receive(:sprintf).with("%{bar}").and_return("resolved_value")
+      expect(event).to receive(:get).with("[bar]").and_return("resolved_value")
       resolved_params = esql_executor.send(:resolve_parameters, event)
       expect(resolved_params).to include("foo" => "resolved_value")
     end
 
     it "executes the query with resolved parameters" do
       allow(logger).to receive(:debug)
+      expect(event).to receive(:get).with("[bar]").and_return("resolved_value")
       expect(client).to receive(:search).with(
-        { body: { query: plugin_config["query"], params: [{ "foo" => "resolved_value" }] }, format: 'json', drop_null_columns: false, },
+        { body: { query: plugin_config["query"], params: [{ "foo" => "resolved_value" }] }, format: 'json', drop_null_columns: true, },
         'esql')
       resolved_params = esql_executor.send(:resolve_parameters, event)
       esql_executor.send(:execute_query, client, resolved_params)
@@ -62,7 +63,7 @@
     end
 
     it "processes the response and adds metadata" do
-      expect(event).to receive(:set).with("[@metadata][total_values]", 1)
+      expect(event).to receive(:set).with("[@metadata][total_hits]", 1)
       expect(event).to receive(:set).with("val_new", "bar")
       esql_executor.send(:process_response, event, response)
     end
@@ -73,19 +74,25 @@
       allow(response).to receive(:headers).and_return({})
       expect(client).to receive(:search).with(
         {
-          body: { query: plugin_config["query"], params: plugin_config["query_params"]["named_params"] },
+          body: { query: plugin_config["query"], params: [{"foo"=>"resolve_me"}] },
           format: 'json',
-          drop_null_columns: false,
+          drop_null_columns: true,
         },
         'esql'
       ).and_return(response)
-      expect { esql_executor.process(client, LogStash::Event.new({ "hello" => "world" })) }.to_not raise_error
+
+      event = LogStash::Event.new({ "hello" => "world", "bar" => "resolve_me" })
+      expect { esql_executor.process(client, event) }.to_not raise_error
+      expect(event.get("[@metadata][total_hits]")).to eq(1)
+      expect(event.get("hello")).to eq("world")
+      expect(event.get("val_new")).to eq("bar")
+      expect(event.get("new_odd")).to be_nil # filters out non-exist fields
     end
 
     it "tags on plugin failures" do
-      expect(event).to receive(:sprintf).with("%{bar}").and_raise("Event#sprintf error")
+      expect(event).to receive(:get).with("[bar]").and_raise("Event#get Invalid FieldReference error")
 
-      expect(logger).to receive(:error).with("Failed to resolve parameter", {:error=>"Event#sprintf error", :key=>"foo", :value=>"%{bar}"})
+      expect(logger).to receive(:error).with("Failed to resolve parameter", {:error=>"Event#get Invalid FieldReference error", :key=>"foo", :value=>"[bar]"})
       expect(logger).to receive(:error).with("Failed to process ES|QL filter", exception: instance_of(RuntimeError))
       expect(event).to receive(:tag).with("_elasticsearch_lookup_failure")
       esql_executor.process(client, event)
diff --git a/spec/filters/elasticsearch_spec.rb b/spec/filters/elasticsearch_spec.rb
@@ -864,6 +864,108 @@ def wait_receive_request
     end
   end
 
+  describe "ES|QL" do
+
+    describe "compatibility" do
+      let(:config) {{ "hosts" => ["localhost:9200"], "query" => "FROM my-index" }}
+
+      context "when LS doesn't support ES|QL" do
+        let(:ls_version) { LogStash::Filters::Elasticsearch::LS_ESQL_SUPPORT_VERSION }
+        before(:each) do
+          stub_const("LOGSTASH_VERSION", "8.17.0")
+        end
+
+        it "raises a runtime error" do
+          expect { plugin.send(:validate_ls_version_for_esql_support!) }
+            .to raise_error(RuntimeError, /Current version of Logstash does not include Elasticsearch client which supports ES|QL. Please upgrade Logstash to at least #{ls_version}/)
+        end
+      end
+
+      context "when ES doesn't support ES|QL" do
+        let(:es_version) { LogStash::Filters::Elasticsearch::ES_ESQL_SUPPORT_VERSION }
+        let(:client) { double(:client) }
+
+        it "raises a runtime error" do
+          allow(plugin).to receive(:get_client).twice.and_return(client)
+          allow(client).to receive(:es_version).and_return("8.8.0")
+
+          expect { plugin.send(:validate_es_for_esql_support!) }
+            .to raise_error(RuntimeError, /Connected Elasticsearch 8.8.0 version does not supports ES|QL. ES|QL feature requires at least Elasticsearch #{es_version} version./)
+        end
+      end
+    end
+
+    context "when non-ES|QL params applied" do
+      let(:config) do
+        {
+          "hosts" => ["localhost:9200"],
+          "query" => "FROM my-index",
+          "index" => "some-index",
+          "docinfo_fields" => { "_index" => "es_index" },
+          "sort" => "@timestamp:desc",
+          "enable_sort" => true,
+          "aggregation_fields" => { "bytes_avg" => "bytes_avg_ls_field" }
+        }
+      end
+      it "raises a config error" do
+        invalid_params_with_esql = %w(index docinfo_fields sort enable_sort aggregation_fields)
+        error_text = /Configured #{invalid_params_with_esql} params cannot be used with ES|QL query/i
+        expect { plugin.register }.to raise_error LogStash::ConfigurationError, error_text
+      end
+    end
+
+    context "when `named_params` isn't array" do
+      let(:config) do
+        {
+          "hosts" => ["localhost:9200"],
+          "query" => "FROM my-index",
+          "query_params" => { "named_params" => {"a" => "b"} },
+        }
+      end
+      it "raises a config error" do
+        expect { plugin.register }.to raise_error LogStash::ConfigurationError, /`query_params => named_params` is required to be array/
+      end
+    end
+
+    context "when `named_params` exists but not placeholder in the query" do
+      let(:config) do
+        {
+          "hosts" => ["localhost:9200"],
+          "query" => "FROM my-index",
+          "query_params" => { "named_params" => [{"a" => "b"}] },
+        }
+      end
+      it "raises a config error" do
+        expect { plugin.register }.to raise_error LogStash::ConfigurationError, /Number of placeholders in `query` and `named_params` do not match/
+      end
+    end
+
+    context "when `named_params` doesn't exist but placeholder found" do
+      let(:config) do
+        {
+          "hosts" => ["localhost:9200"],
+          "query" => "FROM my-index | WHERE a = ?a"
+        }
+      end
+      it "raises a config error" do
+        expect { plugin.register }.to raise_error LogStash::ConfigurationError, /Number of placeholders in `query` and `named_params` do not match/
+      end
+    end
+
+    context "when placeholder and `named_params` do not match" do
+      let(:config) do
+        {
+          "hosts" => ["localhost:9200"],
+          "query" => "FROM my-index | WHERE type = ?type",
+          "query_params" => { "named_params" => [{"b" => "c"}] },
+        }
+      end
+      it "raises a config error" do
+        expect { plugin.register }.to raise_error LogStash::ConfigurationError, /Placeholder type not found in query/
+      end
+    end
+  end
+
   def extract_transport(client)
     # on 7x: client.transport.transport
     # on >=8.x: client.transport
diff --git a/spec/filters/integration/elasticsearch_esql_spec.rb b/spec/filters/integration/elasticsearch_esql_spec.rb
