You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by ka...@apache.org on 2023/08/07 15:19:13 UTC
[druid] branch master updated: Add support for different result formats to MSQ SqlStatementResource (#14571)
This is an automated email from the ASF dual-hosted git repository.
karan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 56ab81f381 Add support for different result formats to MSQ SqlStatementResource (#14571)
56ab81f381 is described below
commit 56ab81f38149f9b849d6f692d54b26c61230f9e2
Author: Adarsh Sanjeev <ad...@gmail.com>
AuthorDate: Mon Aug 7 20:48:59 2023 +0530
Add support for different result formats to MSQ SqlStatementResource (#14571)
* Add support for different result format
* Add tests
* Add tests
* Fix checkstyle
* Remove changes to destination
* Removed some unwanted code
* Address review comments
* Rename parameter
* Fix tests
---
.../msq/sql/resources/SqlStatementResource.java | 105 +++++++++++++----
.../DataSourceMSQDestinationTest.java | 3 +-
.../msq/sql/entity/SqlStatementResultTest.java | 2 +-
.../SqlMSQStatementResourcePostTest.java | 129 ++++++++++++++++++++-
.../{ => resources}/SqlStatementResourceTest.java | 36 +++---
.../org/apache/druid/sql/http/ResultFormat.java | 2 +
.../java/org/apache/druid/sql/http/SqlQuery.java | 2 +-
7 files changed, 232 insertions(+), 47 deletions(-)
diff --git a/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/sql/resources/SqlStatementResource.java b/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/sql/resources/SqlStatementResource.java
index 409aec7f5a..0482c72082 100644
--- a/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/sql/resources/SqlStatementResource.java
+++ b/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/sql/resources/SqlStatementResource.java
@@ -21,6 +21,7 @@ package org.apache.druid.msq.sql.resources;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableMap;
import com.google.common.io.CountingOutputStream;
import com.google.common.util.concurrent.ListenableFuture;
@@ -49,6 +50,7 @@ import org.apache.druid.java.util.common.io.Closer;
import org.apache.druid.java.util.common.logger.Logger;
import org.apache.druid.msq.guice.MultiStageQuery;
import org.apache.druid.msq.indexing.MSQControllerTask;
+import org.apache.druid.msq.indexing.MSQSpec;
import org.apache.druid.msq.indexing.destination.DurableStorageMSQDestination;
import org.apache.druid.msq.indexing.destination.MSQDestination;
import org.apache.druid.msq.indexing.destination.MSQSelectDestination;
@@ -115,6 +117,7 @@ import java.util.stream.Collectors;
public class SqlStatementResource
{
+ public static final String RESULT_FORMAT = "__resultFormat";
private static final Logger log = new Logger(SqlStatementResource.class);
private final SqlStatementFactory msqSqlStatementFactory;
private final AuthorizerMapper authorizerMapper;
@@ -167,12 +170,14 @@ public class SqlStatementResource
@Consumes(MediaType.APPLICATION_JSON)
public Response doPost(final SqlQuery sqlQuery, @Context final HttpServletRequest req)
{
- final HttpStatement stmt = msqSqlStatementFactory.httpStatement(sqlQuery, req);
+ SqlQuery modifiedQuery = createModifiedSqlQuery(sqlQuery);
+
+ final HttpStatement stmt = msqSqlStatementFactory.httpStatement(modifiedQuery, req);
final String sqlQueryId = stmt.sqlQueryId();
final String currThreadName = Thread.currentThread().getName();
boolean isDebug = false;
try {
- QueryContext queryContext = QueryContext.of(sqlQuery.getContext());
+ QueryContext queryContext = QueryContext.of(modifiedQuery.getContext());
isDebug = queryContext.isDebug();
contextChecks(queryContext);
@@ -190,7 +195,7 @@ public class SqlStatementResource
return buildTaskResponse(sequence, stmt.query().authResult().getIdentity());
} else {
// Used for EXPLAIN
- return buildStandardResponse(sequence, sqlQuery, sqlQueryId, rowTransformer);
+ return buildStandardResponse(sequence, modifiedQuery, sqlQueryId, rowTransformer);
}
}
catch (DruidException e) {
@@ -278,6 +283,7 @@ public class SqlStatementResource
public Response doGetResults(
@PathParam("id") final String queryId,
@QueryParam("page") Long page,
+ @QueryParam("resultFormat") String resultFormat,
@Context final HttpServletRequest req
)
{
@@ -328,12 +334,14 @@ public class SqlStatementResource
return Response.ok().build();
}
+ ResultFormat preferredFormat = getPreferredResultFormat(resultFormat, msqControllerTask.getQuerySpec());
return Response.ok((StreamingOutput) outputStream -> resultPusher(
queryId,
signature,
closer,
results,
- new CountingOutputStream(outputStream)
+ new CountingOutputStream(outputStream),
+ preferredFormat
)).build();
}
@@ -658,6 +666,48 @@ public class SqlStatementResource
return msqControllerTask;
}
+ /**
+ * Creates a new sqlQuery from the user submitted sqlQuery after performing required modifications.
+ */
+ private SqlQuery createModifiedSqlQuery(SqlQuery sqlQuery)
+ {
+ Map<String, Object> context = sqlQuery.getContext();
+ if (context.containsKey(RESULT_FORMAT)) {
+ throw InvalidInput.exception("Query context parameter [%s] is not allowed", RESULT_FORMAT);
+ }
+ Map<String, Object> modifiedContext = ImmutableMap.<String, Object>builder()
+ .putAll(context)
+ .put(RESULT_FORMAT, sqlQuery.getResultFormat().toString())
+ .build();
+ return new SqlQuery(
+ sqlQuery.getQuery(),
+ sqlQuery.getResultFormat(),
+ sqlQuery.includeHeader(),
+ sqlQuery.includeTypesHeader(),
+ sqlQuery.includeSqlTypesHeader(),
+ modifiedContext,
+ sqlQuery.getParameters()
+ );
+ }
+
+ private ResultFormat getPreferredResultFormat(String resultFormatParam, MSQSpec msqSpec)
+ {
+ if (resultFormatParam == null) {
+ return QueryContexts.getAsEnum(
+ RESULT_FORMAT,
+ msqSpec.getQuery().context().get(RESULT_FORMAT),
+ ResultFormat.class,
+ ResultFormat.DEFAULT_RESULT_FORMAT
+ );
+ }
+
+ return QueryContexts.getAsEnum(
+ "resultFormat",
+ resultFormatParam,
+ ResultFormat.class
+ );
+ }
+
private Optional<Yielder<Object[]>> getResultYielder(
String queryId,
Long page,
@@ -769,29 +819,15 @@ public class SqlStatementResource
Optional<List<ColumnNameAndTypes>> signature,
Closer closer,
Optional<Yielder<Object[]>> results,
- CountingOutputStream os
+ CountingOutputStream os,
+ ResultFormat resultFormat
) throws IOException
{
try {
- try (final ResultFormat.Writer writer = ResultFormat.OBJECTLINES.createFormatter(os, jsonMapper)) {
+ try (final ResultFormat.Writer writer = resultFormat.createFormatter(os, jsonMapper)) {
Yielder<Object[]> yielder = results.get();
List<ColumnNameAndTypes> rowSignature = signature.get();
- writer.writeResponseStart();
-
- while (!yielder.isDone()) {
- writer.writeRowStart();
- Object[] row = yielder.get();
- for (int i = 0; i < Math.min(rowSignature.size(), row.length); i++) {
- writer.writeRowField(
- rowSignature.get(i).getColName(),
- row[i]
- );
- }
- writer.writeRowEnd();
- yielder = yielder.next(null);
- }
- writer.writeResponseEnd();
- yielder.close();
+ resultPusherInternal(writer, yielder, rowSignature);
}
catch (Exception e) {
log.error(e, "Unable to stream results back for query[%s]", queryId);
@@ -807,6 +843,31 @@ public class SqlStatementResource
}
}
+ @VisibleForTesting
+ static void resultPusherInternal(
+ ResultFormat.Writer writer,
+ Yielder<Object[]> yielder,
+ List<ColumnNameAndTypes> rowSignature
+ ) throws IOException
+ {
+ writer.writeResponseStart();
+
+ while (!yielder.isDone()) {
+ writer.writeRowStart();
+ Object[] row = yielder.get();
+ for (int i = 0; i < Math.min(rowSignature.size(), row.length); i++) {
+ writer.writeRowField(
+ rowSignature.get(i).getColName(),
+ row[i]
+ );
+ }
+ writer.writeRowEnd();
+ yielder = yielder.next(null);
+ }
+ writer.writeResponseEnd();
+ yielder.close();
+ }
+
private static void throwIfQueryIsNotSuccessful(String queryId, TaskStatusPlus statusPlus)
{
SqlStatementState sqlStatementState = SqlStatementResourceHelper.getSqlStatementState(statusPlus);
diff --git a/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/DataSourceMSQDestinationTest.java b/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/destination/DataSourceMSQDestinationTest.java
similarity index 91%
rename from extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/DataSourceMSQDestinationTest.java
rename to extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/destination/DataSourceMSQDestinationTest.java
index 21e2d7ffaa..6d3a5ebfd9 100644
--- a/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/DataSourceMSQDestinationTest.java
+++ b/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/destination/DataSourceMSQDestinationTest.java
@@ -17,11 +17,10 @@
* under the License.
*/
-package org.apache.druid.msq.indexing;
+package org.apache.druid.msq.indexing.destination;
import nl.jqno.equalsverifier.EqualsVerifier;
-import org.apache.druid.msq.indexing.destination.DataSourceMSQDestination;
import org.junit.Test;
public class DataSourceMSQDestinationTest
diff --git a/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/sql/entity/SqlStatementResultTest.java b/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/sql/entity/SqlStatementResultTest.java
index 9fd4a63b52..0434c89ce1 100644
--- a/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/sql/entity/SqlStatementResultTest.java
+++ b/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/sql/entity/SqlStatementResultTest.java
@@ -26,8 +26,8 @@ import org.apache.druid.error.DruidException;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.msq.indexing.error.MSQException;
import org.apache.druid.msq.indexing.error.QueryNotSupportedFault;
-import org.apache.druid.msq.sql.SqlStatementResourceTest;
import org.apache.druid.msq.sql.SqlStatementState;
+import org.apache.druid.msq.sql.resources.SqlStatementResourceTest;
import org.junit.Assert;
import org.junit.Test;
diff --git a/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/sql/SqlMSQStatementResourcePostTest.java b/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/sql/resources/SqlMSQStatementResourcePostTest.java
similarity index 74%
rename from extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/sql/SqlMSQStatementResourcePostTest.java
rename to extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/sql/resources/SqlMSQStatementResourcePostTest.java
index b7421739d1..4b1a53b798 100644
--- a/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/sql/SqlMSQStatementResourcePostTest.java
+++ b/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/sql/resources/SqlMSQStatementResourcePostTest.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.druid.msq.sql;
+package org.apache.druid.msq.sql.resources;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -27,15 +27,17 @@ import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.druid.common.config.NullHandling;
import org.apache.druid.error.DruidException;
import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Sequences;
+import org.apache.druid.java.util.common.guava.Yielders;
import org.apache.druid.msq.indexing.MSQControllerTask;
import org.apache.druid.msq.indexing.destination.MSQSelectDestination;
import org.apache.druid.msq.indexing.error.InsertCannotBeEmptyFault;
import org.apache.druid.msq.indexing.error.MSQException;
+import org.apache.druid.msq.sql.SqlStatementState;
import org.apache.druid.msq.sql.entity.ColumnNameAndTypes;
import org.apache.druid.msq.sql.entity.PageInformation;
import org.apache.druid.msq.sql.entity.ResultSetInformation;
import org.apache.druid.msq.sql.entity.SqlStatementResult;
-import org.apache.druid.msq.sql.resources.SqlStatementResource;
import org.apache.druid.msq.test.MSQTestBase;
import org.apache.druid.msq.test.MSQTestOverlordServiceClient;
import org.apache.druid.msq.util.MultiStageQueryContext;
@@ -43,6 +45,7 @@ import org.apache.druid.query.ExecutionMode;
import org.apache.druid.query.QueryContexts;
import org.apache.druid.segment.column.ValueType;
import org.apache.druid.sql.calcite.util.CalciteTests;
+import org.apache.druid.sql.http.ResultFormat;
import org.apache.druid.sql.http.SqlQuery;
import org.apache.druid.storage.NilStorageConnector;
import org.junit.Assert;
@@ -54,6 +57,7 @@ import javax.ws.rs.core.StreamingOutput;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -142,7 +146,6 @@ public class SqlMSQStatementResourcePostTest extends MSQTestBase
@Test
public void nonSupportedModes()
{
-
SqlStatementResourceTest.assertExceptionMessage(
resource.doPost(new SqlQuery(
"select * from foo",
@@ -241,7 +244,7 @@ public class SqlMSQStatementResourcePostTest extends MSQTestBase
+ "\"resultFormat\":\"compactedList\","
+ "\"columns\":[\"__time\",\"cnt\",\"dim1\",\"dim2\",\"dim3\",\"m1\",\"m2\",\"unique_dim1\"],"
+ "\"legacy\":false,"
- + "\"context\":{\"executionMode\":\"ASYNC\",\"scanSignature\":\"[{\\\"name\\\":\\\"__time\\\",\\\"type\\\":\\\"LONG\\\"},{\\\"name\\\":\\\"cnt\\\",\\\"type\\\":\\\"LONG\\\"},{\\\"name\\\":\\\"dim1\\\",\\\"type\\\":\\\"STRING\\\"},{\\\"name\\\":\\\"dim2\\\",\\\"type\\\":\\\"STRING\\\"},{\\\"name\\\":\\\"dim3\\\",\\\"type\\\":\\\"STRING\\\"},{\\\"name\\\":\\\"m1\\\",\\\"type\\\":\\\"FLOAT\\\"},{\\\"name\\\":\\\"m2\\\",\\\"type\\\":\\\"DOUBLE\\\"},{\\\"name\\\":\\\"unique_dim1\\\",\ [...]
+ + "\"context\":{\"__resultFormat\":\"object\",\"executionMode\":\"ASYNC\",\"scanSignature\":\"[{\\\"name\\\":\\\"__time\\\",\\\"type\\\":\\\"LONG\\\"},{\\\"name\\\":\\\"cnt\\\",\\\"type\\\":\\\"LONG\\\"},{\\\"name\\\":\\\"dim1\\\",\\\"type\\\":\\\"STRING\\\"},{\\\"name\\\":\\\"dim2\\\",\\\"type\\\":\\\"STRING\\\"},{\\\"name\\\":\\\"dim3\\\",\\\"type\\\":\\\"STRING\\\"},{\\\"name\\\":\\\"m1\\\",\\\"type\\\":\\\"FLOAT\\\"},{\\\"name\\\":\\\"m2\\\",\\\"type\\\":\\\"DOUBLE\\\"},{\\\" [...]
+ " RESOURCES=[{\"name\":\"foo\",\"type\":\"DATASOURCE\"}],"
+ " ATTRIBUTES={\"statementType\":\"SELECT\"}}",
String.valueOf(SqlStatementResourceTest.getResultRowsFromResponse(response).get(0))
@@ -329,6 +332,7 @@ public class SqlMSQStatementResourcePostTest extends MSQTestBase
resource.doGetResults(
sqlStatementResult.getQueryId(),
null,
+ ResultFormat.OBJECTLINES.name(),
SqlStatementResourceTest.makeOkRequest()
),
objectMapper);
@@ -344,11 +348,124 @@ public class SqlMSQStatementResourcePostTest extends MSQTestBase
resource.doGetResults(
sqlStatementResult.getQueryId(),
0L,
+ ResultFormat.OBJECTLINES.name(),
SqlStatementResourceTest.makeOkRequest()
),
objectMapper);
}
+ @Test
+ public void testResultFormat() throws Exception
+ {
+ Map<String, Object> context = defaultAsyncContext();
+ context.put(MultiStageQueryContext.CTX_SELECT_DESTINATION, MSQSelectDestination.DURABLESTORAGE.name());
+
+ SqlStatementResult sqlStatementResult = (SqlStatementResult) resource.doPost(
+ new SqlQuery(
+ "select cnt,dim1 from foo",
+ null,
+ false,
+ false,
+ false,
+ context,
+ null
+ ),
+ SqlStatementResourceTest.makeOkRequest()
+ ).getEntity();
+
+ final List<ColumnNameAndTypes> columnNameAndTypes = ImmutableList.of(
+ new ColumnNameAndTypes("cnt", "cnt", "cnt"),
+ new ColumnNameAndTypes("dim1", "dim1", "dim1")
+ );
+
+ List<Object[]> expectedRows = ImmutableList.of(
+ new Object[]{1, ""},
+ new Object[]{1, "10.1"},
+ new Object[]{1, "2"},
+ new Object[]{1, "1"},
+ new Object[]{1, "def"},
+ new Object[]{1, "abc"}
+ );
+
+ for (ResultFormat resultFormat : ResultFormat.values()) {
+ Assert.assertArrayEquals(
+ createExpectedResultsInFormat(resultFormat, expectedRows, columnNameAndTypes, objectMapper),
+ responseToByteArray(
+ resource.doGetResults(
+ sqlStatementResult.getQueryId(),
+ null,
+ resultFormat.name(),
+ SqlStatementResourceTest.makeOkRequest()
+ ), objectMapper
+ )
+ );
+
+ Assert.assertArrayEquals(
+ createExpectedResultsInFormat(resultFormat, expectedRows, columnNameAndTypes, objectMapper),
+ responseToByteArray(
+ resource.doGetResults(
+ sqlStatementResult.getQueryId(),
+ 0L,
+ resultFormat.name(),
+ SqlStatementResourceTest.makeOkRequest()
+ ), objectMapper
+ )
+ );
+ }
+ }
+
+ @Test
+ public void testResultFormatWithParamInSelect() throws IOException
+ {
+ Map<String, Object> context = defaultAsyncContext();
+ context.put(MultiStageQueryContext.CTX_SELECT_DESTINATION, MSQSelectDestination.DURABLESTORAGE.name());
+
+ SqlStatementResult sqlStatementResult = (SqlStatementResult) resource.doPost(
+ new SqlQuery(
+ "select cnt,dim1 from foo",
+ ResultFormat.OBJECTLINES,
+ false,
+ false,
+ false,
+ context,
+ null
+ ),
+ SqlStatementResourceTest.makeOkRequest()
+ ).getEntity();
+
+
+ List<List<Object>> rows = new ArrayList<>();
+ rows.add(ImmutableList.of(1, ""));
+ rows.add(ImmutableList.of(1, "10.1"));
+ rows.add(ImmutableList.of(1, "2"));
+ rows.add(ImmutableList.of(1, "1"));
+ rows.add(ImmutableList.of(1, "def"));
+ rows.add(ImmutableList.of(1, "abc"));
+
+ Assert.assertEquals(rows, SqlStatementResourceTest.getResultRowsFromResponse(resource.doGetResults(
+ sqlStatementResult.getQueryId(),
+ null,
+ ResultFormat.ARRAY.name(),
+ SqlStatementResourceTest.makeOkRequest()
+ )));
+
+ Assert.assertEquals(rows, SqlStatementResourceTest.getResultRowsFromResponse(resource.doGetResults(
+ sqlStatementResult.getQueryId(),
+ 0L,
+ ResultFormat.ARRAY.name(),
+ SqlStatementResourceTest.makeOkRequest()
+ )));
+ }
+
+ private byte[] createExpectedResultsInFormat(ResultFormat resultFormat, List<Object[]> resultsList, List<ColumnNameAndTypes> rowSignature, ObjectMapper jsonMapper) throws Exception
+ {
+ ByteArrayOutputStream os = new ByteArrayOutputStream();
+ try (final ResultFormat.Writer writer = resultFormat.createFormatter(os, jsonMapper)) {
+ SqlStatementResource.resultPusherInternal(writer, Yielders.each(Sequences.simple(resultsList)), rowSignature);
+ }
+ return os.toByteArray();
+ }
+
private void assertExpectedResults(String expectedResult, Response resultsResponse, ObjectMapper objectMapper) throws IOException
{
byte[] bytes = responseToByteArray(resultsResponse, objectMapper);
@@ -400,6 +517,7 @@ public class SqlMSQStatementResourcePostTest extends MSQTestBase
Response resultsResponse = resource.doGetResults(
actual.getQueryId(),
+ 0L,
null,
SqlStatementResourceTest.makeOkRequest()
);
@@ -442,6 +560,7 @@ public class SqlMSQStatementResourcePostTest extends MSQTestBase
Response resultsResponse = resource.doGetResults(
actual.getQueryId(),
+ 0L,
null,
SqlStatementResourceTest.makeOkRequest()
);
@@ -452,7 +571,7 @@ public class SqlMSQStatementResourcePostTest extends MSQTestBase
private static Map<String, Object> defaultAsyncContext()
{
- Map<String, Object> context = new HashMap<String, Object>();
+ Map<String, Object> context = new HashMap<>();
context.put(QueryContexts.CTX_EXECUTION_MODE, ExecutionMode.ASYNC.name());
return context;
}
diff --git a/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/sql/SqlStatementResourceTest.java b/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/sql/resources/SqlStatementResourceTest.java
similarity index 96%
rename from extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/sql/SqlStatementResourceTest.java
rename to extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/sql/resources/SqlStatementResourceTest.java
index 2ff3f643f6..fa4d3c9f40 100644
--- a/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/sql/SqlStatementResourceTest.java
+++ b/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/sql/resources/SqlStatementResourceTest.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.druid.msq.sql;
+package org.apache.druid.msq.sql.resources;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
@@ -57,11 +57,12 @@ import org.apache.druid.msq.indexing.report.MSQStatusReport;
import org.apache.druid.msq.indexing.report.MSQTaskReport;
import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
import org.apache.druid.msq.indexing.report.MSQTaskReportTest;
+import org.apache.druid.msq.sql.MSQTaskQueryMaker;
+import org.apache.druid.msq.sql.SqlStatementState;
import org.apache.druid.msq.sql.entity.ColumnNameAndTypes;
import org.apache.druid.msq.sql.entity.PageInformation;
import org.apache.druid.msq.sql.entity.ResultSetInformation;
import org.apache.druid.msq.sql.entity.SqlStatementResult;
-import org.apache.druid.msq.sql.resources.SqlStatementResource;
import org.apache.druid.msq.test.MSQTestBase;
import org.apache.druid.query.Druids;
import org.apache.druid.query.Query;
@@ -78,6 +79,7 @@ import org.apache.druid.server.security.AuthConfig;
import org.apache.druid.server.security.AuthenticationResult;
import org.apache.druid.sql.calcite.planner.ColumnMappings;
import org.apache.druid.sql.calcite.util.CalciteTests;
+import org.apache.druid.sql.http.ResultFormat;
import org.apache.druid.sql.http.SqlResourceTest;
import org.apache.druid.storage.local.LocalFileStorageConnector;
import org.jboss.netty.handler.codec.http.DefaultHttpResponse;
@@ -158,8 +160,7 @@ public class SqlStatementResourceTest extends MSQTestBase
ColumnType.STRING
)
.build()))
- .destination(
- TaskReportMSQDestination.INSTANCE)
+ .destination(TaskReportMSQDestination.instance())
.tuningConfig(
MSQTuningConfig.defaultConfig())
.build(),
@@ -673,7 +674,7 @@ public class SqlStatementResourceTest extends MSQTestBase
);
assertExceptionMessage(
- resource.doGetResults(ACCEPTED_SELECT_MSQ_QUERY, 0L, makeOkRequest()),
+ resource.doGetResults(ACCEPTED_SELECT_MSQ_QUERY, 0L, null, makeOkRequest()),
StringUtils.format(
"Query[%s] is currently in [%s] state. Please wait for it to complete.",
ACCEPTED_SELECT_MSQ_QUERY,
@@ -707,7 +708,7 @@ public class SqlStatementResourceTest extends MSQTestBase
);
assertExceptionMessage(
- resource.doGetResults(RUNNING_SELECT_MSQ_QUERY, 0L, makeOkRequest()),
+ resource.doGetResults(RUNNING_SELECT_MSQ_QUERY, 0L, null, makeOkRequest()),
StringUtils.format(
"Query[%s] is currently in [%s] state. Please wait for it to complete.",
RUNNING_SELECT_MSQ_QUERY,
@@ -743,7 +744,7 @@ public class SqlStatementResourceTest extends MSQTestBase
null
)), objectMapper.writeValueAsString(response.getEntity()));
- Response resultsResponse = resource.doGetResults(FINISHED_SELECT_MSQ_QUERY, 0L, makeOkRequest());
+ Response resultsResponse = resource.doGetResults(FINISHED_SELECT_MSQ_QUERY, 0L, ResultFormat.OBJECTLINES.name(), makeOkRequest());
Assert.assertEquals(Response.Status.OK.getStatusCode(), resultsResponse.getStatus());
String expectedResult = "{\"_time\":123,\"alias\":\"foo\",\"market\":\"bar\"}\n"
@@ -761,6 +762,7 @@ public class SqlStatementResourceTest extends MSQTestBase
resource.doGetResults(
FINISHED_SELECT_MSQ_QUERY,
0L,
+ ResultFormat.OBJECTLINES.name(),
makeOkRequest()
)
);
@@ -770,13 +772,14 @@ public class SqlStatementResourceTest extends MSQTestBase
resource.doGetResults(
FINISHED_SELECT_MSQ_QUERY,
null,
+ ResultFormat.OBJECTLINES.name(),
makeOkRequest()
)
);
Assert.assertEquals(
Response.Status.BAD_REQUEST.getStatusCode(),
- resource.doGetResults(FINISHED_SELECT_MSQ_QUERY, -1L, makeOkRequest()).getStatus()
+ resource.doGetResults(FINISHED_SELECT_MSQ_QUERY, -1L, null, makeOkRequest()).getStatus()
);
}
@@ -792,7 +795,7 @@ public class SqlStatementResourceTest extends MSQTestBase
for (String queryID : ImmutableList.of(ERRORED_SELECT_MSQ_QUERY, ERRORED_INSERT_MSQ_QUERY)) {
assertExceptionMessage(resource.doGetStatus(queryID, makeOkRequest()), FAILURE_MSG, Response.Status.OK);
assertExceptionMessage(
- resource.doGetResults(queryID, 0L, makeOkRequest()),
+ resource.doGetResults(queryID, 0L, null, makeOkRequest()),
StringUtils.format(
"Query[%s] failed. Check the status api for more details.",
queryID
@@ -824,16 +827,16 @@ public class SqlStatementResourceTest extends MSQTestBase
Assert.assertEquals(
Response.Status.OK.getStatusCode(),
- resource.doGetResults(FINISHED_INSERT_MSQ_QUERY, 0L, makeOkRequest()).getStatus()
+ resource.doGetResults(FINISHED_INSERT_MSQ_QUERY, 0L, null, makeOkRequest()).getStatus()
);
Assert.assertEquals(
Response.Status.OK.getStatusCode(),
- resource.doGetResults(FINISHED_INSERT_MSQ_QUERY, null, makeOkRequest()).getStatus()
+ resource.doGetResults(FINISHED_INSERT_MSQ_QUERY, null, null, makeOkRequest()).getStatus()
);
Assert.assertEquals(
Response.Status.BAD_REQUEST.getStatusCode(),
- resource.doGetResults(FINISHED_INSERT_MSQ_QUERY, -1L, makeOkRequest()).getStatus()
+ resource.doGetResults(FINISHED_INSERT_MSQ_QUERY, -1L, null, makeOkRequest()).getStatus()
);
}
@@ -842,7 +845,7 @@ public class SqlStatementResourceTest extends MSQTestBase
{
for (String queryID : ImmutableList.of(RUNNING_NON_MSQ_TASK, FAILED_NON_MSQ_TASK, FINISHED_NON_MSQ_TASK)) {
assertNotFound(resource.doGetStatus(queryID, makeOkRequest()), queryID);
- assertNotFound(resource.doGetResults(queryID, 0L, makeOkRequest()), queryID);
+ assertNotFound(resource.doGetResults(queryID, 0L, null, makeOkRequest()), queryID);
assertNotFound(resource.deleteQuery(queryID, makeOkRequest()), queryID);
}
}
@@ -866,7 +869,7 @@ public class SqlStatementResourceTest extends MSQTestBase
);
assertExceptionMessage(
- resource.doGetResults(ACCEPTED_INSERT_MSQ_TASK, 0L, makeOkRequest()),
+ resource.doGetResults(ACCEPTED_INSERT_MSQ_TASK, 0L, null, makeOkRequest()),
StringUtils.format(
"Query[%s] is currently in [%s] state. Please wait for it to complete.",
ACCEPTED_INSERT_MSQ_TASK,
@@ -899,7 +902,7 @@ public class SqlStatementResourceTest extends MSQTestBase
);
assertExceptionMessage(
- resource.doGetResults(RUNNING_INSERT_MSQ_QUERY, 0L, makeOkRequest()),
+ resource.doGetResults(RUNNING_INSERT_MSQ_QUERY, 0L, null, makeOkRequest()),
StringUtils.format(
"Query[%s] is currently in [%s] state. Please wait for it to complete.",
RUNNING_INSERT_MSQ_QUERY,
@@ -928,6 +931,7 @@ public class SqlStatementResourceTest extends MSQTestBase
resource.doGetResults(
RUNNING_SELECT_MSQ_QUERY,
1L,
+ null,
makeExpectedReq(CalciteTests.SUPER_USER_AUTH_RESULT)
).getStatus()
);
@@ -959,7 +963,7 @@ public class SqlStatementResourceTest extends MSQTestBase
);
Assert.assertEquals(
Response.Status.NOT_FOUND.getStatusCode(),
- resource.doGetResults(taskIdNotFound, null, makeOkRequest()).getStatus()
+ resource.doGetResults(taskIdNotFound, null, null, makeOkRequest()).getStatus()
);
Assert.assertEquals(
Response.Status.NOT_FOUND.getStatusCode(),
diff --git a/sql/src/main/java/org/apache/druid/sql/http/ResultFormat.java b/sql/src/main/java/org/apache/druid/sql/http/ResultFormat.java
index 0c450b55b5..7a16fb7b88 100644
--- a/sql/src/main/java/org/apache/druid/sql/http/ResultFormat.java
+++ b/sql/src/main/java/org/apache/druid/sql/http/ResultFormat.java
@@ -121,6 +121,8 @@ public enum ResultFormat
return name;
}
+ public static final ResultFormat DEFAULT_RESULT_FORMAT = OBJECT;
+
public interface Writer extends Closeable
{
/**
diff --git a/sql/src/main/java/org/apache/druid/sql/http/SqlQuery.java b/sql/src/main/java/org/apache/druid/sql/http/SqlQuery.java
index 541f6d9fe6..904d4c3871 100644
--- a/sql/src/main/java/org/apache/druid/sql/http/SqlQuery.java
+++ b/sql/src/main/java/org/apache/druid/sql/http/SqlQuery.java
@@ -66,7 +66,7 @@ public class SqlQuery
)
{
this.query = Preconditions.checkNotNull(query, "query");
- this.resultFormat = resultFormat == null ? ResultFormat.OBJECT : resultFormat;
+ this.resultFormat = resultFormat == null ? ResultFormat.DEFAULT_RESULT_FORMAT : resultFormat;
this.header = header;
this.typesHeader = typesHeader;
this.sqlTypesHeader = sqlTypesHeader;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org