fix: stream_type for join (#7348)

### **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

@@ -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);