From 4a65580e86581e07c0c9468a15a1e179e0f79ab7 Mon Sep 17 00:00:00 2001
From: akvlad <akvlad90@gmail.com>
Date: Mon, 26 Aug 2024 19:04:32 +0300
Subject: [PATCH 1/3] ADVANCED_PROFILES_MERGE_LIMIT to limit the merge; limit
 for simultaneous chunking

---
 pyroscope/pyroscope.js | 37 +++++++++++++++++++++++++++++--------
 1 file changed, 29 insertions(+), 8 deletions(-)

diff --git a/pyroscope/pyroscope.js b/pyroscope/pyroscope.js
index 218202ef..57e82543 100644
--- a/pyroscope/pyroscope.js
+++ b/pyroscope/pyroscope.js
@@ -120,6 +120,9 @@ const selectSeries = async (req, res) => {
   return selectSeriesImpl(fromTimeSec, toTimeSec, req.body)
 }
 
+let mergeRequestsCounter = 0
+const mergeRequestsLimit = 10
+
 const selectMergeProfile = async (req, res) => {
   const ctx = newCtxIdx()
   try {
@@ -166,7 +169,7 @@ const selectMergeProfile = async (req, res) => {
       )
     labelSelectorQuery(idxReq, labelSelector)
     const withIdxReq = (new Sql.With('idx', idxReq, !!clusterName))
-    const mainReq = (new Sql.Select())
+    let mainReq = (new Sql.Select())
       .with(withIdxReq)
       .select([new Sql.Raw('payload'), 'payload'])
       .from([`${DATABASE_NAME()}.profiles${dist}`, 'p'])
@@ -174,7 +177,10 @@ const selectMergeProfile = async (req, res) => {
         new Sql.In('p.fingerprint', 'IN', new Sql.WithReference(withIdxReq)),
         Sql.Gte('p.timestamp_ns', new Sql.Raw(`${fromTimeSec}000000000`)),
         Sql.Lt('p.timestamp_ns', new Sql.Raw(`${toTimeSec}000000000`))))
-      .orderBy(new Sql.Raw('timestamp_ns'))
+      .orderBy([new Sql.Raw('timestamp_ns'), 'DESC'], [new Sql.Raw('p.fingerprint'), 'ASC'])
+    if (process.env.ADVANCED_PROFILES_MERGE_LIMIT) {
+      mainReq = mainReq.limit(parseInt(process.env.ADVANCED_PROFILES_MERGE_LIMIT))
+    }
     const approxReq = (new Sql.Select())
       .select(
         [new Sql.Raw('sum(length(payload))'), 'size'],
@@ -196,13 +202,28 @@ const selectMergeProfile = async (req, res) => {
 
     for (let i = 0; i < chunksCount; i++) {
       promises.push((async (i) => {
+        // eslint-disable-next-line no-unmodified-loop-condition
+        while (mergeRequestsCounter >= mergeRequestsLimit) {
+          await (new Promise((resolve) => setTimeout(resolve, 50)))
+        }
         logger.debug(`Processing chunk ${i}`)
-        const profiles = await clickhouse.rawRequest(mainReq.toString() + ` LIMIT ${chunkSize} OFFSET ${i * chunkSize} FORMAT RowBinary`,
-          null,
-          DATABASE_NAME(),
-          {
-            responseType: 'arraybuffer'
-          })
+        mergeRequestsCounter++
+        let profiles = null
+        try {
+          let end = i * chunkSize + chunkSize
+          if (process.env.ADVANCED_PROFILES_MERGE_LIMIT && end > process.env.ADVANCED_PROFILES_MERGE_LIMIT) {
+            end = process.env.ADVANCED_PROFILES_MERGE_LIMIT
+          }
+          mainReq.limit(end - i * chunkSize, i * chunkSize)
+          profiles = await clickhouse.rawRequest(mainReq.toString() + ' FORMAT RowBinary',
+            null,
+            DATABASE_NAME(),
+            {
+              responseType: 'arraybuffer'
+            })
+        } finally {
+          mergeRequestsCounter--
+        }
         const binData = Uint8Array.from(profiles.data)
         logger.debug(`Chunk ${i} - ${binData.length} bytes`)
         const start = process.hrtime.bigint()

From 814a2b066cbfba5aeb15bc0997e1ee7560aa7136 Mon Sep 17 00:00:00 2001
From: Cluas <Cluas@live.cn>
Date: Tue, 27 Aug 2024 16:11:21 +0800
Subject: [PATCH 2/3] fix: from to timestamp

---
 pyroscope/pyroscope.js | 12 +++++++-----
 1 file changed, 7 insertions(+), 5 deletions(-)

diff --git a/pyroscope/pyroscope.js b/pyroscope/pyroscope.js
index 57e82543..21d0564f 100644
--- a/pyroscope/pyroscope.js
+++ b/pyroscope/pyroscope.js
@@ -127,11 +127,13 @@ const selectMergeProfile = async (req, res) => {
   const ctx = newCtxIdx()
   try {
     const _req = req.body
-    const fromTimeSec = Math.floor(req.getStart && req.getStart()
-      ? parseInt(req.getStart()) / 1000
-      : Date.now() / 1000 - HISTORY_TIMESPAN)
-    const toTimeSec = Math.floor(req.getEnd && req.getEnd()
-      ? parseInt(req.getEnd()) / 1000
+    const fromTimeSec =
+    Math.floor(req.body && req.body.getStart
+      ? parseInt(req.body.getStart()) / 1000
+      : (Date.now() - HISTORY_TIMESPAN) / 1000)
+    const toTimeSec =
+    Math.floor(req.body && req.body.getEnd
+      ? parseInt(req.body.getEnd()) / 1000
       : Date.now() / 1000)
     let typeID = _req.getProfileTypeid && _req.getProfileTypeid()
     if (!typeID) {

From 1683efaa8b606464bb3941ac59d9493ab9a1f31f Mon Sep 17 00:00:00 2001
From: Cluas <Cluas@live.cn>
Date: Tue, 27 Aug 2024 16:19:35 +0800
Subject: [PATCH 3/3] chore: use _req

---
 pyroscope/pyroscope.js | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/pyroscope/pyroscope.js b/pyroscope/pyroscope.js
index 21d0564f..522503bd 100644
--- a/pyroscope/pyroscope.js
+++ b/pyroscope/pyroscope.js
@@ -128,12 +128,12 @@ const selectMergeProfile = async (req, res) => {
   try {
     const _req = req.body
     const fromTimeSec =
-    Math.floor(req.body && req.body.getStart
-      ? parseInt(req.body.getStart()) / 1000
+    Math.floor(_req && _req.getStart
+      ? parseInt(_req.getStart()) / 1000
       : (Date.now() - HISTORY_TIMESPAN) / 1000)
     const toTimeSec =
-    Math.floor(req.body && req.body.getEnd
-      ? parseInt(req.body.getEnd()) / 1000
+    Math.floor(_req && _req.getEnd
+      ? parseInt(_req.getEnd()) / 1000
       : Date.now() / 1000)
     let typeID = _req.getProfileTypeid && _req.getProfileTypeid()
     if (!typeID) {