From f64e5e65769729d6f192eadaf79fa5356a2db4c0 Mon Sep 17 00:00:00 2001
From: Loakesh Indiran <66478092+Loaki07@users.noreply.github.com>
Date: Wed, 25 Jun 2025 15:39:29 +0530
Subject: [PATCH] fix: stream_type for join (#7348)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

### **PR Type**
Bug fix


___

### **Description**
- Add fallback for empty stream_type in joins

- Use `sql.stream_type` when no type provided

- Retain existing mapping for provided stream types


___



### **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>http.rs</strong><dd><code>Implement stream_type
fallback logic</code>&nbsp; &nbsp; &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/cluster/http.rs

<li>Add conditional branch for empty stream_type<br> <li> Fallback to
<code>sql.stream_type</code> when missing<br> <li> Retain original
<code>StreamType::from</code> mapping otherwise


</details>


  </td>
<td><a
href="https://github.com/openobserve/openobserve/pull/7348/files#diff-fe0a1bbec7ef6418120544b589a531e5144b28cd2650f99135ed8a70ac17ac03">+5/-1</a>&nbsp;
&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/cluster/http.rs | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/src/service/search/cluster/http.rs b/src/service/search/cluster/http.rs
index f6b0455471..0c534ed1ce 100644
--- a/src/service/search/cluster/http.rs
+++ b/src/service/search/cluster/http.rs
@@ -58,7 +58,11 @@ pub async fn search(
         // Get the schema from `TableReference` for join queries
         // Since it resolves queries where stream_name is prefixed with the stream_type
         // e.g. `logs.my_stream`, `enrich.my_stream`
-        let stream_type = config::meta::stream::StreamType::from(s.stream_type());
+        let stream_type = if s.stream_type().is_empty() {
+            sql.stream_type
+        } else {
+            config::meta::stream::StreamType::from(s.stream_type())
+        };
         let schema = infra::schema::get_cache(&sql.org_id, &s.stream_name(), stream_type).await?;
         if schema.schema().fields().is_empty() {
             let mut result = search::Response::new(sql.offset, sql.limit);