From c9891933ec6397632514a6d9c056357f6b534802 Mon Sep 17 00:00:00 2001 From: Ivan Ivanov Date: Thu, 23 Jul 2026 17:29:43 +0300 Subject: [PATCH] Add timeseriesDataJsonWithDuplicates API --- .../clickhouse/metric/metric_adapter.ex | 22 +++ .../metric/sql_query/metric_sql_query.ex | 46 ++++++ lib/sanbase/metric/behaviour.ex | 16 ++ lib/sanbase/metric/metric.ex | 31 ++++ .../graphql/document/document_provider.ex | 1 + .../access_control/access_control.ex | 1 + .../middlewares/transform_resolution.ex | 1 + .../resolvers/metric/metric_resolver.ex | 37 ++++- .../graphql/schema/types/metric_types.ex | 31 ++++ ...c_timeseries_data_with_duplicates_test.exs | 151 ++++++++++++++++++ 10 files changed, 335 insertions(+), 2 deletions(-) create mode 100644 test/sanbase_web/graphql/metric/api_metric_timeseries_data_with_duplicates_test.exs diff --git a/lib/sanbase/clickhouse/metric/metric_adapter.ex b/lib/sanbase/clickhouse/metric/metric_adapter.ex index d470e6ec58..9c323637e6 100644 --- a/lib/sanbase/clickhouse/metric/metric_adapter.ex +++ b/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 = diff --git a/lib/sanbase/clickhouse/metric/sql_query/metric_sql_query.ex b/lib/sanbase/clickhouse/metric/sql_query/metric_sql_query.ex index 52a9e3faf8..eab18812d6 100644 --- a/lib/sanbase/clickhouse/metric/sql_query/metric_sql_query.ex +++ b/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 diff --git a/lib/sanbase/metric/behaviour.ex b/lib/sanbase/metric/behaviour.ex index 9c7eaa4eda..a994731ab7 100644 --- a/lib/sanbase/metric/behaviour.ex +++ b/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, diff --git a/lib/sanbase/metric/metric.ex b/lib/sanbase/metric/metric.ex index 1c5ddbcc32..0c4360ad9b 100644 --- a/lib/sanbase/metric/metric.ex +++ b/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. diff --git a/lib/sanbase_web/graphql/document/document_provider.ex b/lib/sanbase_web/graphql/document/document_provider.ex index 8c936f0de6..af86bd8764 100644 --- a/lib/sanbase_web/graphql/document/document_provider.ex +++ b/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" diff --git a/lib/sanbase_web/graphql/middlewares/access_control/access_control.ex b/lib/sanbase_web/graphql/middlewares/access_control/access_control.ex index 3210ce15a0..d862b030c1 100644 --- a/lib/sanbase_web/graphql/middlewares/access_control/access_control.ex +++ b/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, diff --git a/lib/sanbase_web/graphql/middlewares/transform_resolution.ex b/lib/sanbase_web/graphql/middlewares/transform_resolution.ex index 5388efd7cb..7e3073fb63 100644 --- a/lib/sanbase_web/graphql/middlewares/transform_resolution.ex +++ b/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" diff --git a/lib/sanbase_web/graphql/resolvers/metric/metric_resolver.ex b/lib/sanbase_web/graphql/resolvers/metric/metric_resolver.ex index df223612f3..ec62f1c6f8 100644 --- a/lib/sanbase_web/graphql/resolvers/metric/metric_resolver.ex +++ b/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, diff --git a/lib/sanbase_web/graphql/schema/types/metric_types.ex b/lib/sanbase_web/graphql/schema/types/metric_types.ex index 2e217c7b78..d252e00801 100644 --- a/lib/sanbase_web/graphql/schema/types/metric_types.ex +++ b/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)) diff --git a/test/sanbase_web/graphql/metric/api_metric_timeseries_data_with_duplicates_test.exs b/test/sanbase_web/graphql/metric/api_metric_timeseries_data_with_duplicates_test.exs new file mode 100644 index 0000000000..a907e7abe9 --- /dev/null +++ b/test/sanbase_web/graphql/metric/api_metric_timeseries_data_with_duplicates_test.exs @@ -0,0 +1,151 @@ +defmodule SanbaseWeb.Graphql.ApiMetricTimeseriesDataWithDuplicatesTest do + use SanbaseWeb.ConnCase, async: false + + import Sanbase.Factory + import SanbaseWeb.Graphql.TestHelpers + + @metric "daily_active_addresses" + + # `timeseriesDataJsonWithDuplicates` returns the data as it is stored in + # Clickhouse - if an asset/metric/dt has multiple values with different + # computed_at, all of them are returned. computedAt is included by default. + + setup do + %{user: user} = + insert(:subscription_pro_sanbase, user: insert(:user, metric_access_level: "alpha")) + + project = insert(:random_project) + conn = setup_jwt_auth(build_conn(), user) + + [ + conn: conn, + slug: project.slug, + from: ~U[2019-01-01 00:00:00Z], + to: ~U[2019-01-03 00:00:00Z] + ] + end + + test "returns all values when a datetime has multiple computed_at", context do + %{conn: conn, slug: slug, from: from, to: to} = context + + rows = [ + row(~U[2019-01-01 00:00:00Z], 100.0, ~U[2019-01-01 12:00:00Z]), + row(~U[2019-01-01 00:00:00Z], 120.0, ~U[2019-01-02 12:00:00Z]), + row(~U[2019-01-02 00:00:00Z], 200.0, ~U[2019-01-02 12:00:00Z]) + ] + + result = run_json(conn, query(slug, from, to, []), rows) + + assert result == [ + %{ + "datetime" => "2019-01-01T00:00:00Z", + "value" => 100.0, + "computedAt" => "2019-01-01T12:00:00Z" + }, + %{ + "datetime" => "2019-01-01T00:00:00Z", + "value" => 120.0, + "computedAt" => "2019-01-02T12:00:00Z" + }, + %{ + "datetime" => "2019-01-02T00:00:00Z", + "value" => 200.0, + "computedAt" => "2019-01-02T12:00:00Z" + } + ] + end + + test "includeComputedAt: false returns only datetime and value", context do + %{conn: conn, slug: slug, from: from, to: to} = context + + rows = [ + row(~U[2019-01-01 00:00:00Z], 100.0, ~U[2019-01-01 12:00:00Z]), + row(~U[2019-01-01 00:00:00Z], 120.0, ~U[2019-01-02 12:00:00Z]) + ] + + result = run_json(conn, query(slug, from, to, include_computed_at: false), rows) + + assert result == [ + %{"datetime" => "2019-01-01T00:00:00Z", "value" => 100.0}, + %{"datetime" => "2019-01-01T00:00:00Z", "value" => 120.0} + ] + end + + test "fields renames the output keys", context do + %{conn: conn, slug: slug, from: from, to: to} = context + + rows = [row(~U[2019-01-01 00:00:00Z], 100.0, ~U[2019-01-01 12:00:00Z])] + + query = + query(slug, from, to, fields: ~s|{datetime: "d", value: "v", computedAt: "c"}|) + + result = run_json(conn, query, rows) + + assert result == [ + %{"d" => "2019-01-01T00:00:00Z", "v" => 100.0, "c" => "2019-01-01T12:00:00Z"} + ] + end + + @tag capture_log: true + test "returns an error for metrics not served by an adapter that supports it", context do + %{conn: conn, slug: slug, from: from, to: to} = context + + error_msg = + conn + |> post("/graphql", query_skeleton(query(slug, from, to, metric: "price_usd"), "getMetric")) + |> json_response(200) + |> get_in(["errors", Access.at(0), "message"]) + + assert error_msg =~ "does not support fetching timeseries data with duplicates" + end + + test "Sanbase.Metric returns an error when the adapter lacks the callback", context do + %{slug: slug, from: from, to: to} = context + + assert {:error, error_msg} = + Sanbase.Metric.timeseries_data_with_duplicates( + "price_usd", + %{slug: slug}, + from, + to, + "1d", + [] + ) + + assert error_msg =~ + "The metric price_usd does not support fetching timeseries data with duplicates" + end + + defp row(dt, value, computed_at), + do: [DateTime.to_unix(dt), value, DateTime.to_unix(computed_at)] + + defp run_json(conn, query, rows) do + Sanbase.Mock.prepare_mock2(&Sanbase.ClickhouseRepo.query/3, {:ok, %{rows: rows}}) + |> Sanbase.Mock.run_with_mocks(fn -> + conn + |> post("/graphql", query_skeleton(query, "getMetric")) + |> json_response(200) + |> get_in(["data", "getMetric", "timeseriesDataJsonWithDuplicates"]) + end) + end + + defp query(slug, from, to, opts) do + metric = Keyword.get(opts, :metric, @metric) + fields = Keyword.get(opts, :fields) + include_computed_at? = Keyword.get(opts, :include_computed_at, true) + + """ + { + getMetric(metric: "#{metric}"){ + timeseriesDataJsonWithDuplicates( + slug: "#{slug}" + from: "#{from}" + to: "#{to}" + #{if fields, do: "fields: #{fields}"} + includeComputedAt: #{include_computed_at?} + ) + } + } + """ + end +end