fix: DROP stream for persistent query doesn't always drop underlying query

[spena]

Description

What behavior do you want to change, why, how does your patch achieve the changes?
This PR fixes an issue when terminating an INSERT query that caused the underlying CREATE_AS query to not be terminated when dropping the stream.

i.e.

ksql> create stream p1 as select * from s1;
ksql> insert into p1 select * from s2;
ksql> show queries;
Query ID       | Query Type | Status    | Sink Name | Sink Kafka Topic | Query String                                                                                            
----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 INSERTQUERY_43 | PERSISTENT | RUNNING:1 | P1        | P1               | insert into p1 select * from s2;                                                                        
 CSAS_P1_41     | PERSISTENT | RUNNING:1 | P1        | P1               | CREATE STREAM P1 WITH (KAFKA_TOPIC='P1', PARTITIONS=1, REPLICAS=1) AS SELECT * FROM S1 S1 EMIT CHANGES; 
----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
ksql> terminate INSERTQUERY_43;
ksql> drop stream p1;
ksql> show queries;
Query ID       | Query Type | Status    | Sink Name | Sink Kafka Topic | Query String                                                                                            
----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
CSAS_P1_41     | PERSISTENT | RUNNING:1 | P1        | P1               | CREATE STREAM P1 WITH (KAFKA_TOPIC='P1', PARTITIONS=1, REPLICAS=1) AS SELECT * FROM S1 S1 EMIT CHANGES; 
----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------

The CSAS_P1_41 should have been terminated automatically with DROP stream.

The problem was in QueryRegistryImpl that was removing query references from both insertQueries and createAsQueries mapping variables. This PR now removes the references from the right map variable.

Testing done

Describe the testing strategy. Unit and integration tests are expected for any behavior changes.

Reviewer checklist

• Ensure docs are updated if necessary. (eg. if a user visible feature is being added or changed).
• Ensure relevant issues are linked (description should include text like "Fixes #")

[wcarlson5]

This makes sense to me, I just got a couple questions about the testing 👍

[wcarlson5]

It seems like this comment was not true?

[spena]

Correct. I thought it was true when I implemented this feature. I don't know why it wasn't it.

[wcarlson5]

Should we have an test for the INSERT type here? I don't think it should have any effect right?

[spena]

Done

[wcarlson5]

Can we test the query gets removed from only the correct list here? It seems that would cover the case that was causing the the problem

[spena]

Done

[wcarlson5]

LGTM thanks for the extra tests @spena !

[guozhangwang]

Thanks @spena . The patch LGTM.

I have an out-of-scope question following this though: back to your example in the description:

create stream p1 as select * from s1;
insert into p1 select * from s2;

if we call drop s1; drop s2; would that cause p1 to be automatically dropped as well?

[spena]

@guozhangwang

create stream p1 as select * from s1;
insert into p1 select * from s2;
if we call drop s1; drop s2; would that cause p1 to be automatically dropped as well?

The DROP commands won't work because another stream/table is reading/writing from S1 or S2. There is a constraint linked from p1 -> s1 as well as p1 -> s2. In order to drop s1 and s2, you must first drop the p1 query and/or terminate the insert query.

[guozhangwang]

Thanks!

In order to drop s1 and s2, you must first drop the p1 query and/or terminate the insert query.

When we drop p1, wouldn't both queries be automatically dropped too? Only s1/s2 would still exist.

[spena]

@guozhangwang

When we drop p1, wouldn't both queries be automatically dropped too? Only s1/s2 would still exist.

No. That was proposed in the KLIP, but it had many discussions. At the end we implemented the klip partially to at least drop a stream+query if there was no other constraints.