feat: introduce branches for pull queries on streams #8045

vvcephei · 2021-08-24T02:29:15Z

This change implements part of KLIP-53

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

This change implements part of [KLIP-53](https://github.com/confluentinc/ksql/blob/master/design-proposals/klip-53-pull-queries-on-streams.md)

vvcephei

Quick PR to start incrementally introducing stream pull queries.

I tried to do it all in one pass, but the three different endpoints really explodes the code complexity.

My plan is to start with this PR, then to add the feature to one endpoint at a time, complete with tests for those endpoints. That way, after each commit, for any given endpoint, the feature will be fully functional or produce a nice error message.

vvcephei · 2021-08-24T02:29:34Z

ksqldb-common/src/main/java/io/confluent/ksql/util/KsqlConstants.java

@@ -46,6 +46,7 @@ private KsqlConstants() {
  public static final String KSQL_QUERY_SOURCE_TAG = "query_source";
  public static final String KSQL_QUERY_PLAN_TYPE_TAG = "query_plan_type";
  public static final String KSQL_QUERY_ROUTING_TYPE_TAG = "query_routing_type";
+  public static final String KSQL_DATA_SOURCE_TYPE_TAG = "data_source_type";


Adding a metric tag for the data source type

vvcephei · 2021-08-24T02:30:29Z

ksqldb-engine/src/main/java/io/confluent/ksql/engine/EngineExecutor.java

@@ -210,7 +213,8 @@ PullQueryResult executeTablePullQuery(
          physicalPlan.getQueryId(), pullQueryQueue);
      final PullQueryResult result = new PullQueryResult(physicalPlan.getOutputSchema(), populator,
          physicalPlan.getQueryId(), pullQueryQueue, pullQueryMetrics, physicalPlan.getSourceType(),
-          physicalPlan.getPlanType(), routingNodeType, physicalPlan::getRowsReadFromDataSource);
+          physicalPlan.getPlanType(), routingNodeType, DataSourceType.KTABLE,


In all the pull request metrics, we are introducing the data type tag, so that we will be able to distinguish between table and stream pull requests.

vvcephei · 2021-08-24T02:31:46Z

ksqldb-engine/src/main/java/io/confluent/ksql/engine/PullQueryExecutionUtil.java

+      throw new KsqlException("Unexpected data source type for table pull query: "
+          + source.getDataSourceType() + " "


This is now just serving as a last-minute check on the source type. As far as we know, though, all code paths leading here will only send us table queries now, so it makes sense to reword the error.

vvcephei · 2021-08-24T02:32:27Z

ksqldb-engine/src/main/java/io/confluent/ksql/internal/PullQueryExecutorMetrics.java

  ) {
-    final MetricsKey key = new MetricsKey(sourceType, planType, routingNodeType);
+    final MetricsKey key = new MetricsKey(sourceType, planType, routingNodeType, dataSourceType);


Here's where we're actually adding the new type tag. Question: is this a compatible change?

vvcephei · 2021-08-24T02:33:06Z

ksqldb-engine/src/main/java/io/confluent/ksql/internal/PullQueryExecutorMetrics.java

+      final MetricsKey that = (MetricsKey) o;
+      return sourceType == that.sourceType && planType == that.planType
+          && routingNodeType == that.routingNodeType && dataSourceType == that.dataSourceType;


Switched this out for auto-generated equals/hashcode/toString.

vvcephei · 2021-08-24T02:33:31Z

ksqldb-engine/src/main/java/io/confluent/ksql/internal/PullQueryExecutorMetrics.java

+          + routingNodeTypeName() + "-"
+          + dataSourceTypeName();


The data source needs to be part of the variant name now.

vvcephei · 2021-08-24T02:34:55Z

ksqldb-rest-app/src/main/java/io/confluent/ksql/api/impl/QueryEndpoint.java

+      final DataSource dataSource = analysis.getFrom().getDataSource();
+      final DataSource.DataSourceType dataSourceType = dataSource.getDataSourceType();
+      switch (dataSourceType) {
+        case KTABLE:
+          return createTablePullQueryPublisher(
+              analysis,
+              context,
+              serviceContext,
+              statement,
+              pullQueryMetrics,
+              workerExecutor,
+              metricsCallbackHolder
+          );
+        case KSTREAM:
+          throw new KsqlStatementException(
+              "Pull queries are not supported on streams.",
+              statement.getStatementText()
+          );
+        default:
+          throw new KsqlStatementException(
+              "Unexpected data source type for pull query: " + dataSourceType,
+              statement.getStatementText()
+          );
+      }


This is the main contribution of this PR: Detect the source type up front and add a switch statement so that we'll have a place to add the stream pull query in later.

vvcephei · 2021-08-24T02:36:51Z

ksqldb-rest-app/src/test/java/io/confluent/ksql/rest/server/execution/PullQueryMetricsTest.java

  ) {
    final Metrics metrics = pullMetrics.getMetrics();
    final Map<String, String> tags = ImmutableMap.<String, String>builder()
        .putAll(CUSTOM_TAGS_WITH_SERVICE_ID)
        .put(KsqlConstants.KSQL_QUERY_SOURCE_TAG, sourceType.name().toLowerCase())
        .put(KsqlConstants.KSQL_QUERY_PLAN_TYPE_TAG, planType.name().toLowerCase())
        .put(KsqlConstants.KSQL_QUERY_ROUTING_TYPE_TAG, routingNodeType.name().toLowerCase())
+        .put(KsqlConstants.KSQL_DATA_SOURCE_TYPE_TAG, dataSourceType.name().toLowerCase())


The fact that we have to modify the expectations of this test is making me nervous that this might not be a compatible change.

It would be safer, though more verbose, to just add a whole new set of identical metrics for stream pull query metrics.

I'm not sure if this is compatible. I would think it should be but might be worth checking. I think this is preferable to creating a whole new set of metrics.

My understanding is that:

for DD, the metrics tag matching is not necessarily exhaustive: i.e. if you have three tags, and the DD collection rule only sets two of them, it would collect all metrics that have the matching values for these two tags. So that should be compatible.

for telemetry, my low-confidence memory is that it is similar to DD.

Cc @cadonna who may help verifying here?

another question that is interesting to answer is that what would break if it's not compatible? do we just clobber our old metrics or ignore old-format metrics that get emitted? that might be acceptable.

Ok, upon further reflection (and upon actually trying to use these metrics for stream pull queries as well), I've decided that it's both too risky and not useful enough to do this by adding this tag, so I've pulled the new tag from this PR.

When I add stream pull queries, I'll just add new metrics to measure them.

AlanConfluent · 2021-08-24T23:27:23Z

ksqldb-rest-app/src/test/java/io/confluent/ksql/rest/server/execution/PullQueryMetricsTest.java

  ) {
    final Metrics metrics = pullMetrics.getMetrics();
    final Map<String, String> tags = ImmutableMap.<String, String>builder()
        .putAll(CUSTOM_TAGS_WITH_SERVICE_ID)
        .put(KsqlConstants.KSQL_QUERY_SOURCE_TAG, sourceType.name().toLowerCase())
        .put(KsqlConstants.KSQL_QUERY_PLAN_TYPE_TAG, planType.name().toLowerCase())
        .put(KsqlConstants.KSQL_QUERY_ROUTING_TYPE_TAG, routingNodeType.name().toLowerCase())
+        .put(KsqlConstants.KSQL_DATA_SOURCE_TYPE_TAG, dataSourceType.name().toLowerCase())


I'm not sure if this is compatible. I would think it should be but might be worth checking. I think this is preferable to creating a whole new set of metrics.

agavra

LGTM too pending the metrics question

guozhangwang

LGTM!

feat: introduce branches for pull queries on streams

ed13134

This change implements part of [KLIP-53](https://github.com/confluentinc/ksql/blob/master/design-proposals/klip-53-pull-queries-on-streams.md)

vvcephei requested a review from a team as a code owner August 24, 2021 02:29

vvcephei commented Aug 24, 2021

View reviewed changes

vvcephei self-assigned this Aug 24, 2021

checkstyle

a5b735d

AlanConfluent approved these changes Aug 24, 2021

View reviewed changes

checkstyle

708d547

agavra approved these changes Aug 25, 2021

View reviewed changes

guozhangwang approved these changes Aug 25, 2021

View reviewed changes

John Roesler added 3 commits August 25, 2021 15:14

fix test

c669b15

revert data type metric tag

3879fa2

fix test

8eb9e10

vvcephei mentioned this pull request Aug 26, 2021

feat: introduce stream pull queries for HTTP/1 #8064

Merged

2 tasks

vvcephei merged commit 886da99 into master Aug 27, 2021

vvcephei deleted the add-branches-for-stream-pull-queries branch August 27, 2021 01:57

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: introduce branches for pull queries on streams #8045

feat: introduce branches for pull queries on streams #8045

vvcephei commented Aug 24, 2021

vvcephei left a comment

vvcephei Aug 24, 2021

vvcephei Aug 24, 2021

vvcephei Aug 24, 2021

vvcephei Aug 24, 2021

vvcephei Aug 24, 2021

vvcephei Aug 24, 2021

vvcephei Aug 24, 2021

vvcephei Aug 24, 2021

AlanConfluent Aug 24, 2021

guozhangwang Aug 25, 2021

guozhangwang Aug 25, 2021

agavra Aug 25, 2021

vvcephei Aug 26, 2021

AlanConfluent Aug 24, 2021

agavra left a comment

guozhangwang left a comment

		throw new KsqlException("Unexpected data source type for table pull query: "
		+ source.getDataSourceType() + " "

feat: introduce branches for pull queries on streams #8045

feat: introduce branches for pull queries on streams #8045

Conversation

vvcephei commented Aug 24, 2021

Reviewer checklist

vvcephei 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

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

agavra left a comment

Choose a reason for hiding this comment

guozhangwang left a comment

Choose a reason for hiding this comment