fix: CREATE IF NOT EXISTS does not work at all #6073

hemantgs · 2020-08-21T15:38:55Z

Description

IF NOT EXISTS were not added to the docs for Streams , even though support has been provided in code.
This came up in discussion of #6036
Added docs for the same

Testing done

None

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 #")

ghost · 2020-08-21T15:38:58Z

@confluentinc It looks like @hemantgs just signed our Contributor License Agreement. 👍

Always at your service,

clabot

spena · 2020-08-31T18:57:02Z

There's a bug in the CREATE STREAM IF NOT EXISTS command. It still returns an error if the stream does exists.
#6050

We should fix the above and the doc.

hemantgs · 2020-09-01T03:35:24Z

@spena Is someone looking at it ? else I can try 👍

big-andy-coates · 2020-09-01T12:43:30Z

@spena Is someone looking at it ? else I can try 👍

May as well fix in this PR if you have the time @hemantgs

hemantgs · 2020-09-01T13:28:35Z

@spena Is someone looking at it ? else I can try +1

May as well fix in this PR if you have the time @hemantgs

Sure 👍

hemantgs · 2020-09-01T17:37:02Z

ksqldb-engine/src/main/java/io/confluent/ksql/ddl/commands/DdlCommandExec.java

@@ -75,6 +75,13 @@ public DdlCommandResult executeCreateStream(final CreateStreamCommand createStre
          withQuery,
          getKsqlTopic(createStream)
      );
+


@big-andy-coates @spena
I noticed that CreateStreamCommand did not have the boolean flag for ifNotExists set here , I thought if this approach is ok , I will add that too and fail the query if its not set and source exists

The place to put the check is in the CreateSourceFactory. You'll need to pass the MetaStore in to the factory's constructor.

See DropSourceFactory for an example of the pattern.

hemantgs · 2020-09-01T17:38:35Z

@spena Is someone looking at it ? else I can try +1

May as well fix in this PR if you have the time @hemantgs

I took a stab at this 😄 , can you check if I am in the right path , if yes then I will fix the tests and add more , and add the ifNotExists boolean to CreateStreamCommand as well

big-andy-coates · 2020-09-01T18:22:05Z

What out for merge clashes with #6041. Would hate to lose some of your doc changes in a merge error.

big-andy-coates

Quick review before I sign off. Thanks!

big-andy-coates · 2020-09-01T18:22:24Z

docs/developer-guide/ksqldb-reference/create-stream.md

+If the IF NOT EXISTS clause is present, the statement doesn't fail if the
+stream doesn't exist.


Suggested change

If the IF NOT EXISTS clause is present, the statement doesn't fail if the

stream doesn't exist.

If the IF NOT EXISTS clause is present, the statement won't fail if a

stream with the same name already exists.

big-andy-coates · 2020-09-01T18:23:16Z

docs/developer-guide/ksqldb-reference/create-table.md

+If the IF NOT EXISTS clause is present, the statement doesn't fail if the
+table doesn't exist.


Suggested change

If the IF NOT EXISTS clause is present, the statement doesn't fail if the

table doesn't exist.

If the IF NOT EXISTS clause is present, the statement won't fail if a

table with the same name already exists.

big-andy-coates · 2020-09-01T18:33:32Z

ksqldb-engine/src/main/java/io/confluent/ksql/ddl/commands/DdlCommandExec.java

@@ -75,6 +75,13 @@ public DdlCommandResult executeCreateStream(final CreateStreamCommand createStre
          withQuery,
          getKsqlTopic(createStream)
      );
+


The place to put the check is in the CreateSourceFactory. You'll need to pass the MetaStore in to the factory's constructor.

See DropSourceFactory for an example of the pattern.

big-andy-coates · 2020-09-01T18:34:33Z

ksqldb-metastore/src/main/java/io/confluent/ksql/metastore/MetaStoreImpl.java

@@ -77,22 +77,20 @@ public DataSource getSource(final SourceName sourceName) {
  }

  @Override
-  public void putSource(final DataSource dataSource, final boolean allowReplace) {
+  public boolean validateSource(final DataSource dataSource, final boolean allowReplace) {


We shouldn't be splitting this into two methods, as this makes it much more likely someone will call putSource and inadvertently overwrite a source that they shouldn't.

That's a good point,
I looked at DropSourceFactory , even there

if (dataSource == null) { if (!ifExists) { throw new KsqlException("Source " + sourceName.text() + " does not exist."); } }

We throw an Exception which will Fail the command which should not be the outcome right ,

The only place I could find other than DdlCommandExec that had capability to pass along a warning was in KsqlResource.handleKsqlStatements()
where if an exception is thrown in MetaStoreImpl we could possibly catch that and throw a warning like

catch (final KsqlStatementException e) { LOG.info("Processed unsuccessfully: " + request + ", reason: " + e.getMessage()); Errors.badStatement("",e.getSqlStatement(),new KsqlEntityList( Arrays.asList(new WarningEntity(e.getSqlStatement(),"Source Exists")) ));

Even here the first parameter to badStatement() seems it needs to be an empty string , else the msg is printed in the client in red . What do you suggest we do

We throw an Exception which will Fail the command which should not be the outcome right ,

That code looks correct to me for dropping a source. It means if I say DROP TABLE IF EXISTS FOO that it won't throw if FOO doesn't exist, which is what we want.

As to where to put this new check - you'll need two changes:

as above, put a check in CreateSourceFactory so that it throws an exception if IF NOT EXIST is not present and a source with the same name already exists. (Following a similar pattern to DropSourceFactory).

In DdlCommandExec.executeCreateStream and executeCreateTable you will need to add a check to see if a source with the same name already exists and isOrReplace() is false, e.g.

@Override public DdlCommandResult executeCreateStream(final CreateStreamCommand createStream) { final SourceName sourceName = createStream.getSourceName(); final DataSource existing = metaStore.getSource(sourceName); if (existing != null && !createStream.isOrReplace()) { return new DdlCommandResult(true, "Stream " + sourceName + " does not exist."); } final KsqlStream<?> ksqlStream = new KsqlStream<>( sql, sourceName, createStream.getSchema(), createStream.getFormats().getOptions(), createStream.getTimestampColumn(), withQuery, getKsqlTopic(createStream) ); metaStore.putSource(ksqlStream, createStream.isOrReplace()); return new DdlCommandResult(true, "Stream created"); }

That code looks correct to me for dropping a source. It means if I say DROP TABLE IF EXISTS FOO that it won't throw if FOO doesn't exist, which is what we want.

@big-andy-coates I always seem to confuse myself with that part :| , so anyway I have made the changes
i have made changes to KsqlEngineTest.shouldNotThrowWhenExecutingDuplicateStream() and KsqlEngineTest.shouldThrowWhenExecutingDuplicateTable() , these tests were failing when the query plans for both create source queries had different Kafka topics one had FOO(stream name) and other had BAR, before this PR the test would have failed much before getting to the planning phase, so I have made changes to the queries in the tests , I am unsure if that is alright

Also build in Jenkins has failed I get 404 when I try to open the link ,
When I try mvn verify I get

[INFO] [ERROR] Failed to execute goal on project my-first-test-udf: Could not resolve dependencies for project com.acme.ksql.functions:my-first-test-udf:jar:0.1.0-SNAPSHOT: Could not find artifact io.confluent.ksql:ksqldb-udf:jar:6.1.0-SNAPSHOT in confluent (https://packages.confluent.io/maven/) -> [Help 1] [INFO] org.apache.maven.lifecycle.LifecycleExecutionException: Failed to execute goal on project my-first-test-udf: Could not resolve dependencies for project com.acme.ksql.functions:my-first-test-udf:jar:0.1.0-SNAPSHOT: Could not find artifact io.confluent.ksql:ksqldb-udf:jar:6.1.0-SNAPSHOT in confluent (https://packages.confluent.io/maven/)

and it fails
I am a bit unsure as to how to run all the tests

big-andy-coates

Thanks @hemantgs, this is looking close, except looks like you've still got some changes in there from the old approach.

Also, would you mind adding some json QTT style tests to cover adding tables/streams when a source (or the opposite type) already exists with the same name:

Positive tests: i.e. with IF NOT EXISTS, then ensure selecting from the source is selecting from the pre-existing source, not the new source, i.e. testing that the source was NOT replaced.
Negative tests: i.e. without IF NOT EXISTS and testing the exception is thrown and the error message is correct.

Thanks!

big-andy-coates · 2020-09-11T10:41:26Z

ksqldb-engine/src/main/java/io/confluent/ksql/ddl/commands/CreateSourceFactory.java

  ) {
    this.serviceContext = Objects.requireNonNull(serviceContext, "serviceContext");
    this.serdeOptionsSupplier =
        Objects.requireNonNull(serdeOptionsSupplier, "serdeOptionsSupplier");
    this.keySerdeFactory = Objects.requireNonNull(keySerdeFactory, "keySerdeFactory");
    this.valueSerdeFactory = Objects.requireNonNull(valueSerdeFactory, "valueSerdeFactory");
+    this.metaStore = metaStore;


nit: validate params that will be stored in object state; ensuring object does not get into an invalid state. i.e. use requireNonNull.

this is done

big-andy-coates · 2020-09-11T10:42:47Z

ksqldb-engine/src/main/java/io/confluent/ksql/ddl/commands/CreateSourceFactory.java

@@ -115,6 +131,17 @@ public CreateTableCommand createTableCommand(
      final KsqlConfig ksqlConfig
  ) {
    final SourceName sourceName = statement.getName();
+    final DataSource dataSource = metaStore.getSource(sourceName);
+
+    if (dataSource != null) {


Suggested change

if (dataSource != null) {

if (dataSource != null && !statement.isOrReplace()) {

this is done

big-andy-coates · 2020-09-11T10:43:32Z

ksqldb-engine/src/main/java/io/confluent/ksql/ddl/commands/CreateSourceFactory.java

+            String.format("Cannot add %s '%s': A %s with the same name already exists",
+                sourceType, sourceName.text(), sourceType));


Suggested change

String.format("Cannot add %s '%s': A %s with the same name already exists",

sourceType, sourceName.text(), sourceType));

String.format("Cannot add stream '%s': A %s with the same name already exists",

sourceName.text(), sourceType.toLowerCase()));

this is done

big-andy-coates · 2020-09-11T10:43:53Z

ksqldb-engine/src/main/java/io/confluent/ksql/ddl/commands/CreateSourceFactory.java

+            String.format("Cannot add %s '%s': A %s with the same name already exists",
+                sourceType, sourceName.text(), sourceType));


Suggested change

String.format("Cannot add %s '%s': A %s with the same name already exists",

sourceType, sourceName.text(), sourceType));

String.format("Cannot add table '%s': A %s with the same name already exists",

sourceName.text(), sourceType.toLowerCase()));

this is done

big-andy-coates · 2020-09-11T10:46:57Z

ksqldb-engine/src/main/java/io/confluent/ksql/ddl/commands/DdlCommandExec.java

+      if (dataSource != null && !createStream.isOrReplace()) {
+        return new DdlCommandResult(true,
+            String.format("Cannot add stream %s: A stream with the same name "
+                    + "already exists.",
+                sourceName));
+      }


Isn't this check now redundant, given there is code in CreateSourceFactory to do this now? We should let the code do what it used to do, which is throw an exception, not return a success code!

@big-andy-coates The code in CreateSourceFactory checks if a dataSource already exists , if isOrReplace() is not set and ifNotExists is also not set and throws exception , which seems right
If the code passes the checks in that gate ,it means in DdlCommandExec it would mean the ifNotExists flag is set to true and we need to ensure that the statement does NOT fail ?

Seems redundant, but it is what DROP IF NOT EXIST does too. It has a check in the DropSourceCommand and another in this class, which returns the DdlCommandResult message without causing a failure.

I figured out why we need this repeated code. Turns out that CreateSourceFactory is only called for CREATE without queries statements. For CREATE AS, a different code path is called, which creates the Create command similar to the CreateSourceFactory, but without the factory validations and with more parameters. That path is EngineExecutor.maybeCreateSinkDdl.

I was thinking on refactor the code, but seems it requires several changes. For now, I leave this temporary and do a check inside the maybeCreateSinkDdl for the isNotExists value. I'll do the refactor in another PR.

big-andy-coates · 2020-09-11T10:50:30Z

ksqldb-engine/src/test/java/io/confluent/ksql/engine/KsqlEngineTest.java

            + "CREATE TABLE FOO WITH (KAFKA_TOPIC='BAR') AS SELECT * FROM TEST2;");

    givenStatementAlreadyExecuted(parsed.get(0));

-    final PreparedStatement<?> prepared = prepare(parsed.get(1));
+    final PreparedStatement<?> prepared = ksqlEngine.prepare(parsed.get(1));


This change should be reverted.

big-andy-coates · 2020-09-11T10:50:35Z

ksqldb-engine/src/test/java/io/confluent/ksql/engine/KsqlEngineTest.java

+    ExecuteResult executeResult = ksqlEngine.execute(
+        serviceContext,
+        of(prepared, new HashMap<>(), KSQL_CONFIG)


This change should be reverted.

big-andy-coates · 2020-09-11T10:50:39Z

ksqldb-engine/src/test/java/io/confluent/ksql/engine/KsqlEngineTest.java

-        "Cannot add table 'FOO': A table with the same name already exists")));
-    assertThat(e, statementText(is(
-        "CREATE TABLE FOO WITH (KAFKA_TOPIC='BAR') AS SELECT * FROM TEST2;")));
+    assertThat(executeResult.getQuery(), is(not(Optional.empty())));


This change should be reverted.

big-andy-coates · 2020-09-11T10:50:47Z

ksqldb-engine/src/test/java/io/confluent/ksql/engine/KsqlEngineTest.java

+  public void shouldNotThrowWhenExecutingDuplicateStream() {
    // Given:
    final List<ParsedStatement> parsed = ksqlEngine.parse(
-        "CREATE STREAM FOO AS SELECT * FROM ORDERS; "
+        "CREATE STREAM FOO WITH (KAFKA_TOPIC='BAR') AS SELECT * FROM ORDERS; "
            + "CREATE STREAM FOO WITH (KAFKA_TOPIC='BAR') AS SELECT * FROM ORDERS;");

    givenStatementAlreadyExecuted(parsed.get(0));

    final PreparedStatement<?> prepared = ksqlEngine.prepare(parsed.get(1));

    // When:
-    final KsqlStatementException e = assertThrows(
-        KsqlStatementException.class,
-        () -> ksqlEngine.execute(
-            serviceContext,
-            of(prepared, new HashMap<>(), KSQL_CONFIG)
-        )
+    ExecuteResult executeResult = ksqlEngine.execute(
+        serviceContext,
+        of(prepared, new HashMap<>(), KSQL_CONFIG)
    );

    // Then:
-    assertThat(e, rawMessage(
-        is(
-            "Cannot add stream 'FOO': A stream with the same name already exists")));
-    assertThat(e, statementText(
-        is(
-            "CREATE STREAM FOO WITH (KAFKA_TOPIC='BAR') AS SELECT * FROM ORDERS;")));
+    assertThat(executeResult.getQuery(), is(not(Optional.empty())));


This change should be reverted.

big-andy-coates · 2020-09-11T10:50:52Z

ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/resources/KsqlResource.java

@@ -63,6 +63,7 @@
 import io.confluent.ksql.version.metrics.ActivenessRegistrar;
 import java.net.URL;
 import java.time.Duration;
+import java.util.Arrays;


This change should be reverted.

JimGalasyn

LGTM!

spena

@hemantgs Sorry for the late review. I left 3 comments only. Are you still working in this, btw?

spena · 2020-11-02T14:40:43Z

ksqldb-engine/src/main/java/io/confluent/ksql/ddl/commands/DdlCommandExec.java

+      if (dataSource != null && !createStream.isOrReplace()) {
+        return new DdlCommandResult(true,
+            String.format("Cannot add stream %s: A stream with the same name "
+                    + "already exists.",
+                sourceName));
+      }


Seems redundant, but it is what DROP IF NOT EXIST does too. It has a check in the DropSourceCommand and another in this class, which returns the DdlCommandResult message without causing a failure.

spena · 2020-11-02T14:43:51Z

ksqldb-engine/src/test/java/io/confluent/ksql/ddl/commands/CreateSourceFactoryTest.java

@@ -911,6 +929,41 @@ public void shouldThrowIfTableIsMissingPrimaryKey() {
        containsString("Tables require a PRIMARY KEY. Please define the PRIMARY KEY."));
  }

+  @Test
+  public void shouldThrowIfStreamExits() {


Could you add two more tests shouldNotThrowWhenCreateStreamIfNotExistsIsSet and same for a table?

spena · 2020-11-02T14:45:20Z

...b-version-metrics-client/src/main/java/io/confluent/support/metrics/BaseMetricsReporter.java

@@ -109,11 +109,11 @@ public void init() {

    final String endpointHttp = supportConfig.getEndpointHttp();
    final String endpointHttps = supportConfig.getEndpointHttps();
-    final String proxyURI = supportConfig.getProxy();
+    final String proxyUri = supportConfig.getProxy();


Is this part of the changes? Should it be reverted?

spena · 2020-11-13T22:05:11Z

ksqldb-engine/src/main/java/io/confluent/ksql/schema/ksql/inference/SchemaRegisterInjector.java

+        createSourceCommand.getSchema(),
+        createSourceCommand.getTopicName(),
+        createSourceCommand.getFormats().getKeyFormat(),
+        createSourceCommand.getFormats().getKeyFeatures(),
+        createSourceCommand.getFormats().getValueFormat(),
+        createSourceCommand.getFormats().getValueFeatures(),


I had to build the query information from the plan instead of executing it. This avoids failing with an error if the IF NOT EXISTS keyword is used in the CREATE_AS statement. Also, this seems a better approach than execution 'cause this injector is only registering the schema in SR.

spena · 2020-11-13T22:05:57Z

ksqldb-engine/src/test/java/io/confluent/ksql/engine/KsqlEngineTest.java

@@ -981,27 +981,21 @@ public void shouldNotThrowWhenPreparingDuplicateTable() {
  public void shouldThrowWhenExecutingDuplicateTable() {
    // Given:
    final List<ParsedStatement> parsed = ksqlEngine.parse(
-        "CREATE TABLE FOO AS SELECT * FROM TEST2; "
+        "CREATE TABLE FOO WITH (KAFKA_TOPIC='BAR') AS SELECT * FROM TEST2; "


This is a new test now that verifies the IF NOT EXISTS works.

jzaralim · 2020-12-07T23:31:07Z

ksqldb-engine/src/main/java/io/confluent/ksql/ddl/commands/DdlCommandExec.java

+
+      if (dataSource != null && !createStream.isOrReplace()) {
+        return new DdlCommandResult(true,
+            String.format("Cannot add stream %s: A stream with the same name "


nit: i think we should use dataSource.getDataSourceType().getKsqlType() here and in executeCreateTable

Thanks. I updated the PR.

Changes reverted and added to CommandFactories Fixed and added tests Fixed and added tests Fixed tests

hemantgs requested review from JimGalasyn and a team as code owners August 21, 2020 15:38

hemantgs mentioned this pull request Aug 21, 2020

feat: Support for IF EXISTS on CREATE CONNECTOR #6036

Merged

2 tasks

hemantgs force-pushed the streams-doc-notexists branch from 760f5d8 to 04546cc Compare September 1, 2020 17:32

hemantgs commented Sep 1, 2020

View reviewed changes

big-andy-coates reviewed Sep 1, 2020

View reviewed changes

spena mentioned this pull request Sep 2, 2020

Create IF NOT EXISTS throws an error if the stream/table exists #6050

Closed

hemantgs force-pushed the streams-doc-notexists branch 3 times, most recently from 2559b8c to fa37637 Compare September 4, 2020 11:00

big-andy-coates reviewed Sep 11, 2020

View reviewed changes

JimGalasyn approved these changes Sep 11, 2020

View reviewed changes

hemantgs force-pushed the streams-doc-notexists branch from fa37637 to 6781631 Compare September 26, 2020 07:45

spena reviewed Nov 2, 2020

View reviewed changes

spena force-pushed the streams-doc-notexists branch from 7e21ffd to b7aada8 Compare November 6, 2020 18:52

spena mentioned this pull request Nov 10, 2020

refactor: re-use CreateSourceFactory to create streams/tables for CREATE_AS statements #6594

Merged

2 tasks

spena force-pushed the streams-doc-notexists branch from b7aada8 to b1ea153 Compare November 13, 2020 22:02

spena reviewed Nov 13, 2020

View reviewed changes

spena requested a review from a team November 13, 2020 22:06

spena force-pushed the streams-doc-notexists branch from b1ea153 to a8a1850 Compare November 19, 2020 15:17

spena force-pushed the streams-doc-notexists branch from a8a1850 to 9a1c440 Compare December 1, 2020 03:22

spena added this to the 0.15.0 milestone Dec 3, 2020

spena changed the title ~~docs: Added missing IF NOT EXISTS to Streams and Tables doc~~ fix: CREATE IF NOT EXISTS does not work at all Dec 4, 2020

spena requested review from a team and removed request for a team December 4, 2020 20:47

jzaralim approved these changes Dec 7, 2020

View reviewed changes

hemantgs and others added 9 commits December 7, 2020 20:07

docs: Added missing IF NOT EXISTS to Streams and Tables doc

2965dcd

Fix for streams IF NOT EXISTS failing

b01aebc

Changes reverted and added to CommandFactories Fixed and added tests Fixed and added tests Fixed tests

Fixed tests

a21ee7f

Review comments

b861426

Fixed tests

6bc0a99

Checkstyle fix

7dfd7f5

fix: check IF NOT EXISTS is set in EngineExecutor

ff9c504

fix: display warning message with C*AS + IF NOT EXISTS statements

9fe2c4a

chore: fix test issue

8a84700

spena force-pushed the streams-doc-notexists branch from 9a1c440 to 8a84700 Compare December 8, 2020 02:08

spena merged commit 6edf7ec into confluentinc:master Dec 8, 2020

wcarlson5 mentioned this pull request Sep 24, 2021

Repeated IF NOT EXISTS used with CSAS/CTAS results in dropped query when ksql is restarted #8173

Closed

		If the IF NOT EXISTS clause is present, the statement doesn't fail if the
		stream doesn't exist.

		If the IF NOT EXISTS clause is present, the statement doesn't fail if the
		table doesn't exist.

	if (dataSource != null) {
	if (dataSource != null && !statement.isOrReplace()) {

		String.format("Cannot add %s '%s': A %s with the same name already exists",
		sourceType, sourceName.text(), sourceType));

fix: CREATE IF NOT EXISTS does not work at all #6073

fix: CREATE IF NOT EXISTS does not work at all #6073

Conversation

hemantgs commented Aug 21, 2020

Description

Testing done

Reviewer checklist

ghost commented Aug 21, 2020

spena commented Aug 31, 2020

hemantgs commented Sep 1, 2020

big-andy-coates commented Sep 1, 2020

hemantgs commented Sep 1, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hemantgs commented Sep 1, 2020

big-andy-coates commented Sep 1, 2020

big-andy-coates left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

big-andy-coates left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hemantgs Sep 26, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

JimGalasyn left a comment

Choose a reason for hiding this comment

spena left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hemantgs Sep 26, 2020 •

edited

Loading