fix: handle offset pagination on leader (#7132)

### **PR Type**
Bug fix


___

### **Description**
- Support offset pagination with `hits_to_skip`

- Adjust cache flow logic on paginated queries

- Trim skipped hits in partitioned search

- Fix delta search truncation condition


___



### **Changes walkthrough** 📝
<table><thead><tr><th></th><th align="left">Relevant
files</th></tr></thead><tbody><tr><td><strong>Bug
fix</strong></td><td><table>
<tr>
  <td>
    <details>
<summary><strong>search_stream.rs</strong><dd><code>Implement offset
pagination skip logic</code>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; </dd></summary>
<hr>

src/service/search/search_stream.rs

<li>Added <code>hits_to_skip</code> calculation, reset
<code>req.query.from</code><br> <li> Updated cache usage for paginated
requests<br> <li> Passed <code>hits_to_skip</code> to partitioned
search, trimmed hits<br> <li> Corrected delta search size truncation
logic


</details>


  </td>
<td><a
href="https://github.com/openobserve/openobserve/pull/7132/files#diff-b952a89375f2745a0daa273f1bdefc80c22997759a9b17a8fccbc70ae294e8e4">+46/-3</a>&nbsp;
&nbsp; </td>

</tr>
</table></td></tr></tr></tbody></table>

___

> <details> <summary> Need help?</summary><li>Type <code>/help how to
...</code> in the comments thread for any questions about PR-Agent
usage.</li><li>Check out the <a
href="https://qodo-merge-docs.qodo.ai/usage-guide/">documentation</a>
for more information.</li></details>

src/service/search/search_stream.rs

@@ -120,7 +120,31 @@ pub async fn process_search_stream_request(
 
     let max_query_range = get_max_query_range(&stream_names, &org_id, &user_id, stream_type).await; // hours
 
-    if req.query.from == 0 && !req.query.track_total_hits {
+    // HACK: always search from the first partition
+    let mut hits_to_skip = req.query.from;
+    let original_size = req.query.size;
+    req.query.from = 0;
+    req.query.size = original_size + hits_to_skip;
+
+    let use_cached_flow = if req.query.from == 0 && !req.query.track_total_hits  {
+        if use_cache && hits_to_skip > 0 {
+            // do not use cached flow as its a paginated query
+            // use partitioned search instead
+            log::info!(
+                "[HTTP2_STREAM] trace_id: {}, Skipping cached flow as paginated query, hits_to_skip: {}",
+                trace_id,
+                hits_to_skip
+            );
+            false
+        } else {
+            // check cache for the first page
+            true
+        }
+    } else {
+        false
+    };
+
+    if use_cached_flow {
         let c_resp = match cache::check_cache_v2(&trace_id, &org_id, stream_type, &req, use_cache)
             .instrument(search_span.clone())
             .await
@@ -296,6 +320,7 @@ pub async fn process_search_stream_request(
                 sender.clone(),
                 values_ctx,
                 fallback_order_by_col,
+                &mut hits_to_skip,
             )
             .instrument(search_span.clone())
             .await
@@ -409,6 +434,7 @@ pub async fn process_search_stream_request(
             sender.clone(),
             values_ctx,
             fallback_order_by_col,
+            &mut hits_to_skip,
         )
         .instrument(search_span.clone())
         .await
@@ -511,6 +537,7 @@ pub async fn do_partitioned_search(
     sender: mpsc::Sender<Result<StreamResponses, infra::errors::Error>>,
     values_ctx: Option<ValuesEventContext>,
     fallback_order_by_col: Option<String>,
+    hits_to_skip: &mut i64,
 ) -> Result<(), infra::errors::Error> {
     // limit the search by max_query_range
     let mut range_error = String::new();
@@ -579,7 +606,23 @@ pub async fn do_partitioned_search(
         // do not use cache for partitioned search without cache
         let mut search_res = do_search(trace_id, org_id, stream_type, &req, user_id, false).await?;
 
-        let total_hits = search_res.total as i64;
+        let mut total_hits = search_res.total as i64;
+
+        let skip_hits = std::cmp::min(*hits_to_skip, total_hits);
+        if skip_hits > 0 {
+            search_res.hits = search_res.hits[skip_hits as usize..].to_vec();
+            search_res.total = search_res.hits.len();
+            total_hits = search_res.total as i64;
+            *hits_to_skip -= skip_hits;
+            log::info!(
+                "[HTTP2_STREAM] trace_id: {}, Skipped {} hits, remaining hits to skip: {}, total hits for partition {}: {}",
+                trace_id,
+                skip_hits,
+                *hits_to_skip,
+                idx,
+                total_hits
+            );
+        }
 
         if req.query.size > 0 && total_hits > req.query.size {
             log::info!(
@@ -1031,7 +1074,7 @@ async fn process_delta(
 
         let total_hits = search_res.total as i64;
 
-        if total_hits > req.query.size {
+        if req.query.size > 0 && total_hits > req.query.size {
             log::info!(
                 "[HTTP2_STREAM] trace_id: {}, Reached requested result size ({}), truncating results",
                 trace_id,