Add timeseriesDataJsonWithDuplicates API

lib/sanbase/clickhouse/metric/metric_adapter.ex

@@ -86,6 +86,28 @@ defmodule Sanbase.Clickhouse.MetricAdapter do
     {:error, unsupported_selector_error(selector)}
   end
 
+  @doc ~s"""
+  Like timeseries_data/6, but without deduplication by computed_at and without
+  interval/aggregation - every stored row in the time range is returned as-is.
+  """
+  @impl Sanbase.Metric.Behaviour
+  def timeseries_data_with_duplicates(_metric, %{slug: []}, _from, _to, _interval, _opts),
+    do: {:ok, []}
+
+  def timeseries_data_with_duplicates(metric, selector, from, to, _interval, opts)
+      when is_supported_selector(selector) do
+    opts = resolve_fixed_parameters(opts, metric)
+    filters = get_filters(metric, opts)
+
+    timeseries_data_with_duplicates_query(metric, selector, from, to, filters, opts)
+    |> exec_timeseries_data_query()
+  end
+
+  def timeseries_data_with_duplicates(_metric, selector, _from, _to, _interval, _opts)
+      when is_map(selector) do
+    {:error, unsupported_selector_error(selector)}
+  end
+
   @impl Sanbase.Metric.Behaviour
   def timeseries_data_per_slug(metric, %{slug: slug}, from, to, interval, opts) do
     aggregation =

lib/sanbase/clickhouse/metric/sql_query/metric_sql_query.ex

@@ -79,6 +79,52 @@ defmodule Sanbase.Clickhouse.MetricAdapter.SqlQuery do
     Sanbase.Clickhouse.Query.new(sql, params)
   end
 
+  # Unlike timeseries_data_query/8, rows are NOT deduplicated by computed_at
+  # (no argMax) and NOT aggregated by interval/asset - if an asset/metric/dt
+  # has multiple values with different computed_at, all of them are returned.
+  def timeseries_data_with_duplicates_query(metric, selector, from, to, filters, opts) do
+    version = Keyword.get(opts, :version) || @default_version
+
+    params = %{
+      metric: Map.get(Registry.name_to_metric_map(), metric),
+      from: dt_to_unix(:from, from),
+      to: dt_to_unix(:to, to),
+      selector: asset_filter_value(selector),
+      version: version,
+      table: deduce_table(metric, opts)
+    }
+
+    only_finalized_data = Keyword.get(opts, :only_finalized_data, false)
+
+    {additional_filters, params} =
+      maybe_get_additional_filters(metric, filters, params, trailing_and: true)
+
+    {fixed_parameters_str, params} =
+      maybe_get_fixed_parameters(metric, version, selector, params, opts ++ [trailing_and: true])
+
+    sql =
+      """
+      SELECT
+        toUnixTimestamp(dt) AS t,
+        value,
+        toUnixTimestamp(computed_at) AS computed_at
+      FROM {{table:inline}}
+      WHERE
+        #{finalized_data_filter_str(params.table, only_finalized_data)}
+        #{fixed_parameters_str}
+        #{additional_filters}
+        #{maybe_add_is_not_nan_check(params.table, column_name: "value", trailing_and: true)}
+        isNotNull(value) AND
+        #{maybe_convert_to_date(:after, metric, "dt", "toDateTime({{from}})")} AND
+        #{maybe_convert_to_date(:before, metric, "dt", "toDateTime({{to}})")} AND
+        #{asset_id_filter(selector, argument_name: "selector", allow_missing_slug: true)} AND
+        #{versioned_metric_id_filter(metric, argument_name: "metric", version: version, version_arg_name: "version")}
+      ORDER BY dt, computed_at
+      """
+
+    Sanbase.Clickhouse.Query.new(sql, params)
+  end
+
   defp maybe_get_fixed_parameters(_metric, _version, selector, params, _opts)
        when is_map_key(selector, :label_fqn) or is_map_key(selector, :label_fqns) do
     # In some cases like 'historical_balance_centralized_exchanges' the

lib/sanbase/metric/behaviour.ex

@@ -156,6 +156,19 @@ defmodule Sanbase.Metric.Behaviour do
             ) ::
               timeseries_data_result
 
+  # Like timeseries_data/6, but the result is not deduplicated by computed_at -
+  # if an asset/metric/dt has multiple values with different computed_at, all
+  # of them are returned. Each point includes a :computed_at key.
+  @callback timeseries_data_with_duplicates(
+              metric :: metric(),
+              selector :: selector,
+              from :: DatetTime.t(),
+              to :: DateTime.t(),
+              interval :: interval(),
+              opts :: opts
+            ) ::
+              timeseries_data_result
+
   @callback timeseries_data_per_slug(
               metric :: metric(),
               selector :: selector,
@@ -281,6 +294,9 @@ defmodule Sanbase.Metric.Behaviour do
   @optional_callbacks [
     histogram_data: 6,
     table_data: 5,
+    # Only the Clickhouse adapter can return data with duplicated dt values
+    # (multiple values with different computed_at for the same dt)
+    timeseries_data_with_duplicates: 6,
     deprecated_metrics_map: 0,
     fixed_labels_parameters_metrics: 0,
     soft_deprecated_metrics_map: 0,

lib/sanbase/metric/metric.ex

@@ -282,6 +282,37 @@ defmodule Sanbase.Metric do
     end
   end
 
+  @doc ~s"""
+  Returns timeseries data without deduplication by computed_at - if an
+  asset/metric/dt has multiple values with different computed_at, all of them
+  are returned. No interval aggregation is applied - the data is returned as
+  it is stored. Supported only by adapters that implement the optional
+  timeseries_data_with_duplicates/6 callback (the Clickhouse adapter).
+  """
+  @spec timeseries_data_with_duplicates(metric, selector, datetime, datetime, interval, opts) ::
+          Type.timeseries_data_result()
+  def timeseries_data_with_duplicates(metric, selector, from, to, interval, opts \\ [])
+
+  def timeseries_data_with_duplicates(metric, selector, from, to, interval, opts) do
+    metric = maybe_replace_metric(metric, selector)
+
+    case get_module(metric, selector: selector, opts: opts) do
+      nil ->
+        metric_not_available_error(metric, type: :timeseries)
+
+      module when is_atom(module) ->
+        with :ok <- check_metric_data_type(metric, :timeseries) do
+          if function_exported?(module, :timeseries_data_with_duplicates, 6) do
+            module.timeseries_data_with_duplicates(metric, selector, from, to, interval, opts)
+            |> maybe_round_floats(:timeseries_data)
+          else
+            {:error,
+             "The metric #{metric} does not support fetching timeseries data with duplicates"}
+          end
+        end
+    end
+  end
+
   @doc ~s"""
   Returns timeseries data (pairs of datetime and float value) for every slug
   separately.

lib/sanbase_web/graphql/document/document_provider.ex

@@ -338,6 +338,7 @@ defmodule SanbaseWeb.Graphql.Phase.Document.Complexity.Preprocess do
         [
           "timeseries_data",
           "timeseries_data_json",
+          "timeseries_data_json_with_duplicates",
           "timeseries_data_per_slug",
           "timeseries_data_per_slug_json",
           "aggregated_timeseries_data"

lib/sanbase_web/graphql/middlewares/access_control/access_control.ex

@@ -620,6 +620,7 @@ defmodule SanbaseWeb.Graphql.Middlewares.AccessControl do
     :aggregated_timeseries_data,
     :timeseries_data,
     :timeseries_data_json,
+    :timeseries_data_json_with_duplicates,
     :timeseries_data_per_slug,
     :timeseries_data_per_slug_json,
     :table_data,

lib/sanbase_web/graphql/middlewares/transform_resolution.ex

@@ -46,6 +46,7 @@ defmodule SanbaseWeb.Graphql.Middlewares.TransformResolution do
   @fields_with_selector [
     "timeseriesData",
     "timeseriesDataJson",
+    "timeseriesDataJsonWithDuplicates",
     "timeseriesDataPerSlug",
     "timeseriesDataPerSlugJson",
     "aggregatedTimeseriesData"

lib/sanbase_web/graphql/resolvers/metric/metric_resolver.ex

@@ -25,7 +25,11 @@ defmodule SanbaseWeb.Graphql.Resolvers.MetricResolver do
   # serialized), while the typed fields let GraphQL field selection filter keys.
   # `timeseries_data/3` and `timeseries_data_per_slug/3` each back both a typed
   # and a JSON field, so we look at which schema field is being resolved.
-  @json_field_identifiers [:timeseries_data_json, :timeseries_data_per_slug_json]
+  @json_field_identifiers [
+    :timeseries_data_json,
+    :timeseries_data_per_slug_json,
+    :timeseries_data_json_with_duplicates
+  ]
 
   def get_metric(_root, %{metric: metric} = args, resolution) do
     # TODO: Check that the version is also deprecated
@@ -314,6 +318,26 @@ defmodule SanbaseWeb.Graphql.Resolvers.MetricResolver do
     e -> reraise(e, __STACKTRACE__)
   end
 
+  def timeseries_data_with_duplicates(
+        _root,
+        args,
+        %{source: %{metric: metric, version: version}} = resolution
+      ) do
+    requested_fields = requested_fields(resolution)
+
+    fetch_timeseries_data(
+      metric,
+      version,
+      args,
+      requested_fields,
+      :timeseries_data_with_duplicates,
+      _json_variant? = true
+    )
+  rescue
+    e in [Sanbase.Metric.CatchableError] -> {:error, Exception.message(e)}
+    e -> reraise(e, __STACKTRACE__)
+  end
+
   def timeseries_data_per_slug(
         _root,
         args,
@@ -531,7 +555,11 @@ defmodule SanbaseWeb.Graphql.Resolvers.MetricResolver do
   # exactly the same way. The only difference is the function that is called
   # from the Metric module.
   defp fetch_timeseries_data(metric, version, args, requested_fields, function, json_variant?)
-       when function in [:timeseries_data, :timeseries_data_per_slug] do
+       when function in [
+              :timeseries_data,
+              :timeseries_data_per_slug,
+              :timeseries_data_with_duplicates
+            ] do
     only_finalized_data = Map.get(args, :only_finalized_data, false)
 
     with {:ok, selector} <- args_to_selector(args, use_process_dictionary: true),
@@ -608,6 +636,11 @@ defmodule SanbaseWeb.Graphql.Resolvers.MetricResolver do
     |> maybe_put_computed_at(point, fields, append?)
   end
 
+  # The with-duplicates rows have the same shape as the regular timeseries_data
+  # ones (multiple rows just share a datetime), so they serialize the same way.
+  defp json_row(point, :timeseries_data_with_duplicates, fields, append?),
+    do: json_row(point, :timeseries_data, fields, append?)
+
   defp json_row(%{value: value} = point, :timeseries_data, fields, append?) do
     %{
       output_key(fields, :datetime) => point.datetime,

lib/sanbase_web/graphql/schema/types/metric_types.ex

@@ -676,6 +676,37 @@ defmodule SanbaseWeb.Graphql.MetricTypes do
       middleware(AfterCacheTransform)
     end
 
+    @desc ~s"""
+    Similar to `timeseriesDataJson`, but the data is not deduplicated by the
+    `computedAt` value - if a given asset/metric/datetime has multiple values
+    with different `computedAt`, all of them are returned. The data is returned
+    as it is stored, without interval aggregation - the `interval` argument is
+    used only for complexity computation. Supported only by metrics stored in
+    Clickhouse.
+    """
+    field :timeseries_data_json_with_duplicates, :json do
+      arg(:slug, :string)
+      arg(:selector, :metric_target_selector_input_object)
+      arg(:from, non_null(:datetime))
+      arg(:to, non_null(:datetime))
+      arg(:interval, :interval, default_value: "1d")
+      arg(:include_incomplete_data, :boolean, default_value: false)
+      arg(:only_finalized_data, :boolean, default_value: false)
+      arg(:caching_params, :caching_params_input_object)
+      arg(:fields, :timeseries_data_json_fields)
+      arg(:include_computed_at, :boolean, default_value: true)
+
+      complexity(&Complexity.from_to_interval/3)
+      middleware(AccessControl, resolve_slugs_list: true)
+
+      cache_resolve(&MetricResolver.timeseries_data_with_duplicates/3,
+        ttl: 300,
+        max_ttl_offset: 90
+      )
+
+      middleware(AfterCacheTransform)
+    end
+
     field :timeseries_data_per_slug_json, :json do
       arg(:selector, :metric_target_selector_input_object)
       arg(:from, non_null(:datetime))