services/horizon: Refactor trade aggregation query.

[erika-sdf]

PR Checklist

PR Structure

• This PR has reasonably narrow scope (if not, break it down into smaller PRs).
• This PR avoids mixing refactoring changes with feature changes (split into two PRs
  otherwise).
• This PR's title starts with name of package that is most changed in the PR, ex.
  services/friendbot, or all or doc if the changes are broad or impact many
  packages.

Thoroughness

• This PR adds tests for the most critical parts of the new functionality or fixes.
• I've updated any docs (developer docs, .md
  files, etc... affected by this change). Take a look in the docs folder for a given service,
  like this one.

Release planning

• I've updated the relevant CHANGELOG (here for Horizon) if
  needed with deprecations, added features, breaking changes, and DB schema changes.
• I've decided if this PR requires a new major/minor version according to
  semver, or if it's mainly a patch change. The PR is targeted at the next
  release branch if it's not a patch change.

What

Refactor trade aggregation query to be more performant.

Why

Expensive queries are hogging DB connections.

[sreuland]

nice work, sequencing all the sql fragments together, it's hard to visualize the final sql state in these types of dynamic assemblies, any chance you could post a sample of what TradeAggregationsQ .GetSql() emits, just to visually see the net effect, or is it logged off a test case that can take a look?

[sydneynotthecity]

Overall looks good. I have a couple questions regarding preserving ordering and limit preferences from the user. I think we make some assumptions based on the specific query we used as a template and need to make sure ASC and DESC are parameterized appropriately.

I agree with @sreuland's idea of printing the SQL built by the code so that we can confirm it works as intended. If we're able to, we should try several input combinations to validate that we do not impact results regardless of user inputs.

[sydneynotthecity]

Should this timestamp be qualified? (ie tr.timestamp). Or does the open_ledger_toid preserve original raw timestamp order enough to guarantee that the ordering would not be incorrect ever?

[erika-sdf]

@paulbellamy do you have any insight here?

[sreuland]

grabbed the emitted sql off the integration test output:

WITH last_200_ts AS (
  SELECT 
    (
      (timestamp - 0) / 86400000
    ) * 86400000 + 0 AS timestamp 
  FROM 
    history_trades_60000 
  WHERE 
    base_asset_id = ? 
    AND counter_asset_id = ? 
    AND timestamp >= ? 
  GROUP BY 
    (
      (timestamp - 0) / 86400000
    ) * 86400000 + 0 
  ORDER BY 
    (
      (timestamp - 0) / 86400000
    ) * 86400000 + 0 DESC FETCH FIRST 200 ROWS ONLY
), 
timestamp_range AS (
  SELECT 
    min(timestamp) as min_ts, 
    max(timestamp) as max_ts 
  FROM 
    last_200_ts
), 
raw_trades AS (
  SELECT 
    (
      (tr.timestamp - 0) / 86400000
    ) * 86400000 + 0 AS timestamp, 
    count, 
    base_volume, 
    counter_volume, 
    avg, 
    high_n, 
    high_d, 
    low_n, 
    low_d, 
    open_n, 
    open_d, 
    close_n, 
    close_d 
  FROM 
    history_trades_60000 AS tr 
    JOIN timestamp_range r ON 1 = 1 
  WHERE 
    base_asset_id = ? 
    AND counter_asset_id = ? 
    AND r.max_ts >= (
      (tr.timestamp - 0) / 86400000
    ) * 86400000 + 0 
    AND r.min_ts <= (
      (tr.timestamp - 0) / 86400000
    ) * 86400000 + 0 
  ORDER BY 
    timestamp ASC, 
    open_ledger_toid ASC
) 
SELECT 
  timestamp, 
  sum(
    \ "count\") as count, sum(base_volume) as base_volume, sum(counter_volume) as counter_volume, sum(counter_volume::numeric)/sum(base_volume::numeric) as avg, (max_price(ARRAY[high_n, high_d]))[1] as high_n, (max_price(ARRAY[high_n, high_d]))[2] as high_d, (min_price(ARRAY[low_n, low_d]))[1] as low_n, (min_price(ARRAY[low_n, low_d]))[2] as low_d, (first(ARRAY[open_n, open_d]))[1] as open_n, (first(ARRAY[open_n, open_d]))[2] as open_d, (last(ARRAY[close_n, close_d]))[1] as close_n, (last(ARRAY[close_n, close_d]))[2] as close_d FROM raw_trades GROUP BY timestamp ORDER BY timestamp  LIMIT 100

[erika-sdf]

Some requests and their resulting sql

/trade_aggregations?base_asset_type=native&counter_asset_type=credit_alphanum4&counter_asset_code=USD&counter_asset_issuer=GDUKMGUGDZQK6YHYA5Z6AY2G4XDSZPSZ3SW5UN3ARVMO6QSRDWP5YLEX&resolution=86400000&order=desc&limit=200

WITH last_range_ts AS (
    SELECT ((timestamp - 0) / 86400000) * 86400000 + 0 AS timestamp
    FROM history_trades_60000
    WHERE base_asset_id = ? AND counter_asset_id = ? AND timestamp >= ?
    GROUP BY ((timestamp - 0) / 86400000) * 86400000 + 0
    ORDER BY ((timestamp - 0) / 86400000) * 86400000 + 0 desc
    FETCH FIRST 200 ROWS ONLY
), timestamp_range AS (
    SELECT min(timestamp) as min_ts,
    max(timestamp) as max_ts FROM last_range_ts
), raw_trades AS (
    SELECT ((tr.timestamp - 0) / 86400000) * 86400000 + 0 AS timestamp, 
    count,
    base_volume,
    counter_volume,
    avg,
    high_n, high_d, low_n, low_d,
    open_n, open_d, close_n, close_d 
    FROM history_trades_60000 AS tr
    JOIN timestamp_range r ON 1=1
    WHERE base_asset_id = ? AND counter_asset_id = ?
    AND r.max_ts >= ((tr.timestamp - 0) / 86400000) * 86400000 + 0
    AND r.min_ts <= ((tr.timestamp - 0) / 86400000) * 86400000 + 0
    ORDER BY timestamp ASC, open_ledger_toid ASC
) 
SELECT timestamp,
    sum(\"count\") as count,
    sum(base_volume) as base_volume,
    sum(counter_volume) as counter_volume,
    sum(counter_volume::numeric)/sum(base_volume::numeric) as avg,
    (max_price(ARRAY[high_n, high_d]))[1] as high_n,
    (max_price(ARRAY[high_n, high_d]))[2] as high_d,
    (min_price(ARRAY[low_n, low_d]))[1] as low_n,
    (min_price(ARRAY[low_n, low_d]))[2] as low_d,
    (first(ARRAY[open_n, open_d]))[1] as open_n,
    (first(ARRAY[open_n, open_d]))[2] as open_d,
    (last(ARRAY[close_n, close_d]))[1] as close_n,
    (last(ARRAY[close_n, close_d]))[2] as close_d
FROM raw_trades
GROUP BY timestamp
ORDER BY timestamp desc
LIMIT 200

/trade_aggregations?base_asset_type=native&counter_asset_type=credit_alphanum4&counter_asset_code=USD&counter_asset_issuer=GDUKMGUGDZQK6YHYA5Z6AY2G4XDSZPSZ3SW5UN3ARVMO6QSRDWP5YLEX&resolution=60000&order=asc&limit=100

WITH last_range_ts AS (
    SELECT ((timestamp - 0) / 60000) * 60000 + 0 AS timestamp
    FROM history_trades_60000
    WHERE base_asset_id = ?
    AND counter_asset_id = ?
    AND timestamp >= ?
    GROUP BY ((timestamp - 0) / 60000) * 60000 + 0
    ORDER BY ((timestamp - 0) / 60000) * 60000 + 0 asc
    FETCH FIRST 100 ROWS ONLY
), timestamp_range AS (
    SELECT min(timestamp) as min_ts,
    max(timestamp) as max_ts FROM last_range_ts
), raw_trades AS (
    SELECT ((tr.timestamp - 0) / 60000) * 60000 + 0 AS timestamp,
    count,
    base_volume,
    counter_volume,
    avg,
    high_n, high_d, low_n, low_d,
    open_n, open_d, close_n, close_d
    FROM history_trades_60000 AS tr
    JOIN timestamp_range r ON 1=1
    WHERE base_asset_id = ? 
    AND counter_asset_id = ? 
    AND r.max_ts >= ((tr.timestamp - 0) / 60000) * 60000 + 0 
    AND r.min_ts <= ((tr.timestamp - 0) / 60000) * 60000 + 0
) 
SELECT timestamp, 
    sum(\"count\") as count,
    sum(base_volume) as base_volume,
    sum(counter_volume) as counter_volume,
    sum(counter_volume::numeric)/sum(base_volume::numeric) as avg,
    (max_price(ARRAY[high_n, high_d]))[1] as high_n,
    (max_price(ARRAY[high_n, high_d]))[2] as high_d,
    (min_price(ARRAY[low_n, low_d]))[1] as low_n,
    (min_price(ARRAY[low_n, low_d]))[2] as low_d,
    (first(ARRAY[open_n, open_d]))[1] as open_n,
    (first(ARRAY[open_n, open_d]))[2] as open_d,
    (last(ARRAY[close_n, close_d]))[1] as close_n,
    (last(ARRAY[close_n, close_d]))[2] as close_d
FROM raw_trades
GROUP BY timestamp
ORDER BY timestamp asc
LIMIT 100

[sydneynotthecity]

Overall looks good. Had a question on the second outputtted SQL statement @erika-sdf. Looks like the second statement doesn't have an ORDER BY clause in the raw_trades subselect query. Do we know why that is happening? We should be ordering that statement by timestamp ASC, open_ledger_toid ASC every time.

[sreuland]

looks accurate per the optimizations, tests passing, nice work.

[erika-sdf]

Overall looks good. Had a question on the second outputtted SQL statement @erika-sdf. Looks like the second statement doesn't have an ORDER BY clause in the raw_trades subselect query. Do we know why that is happening? We should be ordering that statement by timestamp ASC, open_ledger_toid ASC every time.

It seems like timestamp ASC, open_ledger_toid ASC is appended whenever resolution is not 60000.

Is there an assumption that there is only 1 entry for 1 minute resolution, but potentially multiple for longer time frames?

[sydneynotthecity]

Is there an assumption that there is only 1 entry for 1 minute resolution, but potentially multiple for longer time frames?

From the Stellar API docs:

The segment duration represented as milliseconds. Supported values are 1 minute (60000), 
5 minutes (300000), 15 minutes (900000), 1 hour (3600000), 1 day (86400000) and 1 week (604800000).

Looks like lowest value is 1 minute, so i guess if we're ordering by the transformed timestamp, that might work out and not need ordering guarantees? @paulbellamy can you confirm?

[sreuland]

@erika-sdf , @sydneynotthecity can we merge this as-is and then PR another change if @paulbellamy recommends? this currently is just to deploy to staging for the replay performance test.

[sydneynotthecity]

@sreuland I'm fine with that, i think it's good to go to staging as is.