Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: add CREATE SOURCE STREAM syntax and metadata info #8004

Merged
merged 2 commits into from
Aug 18, 2021
Merged

feat: add CREATE SOURCE STREAM syntax and metadata info #8004

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d6360d4
feat: add CREATE SOURCE STREAM syntax and metadata info
Aug 17, 2021
84be63b
revert leftjoin changes in schema
Aug 17, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions ksqldb-engine/src/main/java/io/confluent/ksql/ddl/commands/CreateSourceFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,7 @@ public CreateSourceFactory(final ServiceContext serviceContext, final MetaStore
this.metaStore = requireNonNull(metaStore);
}

// This method is called by CREATE_AS statements
public CreateStreamCommand createStreamCommand(final KsqlStructuredDataOutputNode outputNode) {
return new CreateStreamCommand(
outputNode.getSinkName().get(),
Expand All @@ -100,10 +101,12 @@ public CreateStreamCommand createStreamCommand(final KsqlStructuredDataOutputNod
outputNode.getKsqlTopic().getKafkaTopicName(),
Formats.from(outputNode.getKsqlTopic()),
outputNode.getKsqlTopic().getKeyFormat().getWindowInfo(),
Optional.of(outputNode.getOrReplace())
Optional.of(outputNode.getOrReplace()),
Optional.of(false)
);
}

// This method is called by simple CREATE statements
public CreateStreamCommand createStreamCommand(
final CreateStream statement,
final KsqlConfig ksqlConfig
Expand All @@ -130,7 +133,8 @@ public CreateStreamCommand createStreamCommand(
topicName,
buildFormats(statement.getName(), schema, props, ksqlConfig),
getWindowInfo(props),
Optional.of(statement.isOrReplace())
Optional.of(statement.isOrReplace()),
Optional.of(statement.isSource())
);
}

Expand Down
5 changes: 3 additions & 2 deletions ksqldb-engine/src/main/java/io/confluent/ksql/ddl/commands/DdlCommandExec.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,7 +98,8 @@ public DdlCommandResult executeCreateStream(final CreateStreamCommand createStre
createStream.getSchema(),
createStream.getTimestampColumn(),
withQuery,
getKsqlTopic(createStream)
getKsqlTopic(createStream),
createStream.getIsSource()
);

metaStore.putSource(ksqlStream, createStream.isOrReplace());
Expand Down Expand Up @@ -126,7 +127,7 @@ public DdlCommandResult executeCreateTable(final CreateTableCommand createTable)
createTable.getTimestampColumn(),
withQuery,
getKsqlTopic(createTable),
createTable.isSource()
createTable.getIsSource()
);
metaStore.putSource(ksqlTable, createTable.isOrReplace());
metaStore.addSourceReferences(ksqlTable.getName(), withQuerySources);
Expand Down
6 changes: 4 additions & 2 deletions ksqldb-engine/src/test/java/io/confluent/ksql/analyzer/AnalyzerFunctionalTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -166,7 +166,8 @@ public void shouldNotInheritNamespaceExplicitlySetUpstreamForAvro() {
schema,
Optional.empty(),
false,
ksqlTopic
ksqlTopic,
false
);

newAvroMetaStore.putSource(ksqlStream, false);
Expand Down Expand Up @@ -383,7 +384,8 @@ private void registerKafkaSource() {
schema,
Optional.empty(),
false,
topic
topic,
false
);

jsonMetaStore.putSource(stream, false);
Expand Down
24 changes: 21 additions & 3 deletions ksqldb-engine/src/test/java/io/confluent/ksql/ddl/commands/CommandFactoriesTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -168,7 +168,7 @@ private void givenCommandFactoriesWithMocks() {
@Test
public void shouldCreateCommandForCreateStream() {
// Given:
final CreateStream statement = new CreateStream(SOME_NAME, SOME_ELEMENTS, false, true, withProperties);
final CreateStream statement = new CreateStream(SOME_NAME, SOME_ELEMENTS, false, true, withProperties, false);

// When:
final DdlCommand result = commandFactories
Expand All @@ -178,10 +178,28 @@ public void shouldCreateCommandForCreateStream() {
verify(createSourceFactory).createStreamCommand(statement, ksqlConfig);
}

@Test
public void shouldCreateCommandForCreateSourceStream() {
// Given:
final CreateStream statement = new CreateStream(SOME_NAME,
TableElements.of(
tableElement(Namespace.VALUE, "COL1", new Type(SqlTypes.BIGINT)),
tableElement(Namespace.VALUE, "COL2", new Type(SqlTypes.STRING))),
false, true, withProperties, true);

// When:
final DdlCommand result = commandFactories
.create(sqlExpression, statement, SessionConfig.of(ksqlConfig, emptyMap()));

// Then:
assertThat(result, is(createStreamCommand));
verify(createSourceFactory).createStreamCommand(statement, ksqlConfig);
}

@Test
public void shouldCreateCommandForStreamWithOverriddenProperties() {
// Given:
final CreateStream statement = new CreateStream(SOME_NAME, SOME_ELEMENTS, false, true, withProperties);
final CreateStream statement = new CreateStream(SOME_NAME, SOME_ELEMENTS, false, true, withProperties, false);

// When:
commandFactories.create(sqlExpression, statement, SessionConfig.of(ksqlConfig, OVERRIDES));
Expand Down Expand Up @@ -347,7 +365,7 @@ public void shouldCreateStreamCommandWithSingleValueWrappingFromOverridesNotConf
);

final DdlStatement statement =
new CreateStream(SOME_NAME, SOME_ELEMENTS, false, true, withProperties);
new CreateStream(SOME_NAME, SOME_ELEMENTS, false, true, withProperties, false);

// When:
final DdlCommand cmd = commandFactories
Expand Down
Loading