misc: Refact aggregation query to speedup field presence check

app/services/billable_metrics/aggregations/base_service.rb

@@ -85,6 +85,10 @@ def sanitized_field_name
         sanitized_name(billable_metric.field_name)
       end
 
+      def field_presence_condition
+        "events.properties::jsonb ? '#{ActiveRecord::Base.sanitize_sql_for_conditions(billable_metric.field_name)}'"
+      end
+
       def handle_in_advance_current_usage(total_aggregation)
         if previous_event
           aggregation = total_aggregation -

app/services/billable_metrics/aggregations/latest_service.rb

@@ -17,10 +17,9 @@ def aggregate(options: {})
       private
 
       def events
-        @events ||=
-          events_scope(from_datetime:, to_datetime:)
-            .where("#{sanitized_field_name} IS NOT NULL")
-            .reorder(timestamp: :desc, created_at: :desc)
+        @events ||= events_scope(from_datetime:, to_datetime:)
+          .where(field_presence_condition)
+          .reorder(timestamp: :desc, created_at: :desc)
       end
 
       def compute_aggregation(latest_event)

app/services/billable_metrics/aggregations/max_service.rb

@@ -36,7 +36,7 @@ def compute_per_event_aggregation
       private
 
       def fetch_events(from_datetime:, to_datetime:)
-        events_scope(from_datetime:, to_datetime:).where("#{sanitized_field_name} IS NOT NULL")
+        events_scope(from_datetime:, to_datetime:).where(field_presence_condition)
       end
     end
   end

app/services/billable_metrics/aggregations/sum_service.rb

@@ -99,7 +99,7 @@ def events
             events_scope(from_datetime:, to_datetime:)
           end
 
-          query.where("#{sanitized_field_name} IS NOT NULL")
+          query.where(field_presence_condition)
         end
       end
 
@@ -113,7 +113,7 @@ def previous_event
           else
             events_scope(from_datetime:, to_datetime:)
           end
-          scope = query.where("#{sanitized_field_name} IS NOT NULL")
+          scope = query.where(field_presence_condition)
 
           # Events without attached right metadata are ignored since such events cannot be processed correctly.
           # Could happen in race condition when event is stored but metadata in async job are attached later.

app/services/billable_metrics/aggregations/unique_count_service.rb

@@ -79,7 +79,7 @@ def previous_event
           end
           query = query
             .joins(:quantified_event)
-            .where("#{sanitized_field_name} IS NOT NULL")
+            .where(field_presence_condition)
             .where("events.metadata->>'current_aggregation' IS NOT NULL")
             .where("events.metadata->>'max_aggregation' IS NOT NULL")
             .where('quantified_events.added_at::timestamp(0) >= ?', from_datetime)

app/services/billable_metrics/aggregations/weighted_sum_service.rb

@@ -30,11 +30,11 @@ def fetch_events(from_datetime:, to_datetime:)
           # NOTE: When recurring we need to scope the fetch using the external ID to handle events
           #       sent to upgraded/downgraded subscription
           return recurring_events_scope(from_datetime:, to_datetime:)
-              .where("#{sanitized_field_name} IS NOT NULL")
+              .where(field_presence_condition)
               .order(timestamp: :asc)
         end
 
-        events_scope(from_datetime:, to_datetime:).where("#{sanitized_field_name} IS NOT NULL")
+        events_scope(from_datetime:, to_datetime:).where(field_presence_condition)
       end
 
       def compute_aggregation
@@ -155,7 +155,7 @@ def latest_value_from_events
           .where(created_at: billable_metric.created_at...)
           .where(timestamp: ..from_datetime)
           .where.not(subscription_id: subscription.id)
-          .where("#{sanitized_field_name} IS NOT NULL")
+          .where(field_presence_condition)
 
         return scope.sum("(#{sanitized_field_name})::numeric") unless group
 

app/services/billable_metrics/prorated_aggregations/sum_service.rb

@@ -58,17 +58,11 @@ def aggregation_query
       end
 
       def persisted_query
-        @persisted_query ||= begin
-          query = recurring_events_scope(to_datetime: from_datetime)
-          query.where("#{sanitized_field_name} IS NOT NULL")
-        end
+        @persisted_query ||= recurring_events_scope(to_datetime: from_datetime).where(field_presence_condition)
       end
 
       def period_query
-        @period_query ||= begin
-          query = recurring_events_scope(to_datetime:, from_datetime:)
-          query.where("#{sanitized_field_name} IS NOT NULL")
-        end
+        @period_query ||= recurring_events_scope(to_datetime:, from_datetime:).where(field_presence_condition)
       end
 
       # NOTE: Compute pro-rata of the duration in days between the datetimes over the duration of the billing period