fix: don't allow duplicate stream names

CHANGELOG.md

@@ -31,6 +31,7 @@
 
 ### Fixes
 
+- Don't allow duplicate stream names: #1212
 - Fix memory safety issue when using `merge` or `patch` with `state` as target and reassigning the resulting value to `state` [#1217](https://github.com/tremor-rs/tremor-runtime/pull/1217)
 - Fix delayed event re-execution in case of errors in a branched pipeline
 - Skip instead of fail EQC on out of repo PRs

tests/query_error.rs

@@ -142,6 +142,7 @@ test_cases!(
     pp_embed_unrecognized_token4,
     pp_embed_unrecognized_token5,
     // INSERT
+    duplicate_stream_name,
     window_both_settings,
     window_group_by_event_in_target,
     window_event_in_target,

tests/query_errors/duplicate_stream_name/error.txt

@@ -0,0 +1,3 @@
+Error: 
+    2 | create stream snot; #  this should error
+      | ^^^^^^^^^^^^^^^^^^ duplicate stream name

tests/query_errors/duplicate_stream_name/query.trickle

@@ -0,0 +1,5 @@
+create stream snot;
+create stream snot; # this should error
+
+select event from in into snot;
+select event from snot into out;

tremor-script/src/ast/query/raw.rs

@@ -15,6 +15,8 @@
 // We want to keep the names here
 #![allow(clippy::module_name_repetitions)]
 
+use std::collections::HashSet;
+
 use super::super::raw::{ExprRaw, IdentRaw, ImutExprRaw, ModuleRaw, ScriptRaw, WithExprsRaw};
 use super::{
     error_generic, error_no_consts, error_no_locals, AggrRegistry, BaseExpr, GroupBy, GroupByInt,
@@ -55,13 +57,22 @@ impl<'script> QueryRaw<'script> {
         mut helper: &mut Helper<'script, 'registry>,
     ) -> Result<Query<'script>> {
         let mut stmts = vec![];
+        let mut stream_names = HashSet::new();
+
         for (_i, e) in self.stmts.into_iter().enumerate() {
             match e {
                 StmtRaw::ModuleStmt(m) => {
                     m.define(helper.reg, helper.aggr_reg, &mut vec![], &mut helper)?;
                 }
                 other => {
-                    stmts.push(other.up(&mut helper)?);
+                    let f = other.up(&mut helper)?;
+
+                    if let Stmt::Stream(s) = &f {
+                        if !stream_names.insert(s.id.clone()) {
+                            return error_generic(&f, &f, &"duplicate stream name", &helper.meta);
+                        }
+                    }
+                    stmts.push(f);
                 }
             }
         }