Adding Ability for MSQ to write select results to durable storage. #14527

cryptoe · 2023-07-05T14:31:02Z

Description

One of the most requested features in druid is to have an ability to download big result sets.
As part of #14416 , we added an ability for MSQ to be queried via a query friendly endpoint. This PR builds upon that work and adds the ability for MSQ to write select results to durable storage.

We write the results to the durable storage location <prefix>/results/<queryId> in the druid frame format. This is exposed to users by
/v2/sql/statements/:queryId/results.

The broker now understands how to read frame files from durable storage and return the results as a json format to end users.

We also write the first 3000 rows of the result to task report to enable preview results for clients like the console.

Users can use this by setting the context flag
selectDestination:DURABLE_STORAGE while issuing select q's to MSQ.
Note that DURABLE_STORAGE needs to be configured on broker, overlord, ingestion tasks/indexers captured at : https://druid.apache.org/docs/latest/multi-stage-query/reference.html#enable-durable-storage

Key changed/added classes in this PR

SQLStatementResource
QueryResultsFrameProcessor
ControllerImpl
WorkerImpl
DurableStorageInputChannelFactory
DurableStorageOutputChannelFactory

TODO:

Add more UT's

User facing docs for both # #14416 and this PR would be raised once this is merged.

This PR has:

adarshsanjeev · 2023-07-06T07:26:45Z

...core/multi-stage-query/src/main/java/org/apache/druid/msq/guice/MSQDurableStorageModule.java

@@ -98,6 +99,9 @@ public void configure(Binder binder)
                   .addBinding()
                   .to(DurableStorageCleaner.class);
      }
+    } else {
+      // bind with nil implementation so that configs are not required during service startups.
+      binder.bind(Key.get(StorageConnector.class, MultiStageQuery.class)).toInstance(NilStorageConnector.getInstance());


Is this change just as a convenience for users? Or is this required in some way for this PR?

Since the SQLStatementResource needs a binding of the StorageConnector, brokers would need storage Connector configs to start.

With this change, we bind a nil implementation so that those configs are not necessary.

In that case

We should selectively bind the implementation to the Broker Nodes only. For MSQ stuff residing in the indexers, we have an extensive system in place that determines whether durable storage is enabled.

This still seems slightly out of place, where each method throws a UOE for the user. IMO we should do something like this: Wherever we are using a storage connector in the broker, we should check isDurableStorageEnabled - as we do in the MSQ engine - which would basically check if the connector is a NilStorageConnector and then throw an error which is directed at the ADMIN/OPERATOR of the cluster.

The UOE in the connector's implementation should be a defensive mechanism, and we should ensure that it is never called.

Discussed here :
#14527 (comment)

adarshsanjeev · 2023-07-06T11:01:31Z

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/WorkerImpl.java

@@ -1319,7 +1342,7 @@ private void writeDurableStorageSuccessFileIfNeeded(final int stageNumber)
            e,
            "Unable to create the success file [%s] at the location [%s]",
            DurableStorageUtils.SUCCESS_MARKER_FILENAME,
-            DurableStorageUtils.getSuccessFilePath(
+            DurableStorageUtils.getWorkerOutputSuccessFilePath(


Should we get the worker output path from the DurableStorageOutputChannelFactory instead?

Good catch!!

adarshsanjeev · 2023-07-06T11:04:05Z

...-query/src/main/java/org/apache/druid/msq/indexing/destination/DataSourceMSQDestination.java

@@ -35,7 +36,7 @@

 public class DataSourceMSQDestination implements MSQDestination


Is there a reason to not move MSQDestination to org.apache.druid.msq.indexing.destination as well?

LakshSingla · 2023-07-06T11:51:41Z

processing/src/main/java/org/apache/druid/storage/NilStorageConnector.java

+  @Override
+  public boolean pathExists(String path)
+  {
+    throw new UOE("Please configure durable storage.");


We can refactor this into a method and call that method from everywhere

Suggested change

throw new UOE("Please configure durable storage.");

private void throwError()

{

throw new UOE("Please configure durable storage.");

}

Also, we should be replacing UOE with a DruidException, and the category should be defensive since we don't expect these methods to be called under normal circumstances. The top-level code should detect if the implementation is a NilStorageConnector, and then not call the method

We can refactor this into a method and call that method from everywhere

This seems better. Used a version of this.

The top-level code should detect if the implementation is a NilStorageConnector, and then not call the method

We already do that in multiple places

SqlStatementResource#contextChecks

MSQTasks.makeStorageConnector()

LakshSingla · 2023-07-06T11:53:17Z

processing/src/main/java/org/apache/druid/storage/NilStorageConnector.java

Please add a javadoc for this class explaining the necessity of this

LakshSingla · 2023-07-06T11:56:16Z

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/MSQTasks.java

-      return injector.getInstance(Key.get(StorageConnector.class, MultiStageQuery.class));
+      StorageConnector storageConnector = injector.getInstance(Key.get(StorageConnector.class, MultiStageQuery.class));
+      if (storageConnector instanceof NilStorageConnector) {
+        throw new Exception("Storage connector not configured.");


Once we selectively bind the NilStorageConnector to Broker nodes only, this check won't be required right?

That is also one approach. Since we would need to check this in the broker as well, I thought we can make it similar in all places.

I think what I can do is keep doing this check but also bind the NilStorageConnector only to the broker role.

LakshSingla · 2023-07-06T12:06:35Z

...sions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/MSQDestination.java


 @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
 @JsonSubTypes(value = {
    @JsonSubTypes.Type(name = DataSourceMSQDestination.TYPE, value = DataSourceMSQDestination.class),
-    @JsonSubTypes.Type(name = TaskReportMSQDestination.TYPE, value = TaskReportMSQDestination.class)
+    @JsonSubTypes.Type(name = TaskReportMSQDestination.TYPE, value = TaskReportMSQDestination.class),
+    @JsonSubTypes.Type(name = DurableStorageDestination.TYPE, value = DurableStorageDestination.class)


Suggested change

@JsonSubTypes.Type(name = DurableStorageDestination.TYPE, value = DurableStorageDestination.class)

@JsonSubTypes.Type(name = DurableStorageMSQDestination.TYPE, value = DurableStorageDestination.class)

nit: For consistency with the other two names, this class could be renamed

LakshSingla · 2023-07-06T12:09:47Z

...ns-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/MSQControllerTask.java

@@ -263,4 +265,9 @@ public static boolean isIngestion(final MSQSpec querySpec)
  {
    return querySpec.getDestination() instanceof DataSourceMSQDestination;
  }
+
+  public static boolean writeResultsToDurableStorage(final MSQSpec querySpec)


nit: Naming. From the name, the method seems like it writes the results to durable storage and returns whether the write was successful.

Suggested change

public static boolean writeResultsToDurableStorage(final MSQSpec querySpec)

public static boolean isWriteResultsToDurableStorage(final MSQSpec querySpec)

LakshSingla · 2023-07-06T12:12:16Z

...ry/src/main/java/org/apache/druid/msq/querykit/results/QueryResultFrameProcessorFactory.java

+          partition.getPartitionNumber(),
+          i -> {
+            try {
+              return outputChannelFactory.openChannel(i);


I think there was some discussion to change this to open inside the processor.

I think this would need a change in both GroupByPostFrameProcessorFactory and this new QueryResultsFrameProceesor.

Since I donot want to accidentally break GroupByPostFrameProcessorFactory so close to the cut off , we can change it as part of a follow up PR.
WDYT ?

LakshSingla · 2023-07-06T12:13:51Z

...ry/src/main/java/org/apache/druid/msq/querykit/results/QueryResultFrameProcessorFactory.java

+  )
+  {
+    // Expecting a single input slice from some prior stage.
+    final StageInputSlice slice = (StageInputSlice) Iterables.getOnlyElement(inputSlices);


nit: This returns a pretty bad exception message when it fails.
While we don't expect it to ever hit, it would be good if we can find out the elements that were present in the input slices (if it fails). Can we manually check and extract the slice instead of this helper method?

LakshSingla · 2023-07-06T12:20:38Z

...ge-query/src/main/java/org/apache/druid/msq/querykit/results/QueryResultsFrameProcessor.java

+  }
+
+  @Override
+  public void cleanup() throws IOException


This relates to the above point of opening the channel in the factory.

It is fine if we open the channel in the factory and close it in the factory

Or if we open the channel in the processor and close it in the processor

However, currently we are opening the output channel in the processor and closing it in the factory. If the above point is addressed, then this cleanup method is fine. However currently, we should move it to the place where we are opening the output channel

#14527 (comment)

LakshSingla · 2023-07-06T12:22:24Z

...n/java/org/apache/druid/msq/shuffle/input/DurableStorageQueryResultsInputChannelFactory.java

Please add Javadoc for this and the other new durable storage classes that have been created.

...re/multi-stage-query/src/main/java/org/apache/druid/msq/util/SqlStatementResourceHelper.java

adarshsanjeev

The overall approach seems fine to me. Had a few more minor comments.

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/sql/MSQTaskQueryMaker.java

...ry/src/main/java/org/apache/druid/msq/shuffle/output/DurableStorageOutputChannelFactory.java

...ain/java/org/apache/druid/msq/shuffle/output/DurableStorageTaskOutputChannelFactoryImpl.java

...main/java/org/apache/druid/msq/shuffle/output/DurableStorageResultsOutputChannelFactory.java

adarshsanjeev · 2023-07-06T13:50:39Z

...multi-stage-query/src/main/java/org/apache/druid/msq/sql/resources/SqlStatementResource.java

-          return Response.ok().build();
-        }
+      Optional<List<ColumnNameAndTypes>> signature = SqlStatementResourceHelper.getSignature(msqControllerTask);
+      if (!signature.isPresent() || MSQControllerTask.isIngestion(msqControllerTask.getQuerySpec())) {


Is there a need for both these conditions? Shouldn't signature be present only for selects?

This is more of a defensive check which does not harm us.

...re/multi-stage-query/src/main/java/org/apache/druid/msq/util/SqlStatementResourceHelper.java

2. Addressing review comments.

...multi-stage-query/src/main/java/org/apache/druid/msq/sql/resources/SqlStatementResource.java

adarshsanjeev

Thanks for the update! The changes look good to me % some test failures

cryptoe · 2023-07-07T15:20:20Z

Thanks @adarshsanjeev and @LakshSingla for the review!!

…pache#14527) One of the most requested features in druid is to have an ability to download big result sets. As part of apache#14416 , we added an ability for MSQ to be queried via a query friendly endpoint. This PR builds upon that work and adds the ability for MSQ to write select results to durable storage. We write the results to the durable storage location <prefix>/results/<queryId> in the druid frame format. This is exposed to users by /v2/sql/statements/:queryId/results.

cryptoe added 2 commits July 5, 2023 19:39

Adding Ability for MSQ to write select results to durable storage.

5b2f93d

Removing forbidden API's

482cccd

cryptoe added the Area - MSQ For multi stage queries - https://github.com/apache/druid/issues/12262 label Jul 5, 2023

cryptoe mentioned this pull request Jul 5, 2023

Community roadmap 2023 #14157

Open

cryptoe added 5 commits July 5, 2023 21:49

Attaching result stage always.

43e71bd

CI things

e547230

CI things

855e776

Adding coverage for processing module.

4e80ab5

Intellij inspections

4a2dd36

adarshsanjeev reviewed Jul 6, 2023

View reviewed changes

LakshSingla reviewed Jul 6, 2023

View reviewed changes

adarshsanjeev reviewed Jul 6, 2023

View reviewed changes

...re/multi-stage-query/src/main/java/org/apache/druid/msq/util/SqlStatementResourceHelper.java Outdated Show resolved Hide resolved

adarshsanjeev reviewed Jul 6, 2023

View reviewed changes

1. Cleaning up code.

7c8509b

2. Addressing review comments.

github-advanced-security bot found potential problems Jul 6, 2023

View reviewed changes

...multi-stage-query/src/main/java/org/apache/druid/msq/sql/resources/SqlStatementResource.java Fixed Show fixed Hide fixed

...multi-stage-query/src/main/java/org/apache/druid/msq/sql/resources/SqlStatementResource.java Fixed Show fixed Hide fixed

cryptoe added 2 commits July 7, 2023 09:44

Test case fix.

c5f4205

Adding java docs.

47fc66c

cryptoe added this to the 27.0 milestone Jul 7, 2023

adarshsanjeev approved these changes Jul 7, 2023

View reviewed changes

cryptoe added 4 commits July 7, 2023 11:42

Reverting autocloseable change.

73ed03d

Fixing pageID bug.

c985be5

Reverting debug statement.

6984979

Better naming

81a7ecb

LakshSingla approved these changes Jul 7, 2023

View reviewed changes

Adding durable storage test for async api.

8f1971e

cryptoe merged commit afa8c7b into apache:master Jul 7, 2023

vogievetsky added the Needs web console change Backend API changes that would benefit from frontend support in the web console label Jul 18, 2023

vogievetsky mentioned this pull request Jul 26, 2023

Web console: add durable storage selector #14669

Merged

AmatyaAvadhanula mentioned this pull request Aug 6, 2023

[DRAFT] 27.0.0 release notes #14761

Closed

vogievetsky removed the Needs web console change Backend API changes that would benefit from frontend support in the web console label Aug 8, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Adding Ability for MSQ to write select results to durable storage. #14527

Adding Ability for MSQ to write select results to durable storage. #14527

cryptoe commented Jul 5, 2023

adarshsanjeev Jul 6, 2023

cryptoe Jul 6, 2023

LakshSingla Jul 6, 2023

cryptoe Jul 6, 2023

adarshsanjeev Jul 6, 2023

cryptoe Jul 6, 2023

adarshsanjeev Jul 6, 2023

LakshSingla Jul 6, 2023

LakshSingla Jul 6, 2023

cryptoe Jul 6, 2023

LakshSingla Jul 6, 2023

LakshSingla Jul 6, 2023

cryptoe Jul 6, 2023

LakshSingla Jul 6, 2023

LakshSingla Jul 6, 2023 •

edited

Loading

LakshSingla Jul 6, 2023

cryptoe Jul 7, 2023

LakshSingla Jul 6, 2023

LakshSingla Jul 6, 2023

cryptoe Jul 7, 2023

LakshSingla Jul 6, 2023

adarshsanjeev left a comment

adarshsanjeev Jul 6, 2023

cryptoe Jul 6, 2023

adarshsanjeev left a comment

cryptoe commented Jul 7, 2023

		@@ -35,7 +36,7 @@

		public class DataSourceMSQDestination implements MSQDestination

	@JsonSubTypes.Type(name = DurableStorageDestination.TYPE, value = DurableStorageDestination.class)
	@JsonSubTypes.Type(name = DurableStorageMSQDestination.TYPE, value = DurableStorageDestination.class)

	public static boolean writeResultsToDurableStorage(final MSQSpec querySpec)
	public static boolean isWriteResultsToDurableStorage(final MSQSpec querySpec)

Adding Ability for MSQ to write select results to durable storage. #14527

Adding Ability for MSQ to write select results to durable storage. #14527

Conversation

cryptoe commented Jul 5, 2023

Description

Key changed/added classes in this PR

TODO:

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

LakshSingla Jul 6, 2023 • 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

adarshsanjeev left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

adarshsanjeev left a comment

Choose a reason for hiding this comment

cryptoe commented Jul 7, 2023

LakshSingla Jul 6, 2023 •

edited

Loading